From 439ea4a75af809ffb3b596e6e2b80bb685f94776 Mon Sep 17 00:00:00 2001 From: Atharva Lade Date: Tue, 28 Apr 2026 19:28:50 -0500 Subject: [PATCH 1/5] fix(connectors): replace overloaded InvalidRecord with distinct error variants --- core/connectors/sdk/src/lib.rs | 12 ++++++++++++ core/connectors/sinks/iceberg_sink/src/router/mod.rs | 10 +++++----- 2 files changed, 17 insertions(+), 5 deletions(-) diff --git a/core/connectors/sdk/src/lib.rs b/core/connectors/sdk/src/lib.rs index 2289aa9503..c485cf072f 100644 --- a/core/connectors/sdk/src/lib.rs +++ b/core/connectors/sdk/src/lib.rs @@ -425,4 +425,16 @@ pub enum Error { /// failures so that circuit breakers are not tripped by bad data. #[error("Permanent HTTP error: {0}")] PermanentHttpError(String), + /// The source schema could not be mapped to the destination schema. + /// Indicates a table definition or configuration problem — not a data issue. + #[error("Schema mismatch: {0}")] + SchemaMismatch(String), + /// An I/O failure while writing data (e.g. Parquet serialization, file + /// writer close). Distinct from record-level validation errors. + #[error("Write failure: {0}")] + WriteFailure(String), + /// A catalog or transaction-level failure (e.g. applying or committing an + /// Iceberg transaction). Callers may retry on transient catalog outages. + #[error("Catalog error: {0}")] + CatalogError(String), } diff --git a/core/connectors/sinks/iceberg_sink/src/router/mod.rs b/core/connectors/sinks/iceberg_sink/src/router/mod.rs index 8e60beedfc..a8f677237f 100644 --- a/core/connectors/sinks/iceberg_sink/src/router/mod.rs +++ b/core/connectors/sinks/iceberg_sink/src/router/mod.rs @@ -178,7 +178,7 @@ async fn write_data( table.metadata().uuid(), err ); - Error::InvalidRecord + Error::SchemaMismatch(err.to_string()) })?, )) .build(cursor) @@ -197,13 +197,13 @@ async fn write_data( })?; writer.write(batch_data).await.map_err(|err| { error!("Error while writing record batch: {}", err); - Error::InvalidRecord + Error::WriteFailure(err.to_string()) })?; } let data_files = writer.close().await.map_err(|err| { error!("Error while writing data records to Parquet file: {}", err); - Error::InvalidRecord + Error::WriteFailure(err.to_string()) })?; let table_commit = Transaction::new(table); @@ -216,7 +216,7 @@ async fn write_data( table.metadata().uuid(), err ); - Error::InvalidRecord + Error::CatalogError(err.to_string()) })?; let _table = tx.commit(catalog).await.map_err(|err| { @@ -225,7 +225,7 @@ async fn write_data( table.metadata().uuid(), err ); - Error::InvalidRecord + Error::CatalogError(err.to_string()) })?; Ok(()) } From d67a0ee6ca632fa9fa277b43973b58a63e12d0fb Mon Sep 17 00:00:00 2001 From: Atharva Lade <92752921+atharvalade@users.noreply.github.com> Date: Sun, 3 May 2026 18:14:39 -0500 Subject: [PATCH 2/5] Refine SchemaMismatch error documentation Removed redundant phrasing from SchemaMismatch error documentation. --- core/connectors/sdk/src/lib.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/core/connectors/sdk/src/lib.rs b/core/connectors/sdk/src/lib.rs index c485cf072f..8aa68a090f 100644 --- a/core/connectors/sdk/src/lib.rs +++ b/core/connectors/sdk/src/lib.rs @@ -426,7 +426,7 @@ pub enum Error { #[error("Permanent HTTP error: {0}")] PermanentHttpError(String), /// The source schema could not be mapped to the destination schema. - /// Indicates a table definition or configuration problem — not a data issue. + /// Indicates a table definition or configuration problem #[error("Schema mismatch: {0}")] SchemaMismatch(String), /// An I/O failure while writing data (e.g. Parquet serialization, file From f282de2e076e1f078fcef68b2616bdf561d16be8 Mon Sep 17 00:00:00 2001 From: Atharva Lade Date: Sun, 17 May 2026 01:51:54 -0500 Subject: [PATCH 3/5] fix(connectors): split CatalogError into TransactionApplyError and CatalogCommitError --- core/connectors/sdk/src/lib.rs | 12 ++++++++---- core/connectors/sinks/iceberg_sink/src/router/mod.rs | 4 ++-- 2 files changed, 10 insertions(+), 6 deletions(-) diff --git a/core/connectors/sdk/src/lib.rs b/core/connectors/sdk/src/lib.rs index 8aa68a090f..aee7d8c78b 100644 --- a/core/connectors/sdk/src/lib.rs +++ b/core/connectors/sdk/src/lib.rs @@ -433,8 +433,12 @@ pub enum Error { /// writer close). Distinct from record-level validation errors. #[error("Write failure: {0}")] WriteFailure(String), - /// A catalog or transaction-level failure (e.g. applying or committing an - /// Iceberg transaction). Callers may retry on transient catalog outages. - #[error("Catalog error: {0}")] - CatalogError(String), + /// In-memory transaction preparation failed (e.g. invalid partition spec, + /// schema validation). These failures are deterministic and not retryable. + #[error("Transaction apply error: {0}")] + TransactionApplyError(String), + /// A catalog commit failed over the network (e.g. REST catalog I/O, + /// conflict). Transient failures may be retried. + #[error("Catalog commit error: {0}")] + CatalogCommitError(String), } diff --git a/core/connectors/sinks/iceberg_sink/src/router/mod.rs b/core/connectors/sinks/iceberg_sink/src/router/mod.rs index a8f677237f..ed5ca2bb3d 100644 --- a/core/connectors/sinks/iceberg_sink/src/router/mod.rs +++ b/core/connectors/sinks/iceberg_sink/src/router/mod.rs @@ -216,7 +216,7 @@ async fn write_data( table.metadata().uuid(), err ); - Error::CatalogError(err.to_string()) + Error::TransactionApplyError(err.to_string()) })?; let _table = tx.commit(catalog).await.map_err(|err| { @@ -225,7 +225,7 @@ async fn write_data( table.metadata().uuid(), err ); - Error::CatalogError(err.to_string()) + Error::CatalogCommitError(err.to_string()) })?; Ok(()) } From b19bbedfa6772d368199d22d105ee3a4f1a8dc52 Mon Sep 17 00:00:00 2001 From: Atharva Lade Date: Fri, 22 May 2026 21:51:16 -0700 Subject: [PATCH 4/5] fix(connectors): address reviewer feedback on error variant docs and diagnostics --- core/connectors/sdk/src/lib.rs | 15 +++-- .../sinks/iceberg_sink/src/router/mod.rs | 63 ++++++++++++++----- 2 files changed, 56 insertions(+), 22 deletions(-) diff --git a/core/connectors/sdk/src/lib.rs b/core/connectors/sdk/src/lib.rs index aee7d8c78b..627bc2b3d7 100644 --- a/core/connectors/sdk/src/lib.rs +++ b/core/connectors/sdk/src/lib.rs @@ -426,19 +426,24 @@ pub enum Error { #[error("Permanent HTTP error: {0}")] PermanentHttpError(String), /// The source schema could not be mapped to the destination schema. - /// Indicates a table definition or configuration problem + /// Indicates a table definition or configuration problem. #[error("Schema mismatch: {0}")] SchemaMismatch(String), /// An I/O failure while writing data (e.g. Parquet serialization, file - /// writer close). Distinct from record-level validation errors. + /// writer close). Distinct from record-level validation errors. May leave + /// orphaned partial files on the object store; cleanup is caller-side. #[error("Write failure: {0}")] WriteFailure(String), /// In-memory transaction preparation failed (e.g. invalid partition spec, - /// schema validation). These failures are deterministic and not retryable. + /// schema validation). Typically deterministic in the current Iceberg + /// version; check the underlying Iceberg error to decide retryability. #[error("Transaction apply error: {0}")] TransactionApplyError(String), - /// A catalog commit failed over the network (e.g. REST catalog I/O, - /// conflict). Transient failures may be retried. + /// A catalog commit failed. `Transaction::commit()` consumes the + /// transaction, so retrying requires rebuilding it from new data files. + /// Retry is not idempotent: callers must verify via the catalog whether + /// the original commit was applied before retrying, otherwise data may + /// be duplicated. #[error("Catalog commit error: {0}")] CatalogCommitError(String), } diff --git a/core/connectors/sinks/iceberg_sink/src/router/mod.rs b/core/connectors/sinks/iceberg_sink/src/router/mod.rs index ed5ca2bb3d..3160fb135e 100644 --- a/core/connectors/sinks/iceberg_sink/src/router/mod.rs +++ b/core/connectors/sinks/iceberg_sink/src/router/mod.rs @@ -41,6 +41,17 @@ use std::sync::Arc; use tracing::{error, warn}; use uuid::Uuid; +fn format_error_chain(err: &dyn std::error::Error) -> String { + let mut chain = err.to_string(); + let mut source = err.source(); + while let Some(cause) = source { + chain.push_str(": "); + chain.push_str(&cause.to_string()); + source = cause.source(); + } + chain +} + mod arrow_streamer; pub mod dynamic_router; pub mod static_router; @@ -173,12 +184,13 @@ async fn write_data( let cursor = JsonArrowReader::new(msgs.as_slice()); let reader = ReaderBuilder::new(Arc::new( schema_to_arrow_schema(&table.metadata().current_schema().clone()).map_err(|err| { + let chain = format_error_chain(&err); error!( "Error while mapping records to Iceberg table with uuid: {}. Error {}", table.metadata().uuid(), - err + chain ); - Error::SchemaMismatch(err.to_string()) + Error::SchemaMismatch(chain) })?, )) .build(cursor) @@ -190,20 +202,35 @@ async fn write_data( Error::InitError(err.to_string()) })?; - for batch in reader { - let batch_data = batch.map_err(|err| { - error!("Error while getting record batch: {}", err); - Error::InvalidRecord - })?; - writer.write(batch_data).await.map_err(|err| { - error!("Error while writing record batch: {}", err); - Error::WriteFailure(err.to_string()) - })?; + 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.map_err(|err| { + let chain = format_error_chain(&err); + error!("Error while writing record batch: {}", chain); + Error::WriteFailure(chain) + })?; + } + Ok(()) + } + .await; + + if let Err(e) = &write_result { + error!("Batch loop failed ({}), closing writer to release resources", e); + if let Err(close_err) = writer.close().await { + error!("Failed to close writer after batch error: {}", close_err); + } + return Err(write_result.unwrap_err()); } let data_files = writer.close().await.map_err(|err| { - error!("Error while writing data records to Parquet file: {}", err); - Error::WriteFailure(err.to_string()) + let chain = format_error_chain(&err); + error!("Error while writing data records to Parquet file: {}", chain); + Error::WriteFailure(chain) })?; let table_commit = Transaction::new(table); @@ -211,21 +238,23 @@ async fn write_data( let action = table_commit.fast_append().add_data_files(data_files); let tx = action.apply(table_commit).map_err(|err| { + let chain = format_error_chain(&err); error!( "Failed to apply transaction on table with UUID: {}, Error: {}", table.metadata().uuid(), - err + chain ); - Error::TransactionApplyError(err.to_string()) + Error::TransactionApplyError(chain) })?; let _table = tx.commit(catalog).await.map_err(|err| { + let chain = format_error_chain(&err); error!( "Failed to commit transaction on table with UUID: {}, Error: {}", table.metadata().uuid(), - err + chain ); - Error::CatalogCommitError(err.to_string()) + Error::CatalogCommitError(chain) })?; Ok(()) } From 9ffe94decff1ca636d0e6d1ec3b384fb6c4ede0b Mon Sep 17 00:00:00 2001 From: Atharva Lade Date: Fri, 22 May 2026 21:55:20 -0700 Subject: [PATCH 5/5] style(connectors): fix rustfmt formatting in error log macros --- core/connectors/sinks/iceberg_sink/src/router/mod.rs | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/core/connectors/sinks/iceberg_sink/src/router/mod.rs b/core/connectors/sinks/iceberg_sink/src/router/mod.rs index 3160fb135e..67a4a555cd 100644 --- a/core/connectors/sinks/iceberg_sink/src/router/mod.rs +++ b/core/connectors/sinks/iceberg_sink/src/router/mod.rs @@ -220,7 +220,10 @@ async fn write_data( .await; if let Err(e) = &write_result { - error!("Batch loop failed ({}), closing writer to release resources", e); + error!( + "Batch loop failed ({}), closing writer to release resources", + e + ); if let Err(close_err) = writer.close().await { error!("Failed to close writer after batch error: {}", close_err); } @@ -229,7 +232,10 @@ async fn write_data( let data_files = writer.close().await.map_err(|err| { let chain = format_error_chain(&err); - error!("Error while writing data records to Parquet file: {}", chain); + error!( + "Error while writing data records to Parquet file: {}", + chain + ); Error::WriteFailure(chain) })?;