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-least-once. Buffers records and PutObjects a rolling part object whenever the buffer reaches rollover_bytes, and again on flush.
• s3_2pc_string_sink - exactly-once via multipart-upload-complete-on-commit (selected by delivery_guarantee='exactly_once').

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-least-once
s3_2pc_string_sink	string	exactly-once (2PC)

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: atomic multi-sink commit group.

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.

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-least-once. A replay after a failed checkpoint re-PutObjects the buffered part, appending a duplicate object. On a put error the sink throws so the job replays from the last checkpoint.
• 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.