EPIC: Support parallel scan in iceberg-datafusion

[ZENOTME]

What's the feature are you trying to implement?

As @colinmarc mention in https://apache-iceberg.slack.com/archives/C05HTENMJG4/p1753476472857519 , performance of iceberg-datafusion integration is lower than pure datafusion reading parquet file directly. As mention by @liurenjie1024 , it's caused by iceberg-datafusion integration only use one thread now. This issue propose a design to support parallel scan in iceberg-datafusion integration. And thanks to https://github.com/colinmarc/iceberg-datafusion-benchmarks from @colinmarc let us can dive into the bottleneck!

Row group based parallel scan

This parallel scan is row group based. The basic idea is to prune the file need scan into several group and pack them based on the parallism set by datafusion. The benefit of row group base parallel:

1. Parallel scan even in less file scene
2. More even distribute the read load after prune some row group in the file, e.g.

   iceberg-rust/crates/iceberg/src/arrow/reader.rs

   Line 241 in d9fbc5c

   let mut selected_row_group_indices = None;

The process can be describe as following:

• 1 is our plan_file now, prune the iceberg metadata and get the FileScanTask finally. It's one data file bind with several delete file related to it.
• 2 We introduce a GroupPruner, which can attach the group info at FileScanTask and prune the some group based on predicate. We can implement group pruner for specific file format, e.g. ParquetGroupPruner to process different format of FileScanTask.
• 3 FileScanTask with group info can be split into multilple FileScanTask or merge. Base on the parallesim, we can repartition the FileScanTask into multiple PartitionTask which contain several FileScanTask.

Each partition task return as a RecordBatchStream when we execute using corresponding partition.

Willingness to contribute

Yes

[ZENOTME]

I draft the implementation of design above #1610, test on local machine(mac m1 pro). Although the latency is high and not stable but we can see that the performance of iceberg-rust is much closer to datafusion-parquet. I will build a local test env to bench it next step.

====== full scan tpch_10.customer (1500000 rows, 1 files) =======
> before parallel scan
scan/iceberg-rust/tpch_10.customer
                        time:   [10.834 s 11.197 s 11.593 s]
                        thrpt:  [129.38 Kelem/s 133.97 Kelem/s 138.46 Kelem/s]
scan/datafusion-parquet/tpch_10.customer
                        time:   [4.9707 s 5.1641 s 5.3602 s]
                        thrpt:  [279.84 Kelem/s 290.46 Kelem/s 301.77 Kelem/s]

> after parallel scan
scan/iceberg-rust/tpch_10.customer
                        time:   [6.4354 s 6.8607 s 7.3866 s]
                        thrpt:  [203.07 Kelem/s 218.64 Kelem/s 233.09 Kelem/s]
scan/datafusion-parquet/tpch_10.customer
                        time:   [5.1607 s 5.5025 s 5.9684 s]
                        thrpt:  [251.33 Kelem/s 272.61 Kelem/s 290.66 Kelem/s]

====== full scan tpch_30.customer (4500000 rows, 3 files) =======
> before parallel scan
scan/iceberg-rust/tpch_30.customer
                        time:   [15.670 s 18.931 s 24.221 s]
                        thrpt:  [185.79 Kelem/s 237.71 Kelem/s 287.18 Kelem/s]
scan/datafusion-parquet/tpch_30.customer
                        time:   [7.8278 s 8.3205 s 9.0664 s]
                        thrpt:  [496.34 Kelem/s 540.83 Kelem/s 574.88 Kelem/s]

> after parallel scan
scan/iceberg-rust/tpch_30.customer
                        time:   [8.2926 s 10.530 s 14.204 s]
                        thrpt:  [316.80 Kelem/s 427.37 Kelem/s 542.65 Kelem/s]
scan/datafusion-parquet/tpch_30.customer
                        time:   [7.9077 s 8.2183 s 8.5933 s]
                        thrpt:  [523.67 Kelem/s 547.56 Kelem/s 569.07 Kelem/s]

====== full scan tpch_50.customer (7500000 rows, 5 files) =======
> before parallel scan
scan/iceberg-rust/tpch_50.customer
                        time:   [15.612 s 19.579 s 26.554 s]
                        thrpt:  [282.44 Kelem/s 383.06 Kelem/s 480.39 Kelem/s]
scan/datafusion-parquet/tpch_50.customer
                        time:   [12.567 s 14.950 s 18.726 s]
                        thrpt:  [400.50 Kelem/s 501.67 Kelem/s 596.80 Kelem/s]

> after parallel scan
scan/iceberg-rust/tpch_50.customer
                        time:   [12.020 s 13.179 s 14.382 s]
                        thrpt:  [521.48 Kelem/s 569.10 Kelem/s 623.94 Kelem/s]
scan/datafusion-parquet/tpch_50.customer
                        time:   [12.035 s 21.058 s 38.281 s]
                        thrpt:  [195.92 Kelem/s 356.16 Kelem/s 623.18 Kelem/s]

[ZENOTME]

Feel free for any suggestion for this cc @liurenjie1024 @sdd @Xuanwo @Fokko @kevinjqliu

[liurenjie1024]

Thanks @ZENOTME for raising this, but I have concern to do the filter in task planning phase, as opening data files is typically slow. There are several places to improve:

1. Size base planning, see Plan file scan task according scan file size. #128
2. Current data fusion table scan has only one partition:

   iceberg-rust/crates/integrations/datafusion/src/physical_plan/scan.rs

   Line 105 in 6e21b7f

   Partitioning::UnknownPartitioning(1),

3. Filter push down not working:

   iceberg-rust/crates/integrations/datafusion/src/table/mod.rs

   Line 152 in 36cc120

   fn supports_filters_pushdown(

As with row group pruning, it should be pushed to parquet reader.

[ZENOTME]

Thanks feedback from @liurenjie1024! Here are my consideration: Datafusion is static partition at plan phase, so comparing size base planning, row group plan may achieve a more balanced partitioning at data skew scene.

but I have concern to do the filter in task planning phase, as opening data files is typically slow

I think the overall execution time remains the same—we’re essentially moving the group pruning logic from the reader stage to the planning stage. If metadata is retrieved during planning, it can be reused by the reader to avoid fetching it again (though we need to weigh the memory overhead).

I think this could be an optional behavior, allowing users to make the tradeoff: either spend more time during planning and to achieve a more balanced partitioning, or use a size-based plan for faster planning. The best choice will depend on the actual data distribution.

I also agree we don’t need to introduce too many optimizations upfront. Still, I’m curious whether this is a realistic need in practice—if so, it may be worth making some room for it in the design so that we can extend in this direction later if necessary.

[liurenjie1024]

Here are my consideration: Datafusion is static partition at plan phase, so comparing size base planning, row group plan may achieve a more balanced partitioning at data skew scene.

Please remember that this library is not designed for datafusion only, but for all compute engines. Pruning row group at planning phase helps a little for data skew, but not that much in most cases. Remember that we have more filters such as page level filtering, deletion vector, eq deletes to further remove records. There are many problems with row group pruning at planning phase:

1. Planning phase typically happens master/driver node in a distributed compute engine. So the cost of reading manifest only vs reading parquet files is different. Number of data files is much more than manifest files.
2. The cost of opening data files can't be neglected, not even to mention parse footer and run the filter logic. Mainfest files typically are cached in compute engine, while data files are not.
3. Even if some advanced compute engine could do distributed planning, opening data file twice costs much than necessary.
4. It makes the planning phase format specific.

Compared with the problems it brings, I don't see much benefit it brings. There are many ways to deal with data skew, and this approach doesn't help that much.

[Fokko]

@ZENOTME Are you benchmarking against a local FS? I think one of the important objectives is to be efficient against Object-Stores, something that isn't taken into consideration with a local FS.

I share @liurenjie1024's concern of opening up the Parquet file in the driver/kernel, since the overhead of opening a file is significant (miliseconds until first byte).

Planning phase typically happens master/driver node in a distributed compute engine.

That's true, but not always the case. For exameple, Spark leverages distributed planning. Each manifest file has a target size of 8MB, which could be dispatched for distributed planning.

I would flip it around, what's the problem if an executor opens up a Parquet file and reads very little data. This is unlikely since we have the min/max of the file, but the row-group could still be skipped if it doesn't hit the min/max there. Parsing the Parquet footer is also pretty slow, especially for wide schema's, so I would be reluctant to send that over the wire.

[ZENOTME]

Thanks for the suggestions from @Fokko and @liurenjie1024, I learned a lot!

That's true, but not always the case. For exameple, Spark leverages distributed planning. Each manifest file has a target size of 8MB, which could be dispatched for distributed planning.

It's close to my initial thought for the distributed case. Wee can push the row group pruning down to the executors, and at the same time cache the metadata they read there, so we don’t have to pay the cost of opening the data file twice. But number of data files is much more than manifest files so the cache size need to be consideration.

This does make things more complex and the benefit is still uncertain. Let’s stick to the size-based planning for now. I'd like to implement it if no one implement it now and bench it again to see what happen.

[ZENOTME]

Let’s stick to the size-based planning for now. #128