Skip to content

fix(connector): iceberg dummy partition for every data #3860

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
ashudeqode wants to merge 3 commits into
base: master
Choose a base branch
from
+102 −99
Open

fix(connector): iceberg dummy partition for every data #3860

Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
92ef77f
fix: iceberg dummy partition for every data
ashudeqode Aug 11, 2026
dc10410
chore: cargo fmt
ashudeqode Aug 11, 2026
11ac098
Merge branch 'master' into fix/iceberg-partition-issue
ashudeqode Aug 11, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
201 changes: 102 additions & 99 deletions core/connectors/sinks/iceberg_sink/src/router/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,15 +20,15 @@ use crate::slice_user_table;
use arrow_json::ReaderBuilder;
use async_trait::async_trait;
use iceberg::TableIdent;
use iceberg::arrow::RecordBatchPartitionSplitter;
use iceberg::arrow::schema_to_arrow_schema;
use iceberg::spec::{
Literal, PartitionKey, PartitionSpec, PrimitiveLiteral, PrimitiveType, Struct, StructType,
};
use iceberg::table::Table;
use iceberg::transaction::{ApplyTransactionAction, Transaction};
use iceberg::writer::base_writer::data_file_writer::DataFileWriterBuilder;
use iceberg::writer::file_writer::ParquetWriterBuilder;
use iceberg::writer::file_writer::rolling_writer::RollingFileWriterBuilder;
use iceberg::writer::partitioning::PartitioningWriter;
use iceberg::writer::partitioning::fanout_writer::FanoutWriter;
use iceberg::writer::{IcebergWriter, IcebergWriterBuilder};
use iceberg::{
Catalog,
Expand Down Expand Up @@ -67,49 +67,6 @@ async fn table_exists(route_field_val: &str, catalog: &dyn Catalog) -> Option<Ta
catalog.load_table(&table_ident).await.ok()
}

pub fn primitive_type_to_literal(pt: &PrimitiveType) -> Result<PrimitiveLiteral, Error> {
match pt {
PrimitiveType::Boolean => Ok(PrimitiveLiteral::Boolean(false)),
PrimitiveType::Int => Ok(PrimitiveLiteral::Int(0)),
PrimitiveType::Long => Ok(PrimitiveLiteral::Long(0)),
PrimitiveType::Decimal { .. } => Ok(PrimitiveLiteral::Int128(0)),
PrimitiveType::Date => Ok(PrimitiveLiteral::Int(0)), // e.g. days since epoch
PrimitiveType::Time => Ok(PrimitiveLiteral::Long(0)), // microseconds since midnight
PrimitiveType::Timestamp => Ok(PrimitiveLiteral::Long(0)), // microseconds since epoch
PrimitiveType::Timestamptz => Ok(PrimitiveLiteral::Long(0)),
PrimitiveType::TimestampNs => Ok(PrimitiveLiteral::Long(0)),
PrimitiveType::TimestamptzNs => Ok(PrimitiveLiteral::Long(0)),
PrimitiveType::String => Ok(PrimitiveLiteral::String(String::new())),
PrimitiveType::Uuid => Ok(PrimitiveLiteral::Binary(vec![0; 16])),
PrimitiveType::Fixed(len) => Ok(PrimitiveLiteral::Binary(vec![0; *len as usize])),
PrimitiveType::Binary => Ok(PrimitiveLiteral::Binary(Vec::new())),
_ => {
error!("Partition type not supported");
Err(Error::InvalidConfig)
}
}
}

fn get_partition_type_value(default_partition_type: &StructType) -> Result<Option<Struct>, Error> {
let mut fields: Vec<Option<Literal>> = Vec::new();

if default_partition_type.fields().is_empty() {
return Ok(None);
};

for field in default_partition_type.fields() {
let field_type = field.field_type.as_primitive_type().ok_or_else(|| {
error!("The partition type of the configured iceberg table is not a primitive type");
Error::InvalidConfig
})?;

let value = Some(Literal::Primitive(primitive_type_to_literal(field_type)?));

fields.push(value);
}
Ok(Some(Struct::from_iter(fields)))
}

async fn write_data(
messages: &[Payload],
table: &Table,
Expand Down Expand Up @@ -145,27 +102,6 @@ async fn write_data(

let data_file_writer_builder = DataFileWriterBuilder::new(rolling_file_writer_builder);

let partition_spec = PartitionSpec::builder(table.current_schema_ref());

let partition_type = get_partition_type_value(table.metadata().default_partition_type())?;

let mut writer = data_file_writer_builder
.build(match partition_type {
None => None,
Some(p_type) => Some(PartitionKey::new(
partition_spec
.build()
.map_err(|err| Error::InitError(err.to_string()))?,
table.current_schema_ref(),
p_type,
)),
})
.await
.map_err(|err| {
error!("Error while constructing data file writer: {}", err);
Error::InitError(err.to_string())
})?;

let msgs: Vec<&simd_json::OwnedValue> = messages
.iter()
.filter_map(|payload| match payload {
Expand Down Expand Up @@ -201,42 +137,109 @@ async fn write_data(
Error::InitError(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)
})?;
let partition_spec = table.metadata().default_partition_spec();

let data_files = if partition_spec.is_unpartitioned() {

@EdgarModesto23 EdgarModesto23 Aug 11, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Could we extract the shared batch-processing and writer-closing logic here? The partitioned and unpartitioned branches are nearly identical, with only the write operation differing. A small writer abstraction or helper could encapsulate that variance and make this easier to maintain.

@ashudeqode ashudeqode Aug 12, 2026

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

sure, I'll do it.

let mut writer = data_file_writer_builder.build(None).await.map_err(|err| {
error!("Error while constructing data file writer: {}", err);
Error::InitError(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(())
}
Ok(())
}
.await;
.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);
if let Err(e) = &write_result {
error!(
"Batch loop failed ({}), closing writer to release resources",
e
);
if let Err(close_err) = writer.close().await {

@mattp5657 mattp5657 Aug 12, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Same thing here as above comment:

On a write failure we're possibly leaving orphaned files in the Iceberg table location. close() still finalizes whatever partitions already succeeded, but we never commit them and just throw away the result. Could we grab the files from close()'s return and delete them before returning the error?

@ashudeqode ashudeqode Aug 13, 2026

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

well in iceberg data warehouse we have to run compaction for tables as streaming data make tons of parquet files so we have to run compaction time to time which combines multiple files and leave the other files as it is and then we have to expire snapshots and then remove the orphaned file time to time so it is a three step process so this will cover in it but in this PR our main concern is that the data should go in correct partition

error!("Failed to close writer after batch error: {}", close_err);
}
return Err(write_result.unwrap_err());
}
return Err(write_result.unwrap_err());
}

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::WriteFailure(chain)
})?;
writer.close().await.map_err(|err| {
let chain = format_error_chain(&err);
error!(
"Error while writing data records to Parquet file: {}",
chain
);
Error::WriteFailure(chain)
})?
} else {
let splitter = RecordBatchPartitionSplitter::try_new_with_computed_values(
table.metadata().current_schema().clone(),
partition_spec.clone(),
)
.map_err(|err| {
error!("Failed to create partition splitter: {}", err);
Error::InitError(err.to_string())
})?;

let mut fanout_writer = FanoutWriter::new(data_file_writer_builder);

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)
})?;
let partitioned_batches = splitter.split(&batch_data).map_err(|err| {
let chain = format_error_chain(&err);
error!("Error while splitting batch by partition: {}", chain);
Error::InvalidRecordValue(chain)
})?;
for (partition_key, partition_batch) in partitioned_batches {
fanout_writer
.write(partition_key, partition_batch)
.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) = fanout_writer.close().await {

@mattp5657 mattp5657 Aug 12, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

On a write failure we're possibly leaving orphaned files in the Iceberg table location. close() still finalizes whatever partitions already succeeded, but we never commit them and just throw away the result. Could we grab the files from close()'s return and delete them before returning the error?

@EdgarModesto23 EdgarModesto23 Aug 13, 2026 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

You are correct! This has been noted before as seen in #3194 (comment) and it's currently documented under https://github.com/apache/iggy/blob/master/core/connectors/sdk/src/lib.rs#L429. It's not an issue from this PR specifically so I wouldn't block for it, tho we would love to see this fixed either here or on a separate issue for sure :) Thank you for bringing this up! ❤️

error!("Failed to close writer after batch error: {}", close_err);
}
return Err(write_result.unwrap_err());
}

fanout_writer.close().await.map_err(|err| {
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);

Expand Down
Loading