Add file-per-partition Iceberg scan#2
Open
toutane wants to merge 27 commits intomain-upstream-dd-17-03-2026-plus-display-limitfrom
Open
Add file-per-partition Iceberg scan#2toutane wants to merge 27 commits intomain-upstream-dd-17-03-2026-plus-display-limitfrom
toutane wants to merge 27 commits intomain-upstream-dd-17-03-2026-plus-display-limitfrom
Conversation
toutane
force-pushed
the
charlesantoine.leger/partitioned-scan
branch
from
91d1c62 to
f6e1a21
Compare
toutane
changed the title
toutane
changed the title
toutane
force-pushed
the
charlesantoine.leger/partitioned-scan
branch
3 times, most recently
from
8e2254f to
61968df
Compare
notfilippo
approved these changes
Mar 26, 2026
Member
notfilippo
left a comment
notfilippo
left a comment
There was a problem hiding this comment.
Looks great, just some questions and nitpicks!
Required by external codecs (e.g. dd-datafusion-iceberg) that need to construct IcebergPartitionedScan from deserialized scan tasks and FileIO, and to read the FileIO config for serialization.
…can.rs Co-authored-by: Filippo <[email protected]>
toutane
force-pushed
the
charlesantoine.leger/partitioned-scan
branch
from
25e7166 to
c332334
Compare
gabotechs
pushed a commit
that referenced
this pull request
Apr 16, 2026
…chTransformer (apache#1821) ## Which issue does this PR close? Partially address apache#1749. ## What changes are included in this PR? This PR adds partition spec handling to `FileScanTask` and `RecordBatchTransformer` to correctly implement the Iceberg spec's "Column Projection" rules for fields "not present" in data files. ### Problem Statement Prior to this PR, `iceberg-rust`'s `FileScanTask` had no mechanism to pass partition information to `RecordBatchTransformer`, causing two issues: 1. **Incorrect handling of bucket partitioning**: Couldn't distinguish identity transforms (which should use partition metadata constants) from non-identity transforms like bucket/truncate/year/month (which must read from data file). For example, `bucket(4, id)` stores `id_bucket = 2` (bucket number) in partition metadata, but actual `id` values (100, 200, 300) are only in the data file. iceberg-rust was incorrectly treating bucket-partitioned source columns as constants, breaking runtime filtering and returning incorrect query results. 2. **Field ID conflicts in add_files scenarios**: When importing Hive tables via `add_files`, partition columns could have field IDs conflicting with Parquet data columns. Example: Parquet has field_id=1→"name", but Iceberg expects field_id=1→"id" (partition). Per spec, the correct field is "not present" and requires name mapping fallback. ### Iceberg Specification Requirements Per the Iceberg spec (https://iceberg.apache.org/spec/#column-projection), when a field ID is "not present" in a data file, it must be resolved using these rules: 1. Return the value from partition metadata if an **Identity Transform** exists 2. Use `schema.name-mapping.default` metadata to map field id to columns without field id 3. Return the default value if it has a defined `initial-default` 4. Return null in all other cases **Why this matters:** - **Identity transforms** (e.g., `identity(dept)`) store actual column values in partition metadata that can be used as constants without reading the data file - **Non-identity transforms** (e.g., `bucket(4, id)`, `day(timestamp)`) store transformed values in partition metadata (e.g., bucket number 2, not the actual `id` values 100, 200, 300) and must read source columns from the data file ### Changes Made 1. **Added partition fields to `FileScanTask`** (`scan/task.rs`): - `partition: Option<Struct>` - Partition data from manifest entry - `partition_spec: Option<Arc<PartitionSpec>>` - For transform-aware constant detection - `name_mapping: Option<Arc<NameMapping>>` - Name mapping from table metadata 2. **Implemented `constants_map()` function** (`arrow/record_batch_transformer.rs`): - Replicates Java's `PartitionUtil.constantsMap()` behavior - Only includes fields where transform is `Transform::Identity` - Used to determine which fields use partition metadata constants vs. reading from data files 3. **Enhanced `RecordBatchTransformer`** (`arrow/record_batch_transformer.rs`): - Added `build_with_partition_data()` method to accept partition spec, partition data, and name mapping - Implements all 4 spec rules for column resolution with identity-transform awareness - Detects field ID conflicts by verifying both field ID AND name match - Falls back to name mapping when field IDs are missing/conflicting (spec rule #2) 4. **Updated `ArrowReader`** (`arrow/reader.rs`): - Uses `build_with_partition_data()` when partition information is available - Falls back to `build()` when not available 5. **Updated manifest entry processing** (`scan/context.rs`): - Populates partition fields in `FileScanTask` from manifest entry data ### Tests Added 1. **`bucket_partitioning_reads_source_column_from_file`** - Verifies that bucket-partitioned source columns are read from data files (not treated as constants from partition metadata) 2. **`identity_partition_uses_constant_from_metadata`** - Verifies that identity-transformed fields correctly use partition metadata constants 3. **`test_bucket_partitioning_with_renamed_source_column`** - Verifies field-ID-based mapping works despite column rename 4. **`add_files_partition_columns_without_field_ids`** - Verifies name mapping resolution for Hive table imports without field IDs (spec rule #2) 5. **`add_files_with_true_field_id_conflict`** - Verifies correct field ID conflict detection with name mapping fallback (spec rule #2) 6. **`test_all_four_spec_rules`** - Integration test verifying all 4 spec rules work together ## Are these changes tested? Yes, there are 6 new unit tests covering all 4 Iceberg spec rules. This also resolved approximately 50 Iceberg Java tests when running with DataFusion Comet's experimental apache/datafusion-comet#2528 PR. --------- Co-authored-by: Renjie Liu <[email protected]>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
QECO-981
Summary
Adds two new public types to
iceberg-datafusion:IcebergPartitionedScan: a DataFusionExecutionPlanwhere eachFileScanTaskmaps to exactly one partition, enabling DataFusion to dispatch file reads in parallel.IcebergPartitionedTableProvider: a catalog-backedTableProviderthat builds anIcebergPartitionedScanon every query, always fetching the latest snapshot.Design choices
One file = one partition
IcebergTableScanuses UnknownPartitioning(1) and streams all files sequentially through a single partition.IcebergPartitionedScanuses UnknownPartitioning(n_files), giving DataFusion the information it needs to schedule execute(i) calls concurrently, one per file.Table reloaded on every scan
IcebergPartitionedTableProviderloads the table twice: once at construction to snapshot the Arrow schema for DataFusion planning, and once at scan time to guarantee the freshest snapshot. This mirrors the behavior ofIcebergTableProviderand ensures scans always reflect the current table state at the cost of an extra catalog round-trip per query.Known limitation: schema staleness on projection
Projection indices are resolved against the schema captured at construction time. If the table schema evolves between construction and scan, projected column names may be incorrect. This is inherited behavior from
IcebergTableProviderand is commented, this is inherent to the DataFusion's model.No stored projection/predicate fields
The struct is intentionally self-contained: its full state is (tasks, file_io, schema). All three are serializable, which keeps the distributed protobuf codec simple, no extra fields to encode separately in the round-trip. Display information is derived at runtime from the schema and tasks.
No limit pushdown
DataFusion inserts a
GlobalLimitExecabove any leaf that does not implement pushdown, so correctness is maintained.No writes
insert_into returns FeatureUnsupported. Use
IcebergTableProviderfor write operations.Tests