feat: Per-Partition Plan Building for Native Iceberg Scans [iceberg-rust]

[andygrove]

This PR ensures that when Iceberg scans are executed natively, each Spark task's native plan only contains file scan tasks for that specific partition, rather than all partitions.

Problem

Previously, when CometIcebergNativeScanExec was a child of another operator (e.g., CometFilterExec), the parent operator's serialized native plan (nativeOp) contained the entire IcebergScan message with data for all partitions. This meant:

• Each task received metadata for all partitions when it only needed one
• The serialized plan size grew linearly with partition count
• Every task deserialized unnecessary partition data

Solution

This PR implements per-partition plan building that injects only the relevant partition's data at execution time:

New Components

CometIcebergSplitRDD - A custom RDD that:

• Holds common data (deduplication pools, catalog properties) in the closure
• Stores each partition's file scan tasks in its Partition objects
• Combines common + partition data at compute time to build partition-specific native plans

IcebergPartitionInjector - A helper that traverses an Operator tree and injects partition data into IcebergScan nodes that are missing it

findIcebergSplitData() - Locates CometIcebergNativeScanExec descendants in the plan tree and retrieves their per-partition data

Modified Execution Flow

In CometNativeExec.doExecuteColumnar():

1. Check if the plan tree contains an IcebergScan with per-partition data available
2. If so, for each partition:
   • Parse the base operator tree from the serialized plan
   • Inject that partition's file scan task data into the IcebergScan node
   • Re-serialize and pass to native execution
3. Each task's native plan now only contains its own partition's data

Protobuf Changes

Added IcebergScanCommon message to hold shared data (pools, metadata) separately from per-partition file scan tasks. The IcebergScan message now has:

• common field for shared deduplication pools
• partition field for a single partition's file tasks

Rust Changes

Simplified the Iceberg scan handling in planner.rs to expect common + partition data, removing the code path that read from a list of all partitions.

Test Plan

• All 62 existing Iceberg tests pass
• Filter pushdown tests verify parent operators work correctly
• MOR (Merge-On-Read) tests with positional and equality deletes
• Schema evolution, complex types, partition pruning tests
• REST catalog integration test

[codecov-commenter]

Codecov Report

❌ Patch coverage is 89.17197% with 17 lines in your changes missing coverage. Please review.
✅ Project coverage is 60.19%. Comparing base (f09f8af) to head (0a02bb7).
⚠️ Report is 902 commits behind head on main.

Files with missing lines	Patch %	Lines
...n/scala/org/apache/spark/sql/comet/operators.scala	79.24%	7 Missing and 4 partials ⚠️
.../comet/serde/operator/CometIcebergNativeScan.scala	88.88%	4 Missing ⚠️
...e/spark/sql/comet/CometIcebergNativeScanExec.scala	96.15%	0 Missing and 1 partial ⚠️
.../apache/spark/sql/comet/CometIcebergSplitRDD.scala	97.61%	0 Missing and 1 partial ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##               main    #3295      +/-   ##
============================================
+ Coverage     56.12%   60.19%   +4.07%     
- Complexity      976     1470     +494     
============================================
  Files           119      176      +57     
  Lines         11743    16301    +4558     
  Branches       2251     2698     +447     
============================================
+ Hits           6591     9813    +3222     
- Misses         4012     5136    +1124     
- Partials       1140     1352     +212     

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[andygrove]

@hsiang-c fyi

[mbutrovich]

Closing in favor of #3349.