fix(connector): iceberg dummy partition for every data - #3860
Conversation
|
Thanks for the PR. It is labeled Slash commands (own line, regular comment) move it around the queue:
See CONTRIBUTING.md for details. |
ashudeqode
changed the title
|
@EdgarModesto23 could you please check this? |
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## master #3860 +/- ##
============================================
- Coverage 76.95% 76.81% -0.15%
Complexity 1021 1021
============================================
Files 1366 1366
Lines 176130 175838 -292
Branches 146173 145881 -292
============================================
- Hits 135547 135072 -475
- Misses 36707 36885 +178
- Partials 3876 3881 +5
🚀 New features to boost your workflow:
|
|
@ashudeqode please dont remove the PR template, fill it so we can see reasoning behind this change. |
@hubcio Updated the description |
| })?; | ||
| let partition_spec = table.metadata().default_partition_spec(); | ||
|
|
||
| let data_files = if partition_spec.is_unpartitioned() { |
There was a problem hiding this comment.
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.
github-actions
Bot
added
S-waiting-on-author
mattp5657
left a comment
mattp5657
left a comment
There was a problem hiding this comment.
@EdgarModesto23 I'm still pretty new here, but wanted to jump in and help review. Let me know if you agree with the comments below.
Also think we should probably add some regression testing here.
| "Batch loop failed ({}), closing writer to release resources", | ||
| e | ||
| ); | ||
| if let Err(close_err) = fanout_writer.close().await { |
There was a problem hiding this 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?
There was a problem hiding this comment.
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! ❤️
| "Batch loop failed ({}), closing writer to release resources", | ||
| e | ||
| ); | ||
| if let Err(close_err) = writer.close().await { |
There was a problem hiding this comment.
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?
There was a problem hiding this comment.
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
4 participants
Which issue does this PR address?
Closes #3853
Rationale
The Iceberg sink's write_data function had three critical bugs that caused all records to land in a single fake partition, regardless of the table's actual partition spec. Partition values were hardcoded to dummy zeros (e.g., Int(0), String("")), the partition spec was built from scratch as empty instead of read from table metadata, and there was no fan-out logic to route records from a mixed-partition batch to separate data files.
What changed?
The sink was populating a PartitionKey with static zero-value literals and pairing it with a newly-built empty PartitionSpec, so every record — regardless of its actual partition column values — was written under a single dummy partition (e.g., year=1970).
Now the sink reads the real partition spec from table.metadata().default_partition_spec(). For unpartitioned tables, the writer is built with None partition key (unchanged write path, zero overhead). For partitioned tables, a RecordBatchPartitionSplitter evaluates the table's partition transforms (day(), bucket(), identity(), etc.) against each Arrow RecordBatch, and a FanoutWriter fans records out to per-partition DataFileWriter instances. primitive_type_to_literal and get_partition_type_value were removed as dead code.
Local Execution
Passed: tested with a local iggy-connect runtime writing to a partitioned Iceberg table (REST catalog)
Verified that records with different partition values land in separate Parquet files under correct partition paths
Verified unpartitioned tables still work with no behavioral change
Pre-commit hooks ran
AI Usage
GitHub Copilot (VS Code)
API exploration of the iceberg 0.9.1 crate, code review of the fix, and generating this description
Compiled with cargo check -p iggy_connector_iceberg_sink, then built via Docker and tested against a live local Iceberg catalog
Yes