hubcio commented on code in PR #4219:
URL: https://github.com/apache/iggy/pull/4219#discussion_r4071042749
##########
core/connectors/sinks/delta_sink/README.md:
##########
@@ -10,40 +10,166 @@ 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 wrtitten 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. The write happens when there is either `batch_length` number of records
in the buffer or we hit `poll_interval` timeout. For Delta, writing means
creating a log entry and a separate Parquet file, so if these values are too
low (roughly `batch_length` < 1000 and `poll_interval` < 1s), the connector is
going to write 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 these values 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` + `poll_interval` should cut off the files in the
way that their size is in the suggested range. 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): Path or URI to the Delta table. Supported schemes:
`file://`, `s3://`, `az://`, `gs://`.
Review Comment:
warning: `table_uri` says "Path or URI", but `open()` parses it with
`url::Url::parse`, so a bare path fails - your own malformed-uri test proves
it. say absolute URI only, bare filesystem paths are not accepted.
##########
core/connectors/sinks/delta_sink/README.md:
##########
@@ -88,6 +186,6 @@ Required when `storage_backend_type = "gcs"`.
The connector automatically coerces JSON values to match the Delta table
schema:
-- **Timestamp fields**: ISO 8601 / RFC 3339 formatted strings (e.g.
`"2021-11-11T22:11:58Z", "2021-11-11 22:11:58"`) are converted to microsecond
timestamps. Integer epoch-microsecond timestamps pass through unchanged.
Space-separated timestamps without an offset are interpreted as UTC. Invalid
timestamp strings fail the batch.
-- **String fields**: Non-null, non-string values (numbers, booleans, objects,
arrays) are converted to their string representation.
-- **Nested fields**: Coercions cover nested structs, arrays of strings or
timestamps, and arrays of structs. Nested arrays, maps, and variant columns
pass through without these coercions. Nulls remain null.
+- **Timestamp fields**: ISO 8601 / RFC 3339 formatted strings (e.g.
`"2021-11-11T22:11:58Z", "2021-11-11 22:11:58"`) are converted to microsecond
timestamps. Numeric timestamps pass through unchanged.
+- **String fields**: Non-string values (numbers, booleans, objects, arrays)
are converted to their string representation.
+- **Nested fields**: Coercions are applied recursively to nested structs and
arrays.
Review Comment:
warning: the merge dropped the coercion caveats: map and variant columns
pass through uncoerced (`coercions.rs:82`), invalid timestamp strings fail the
whole batch, nulls stay null. restore those three bullets from master.
##########
core/connectors/sinks/delta_sink/README.md:
##########
@@ -10,40 +10,166 @@ 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 wrtitten 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. The write happens when there is either `batch_length` number of records
in the buffer or we hit `poll_interval` timeout. For Delta, writing means
creating a log entry and a separate Parquet file, so if these values are too
low (roughly `batch_length` < 1000 and `poll_interval` < 1s), the connector is
going to write 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 these values 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` + `poll_interval` should cut off the files in the
way that their size is in the suggested range. 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): Path or URI to the Delta table. Supported schemes:
`file://`, `s3://`, `az://`, `gs://`.
+- **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. Temporary security credentials issued by AWS STS, which is a best practice
recommended by AWS. The role assumed by the writer should allow these actions
on the bucket, here is how a working policy looks in HCL:
+
+ ```hcl
+ data "aws_iam_policy_document" "s3_write" {
+ statement {
+ sid = "BucketLevelActions"
+ effect = "Allow"
+ actions = [
+ "s3:ListBucket"
+ ]
+ resources = [aws_s3_bucket.unity_catalog["iggy-sandbox"].arn]
+ }
+
+ statement {
+ sid = "ObjectLevelActions"
+ effect = "Allow"
+ actions = [
+ "s3:GetObject",
+ "s3:PutObject",
+ "s3:PutObjectAcl",
+ "s3:AbortMultipartUpload"
+ ]
+ resources = ["${aws_s3_bucket.unity_catalog["iggy-sandbox"].arn}/*"]
+ }
+ }
+ ```
+
+ The corresponding plugin configuration will look like this:
+
+ ```toml
+ [plugin_config]
+ table_uri = "s3://iggy-sandbox/tables/test"
+ storage_backend_type = "s3"
+ aws_s3_region = "us-east-1"
+ ```
+
+2. Long-term access key pairs (access + secret key), which AWS recommends to
avoid for security reasons. The example configuration:
+
+ ```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"
+ ```
+
+Parameter descriptions:
+
+- **aws_s3_access_key**: Optional. AWS access key ID. Can only be passed
together with the secret key.
+- **aws_s3_secret_key**: Optional. AWS secret access key. Can only be passed
together with the access key.
+- **aws_s3_region**: Required. AWS region (e.g. `us-east-1`).
+- **aws_s3_endpoint_url**: Optional. S3 endpoint URL. Use for S3-compatible
services. Make sure that the URL implies the same regions that is set in
**aws_s3_region**, otherwise you'll have an error.
+- **aws_s3_allow_http**: Optional. Set to `true` to allow HTTP connections
(for local development).
+
+#### Azure Blob Storage
```toml
[plugin_config]
table_uri = "az://my-container/delta-tables/users"
storage_backend_type = "azure"
azure_storage_account_name = "mystorageaccount"
azure_storage_account_key = "account-key"
+azure_storage_sas_token = "sas-token"
Review Comment:
warning: this example sets both `azure_storage_account_key` and
`azure_storage_sas_token`, and `build_storage_options` rejects that pair at
`open()`. keep one credential here and restore the "exactly one of" sentence at
line 165 that the merge dropped.
##########
core/connectors/sinks/delta_sink/src/storage.rs:
##########
@@ -27,25 +27,33 @@ 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: the keyless branch is not isolated - deltalake-aws exports a keyed
sink's credentials into the process env (`ensure_env_var`), and `with_env_s3`
reads them back here. a keyless sink in the same runtime can pick up the keyed
sink's keys, document that.
##########
core/connectors/sinks/delta_sink/src/sink.rs:
##########
@@ -38,33 +38,69 @@ 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_url_parsed = &self.config.table_uri;
- info!("Parsed table URI: {}", table_url);
+ info!("Parsed table URI: {}", table_url_parsed);
let storage_options = build_storage_options(&self.config).map_err(|e| {
- error!("Invalid storage configuration: {e}");
- Error::InitError(format!("Invalid storage configuration: {e}"))
+ error!("Connector configuration: invalid storage configuration.
Error message: {e}");
+ Error::InitError(format!(
+ "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_url_parsed}'. Check deltalake::DeltaTableBuilder::from_url
docs and code to correct your table_uri. Error message: {e}");
+ Error::InvalidConfigValue(format!("table_uri =
'{table_url_parsed}' 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_url_parsed}' 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}");
+ Error::InitError(format!("deltalake-rs interface: failed
to list table_url '{table_url_parsed}' 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}"))
}
- };
+ )?;
+ if !table_exists {
+ error!(
+ "No delta table found in '{table_url_parsed}. Make sure to
create the delta table in the destination manually or verify the validity of
such table."
Review Comment:
nit: the quote before `{table_url_parsed}` is never closed, here and in the
error at line 84.
##########
core/connectors/sinks/delta_sink/src/sink.rs:
##########
@@ -122,9 +158,6 @@ impl Sink for DeltaSink {
return Ok(());
}
- // TODO: all partition consume() calls serialize on this single lock,
holding it
- // through flush_and_commit() I/O. fix: per-partition writers keyed by
partition_id.
- // Ref:
https://github.com/apache/iggy/pull/2889/#discussion_r2936719763
let mut state_guard = self.state.lock().await;
Review Comment:
nit: deleting the TODO does not fix #3839 - `consume()` still holds the
state lock across write and commit, so topic tasks of one sink serialize.
change the PR body to "Relates to #3839", or merging auto-closes the issue.
##########
core/connectors/sinks/delta_sink/README.md:
##########
@@ -10,40 +10,166 @@ 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 wrtitten 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. The write happens when there is either `batch_length` number of records
in the buffer or we hit `poll_interval` timeout. For Delta, writing means
creating a log entry and a separate Parquet file, so if these values are too
low (roughly `batch_length` < 1000 and `poll_interval` < 1s), the connector is
going to write lots of small files.
Review Comment:
warning: `poll_interval` is the minimum gap between polls, not a flush
timeout, and each commit holds at most `batch_length` rows. so file size is
capped by `batch_length` times row size, and intake by `batch_length` per
interval per topic.
##########
core/connectors/sinks/delta_sink/src/sink.rs:
##########
@@ -38,33 +38,69 @@ 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_url_parsed = &self.config.table_uri;
Review Comment:
nit: `table_url_parsed` holds the raw config string, so the "Parsed table
URI" log prints the unparsed value. log `table_url` and rename this to
`table_uri`.
##########
core/connectors/sinks/delta_sink/README.md:
##########
@@ -10,40 +10,166 @@ 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 wrtitten 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. The write happens when there is either `batch_length` number of records
in the buffer or we hit `poll_interval` timeout. For Delta, writing means
creating a log entry and a separate Parquet file, so if these values are too
low (roughly `batch_length` < 1000 and `poll_interval` < 1s), the connector is
going to write 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 these values 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` + `poll_interval` should cut off the files in the
way that their size is in the suggested range. 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): Path or URI to the Delta table. Supported schemes:
`file://`, `s3://`, `az://`, `gs://`.
+- **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. Temporary security credentials issued by AWS STS, which is a best practice
recommended by AWS. The role assumed by the writer should allow these actions
on the bucket, here is how a working policy looks in HCL:
Review Comment:
nit: STS is not the only keyless option, with no keys set deltalake-aws uses
the AWS SDK default chain (env vars, shared profile incl. SSO, web identity,
ECS, instance metadata). say that here.
##########
core/connectors/sinks/delta_sink/README.md:
##########
@@ -10,40 +10,166 @@ 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 wrtitten 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. The write happens when there is either `batch_length` number of records
in the buffer or we hit `poll_interval` timeout. For Delta, writing means
creating a log entry and a separate Parquet file, so if these values are too
low (roughly `batch_length` < 1000 and `poll_interval` < 1s), the connector is
going to write 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 these values 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` + `poll_interval` should cut off the files in the
way that their size is in the suggested range. 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): Path or URI to the Delta table. Supported schemes:
`file://`, `s3://`, `az://`, `gs://`.
+- **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. Temporary security credentials issued by AWS STS, which is a best practice
recommended by AWS. The role assumed by the writer should allow these actions
on the bucket, here is how a working policy looks in HCL:
+
+ ```hcl
+ data "aws_iam_policy_document" "s3_write" {
+ statement {
+ sid = "BucketLevelActions"
+ effect = "Allow"
+ actions = [
+ "s3:ListBucket"
+ ]
+ resources = [aws_s3_bucket.unity_catalog["iggy-sandbox"].arn]
+ }
+
+ statement {
+ sid = "ObjectLevelActions"
+ effect = "Allow"
+ actions = [
+ "s3:GetObject",
+ "s3:PutObject",
+ "s3:PutObjectAcl",
+ "s3:AbortMultipartUpload"
+ ]
+ resources = ["${aws_s3_bucket.unity_catalog["iggy-sandbox"].arn}/*"]
+ }
+ }
+ ```
+
+ The corresponding plugin configuration will look like this:
+
+ ```toml
+ [plugin_config]
+ table_uri = "s3://iggy-sandbox/tables/test"
+ storage_backend_type = "s3"
+ aws_s3_region = "us-east-1"
+ ```
+
+2. Long-term access key pairs (access + secret key), which AWS recommends to
avoid for security reasons. The example configuration:
+
+ ```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"
+ ```
+
+Parameter descriptions:
+
+- **aws_s3_access_key**: Optional. AWS access key ID. Can only be passed
together with the secret key.
+- **aws_s3_secret_key**: Optional. AWS secret access key. Can only be passed
together with the access key.
+- **aws_s3_region**: Required. AWS region (e.g. `us-east-1`).
+- **aws_s3_endpoint_url**: Optional. S3 endpoint URL. Use for S3-compatible
services. Make sure that the URL implies the same regions that is set in
**aws_s3_region**, otherwise you'll have an error.
Review Comment:
nit: a custom endpoint makes deltalake-aws skip the AWS SDK credential
chain, so profile and SSO credentials stop working. only static keys, web
identity, container credentials or instance metadata remain unless
`AWS_FORCE_CREDENTIAL_LOAD` is set in the runtime env.
##########
core/connectors/sinks/delta_sink/README.md:
##########
@@ -10,40 +10,166 @@ 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 wrtitten 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. The write happens when there is either `batch_length` number of records
in the buffer or we hit `poll_interval` timeout. For Delta, writing means
creating a log entry and a separate Parquet file, so if these values are too
low (roughly `batch_length` < 1000 and `poll_interval` < 1s), the connector is
going to write 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 these values 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` + `poll_interval` should cut off the files in the
way that their size is in the suggested range. 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): Path or URI to the Delta table. Supported schemes:
`file://`, `s3://`, `az://`, `gs://`.
+- **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. Temporary security credentials issued by AWS STS, which is a best practice
recommended by AWS. The role assumed by the writer should allow these actions
on the bucket, here is how a working policy looks in HCL:
+
+ ```hcl
+ data "aws_iam_policy_document" "s3_write" {
+ statement {
+ sid = "BucketLevelActions"
+ effect = "Allow"
+ actions = [
+ "s3:ListBucket"
+ ]
+ resources = [aws_s3_bucket.unity_catalog["iggy-sandbox"].arn]
Review Comment:
nit: `aws_s3_bucket.unity_catalog["iggy-sandbox"]` is a resource address
from a specific terraform setup, use a generic bucket ARN placeholder here and
at line 120.
##########
core/connectors/sinks/delta_sink/README.md:
##########
@@ -10,40 +10,166 @@ 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 wrtitten 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:
Review Comment:
nit: typo - "wrtitten".
##########
core/connectors/sinks/delta_sink/src/storage.rs:
##########
@@ -27,25 +27,33 @@ 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) => {}
+ _ => {
+ return Err(Error::InitError(
Review Comment:
nit: the sibling sinks return `Error::InvalidConfigValue` naming the fields
here (`iceberg_sink/src/props.rs:47`). do the same with `aws_s3_access_key` and
`aws_s3_secret_key`, and let `open()` pass it through instead of wrapping it in
a second `InitError`.
##########
core/integration/tests/connectors/fixtures/delta/fixture.rs:
##########
@@ -200,13 +200,72 @@ impl TestFixture for DeltaFixture {
}
}
-pub struct DeltaS3Fixture {
- #[allow(dead_code)]
- minio: ContainerAsync<GenericImage>,
- minio_endpoint: String,
+pub struct DeltaCorruptedLogFixture {
+ _temp_dir: TempDir,
+ table_path: PathBuf,
}
-impl DeltaS3Fixture {
+impl DeltaCorruptedLogFixture {
+ // Overwrites the version-0 commit file with garbage bytes but keeps its
+ // name intact: `is_delta_table_location` only lists filenames under
+ // `_delta_log`, so the table still "exists"; `DeltaTable::load` is what
+ // then fails parsing this file during log replay.
+ async fn corrupt_commit_log(table_path: &Path) -> Result<(),
TestBinaryError> {
+ let commit_path = table_path
+ .join("_delta_log")
+ .join("00000000000000000000.json");
+ tokio::fs::write(&commit_path, b"not a valid delta log commit entry")
+ .await
+ .map_err(|error| TestBinaryError::FixtureSetup {
+ fixture_type: "DeltaCorruptedLogFixture".to_string(),
+ message: format!(
+ "Failed to corrupt commit log at {}: {error}",
+ commit_path.display()
+ ),
+ })
+ }
+}
+
+#[async_trait]
+impl TestFixture for DeltaCorruptedLogFixture {
+ async fn setup() -> Result<Self, TestBinaryError> {
+ let temp_dir = TempDir::new().map_err(|error|
TestBinaryError::FixtureSetup {
+ fixture_type: "DeltaCorruptedLogFixture".to_string(),
+ message: format!("Failed to create temp directory: {error}"),
+ })?;
+
+ let table_path = temp_dir.path().join("delta_table");
+ let table_uri = format!("file://{}", table_path.display());
+ DeltaFixture::create_table(&table_uri).await?;
+ Self::corrupt_commit_log(&table_path).await?;
+ info!(
+ "Delta corrupted-log fixture created with table path: {}",
+ table_path.display()
+ );
+
+ Ok(Self {
+ _temp_dir: temp_dir,
+ table_path,
+ })
+ }
+
+ fn connectors_runtime_envs(&self) -> HashMap<String, String> {
+ let table_uri = format!("file://{}", self.table_path.display());
+
+ let mut envs = HashMap::new();
+ envs.insert(ENV_SINK_TABLE_URI.to_string(), table_uri);
+ envs.insert(
+ ENV_SINK_PATH.to_string(),
+ "../../target/debug/libiggy_connector_delta_sink".to_string(),
+ );
+ envs
+ }
+}
+
+#[async_trait]
+pub trait DeltaS3SinkOps {
Review Comment:
simplification: `DeltaS3SinkOps` moves the inherent `DeltaS3Fixture` helpers
into a trait, and the four wrapper impls exist to expose `minio_endpoint()`,
which no test calls. keep the inherent impl, drop the trait, its re-exports,
the two `.clone()` on owned strings and the `Box` around minio.
##########
core/integration/tests/connectors/fixtures/delta/fixture.rs:
##########
@@ -382,9 +453,137 @@ impl TestFixture for DeltaS3Fixture {
envs.insert(ENV_SINK_AWS_S3_REGION.to_string(),
"us-east-1".to_string());
envs.insert(
ENV_SINK_AWS_S3_ENDPOINT_URL.to_string(),
- self.minio_endpoint.clone(),
+ self.minio_endpoint().clone(),
);
envs.insert(ENV_SINK_AWS_S3_ALLOW_HTTP.to_string(),
"true".to_string());
envs
}
}
+
+pub struct DeltaS3NoTableFixture {
+ inner: DeltaS3Fixture,
+}
+
+impl DeltaS3SinkOps for DeltaS3NoTableFixture {
+ fn minio_endpoint(&self) -> String {
+ self.inner.minio_endpoint.clone()
+ }
+}
+
+#[async_trait]
+impl TestFixture for DeltaS3NoTableFixture {
+ async fn setup() -> Result<Self, TestBinaryError> {
+ let id = Uuid::new_v4();
+ let network = format!("iggy-delta-s3-{id}");
+ let minio_name =
fixtures::unique_container_name("minio-delta-no-table");
+
+ let (minio, minio_endpoint) = DeltaS3Fixture::start_minio(&network,
&minio_name).await?;
+ DeltaS3Fixture::create_bucket(&minio_endpoint).await?;
+
+ info!("Delta S3 'no table' fixture ready with MinIO at
{minio_endpoint}");
+
+ Ok(Self {
+ inner: DeltaS3Fixture {
+ minio: Box::new(minio),
+ minio_endpoint,
+ },
+ })
+ }
+
+ fn connectors_runtime_envs(&self) -> HashMap<String, String> {
+ self.inner.connectors_runtime_envs()
+ }
+}
+
+pub struct DeltaS3NoBucketFixture {
+ inner: DeltaS3Fixture,
+}
+
+impl DeltaS3SinkOps for DeltaS3NoBucketFixture {
+ fn minio_endpoint(&self) -> String {
+ self.inner.minio_endpoint.clone()
+ }
+}
+
+#[async_trait]
+impl TestFixture for DeltaS3NoBucketFixture {
+ async fn setup() -> Result<Self, TestBinaryError> {
+ let id = Uuid::new_v4();
+ let network = format!("iggy-delta-s3-{id}");
+ let minio_name =
fixtures::unique_container_name("minio-delta-no-bucket");
+
+ let (minio, minio_endpoint) = DeltaS3Fixture::start_minio(&network,
&minio_name).await?;
+
+ info!("Delta S3 'no bucket' fixture ready with MinIO at
{minio_endpoint}");
+
+ Ok(Self {
+ inner: DeltaS3Fixture {
+ minio: Box::new(minio),
+ minio_endpoint,
+ },
+ })
+ }
+
+ fn connectors_runtime_envs(&self) -> HashMap<String, String> {
+ self.inner.connectors_runtime_envs()
+ }
+}
+
+pub struct DeltaS3MissingSecretKeyFixture {
Review Comment:
simplification: this fixture starts minio, a bucket and a table to hit a
branch that fails before any storage call (`storage.rs:45`), and
`s3_backend_missing_secret_key_errors` already covers it. make the test
config-only like the malformed-uri one.
--
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]
