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 deterministic, checkpoint-tagged key <key_prefix>/sub<N>-<ckpt>.ndjson, and the multipart handle (key + uploadId + part ETags) is the committable. 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.

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, and a replay from the checkpoint can also re-PutObject a part that had already rolled, appending a duplicate object. 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.

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.