hubcio commented on code in PR #4219:
URL: https://github.com/apache/iggy/pull/4219#discussion_r4091162861
##########
core/connectors/sinks/delta_sink/README.md:
##########
@@ -10,29 +10,149 @@ The Delta Lake Sink Connector allows you to consume
messages from Iggy topics an
The table must already exist. The connector appends each successful nonempty
batch in one Delta transaction and keeps its schema snapshot until restart. The
plugin has no failed-batch retry loop; the Delta library can retry eligible
commit conflicts and storage requests. Write or commit errors clear the writer
buffers and return an error. The runtime uses consumer auto-commit and does not
replay failed sink batches, so end-to-end at-least-once delivery is not
guaranteed.
-## Configuration example
-
-### Local filesystem
-
-```toml
-[plugin_config]
-table_uri = "file:///tmp/iggy_delta_table"
+## How to configure a Delta Sink connector
+
+First, make sure that the Delta table already exists in the location you're
providing. You can use this script for an example workload:
+
+```python
+import pyarrow as pa
+from deltalake import DeltaTable
+
+table_uri = "s3://test_location/tables/test"
+
+schema = pa.schema([
+ pa.field("user_id", pa.string(), nullable=True),
+ pa.field("user_type", pa.uint8(), nullable=True),
+ pa.field("email", pa.string(), nullable=True),
+ pa.field("source", pa.string(), nullable=True),
+ pa.field("state", pa.string(), nullable=True),
+ pa.field("message", pa.string(), nullable=True),
+ pa.field("created_at", pa.timestamp("us"), nullable=True),
+])
+
+DeltaTable.create(
+ table_uri,
+ schema,
+ name="test",
+ storage_options={"AWS_REGION": "us-east-1"},
+)
+
+print(f"Created table at {table_uri}")
```
-### AWS S3
-
-```toml
-[plugin_config]
-table_uri = "s3://my-bucket/delta-tables/users"
-storage_backend_type = "s3"
-aws_s3_access_key = "your-access-key"
-aws_s3_secret_key = "your-secret-key"
-aws_s3_region = "us-east-1"
-aws_s3_endpoint_url = "https://s3.amazonaws.com";
-aws_s3_allow_http = false
-```
+The configuration is usually written individually for every connector and
consists of two parts: the runtime settings which are registering the sink and
telling which streams should plug into it, and the plugin's settings
themselves. Here's an example of a working configuration:
+
+ ```toml
+ type = "sink"
+ key = "delta"
+ enabled = true
+ version = 0
+ name = "Delta Lake sink"
+ path = "target/release/libiggy_connector_delta_sink" # make sure you have a
compiled plugin binary in this path
+ verbose = true
+
+ # these settings are common between all sinks
+ [[streams]]
+ stream = "your_stream"
+ topics = ["topic_inside_of_your_stream"]
+ schema = "json"
+ # important: read about batch and poll interval settings below
+ batch_length = 10000
+ poll_interval = "3s"
+ consumer_group = "delta_sink_connector"
+
+ # these settings are specific to each plugin, and in case of Delta sink, to
the type of storage used
+ [plugin_config]
+ # the table must exist in the given location
+ table_uri = "s3://iggy-sandbox/tables/test"
+ storage_backend_type = "s3"
+ aws_s3_region = "eu-central-1"
+ aws_s3_allow_http = false
+ ```
+
+### `[[streams]]` section
+
+- Find a topic that you need and the corresponding stream that this topic is
in and add it into the configuration.
+- `batch_length` and `poll_interval` have to be carefully set for Delta
tables. `poll_interval` sets the minimum gap between polls. Each poll returns
immediately with whatever messages are currently available, up to
`batch_length`. The Delta sink writes and commits exactly what one poll
returns, so each Delta commit holds at most `batch_length` rows, and topic
intake is capped at roughly `batch_length` messages per `poll_interval` per
topic.
+ `batch_length` is a ceiling: it only produces large files when messages
arrive fast enough between polls to fill it, so a low-throughput topic produces
small, frequent commits regardless of how high `batch_length` is set. For
Delta, writing means creating a log entry and a separate Parquet file, so a low
`batch_length` (roughly < 1000) or a low-throughput topic results in lots of
small files.
+ This is highly undesirable for the reader as it will have to read from many
small files instead of a few larger ones. For query performance optimization,
the system consuming these files will have to apply `OPTIMIZE` query in order
to consolidate the files. We recommend setting `batch_length` pretty high based
on your workload so that ideally the files are already optimized for reading.
+ The guideline recommended by the developers of Delta is to keep individual
file sizes between 128 MB and 1 GB. In the context of this document, it means
that ideally `batch_length` should be tuned so that each commit's file size
falls within the suggested range, assuming your topic has enough throughput to
fill it. The given range is only a rough guideline and you need to find a good
setting based on the patterns of reading and writing in your systems.
Review Comment:
warning: this sizes `batch_length` for 128 MB files but skips two costs: the
poll sits in memory in several copies until commit, and offsets commit at poll
time, so a failed write loses the whole batch. name both here.
##########
core/connectors/sinks/delta_sink/README.md:
##########
@@ -10,29 +10,149 @@ The Delta Lake Sink Connector allows you to consume
messages from Iggy topics an
The table must already exist. The connector appends each successful nonempty
batch in one Delta transaction and keeps its schema snapshot until restart. The
plugin has no failed-batch retry loop; the Delta library can retry eligible
commit conflicts and storage requests. Write or commit errors clear the writer
buffers and return an error. The runtime uses consumer auto-commit and does not
replay failed sink batches, so end-to-end at-least-once delivery is not
guaranteed.
-## Configuration example
-
-### Local filesystem
-
-```toml
-[plugin_config]
-table_uri = "file:///tmp/iggy_delta_table"
+## How to configure a Delta Sink connector
+
+First, make sure that the Delta table already exists in the location you're
providing. You can use this script for an example workload:
+
+```python
+import pyarrow as pa
+from deltalake import DeltaTable
+
+table_uri = "s3://test_location/tables/test"
+
+schema = pa.schema([
+ pa.field("user_id", pa.string(), nullable=True),
+ pa.field("user_type", pa.uint8(), nullable=True),
+ pa.field("email", pa.string(), nullable=True),
+ pa.field("source", pa.string(), nullable=True),
+ pa.field("state", pa.string(), nullable=True),
+ pa.field("message", pa.string(), nullable=True),
+ pa.field("created_at", pa.timestamp("us"), nullable=True),
+])
+
+DeltaTable.create(
+ table_uri,
+ schema,
+ name="test",
+ storage_options={"AWS_REGION": "us-east-1"},
+)
+
+print(f"Created table at {table_uri}")
```
-### AWS S3
-
-```toml
-[plugin_config]
-table_uri = "s3://my-bucket/delta-tables/users"
-storage_backend_type = "s3"
-aws_s3_access_key = "your-access-key"
-aws_s3_secret_key = "your-secret-key"
-aws_s3_region = "us-east-1"
-aws_s3_endpoint_url = "https://s3.amazonaws.com";
-aws_s3_allow_http = false
-```
+The configuration is usually written individually for every connector and
consists of two parts: the runtime settings which are registering the sink and
telling which streams should plug into it, and the plugin's settings
themselves. Here's an example of a working configuration:
+
+ ```toml
+ type = "sink"
+ key = "delta"
+ enabled = true
+ version = 0
+ name = "Delta Lake sink"
+ path = "target/release/libiggy_connector_delta_sink" # make sure you have a
compiled plugin binary in this path
+ verbose = true
+
+ # these settings are common between all sinks
+ [[streams]]
+ stream = "your_stream"
+ topics = ["topic_inside_of_your_stream"]
+ schema = "json"
+ # important: read about batch and poll interval settings below
+ batch_length = 10000
+ poll_interval = "3s"
+ consumer_group = "delta_sink_connector"
+
+ # these settings are specific to each plugin, and in case of Delta sink, to
the type of storage used
+ [plugin_config]
+ # the table must exist in the given location
+ table_uri = "s3://iggy-sandbox/tables/test"
+ storage_backend_type = "s3"
+ aws_s3_region = "eu-central-1"
+ aws_s3_allow_http = false
+ ```
+
+### `[[streams]]` section
+
+- Find a topic that you need and the corresponding stream that this topic is
in and add it into the configuration.
+- `batch_length` and `poll_interval` have to be carefully set for Delta
tables. `poll_interval` sets the minimum gap between polls. Each poll returns
immediately with whatever messages are currently available, up to
`batch_length`. The Delta sink writes and commits exactly what one poll
returns, so each Delta commit holds at most `batch_length` rows, and topic
intake is capped at roughly `batch_length` messages per `poll_interval` per
topic.
+ `batch_length` is a ceiling: it only produces large files when messages
arrive fast enough between polls to fill it, so a low-throughput topic produces
small, frequent commits regardless of how high `batch_length` is set. For
Delta, writing means creating a log entry and a separate Parquet file, so a low
`batch_length` (roughly < 1000) or a low-throughput topic results in lots of
small files.
+ This is highly undesirable for the reader as it will have to read from many
small files instead of a few larger ones. For query performance optimization,
the system consuming these files will have to apply `OPTIMIZE` query in order
to consolidate the files. We recommend setting `batch_length` pretty high based
on your workload so that ideally the files are already optimized for reading.
+ The guideline recommended by the developers of Delta is to keep individual
file sizes between 128 MB and 1 GB. In the context of this document, it means
that ideally `batch_length` should be tuned so that each commit's file size
falls within the suggested range, assuming your topic has enough throughput to
fill it. The given range is only a rough guideline and you need to find a good
setting based on the patterns of reading and writing in your systems.
+
+### Plugin configuration
+
+#### Attributes common to all types of storage
+
+- **table_uri** (required): Absolute URI to the Delta table. Supported
schemes: `file://`, `s3://`, `az://`, `gs://`. Bare filesystem paths are not
accepted; local tables must use the `file://` scheme.
+- **storage_backend_type** (optional): The cloud storage backend to use. One
of `"s3"`, `"azure"`, or `"gcs"`. Omit for local filesystem tables.
-### Azure Blob Storage
+#### Local filesystem
+
+ ```toml
+ [plugin_config]
+ table_uri = "file:///tmp/iggy_delta_table"
+ ```
+
+#### AWS S3
+
+Currently the implementation offers two possible ways of accessing the bucket.
+
+1. No static keys — the AWS SDK discovers credentials on its own via its
default credential chain (environment variables, shared config/profile
including SSO, web identity federation, ECS/EKS container credentials, or EC2
instance metadata). This is the best practice recommended by AWS. For example,
if running with an attached IAM role, the role assumed by the writer should
allow these actions on the bucket, here is how a working policy looks in HCL:
Review Comment:
warning: a keyed delta sink in this runtime writes its keys into the process
env as `AWS_ACCESS_KEY_ID` / `AWS_SECRET_ACCESS_KEY`, and a keyless sink then
picks them up. say so here - init order decides which identity writes.
##########
core/connectors/sinks/delta_sink/src/storage.rs:
##########
@@ -27,25 +27,38 @@ pub(crate) fn build_storage_options(
match config.storage_backend_type {
Some(StorageBackendType::S3) => {
- let access_key = config.aws_s3_access_key.as_ref().ok_or_else(|| {
- Error::InitError("S3 backend requires
'aws_s3_access_key'".into())
- })?;
- let secret_key = config.aws_s3_secret_key.as_ref().ok_or_else(|| {
- Error::InitError("S3 backend requires
'aws_s3_secret_key'".into())
+ match (
+ config.aws_s3_access_key.as_ref(),
+ config.aws_s3_secret_key.as_ref(),
+ ) {
+ (Some(access_key), Some(secret_key)) => {
+ opts.insert(
+ "AWS_ACCESS_KEY_ID".into(),
+ access_key.expose_secret().to_owned(),
+ );
+ opts.insert(
+ "AWS_SECRET_ACCESS_KEY".into(),
+ secret_key.expose_secret().to_owned(),
+ );
+ }
+ (None, None) => {
Review Comment:
warning: with both key names misspelled this arm silently goes keyless and
writes under the ambient AWS identity, where the old code errored out. log the
credential mode at open in place of the "Successfully composed" info line.
optionally add `#[serde(deny_unknown_fields)]` to `DeltaSinkConfig` so the typo
fails fast.
##########
core/connectors/sinks/delta_sink/src/sink.rs:
##########
@@ -37,33 +37,66 @@ impl Sink for DeltaSink {
);
let table_url = url::Url::parse(&self.config.table_uri).map_err(|e| {
- error!("Failed to parse table URI '{}': {e}",
self.config.table_uri);
- Error::InitError(format!("Invalid table URI: {e}"))
+ error!(
+ "Connector configuration: failed to parse table_uri = '{}':
{e}.",
+ self.config.table_uri
+ );
+ Error::InvalidConfigValue(format!("table_uri: {e}"))
})?;
+ let table_uri = &self.config.table_uri;
info!("Parsed table URI: {}", table_url);
- let storage_options = build_storage_options(&self.config).map_err(|e| {
- error!("Invalid storage configuration: {e}");
- Error::InitError(format!("Invalid storage configuration: {e}"))
+ let storage_options =
build_storage_options(&self.config).inspect_err(|e| {
+ error!("Connector configuration: invalid storage configuration.
Error message: {e}");
})?;
- let table =
- match deltalake::open_table_with_storage_options(table_url,
storage_options).await {
- Ok(table) => table,
- Err(e) => {
- error!("Failed to load Delta table: {e}");
- return Err(Error::InitError(format!("Failed to load Delta
table: {e}")));
+ info!("Successfully composed the storage options for accessing the
storage backend");
+
+ let builder = deltalake::DeltaTableBuilder::from_url(table_url)
+ .map_err(|e| {
+ error!("deltalake-rs interface: failed to configure with
table_uri = '{table_uri}'. Check deltalake::DeltaTableBuilder::from_url docs
and code to correct your table_uri. Error message: {e}");
+ Error::InvalidConfigValue(format!("table_uri = '{table_uri}'
caused an error in deltalake-rs interface. Check
deltalake::DeltaTableBuilder::from_url docs and code to correct your table_uri.
Error message: {e}"))
+ })?
+ .with_storage_options(storage_options);
+ let mut table = builder.build().map_err(|e| {
+ error!("deltalake-rs interface: failed to configure with provided
storage configuration. Error message: {e}");
+ Error::InitError(format!("deltalake-rs interface: failed to
configure with provided storage configuration. Error message: {e}"))
+ })?;
+ let table_exists = table
+ .verify_deltatable_existence()
+ .await
+ .map_err(
+ |e| {
+ error!("deltalake-rs interface: failed to list table_url
'{table_uri}' directory to verify delta table existence. Make sure the
destination exists and the access to the destination is set up correctly - read
the Iggy delta connector docs for more information. Error message: {e}");
Review Comment:
nit: the message says `table_url` but the config field is `table_uri`, so an
operator grepping their config finds nothing. rename the label here and in the
two tests that match it (delta_sink.rs:256 and :331).
##########
core/connectors/sinks/delta_sink/README.md:
##########
@@ -10,29 +10,149 @@ The Delta Lake Sink Connector allows you to consume
messages from Iggy topics an
The table must already exist. The connector appends each successful nonempty
batch in one Delta transaction and keeps its schema snapshot until restart. The
plugin has no failed-batch retry loop; the Delta library can retry eligible
commit conflicts and storage requests. Write or commit errors clear the writer
buffers and return an error. The runtime uses consumer auto-commit and does not
replay failed sink batches, so end-to-end at-least-once delivery is not
guaranteed.
-## Configuration example
-
-### Local filesystem
-
-```toml
-[plugin_config]
-table_uri = "file:///tmp/iggy_delta_table"
+## How to configure a Delta Sink connector
+
+First, make sure that the Delta table already exists in the location you're
providing. You can use this script for an example workload:
+
+```python
+import pyarrow as pa
+from deltalake import DeltaTable
+
+table_uri = "s3://test_location/tables/test"
Review Comment:
nit: `test_location` is not a valid bucket name (no underscores), and the
examples use three bucket placeholders and two regions, `iggy-sandbox` looks
like a real one. pick one valid placeholder and one region for all of them.
also line 151: "the same regions that is set" wants "region".
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
