feat: serialize Iceberg data source plans - #672
Closed
alexanderbianchi wants to merge 11 commits into
Closed
Conversation
## Summary Closes datafusion-contrib#530 - This PR updates the upstream datafusion SHA to the HEAD of https://github.com/apache/datafusion/commits/branch-55/ (edit: this branch is continuously being updated. I will make sure this PR is at the head before merging) - Rust upgade to 1.94 ## Changes 1. In `src/protobuf/distributed_codec.rs` we now use the `proto_converter` argument during serde - We still don't use the `DeduplicatingProtoConverter`, so dynamic filters don't necessarily work. I think this is outside the scope of this PR will be addressed in datafusion-contrib#623, which will be rebased after the upgrade. 3. `ExecutionPlan::apply_expressions` is added for every custom `ExecutionPlan` in this repo - Wrapper types (`MetricsWrapperExec`, `WorkUnitFileScanConfig`, `DistributedLeafExec`) delegate to the inner type - Other plans take`TreeNodeRecursion::Continue` because they have no expressions (ex. `SamplerExec`) - Note that `apply_expressions` does not need to yield sort or partitioning expressions in the plan properties 3. We migrate from `partition_statistics` to `statistics_from_inputs` for every `ExecutionPlan`. - `src/distributed_planner/statistics/plan_statistics.rs` can just use `statistics_from_inputs` directly instead of doing the `StatisticsWrapper` workaround. 5. Range partitioning is now supported. - CPU costing now includes range-key comparison cost and has a new unit test. See src/distributed_planner/ statistics/complexity_cpu.rs:238. - I think there's open questions about range partitioning. I've opened an issue here to make sure it behaves as expected after the upgrade: datafusion-contrib#628 (comment) 6. Peak-memory metrics use the existing gauge wire representation. DataFusion added MetricValue::PeakMemoryUsage. It is serialized as the existing named-gauge protobuf variant to avoid a wire-format change. See src/protocol/grpc/ metrics_proto.rs:124. The value and name survive, and aggregation is still additive, but decoding produces a generic Gauge, not PeakMemoryUsage. The practical difference is mainly display formatting: it may render as a count rather than human-readable bytes. This is the clearest remaining compromise/risk in the upgrade. 7. File-scan rebalancing changed its discriminator. DataFusion removed partitioned_by_file_group; output_partitioning.is_some() is now the source of truth. See src/events/defaults/file_scan_config.rs:43. This decides whether files are round-robin rebalanced or split through FileGroupPartitioner, so it is behavior-sensitive even though it is a one-line migration. 8. Two previously ignored correctness tests were enabled. - See `tests/multi_task_collect_join_repros.rs` - These were upstream DataFusion correctness fixes, not fixes made locally in this upgrade. 9. drop(reporter) was made explicit on the sampler’s empty-input path. The reporter sends its result on Drop; explicitly dropping it both satisfies the new compiler/lint behavior and guarantees the zero-row EOS report is sent before returning. See src/ execution_plans/sampler.rs:259. 10. Plan changes - `dynamic_rg_pruning=eligible` is now displayed on eligible scans: 1,354 occurrences in TPC-DS, 188 in TPC-H, and 12 in ClickBench - `DataSourceExec` now displays its output partitioning. See `tests/join.rs` (eventually, someone should delete this test datafusion-contrib#628) - Project after sort. This looks like some upstream optimizer rule change ex. `tests/distributed_unions.rs` and `tests/distributed_aggregation.rs`. ``` - │ SortExec: expr=[MinTemp@0 ASC NULLS LAST, RainToday@1 ASC NULLS LAST], preserve_partitioning=[true] - │ ProjectionExec: expr=[MaxTemp@0 as MinTemp, RainToday@1 as RainToday] + │ ProjectionExec: expr=[MaxTemp@0 as MinTemp, RainToday@1 as RainToday] + │ SortExec: expr=[MaxTemp@0 ASC NULLS LAST, RainToday@1 ASC NULLS LAST], preserve_partitioning=[true] ``` - LocalLimitExec became more common: TPC-DS went from 0 to 20 occurrences and ClickBench from 1 to 21, reflecting additional local limit pushdown. - Subquery/semi-join plans became more distributed: - TPC-DS CollectLeft hash joins: 615 → 610 - TPC-DS partitioned hash joins: 98 → 103 - TPC-DS left-semi occurrences: 11 → 25 - TPC-DS network shuffles: 368 → 378 - TPC-H - just a few - These are meaningful topology changes: some subqueries now use partitioned left-semi joins and therefore introduce hash shuffles instead of collecting/broadcasting one side. - Scalar rendering improved, especially decimal literals: internal forms such as Some(0),7,2 now display as CAST(0.00 AS Decimal128(7, 2)). - Minor changes (Ex. tpcds 21) - `__common_expr_4` became `__common_expr_3`; that is only an internal alias renumbering. - The projection that renamed `d_date` to `__common_expr_2` disappeared. - `d_date` is retained directly in the join output and referenced directly by partial/final aggregates. - Column positions changed - File-group allocation changed substantially - Some explicit RoundRobinBatch repartitions disappeared and scans gained different numbers of file groups - Distribute byte ranges across partitions: apache/datafusion#22439 - Lowers `repartition_file_min_size` from 10 MiB to 1 MiB. The PR explicitly calls out TPC-DS SF1 dimension tables. Files may be duplicated across multiple partitions where but each partition reads a different byte range (this is hidden by <int>....<int>, but we know from the correctness tests that nothing broke). A lot of tpcds queries now split across `target_partitions` instead of staying under-partitioned. In the `tpcds` plan tests, we use `target_partitions=3`. Example: ``` - │ t0: DataSourceExec: file_groups={2 groups: [[/testdata/tpcds/plans_sf1_partitions4/date_dim/part-0.parquet:<int>..<int>], [/testdata/tpcds/plans_sf1_partitions4/date_dim/part-1.parquet:<int>..<int>, /testdata/tpcds/plans_sf1_partitions4/date_dim/part-2.parquet:<int>..<int>]]}, projection=[d_date_sk, d_week_seq, d_day_name], file_type=parquet, predicate=DynamicFilter [ empty ] - │ t1: DataSourceExec: file_groups={2 groups: [[/testdata/tpcds/plans_sf1_partitions4/date_dim/part-0.parquet:<int>..<int>], [/testdata/tpcds/plans_sf1_partitions4/date_dim/part-2.parquet:<int>..<int>]]}, projection=[d_date_sk, d_week_seq, d_day_name], file_type=parquet, predicate=DynamicFilter [ empty ] - │ t2: DataSourceExec: file_groups={2 groups: [[/testdata/tpcds/plans_sf1_partitions4/date_dim/part-0.parquet:<int>..<int>, /testdata/tpcds/plans_sf1_partitions4/date_dim/part-1.parquet:<int>..<int>], [/testdata/tpcds/plans_sf1_partitions4/date_dim/part-2.parquet:<int>..<int>, /testdata/tpcds/plans_sf1_partitions4/date_dim/part-3.parquet:<int>..<int>]]}, projection=[d_date_sk, d_week_seq, d_day_name], file_type=parquet, predicate=DynamicFilter [ empty ] - │ t3: DataSourceExec: file_groups={2 groups: [[/testdata/tpcds/plans_sf1_partitions4/date_dim/part-1.parquet:<int>..<int>], [/testdata/tpcds/plans_sf1_partitions4/date_dim/part-3.parquet:<int>..<int>]]}, projection=[d_date_sk, d_week_seq, d_day_name], file_type=parquet, predicate=DynamicFilter [ empty ] + │ t0: DataSourceExec: file_groups={3 groups: [[/testdata/tpcds/plans_sf1_partitions4/date_dim/part-0.parquet:<int>..<int>], [/testdata/tpcds/plans_sf1_partitions4/date_dim/part-1.parquet:<int>..<int>], [/testdata/tpcds/plans_sf1_partitions4/date_dim/part-2.parquet:<int>..<int>]]}, projection=[d_date_sk, d_week_seq, d_day_name], file_type=parquet, predicate=DynamicFilter [ empty ] + │ t1: DataSourceExec: file_groups={3 groups: [[/testdata/tpcds/plans_sf1_partitions4/date_dim/part-0.parquet:<int>..<int>], [/testdata/tpcds/plans_sf1_partitions4/date_dim/part-1.parquet:<int>..<int>], [/testdata/tpcds/plans_sf1_partitions4/date_dim/part-2.parquet:<int>..<int>, /testdata/tpcds/plans_sf1_partitions4/date_dim/part-3.parquet:<int>..<int>]]}, projection=[d_date_sk, d_week_seq, d_day_name], file_type=parquet, predicate=DynamicFilter [ empty ] + │ t2: DataSourceExec: file_groups={3 groups: [[/testdata/tpcds/plans_sf1_partitions4/date_dim/part-0.parquet:<int>..<int>], [/testdata/tpcds/plans_sf1_partitions4/date_dim/part-1.parquet:<int>..<int>, /testdata/tpcds/plans_sf1_partitions4/date_dim/part-2.parquet:<int>..<int>], [/testdata/tpcds/plans_sf1_partitions4/date_dim/part-3.parquet:<int>..<int>]]}, projection=[d_date_sk, d_week_seq, d_day_name], file_type=parquet, predicate=DynamicFilter [ empty ] + │ t3: DataSourceExec: file_groups={3 groups: [[/testdata/tpcds/plans_sf1_partitions4/date_dim/part-0.parquet:<int>..<int>, /testdata/tpcds/plans_sf1_partitions4/date_dim/part-1.parquet:<int>..<int>, /testdata/tpcds/plans_sf1_partitions4/date_dim/part-1.parquet:<int>..<int>], [/testdata/tpcds/plans_sf1_partitions4/date_dim/part-2.parquet:<int>..<int>, /testdata/tpcds/plans_sf1_partitions4/date_dim/part-2.parquet:<int>..<int>], [/testdata/tpcds/plans_sf1_partitions4/date_dim/part-3.parquet:<int>..<int>]]}, projection=[d_date_sk, d_week_seq, d_day_name], file_type=parquet, predicate=DynamicFilter [ empty ] ``` --------- Co-authored-by: Gabriel <45515538+gabotechs@users.noreply.github.com> Co-authored-by: Gabriel <gabriel.musatmestre@datadoghq.com>
…ib#661) ## Summary - When pushing a fetch into a `NetworkCoalesceExec` producer stage, walk through `ProjectionExec` (it does not change row count) and apply the bound to the child. - Skip inserting a redundant `LocalLimitExec` when that child already has a sufficient fetch, for example `SortExec: TopK`. - `GlobalLimitExec` still pushes `skip + fetch`. Unbounded producers such as ClickBench q17 still get a per-task `LocalLimitExec`. Closes datafusion-contrib#659.
Successor of datafusion-contrib#624 but rebased with `main` to benefit from the DataFusion 55 upgrade.
## Summary Adds a deterministic `tpch/sorted_sf1` benchmark dataset variant so physical data ordering is a controlled dimension while keeping the standard TPC-H query workload. Part of datafusion-contrib#629, closes datafusion-contrib#632. Uses the same `tpchgen` generators and Arrow writer as unsorted TPC-H, recording table-specific Parquet `sorting_columns` metadata in one pass (`prepare-tpch --sorted`). ## Sort keys | Table | Sort columns | | --- | --- | | region | `r_regionkey` | | nation | `n_nationkey` | | customer | `c_custkey` | | supplier | `s_suppkey` | | part | `p_partkey` | | partsupp | `ps_partkey` | | orders | `o_orderkey` | | lineitem | `l_orderkey`, `l_linenumber` | `partsupp` records only `ps_partkey` because `ps_suppkey` wraps within each part. Data is written to `testdata/tpch/sorted_sf<scale-factor>/`. ```bash # Default SF1, 16 files per table ./benchmarks/gen-tpch-sorted.sh # Tiny smoke-scale dataset SCALE_FACTOR=0.01 ./benchmarks/gen-tpch-sorted.sh WORKERS=8 ./benchmarks/run.sh --threads 2 --dataset tpch/sorted_sf1 ``` Queries stay in `testdata/tpch/queries`. `run_local` extracts the `tpch` suite from `tpch/sorted_sf1`, so `queries_for_dataset` is unchanged. `register_tables` is unchanged. ## Implementation - `benchmarks/src/datasets/tpch.rs`: optional `sorting_columns` on the existing writer - `prepare-tpch --sorted` instead of a separate async rewrite command - `benchmarks/gen-tpch-sorted.sh` writes `testdata/tpch/sorted_sf<scale-factor>` SF1 was not generated in this PR (heavy, and blocked here by the OpenSSL build). Use `SCALE_FACTOR=0.01` to smoke the path, then `./benchmarks/gen-tpch-sorted.sh` for SF1.
gabotechs
force-pushed
the
gabrielmusat/add-basic-iceberg-integration
branch
from
August 24, 2026 20:02
61c91e2 to
1408dd7
Compare
alexanderbianchi
force-pushed
the
iceberg/issue-603
branch
4 times, most recently
from
August 24, 2026 21:37
3da0a24 to
ecaa32e
Compare
alexanderbianchi
force-pushed
the
iceberg/issue-603
branch
from
August 24, 2026 23:48
ecaa32e to
751e7bd
Compare
gabotechs
force-pushed
the
gabrielmusat/add-basic-iceberg-integration
branch
5 times, most recently
from
August 25, 2026 06:31
6a5767e to
a6cb6af
Compare
gabotechs
deleted the branch
datafusion-contrib:gabrielmusat/add-basic-iceberg-integration
August 25, 2026 08:41
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
4 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.
Do not review. Closes #603.
Summary
FileIOandIcebergDataSourcewith worker-local storage/runtime configurationStack
Validation
cargo check -p datafusion-distributed-icebergcurrently reaches an inherited #595/DataFusion 55 compile error aticeberg/src/table_provider/factory.rs(datafusion::sql::TableReferencemoved).