This is an automated email from the ASF dual-hosted git repository. numinnex pushed a commit to branch iceberg_invalid_rows in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 8783b9ec50e809fb704cef0437e301a7e40c39ab Author: Grzegorz Koszyk <[email protected]> AuthorDate: Tue Nov 3 17:40:56 2026 +0100 fix(connectors): skip Iceberg rows that do not match the table schema --- Cargo.lock | 3 +- Cargo.toml | 1 + .../example_config/connectors/iceberg_sink.toml | 1 + core/connectors/sinks/iceberg_sink/Cargo.toml | 3 +- core/connectors/sinks/iceberg_sink/README.md | 7 +- core/connectors/sinks/iceberg_sink/config.toml | 1 + core/connectors/sinks/iceberg_sink/src/lib.rs | 14 ++ core/connectors/sinks/iceberg_sink/src/props.rs | 1 + .../iceberg_sink/src/router/dynamic_router.rs | 12 +- .../sinks/iceberg_sink/src/router/mod.rs | 200 +++++++++++++++++---- .../sinks/iceberg_sink/src/router/static_router.rs | 21 ++- core/connectors/sinks/iceberg_sink/src/sink.rs | 4 +- .../tests/connectors/fixtures/iceberg/container.rs | 31 ++++ .../tests/connectors/fixtures/iceberg/mod.rs | 2 +- core/integration/tests/connectors/fixtures/mod.rs | 2 +- .../tests/connectors/iceberg/iceberg_sink.rs | 71 +++++++- 16 files changed, 320 insertions(+), 54 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 130f544a8..d6391c1ac 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -7590,10 +7590,11 @@ dependencies = [ [[package]] name = "iggy_connector_iceberg_sink" -version = "0.5.1-edge.1" +version = "0.5.1-edge.2" dependencies = [ "arrow-array 58.4.1", "arrow-json 58.4.1", + "arrow-schema 58.4.1", "async-trait", "dashmap", "iceberg", diff --git a/Cargo.toml b/Cargo.toml index de21ccbe0..407cdee53 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -102,6 +102,7 @@ argon2 = "0.6.0" arrow = "58.4.1" arrow-array = "58.4.1" arrow-json = "58.4.1" +arrow-schema = "58.4.1" assert_cmd = "2.2.2" async-broadcast = "0.7.2" async-channel = "2.5.0" diff --git a/core/connectors/runtime/example_config/connectors/iceberg_sink.toml b/core/connectors/runtime/example_config/connectors/iceberg_sink.toml index 0f1079614..f6c8285e2 100644 --- a/core/connectors/runtime/example_config/connectors/iceberg_sink.toml +++ b/core/connectors/runtime/example_config/connectors/iceberg_sink.toml @@ -43,6 +43,7 @@ store_access_key_id = "admin" store_secret_access_key = "password" store_region = "us-east-1" store_class = "s3" +invalid_rows = "fail" [transforms.add_fields] enabled = true diff --git a/core/connectors/sinks/iceberg_sink/Cargo.toml b/core/connectors/sinks/iceberg_sink/Cargo.toml index a8ecc7a54..292afdba4 100644 --- a/core/connectors/sinks/iceberg_sink/Cargo.toml +++ b/core/connectors/sinks/iceberg_sink/Cargo.toml @@ -17,7 +17,7 @@ [package] name = "iggy_connector_iceberg_sink" -version = "0.5.1-edge.1" +version = "0.5.1-edge.2" edition = "2024" license = "Apache-2.0" keywords = ["iggy", "messaging", "streaming"] @@ -37,6 +37,7 @@ crate-type = ["cdylib", "lib"] [dependencies] arrow-array = { workspace = true } arrow-json = { workspace = true } +arrow-schema = { workspace = true } async-trait = { workspace = true } dashmap = { workspace = true } iceberg = { workspace = true } diff --git a/core/connectors/sinks/iceberg_sink/README.md b/core/connectors/sinks/iceberg_sink/README.md index 093cf18d5..aacbbef38 100644 --- a/core/connectors/sinks/iceberg_sink/README.md +++ b/core/connectors/sinks/iceberg_sink/README.md @@ -30,6 +30,7 @@ store_secret_access_key = "password" store_region = "us-east-1" store_class = "s3" store_path_style_access = true +invalid_rows = "fail" ``` ## Configuration Options @@ -46,8 +47,9 @@ store_path_style_access = true - **store_region**: The region of the object storage. Required, including for S3-compatible stores that ignore it. - **store_class**: The storage class to use. **Currently, only S3-compatible storage is supported.** - **store_path_style_access**: Use path-style S3 URLs (`http://host/bucket/key`). Defaults to `true`, which MinIO-style endpoints require; set to `false` for stores that only accept virtual-hosted-style URLs. +- **invalid_rows**: What to do with a row that does not match the table schema. `fail`, the default, fails the whole batch. `skip` writes the rows that match and logs each skipped row as a warning with its offset. -All options above are required except the credential pair, `store_path_style_access`, and the unused routing field. +All options above are required except the credential pair, `store_path_style_access`, `invalid_rows`, and the unused routing field. Static routing requires a non-empty `tables` list; dynamic routing requires a non-empty `dynamic_route_field`. ## Static Routing @@ -121,7 +123,8 @@ can be lost after failed or skipped writes. Replaying committed rows can create Use `schema = "json"` on the stream. Each JSON object represents a table row; nested objects and arrays are supported when they match the table schema. Unknown fields are ignored. Missing -nullable fields become null; missing required fields or incompatible values fail the table write. +nullable fields become null. A missing required field or an incompatible value fails the table +write for the whole batch, unless `invalid_rows = "skip"` leaves out only that row. Sources that wrap row data in an envelope need a transform when the table schema describes the inner row. Otherwise, the Arrow JSON reader maps envelope keys to table columns, producing nulls diff --git a/core/connectors/sinks/iceberg_sink/config.toml b/core/connectors/sinks/iceberg_sink/config.toml index a4fa55c21..c0784cbd7 100644 --- a/core/connectors/sinks/iceberg_sink/config.toml +++ b/core/connectors/sinks/iceberg_sink/config.toml @@ -43,3 +43,4 @@ store_secret_access_key = "password" store_region = "us-east-1" store_class = "s3" store_path_style_access = true +invalid_rows = "fail" diff --git a/core/connectors/sinks/iceberg_sink/src/lib.rs b/core/connectors/sinks/iceberg_sink/src/lib.rs index f61ba1cd8..b09b3e6f5 100644 --- a/core/connectors/sinks/iceberg_sink/src/lib.rs +++ b/core/connectors/sinks/iceberg_sink/src/lib.rs @@ -25,6 +25,18 @@ mod props; mod router; mod sink; +/// What the sink does with a row that does not match the table schema. +#[derive(Debug, Default, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, StrumDisplay)] +#[serde(rename_all = "lowercase")] +#[strum(serialize_all = "lowercase")] +pub enum InvalidRows { + /// Fail the whole batch. + #[default] + Fail, + /// Write the rows that match, and log each skipped row with its offset. + Skip, +} + #[derive(Debug, Serialize, Deserialize, StrumDisplay)] #[serde(rename_all = "lowercase")] pub enum IcebergSinkTypes { @@ -69,6 +81,8 @@ pub struct IcebergSinkConfig { /// need it; iceberg 0.10 defaults to the virtual-host style otherwise. #[serde(default = "default_true")] pub store_path_style_access: bool, + #[serde(default)] + pub invalid_rows: Option<InvalidRows>, } fn default_true() -> bool { diff --git a/core/connectors/sinks/iceberg_sink/src/props.rs b/core/connectors/sinks/iceberg_sink/src/props.rs index 4c1635aa4..2eac7e67a 100644 --- a/core/connectors/sinks/iceberg_sink/src/props.rs +++ b/core/connectors/sinks/iceberg_sink/src/props.rs @@ -69,6 +69,7 @@ mod tests { store_region: "us-east-1".to_string(), store_class: IcebergSinkStoreClass::S3, store_path_style_access: true, + invalid_rows: None, } } diff --git a/core/connectors/sinks/iceberg_sink/src/router/dynamic_router.rs b/core/connectors/sinks/iceberg_sink/src/router/dynamic_router.rs index 17a717bd9..798658255 100644 --- a/core/connectors/sinks/iceberg_sink/src/router/dynamic_router.rs +++ b/core/connectors/sinks/iceberg_sink/src/router/dynamic_router.rs @@ -15,6 +15,7 @@ // specific language governing permissions and limitations // under the License. +use crate::InvalidRows; use crate::router::{Router, is_valid_namespaced_table, table_exists, write_data}; use async_trait::async_trait; use iceberg::Catalog; @@ -28,6 +29,7 @@ use tracing::{info, warn}; pub struct DynamicRouter { catalog: Box<dyn Catalog>, route_field: String, + invalid_rows: InvalidRows, } pub struct DynamicWriter { @@ -75,10 +77,11 @@ impl DynamicWriter { } impl DynamicRouter { - pub fn new(catalog: Box<dyn Catalog>, route_field: String) -> Self { + pub fn new(catalog: Box<dyn Catalog>, route_field: String, invalid_rows: InvalidRows) -> Self { Self { catalog, route_field, + invalid_rows, } } @@ -155,18 +158,17 @@ impl Router for DynamicRouter { None => continue, }; - let data: Vec<Payload> = batch_messages.into_iter().map(|m| m.payload).collect(); - write_data( - &data, + &batch_messages, table_obj, self.catalog.as_ref(), messages_metadata.schema, + self.invalid_rows, ) .await?; info!( "Dynamically routed {} messages to {} iceberg table", - data.len(), + batch_messages.len(), table_name ); } diff --git a/core/connectors/sinks/iceberg_sink/src/router/mod.rs b/core/connectors/sinks/iceberg_sink/src/router/mod.rs index 43fecfbfd..96a051298 100644 --- a/core/connectors/sinks/iceberg_sink/src/router/mod.rs +++ b/core/connectors/sinks/iceberg_sink/src/router/mod.rs @@ -15,10 +15,12 @@ // specific language governing permissions and limitations // under the License. +use crate::InvalidRows; use crate::router::arrow_streamer::JsonArrowReader; use crate::slice_user_table; use arrow_array::RecordBatch; use arrow_json::ReaderBuilder; +use arrow_schema::SchemaRef; use async_trait::async_trait; use iceberg::TableIdent; use iceberg::arrow::{RecordBatchPartitionSplitter, schema_to_arrow_schema}; @@ -34,7 +36,7 @@ use iceberg::{ Catalog, writer::file_writer::location_generator::{DefaultFileNameGenerator, DefaultLocationGenerator}, }; -use iggy_connector_sdk::{ConsumedMessage, Error, MessagesMetadata, Payload, Schema}; +use iggy_connector_sdk::{ConsumedMessage, Error, MessagesMetadata, Schema}; use parquet::file::properties::WriterProperties; use std::borrow::Cow; use std::collections::HashMap; @@ -75,12 +77,17 @@ async fn table_exists(route_field_val: &str, catalog: &dyn Catalog) -> Option<Ta } async fn write_data( - messages: &[Payload], + messages: &[ConsumedMessage], table: &Table, catalog: &dyn Catalog, messages_schema: Schema, + invalid_rows: InvalidRows, ) -> Result<(), Error> { - let data_files = write_data_files(messages, table, messages_schema).await?; + let data_files = write_data_files(messages, table, messages_schema, invalid_rows).await?; + if data_files.is_empty() { + // Every row was skipped, and an empty append would still add a snapshot. + return Ok(()); + } // Data files are kept on commit failure: the catalog may have applied the // commit before the error surfaced, and deleting files referenced by a @@ -114,9 +121,10 @@ async fn write_data( /// Writes the JSON payloads as Parquet data files under the table location, /// laid out by the table's default partition spec. Nothing is committed here. async fn write_data_files( - messages: &[Payload], + messages: &[ConsumedMessage], table: &Table, messages_schema: Schema, + invalid_rows: InvalidRows, ) -> Result<Vec<DataFile>, Error> { let location = DefaultLocationGenerator::new(table.metadata()).map_err(|err| { error!( @@ -150,17 +158,17 @@ async fn write_data_files( // Proto text holding JSON is the descriptor-less `proto_convert` fallback // and is written as the document it holds. Proto text that is not JSON is // skipped like any other payload without a document. - let documents: Vec<Cow<'_, simd_json::OwnedValue>> = messages + let documents: Vec<(u64, Cow<'_, simd_json::OwnedValue>)> = messages .iter() - .filter_map(|payload| { - let document = payload.json_document(); + .filter_map(|message| { + let document = message.payload.json_document(); if document.is_none() { warn!( "Unsupported type of payload, expected JSON, got {}", messages_schema.to_string() ); } - document + document.map(|document| (message.offset, document)) }) .collect(); @@ -172,9 +180,7 @@ async fn write_data_files( return Err(Error::InvalidPayloadType); } - let msgs: Vec<&simd_json::OwnedValue> = documents.iter().map(Cow::as_ref).collect(); - let cursor = JsonArrowReader::new(msgs.as_slice()); - let reader = ReaderBuilder::new(Arc::new( + let arrow_schema = Arc::new( schema_to_arrow_schema(&table.metadata().current_schema().clone()).map_err(|err| { let chain = format_error_chain(&err); error!( @@ -184,15 +190,11 @@ async fn write_data_files( ); Error::SchemaMismatch(chain) })?, - )) - .build(cursor) - .map_err(|err| { - error!( - "Error while building Iceberg reader from message payload: {}", - err - ); - Error::InitError(err.to_string()) - })?; + ); + let batches = decode_rows(&arrow_schema, &documents, invalid_rows)?; + if batches.is_empty() { + return Ok(Vec::new()); + } let mut writer = TableWriter::new(table, data_file_writer_builder).map_err(|err| { let chain = format_error_chain(&err); @@ -201,13 +203,8 @@ async fn write_data_files( })?; let write_result: Result<(), Error> = async { - for batch in reader { - let batch_data = batch.map_err(|err| { - let chain = format_error_chain(&err); - error!("Error while getting record batch: {}", chain); - Error::InvalidRecordValue(chain) - })?; - writer.write(batch_data).await?; + for batch in batches { + writer.write(batch).await?; } Ok(()) } @@ -227,6 +224,59 @@ async fn write_data_files( /// Files finalized by a failed batch are never committed, so they are removed /// instead of being left as orphans on the object store. +/// Decodes the JSON documents into record batches of `schema`. A row that does not match fails +/// the whole batch, unless `invalid_rows` skips it: then each such row is logged with its offset +/// and left out, and the others are decoded. +fn decode_rows( + schema: &SchemaRef, + documents: &[(u64, Cow<'_, simd_json::OwnedValue>)], + invalid_rows: InvalidRows, +) -> Result<Vec<RecordBatch>, Error> { + let rows: Vec<&simd_json::OwnedValue> = documents + .iter() + .map(|(_, document)| document.as_ref()) + .collect(); + let error = match read_batches(schema, &rows) { + Ok(batches) => return Ok(batches), + Err(error) => error, + }; + if invalid_rows == InvalidRows::Fail { + error!("Error while getting record batch: {error}"); + return Err(error); + } + // The batch read only says that some row does not match, so each row is read on its own. + let mut valid = Vec::with_capacity(rows.len()); + for (offset, document) in documents { + match read_batches(schema, &[document.as_ref()]) { + Ok(_) => valid.push(document.as_ref()), + Err(row_error) => warn!( + "Skipping the row at offset: {offset}, which does not match the table schema. {row_error}" + ), + } + } + if valid.is_empty() { + return Ok(Vec::new()); + } + read_batches(schema, &valid).inspect_err(|error| { + error!("Error while getting record batch: {error}"); + }) +} + +fn read_batches( + schema: &SchemaRef, + rows: &[&simd_json::OwnedValue], +) -> Result<Vec<RecordBatch>, Error> { + let reader = ReaderBuilder::new(Arc::clone(schema)) + .build(JsonArrowReader::new(rows)) + .map_err(|err| { + error!("Error while building Iceberg reader from message payload: {err}"); + Error::InitError(err.to_string()) + })?; + reader + .map(|batch| batch.map_err(|err| Error::InvalidRecordValue(format_error_chain(&err)))) + .collect() +} + async fn delete_uncommitted_files(file_io: &FileIO, data_files: &[DataFile]) { for data_file in data_files { if let Err(err) = file_io.delete(data_file.file_path()).await { @@ -379,6 +429,7 @@ mod tests { Schema as IcebergSchema, SortOrder, TableMetadataBuilder, Transform, Type, UnboundPartitionSpec, }; + use iggy_connector_sdk::Payload; use std::collections::HashMap; use std::sync::OnceLock; @@ -431,10 +482,49 @@ mod tests { .collect() } + /// Messages carrying `payloads`, with offsets counting from 0. + fn messages(payloads: &[Payload]) -> Vec<ConsumedMessage> { + payloads + .iter() + .enumerate() + .map(|(offset, payload)| ConsumedMessage { + id: offset as u128, + offset: offset as u64, + checksum: 0, + timestamp: 0, + origin_timestamp: 0, + headers: None, + payload: payload.clone(), + }) + .collect() + } + + fn write_with( + table: &Table, + payloads: &[Payload], + invalid_rows: InvalidRows, + ) -> Result<Vec<DataFile>, Error> { + test_runtime().block_on(write_data_files( + &messages(payloads), + table, + Schema::Json, + invalid_rows, + )) + } + fn write(table: &Table, payloads: &[Payload]) -> Vec<DataFile> { - test_runtime() - .block_on(write_data_files(payloads, table, Schema::Json)) - .expect("Failed to write data files") + write_with(table, payloads, InvalidRows::Fail).expect("Failed to write data files") + } + + /// Two rows that match the table and two that do not: a wrong type and a missing required + /// field. + fn rows_with_two_invalid() -> Vec<Payload> { + vec![ + Payload::Json(simd_json::json!({ "id": 1, "region": "eu" })), + Payload::Json(simd_json::json!({ "id": "not a number", "region": "us" })), + Payload::Json(simd_json::json!({ "id": 3 })), + Payload::Json(simd_json::json!({ "id": 4, "region": "us" })), + ] } fn partition_region(data_file: &DataFile) -> String { @@ -464,7 +554,12 @@ mod tests { let payloads = vec![Payload::Proto("id: 1".to_owned())]; let error = test_runtime() - .block_on(write_data_files(&payloads, &table, Schema::Proto)) + .block_on(write_data_files( + &messages(&payloads), + &table, + Schema::Proto, + InvalidRows::Fail, + )) .expect_err("proto text that is not JSON has no document to write"); assert_eq!(error, Error::InvalidPayloadType); @@ -543,8 +638,49 @@ mod tests { let table = in_memory_table(UnboundPartitionSpec::builder().build()); let payloads = vec![Payload::Text("not json".to_string())]; - let result = test_runtime().block_on(write_data_files(&payloads, &table, Schema::Text)); + let result = test_runtime().block_on(write_data_files( + &messages(&payloads), + &table, + Schema::Text, + InvalidRows::Fail, + )); assert!(matches!(result, Err(Error::InvalidPayloadType))); } + + #[test] + fn given_rows_that_do_not_match_when_invalid_rows_fail_should_fail_the_batch() { + let table = in_memory_table(UnboundPartitionSpec::builder().build()); + + let result = write_with(&table, &rows_with_two_invalid(), InvalidRows::Fail); + + assert!( + matches!(result, Err(Error::InvalidRecordValue(_))), + "got {result:?}" + ); + } + + #[test] + fn given_rows_that_do_not_match_when_invalid_rows_skip_should_write_the_other_rows() { + let table = in_memory_table(UnboundPartitionSpec::builder().build()); + + let data_files = write_with(&table, &rows_with_two_invalid(), InvalidRows::Skip) + .expect("the matching rows should be written"); + + let written: u64 = data_files.iter().map(DataFile::record_count).sum(); + assert_eq!(written, 2); + } + + #[test] + fn given_only_rows_that_do_not_match_when_invalid_rows_skip_should_write_no_data_file() { + let table = in_memory_table(UnboundPartitionSpec::builder().build()); + let payloads = vec![Payload::Json( + simd_json::json!({ "id": "not a number", "region": "eu" }), + )]; + + let data_files = write_with(&table, &payloads, InvalidRows::Skip) + .expect("skipping every row is not an error"); + + assert!(data_files.is_empty()); + } } diff --git a/core/connectors/sinks/iceberg_sink/src/router/static_router.rs b/core/connectors/sinks/iceberg_sink/src/router/static_router.rs index 1c8e2a6bf..ba64691be 100644 --- a/core/connectors/sinks/iceberg_sink/src/router/static_router.rs +++ b/core/connectors/sinks/iceberg_sink/src/router/static_router.rs @@ -15,23 +15,26 @@ // specific language governing permissions and limitations // under the License. +use crate::InvalidRows; use crate::router::{Router, is_valid_namespaced_table, table_exists, write_data}; use async_trait::async_trait; use iceberg::Catalog; use iceberg::table::Table; -use iggy_connector_sdk::{ConsumedMessage, Error, MessagesMetadata, Payload}; +use iggy_connector_sdk::{ConsumedMessage, Error, MessagesMetadata}; use tracing::{error, info, warn}; #[derive(Debug)] pub(crate) struct StaticRouter { tables: Vec<Table>, catalog: Box<dyn Catalog>, + invalid_rows: InvalidRows, } impl StaticRouter { pub async fn new( catalog: Box<dyn Catalog>, declared_tables: &Vec<String>, + invalid_rows: InvalidRows, ) -> Result<Self, Error> { let mut tables: Vec<Table> = Vec::with_capacity(declared_tables.len()); for declared_table in declared_tables { @@ -64,7 +67,11 @@ impl StaticRouter { error!("No valid tables found. Can't initiate Iceberg connector"); return Err(Error::InvalidConfig); } - Ok(StaticRouter { tables, catalog }) + Ok(StaticRouter { + tables, + catalog, + invalid_rows, + }) } } @@ -75,22 +82,18 @@ impl Router for StaticRouter { messages_metadata: MessagesMetadata, messages: Vec<ConsumedMessage>, ) -> Result<(), crate::Error> { - let data: Vec<Payload> = messages - .into_iter() - .map(|m: ConsumedMessage| m.payload) - .collect(); - for table in &self.tables { write_data( - &data, + &messages, table, self.catalog.as_ref(), messages_metadata.schema, + self.invalid_rows, ) .await?; info!( "Routed {} messages to iceberg table {} successfully", - data.len(), + messages.len(), table.identifier().name() ); } diff --git a/core/connectors/sinks/iceberg_sink/src/sink.rs b/core/connectors/sinks/iceberg_sink/src/sink.rs index 1dd33cd39..ed59e9e8e 100644 --- a/core/connectors/sinks/iceberg_sink/src/sink.rs +++ b/core/connectors/sinks/iceberg_sink/src/sink.rs @@ -73,15 +73,17 @@ impl Sink for IcebergSink { ); let catalog: Box<dyn Catalog> = init_catalog(&self.config).await?; + let invalid_rows = self.config.invalid_rows.unwrap_or_default(); if self.config.dynamic_routing { self.router = Some(Box::new(DynamicRouter::new( catalog, self.config.dynamic_route_field.clone(), + invalid_rows, ))) } else { self.router = Some(Box::new( - StaticRouter::new(catalog, &self.config.tables).await?, + StaticRouter::new(catalog, &self.config.tables, invalid_rows).await?, )); } diff --git a/core/integration/tests/connectors/fixtures/iceberg/container.rs b/core/integration/tests/connectors/fixtures/iceberg/container.rs index 6e1c879bb..22284477e 100644 --- a/core/integration/tests/connectors/fixtures/iceberg/container.rs +++ b/core/integration/tests/connectors/fixtures/iceberg/container.rs @@ -49,6 +49,7 @@ pub const ENV_SINK_STORE_SECRET: &str = "IGGY_CONNECTORS_SINK_ICEBERG_PLUGIN_CONFIG_STORE_SECRET_ACCESS_KEY"; pub const ENV_SINK_STORE_REGION: &str = "IGGY_CONNECTORS_SINK_ICEBERG_PLUGIN_CONFIG_STORE_REGION"; pub const ENV_SINK_PATH: &str = "IGGY_CONNECTORS_SINK_ICEBERG_PATH"; +pub const ENV_SINK_INVALID_ROWS: &str = "IGGY_CONNECTORS_SINK_ICEBERG_PLUGIN_CONFIG_INVALID_ROWS"; pub const ENV_AWS_ACCESS_KEY_ID: &str = "AWS_ACCESS_KEY_ID"; pub const ENV_AWS_SECRET_ACCESS_KEY: &str = "AWS_SECRET_ACCESS_KEY"; @@ -457,6 +458,36 @@ impl TestFixture for IcebergPreCreatedFixture { } } +/// Same table as `IcebergPreCreatedFixture`, with a sink that skips rows that do not match it. +pub struct IcebergSkipInvalidRowsFixture { + inner: IcebergPreCreatedFixture, +} + +impl IcebergOps for IcebergSkipInvalidRowsFixture { + fn catalog_url(&self) -> &str { + self.inner.catalog_url() + } + + fn http_client(&self) -> &HttpClient { + self.inner.http_client() + } +} + +#[async_trait] +impl TestFixture for IcebergSkipInvalidRowsFixture { + async fn setup() -> Result<Self, TestBinaryError> { + Ok(Self { + inner: IcebergPreCreatedFixture::setup().await?, + }) + } + + fn connectors_runtime_envs(&self) -> HashMap<String, String> { + let mut envs = self.inner.connectors_runtime_envs(); + envs.insert(ENV_SINK_INVALID_ROWS.to_string(), "skip".to_string()); + envs + } +} + /// Same table as `IcebergPreCreatedFixture`, partitioned by identity(`active`). pub struct IcebergPartitionedTableFixture { inner: IcebergPreCreatedFixture, diff --git a/core/integration/tests/connectors/fixtures/iceberg/mod.rs b/core/integration/tests/connectors/fixtures/iceberg/mod.rs index 0b5dd1877..48f198371 100644 --- a/core/integration/tests/connectors/fixtures/iceberg/mod.rs +++ b/core/integration/tests/connectors/fixtures/iceberg/mod.rs @@ -19,5 +19,5 @@ mod container; pub use container::{ DEFAULT_NAMESPACE, DEFAULT_TABLE, IcebergEnvAuthFixture, IcebergOps, - IcebergPartitionedTableFixture, IcebergPreCreatedFixture, + IcebergPartitionedTableFixture, IcebergPreCreatedFixture, IcebergSkipInvalidRowsFixture, }; diff --git a/core/integration/tests/connectors/fixtures/mod.rs b/core/integration/tests/connectors/fixtures/mod.rs index 3912d0a61..5d067d386 100644 --- a/core/integration/tests/connectors/fixtures/mod.rs +++ b/core/integration/tests/connectors/fixtures/mod.rs @@ -72,7 +72,7 @@ pub use http::{ }; pub use iceberg::{ DEFAULT_NAMESPACE, DEFAULT_TABLE, IcebergEnvAuthFixture, IcebergOps, - IcebergPartitionedTableFixture, IcebergPreCreatedFixture, + IcebergPartitionedTableFixture, IcebergPreCreatedFixture, IcebergSkipInvalidRowsFixture, }; pub use influxdb::{ InfluxDb3SinkFixture, InfluxDb3SourceFixture, InfluxDbSinkBase64Fixture, InfluxDbSinkFixture, diff --git a/core/integration/tests/connectors/iceberg/iceberg_sink.rs b/core/integration/tests/connectors/iceberg/iceberg_sink.rs index 2e969775c..029a75680 100644 --- a/core/integration/tests/connectors/iceberg/iceberg_sink.rs +++ b/core/integration/tests/connectors/iceberg/iceberg_sink.rs @@ -18,7 +18,7 @@ use crate::connectors::create_test_messages; use crate::connectors::fixtures::{ DEFAULT_NAMESPACE, DEFAULT_TABLE, IcebergEnvAuthFixture, IcebergOps, - IcebergPartitionedTableFixture, IcebergPreCreatedFixture, + IcebergPartitionedTableFixture, IcebergPreCreatedFixture, IcebergSkipInvalidRowsFixture, }; use bytes::Bytes; use iggy::prelude::{IggyMessage, Partitioning}; @@ -337,3 +337,72 @@ async fn iceberg_sink_uses_default_credential_chain( assert!(snapshot_count >= 1); drop(fixture); } + +#[iggy_harness( + server(connectors_runtime(config_path = "tests/connectors/iceberg/sink.toml")), + seed = seeds::connector_stream +)] +async fn given_invalid_rows_skip_when_a_batch_holds_a_row_that_does_not_match_should_write_the_other_rows( + harness: &TestHarness, + fixture: IcebergSkipInvalidRowsFixture, +) { + let client = harness.root_client().await.unwrap(); + let stream_id: Identifier = seeds::names::STREAM.try_into().unwrap(); + let topic_id: Identifier = seeds::names::TOPIC.try_into().unwrap(); + + let valid_rows = create_test_messages(4); + let mut payloads: Vec<Vec<u8>> = valid_rows + .iter() + .map(|row| serde_json::to_vec(row).expect("Failed to serialize message")) + .collect(); + // `id` is a required long column, so this row does not match the table. + payloads.insert( + 2, + serde_json::to_vec(&serde_json::json!({ "id": "not a number", "name": "invalid" })) + .expect("Failed to serialize message"), + ); + let mut messages: Vec<IggyMessage> = payloads + .into_iter() + .enumerate() + .map(|(i, payload)| { + IggyMessage::builder() + .id((i + 1) as u128) + .payload(Bytes::from(payload)) + .build() + .expect("Failed to build message") + }) + .collect(); + client + .send_messages( + &stream_id, + &topic_id, + &Partitioning::partition_id(0), + &mut messages, + ) + .await + .expect("Failed to send messages"); + + fixture + .wait_for_snapshots( + DEFAULT_NAMESPACE, + DEFAULT_TABLE, + 1, + SNAPSHOT_POLL_ATTEMPTS, + SNAPSHOT_POLL_INTERVAL_MS, + ) + .await + .expect("The matching rows should be written to Iceberg"); + let metadata = fixture + .get_table_metadata(DEFAULT_NAMESPACE, DEFAULT_TABLE) + .await + .expect("Failed to get table metadata"); + let written: u64 = metadata + .metadata + .snapshots + .unwrap_or_default() + .iter() + .filter_map(|snapshot| snapshot.summary.as_ref()?.added_records.as_deref()) + .map(|records| records.parse::<u64>().expect("added-records is a count")) + .sum(); + assert_eq!(written, valid_rows.len() as u64); +}
