Amazon S3 (raw objects)

The s3 connector writes stream records to S3 as line-delimited text objects via the AWS SDK directly. It is distinct from the s3_parquet connector (Parquet files over Arrow's S3FileSystem): this one is a plain object sink, one record per line.

Overview

Each record is appended as a line (newline-terminated) to an object body. Two sinks are registered, selected by the delivery guarantee:

• s3_text_sink - at-most-once, as its capability record declares. Buffers records and PutObjects a rolling part object whenever the buffer reaches rollover_bytes, and again at end of stream. It has no checkpoint-barrier hook, so records buffered below rollover_bytes when a process dies are lost even though a checkpoint has covered them; use the 2PC sink for output tied to checkpoint completion.
• s3_2pc_string_sink - exactly-once via multipart-upload-complete-on-commit (selected by delivery_guarantee='exactly_once'). One NDJSON object per (subtask, checkpoint), staged at the barrier and made visible atomically at commit; recovery re-completes a persisted handle idempotently, so the guarantee holds across worker loss (exercised end to end, kills included, by the S3 exactly-once integration suite, and qualified on a multi-host cluster under two hours of continuous faults by QUAL-03). A crash can orphan an incomplete multipart upload whose checkpoint never became durable - invisible and harmless for correctness; configure a lifecycle rule (AbortIncompleteMultipartUpload) to expire the parts.

Dependency and version

aws-sdk-cpp (1.11.795). Built when the AWS SDK is available; otherwise the sink constructor throws a configuration error so submission fails loudly. Credentials resolve through the standard AWS chain (environment variables, instance profile, ~/.aws/credentials, IAM role). endpoint_override redirects to MinIO or LocalStack (path-style addressing is used automatically when it is set).

Factories

Factory	Channel	Delivery
s3_text_sink	string	at-most-once (buffer written on rollover and at end of stream only)
s3_2pc_string_sink	string	exactly-once (2PC)

There is also a programmatic raw-object source, S3Source (impls/s3/include/clink/connectors/s3_source.hpp): it reads newline-delimited objects under a prefix and emits one std::string record per line. It has no registered factory name and no SQL binding, keeps no intra-object position (a restart re-reads from the start of an object), and is what the s3 capability record's source half refers to. For Parquet objects use the S3 Parquet source instead.

Configuration

Option	Required	Default	Meaning
bucket	Yes	(none)	Target bucket.
key_prefix	No	""	Key prefix for written objects.
region	No	(SDK default)	AWS region.
endpoint_override	No	(none)	Custom endpoint (MinIO / LocalStack); enables path-style addressing.
rollover_bytes	No	16 MiB	s3_text_sink only: start a new object at this buffer size.
part_size	No	5 MiB	s3_2pc_string_sink only: multipart part size (S3 minimum is 5 MiB).
commit_group	No	""	s3_2pc_string_sink only: commit-group membership. Not a multi-sink atomicity switch - see ../internals/checkpointing.md for what it does and does not do.

SQL usage

connector='s3' binds to the string-channel sink. delivery_guarantee='exactly_once' selects s3_2pc_string_sink; otherwise s3_text_sink.

CREATE TABLE out_s3 (line STRING) WITH (
  connector          = 's3',
  bucket             = 'my-bucket',
  key_prefix         = 'events',
  delivery_guarantee = 'exactly_once'
);

Enabling it

Gated by CLINK_WITH_AWS_S3 (AUTO by default: built when find_package(AWSSDK CONFIG COMPONENTS s3) succeeds; ON makes a missing SDK a hard configure error; OFF always skips the impl). The pinned toolchain builds aws-sdk-cpp from source into CLINK_DEPS_PREFIX on the host (scripts/build-arrow.sh handles it) and bakes it into the Debian image, so the SDK is present on both supported build paths.

Example

Programmatic use of the exactly-once sink via its option struct, mirroring impls/s3/tests/test_s3_2pc_sink.cpp:

#include "clink/connectors/s3_sink_2pc.hpp"

clink::S3Sink2PC::Options o;
o.bucket = "my-bucket";
o.key_prefix = "events";
o.region = "eu-west-1";               // optional; SDK default otherwise
// o.endpoint_override = "http://localhost:9000";  // MinIO / LocalStack
// o.part_size = 5 * 1024 * 1024;     // S3 minimum

clink::S3Sink2PC sink(std::move(o));

In a pipeline the sink is normally selected by factory name (s3_2pc_string_sink / s3_text_sink) with the options passed as operator params, or from SQL as shown above.

Exactly-once sink (s3_2pc_string_sink)

Records since the last checkpoint barrier are buffered; at the barrier the whole interval is uploaded as the parts of an S3 multipart upload under a key tagged with the run and the checkpoint, <key_prefix>/sub<N>-<run>-<ckpt>.ndjson, and the multipart handle (key + uploadId + part ETags) is the committable. <run> is unique to each run of the sink, so a later run under the same prefix never completes an upload over an object an earlier run published, even where it reuses a checkpoint id. The object does not exist until CompleteMultipartUpload, so the framework makes it appear atomically once the checkpoint is globally durable; on abort it AbortMultipartUploads the parts. It participates in commit_group (see ../internals/checkpointing.md for the limits of that).

A multipart upload survives the session, so a crash between the barrier (parts uploaded) and commit does not lose data: on restart the framework CompleteMultipartUploads any handle in the restored checkpoint state.

An upload whose handle never reached a durable checkpoint (a crash before the snapshot) is a benign orphan: it produces no visible object, only staged part storage. It is not reconciled here; configure a bucket lifecycle rule (AbortIncompleteMultipartUpload) to expire such parts.

Memory: one checkpoint interval is buffered before the barrier uploads it (bounded by the checkpoint interval); commit is then a cheap metadata-only CompleteMultipartUpload.

On MinIO, a CompleteMultipartUpload whose client disconnects partway through can leave the upload unable to complete: every later complete is answered InvalidPart and the object never appears. A worker that dies while its complete is in flight does it, and so does a client that gives up on a stalled store (the AWS SDK gives up after 3 s without a reply). Measured on upstream MinIO's final release and on SILO, a maintained fork of it: with the store paused and clients giving up on their completes, the upload broke in most of 60 rounds under CPU load, and in none of 30 with every client waiting. The sink cannot rebuild the interval from what the store kept, so the job fails rather than publish part of it or skip it.

Delivery semantics

• s3_text_sink: at-most-once. The buffer is written only when it reaches rollover_bytes and at end of stream (flush() is the runtime's end-of-input hook, not a barrier hook), so a crash loses the records buffered since the last part was put. Parts are named <key_prefix>/part-<n>, or <key_prefix>/sub<i>-part-<n> when the sink runs at parallelism above 1, so subtasks never name the same object. A run restored from a checkpoint lists its own parts and numbers new ones after them, which needs s3:ListBucket, so it never puts a part over one whose records its sources will not replay; a part that rolled after the checkpoint can be written again under a new number, a duplicate. A fresh run starts at part-0. Before this naming, every subtask wrote part-0, part-1 and so on, so at parallelism above 1 subtasks replaced each other's objects, and a restored run replaced the first parts of the run before it. On a put error the sink throws so the job replays from the last checkpoint. This matches the connector's own capability record (delivery at_most_once).
• s3_2pc_string_sink: exactly-once. An object appears iff its checkpoint completes globally; a prepared-but-uncommitted multipart upload survives a crash and is completed on restart.
• s3_2pc_string_sink must be the only sink on its chain: at every checkpoint barrier it prepares the interval's upload and stages its handle in the checkpoint, which holds only while it owns its chain's checkpoint, so Dag::add_sink refuses another sink beside it, whichever is added first. SQL and job graphs plan every sink as a subtask of its own; with the Dag API, put the other sink in a separate Dag (see checkpointing).

Testing

impls/s3/tests/test_s3_2pc_sink.cpp covers the committable codec and, gated on CLINK_S3_TEST_ENDPOINT + CLINK_S3_TEST_BUCKET (MinIO / LocalStack; credentials via the AWS env chain), live integration: commit round-trip, crash recovery, abort, idempotent commit, and empty interval.

The exactly-once suite (S3ExactlyOnceTest.* in clink_integration_tests) runs the sink in a cluster, a coordinator and workers, and checks that every record is committed exactly once: a clean run, a worker killed, a worker killed inside the commit window, the final commit waited for at end of stream, the store stopped during recovery, and two kills on a longer run. Staged but uncommitted output is witnessed through the store's pending multipart uploads. The suite starts its own store: SILO, Pigsty's maintained fork of the MinIO server, pinned by digest in tests/integration/docker_minio.hpp. When its clients wait for the answer it completes multipart uploads as S3 does, which the suite depends on: a duplicate complete gets NoSuchUpload and never replaces the committed object. A client that disconnects partway through a complete is another matter (see Exactly-once sink), so the suite kills a worker only at the sink's sink.before_commit fault point, never with a complete in flight, and takes the store down by stopping it rather than pausing it. CLINK_TEST_S3_TRACE=1 records every S3 call the store serves, one line each, to a file in the temporary directory whose path the suite prints on stderr.

scripts/docker-suites-live.sh runs the suite with every other suite that starts its own servers in Docker, each with its googletest JSON report, then scripts/check-gtest-reports.py, which fails the run on any skipped or failed case and on fewer cases than a suite has: the suites skip without Docker, and googletest counts a skip as a pass. The docker-suites workflow (.github/workflows/docker-suites.yml) runs it on every push and pull request, and keeps the reports and the store's trace.

scripts/docker-suites-live.sh build