[comphead]

Which issue does this PR close?

Closes #7422.

Rationale for this change

On Spark 3.4, DataSourceV2ScanExecBase.outputPartitioning reports SinglePartition for a scan of one input partition. Spark 3.5 dropped that case. In the SPARK-34637 test, the two files that pass the units_sold = 70 filter form one split. Spark therefore plans the final aggregate of the DPP build side directly over the partial one, with no exchange between them. Comet then fuses both aggregates, the project, the filter and the native Iceberg scan into one native block. The scan reports UnknownPartitioning(0) (#6268).

AQE builds the DPP subquery's AdaptiveSparkPlanExec over that converted build side and runs EnsureRequirements on it again. Since #7231, the final CometHashAggregate reports Spark's ClusteredDistribution(store_id), while the partial aggregate below passes on the scan's UnknownPartitioning(0). So EnsureRequirements inserts an exchange inside the native block, which is already serialized with the scan inlined. findAllPlanData no longer reaches the scan through the new exchange, and the native IcebergScan gets no schema or tasks: Column index 1 is out of bound. This is the Spark 3.4 [scans] failure in the nightly run since #7231 landed.

What changes are included in this PR?

CometIcebergNativeScanExec.outputPartitioning reports SinglePartition when the BatchScanExec it replaces does, and UnknownPartitioning(0) otherwise, as before. Running EnsureRequirements again then finds the requirement satisfied, as Spark did.

• Spark counts that one partition from planInputPartitions(), before runtime filtering, so reading it resolves no DPP subquery, which is the problem fix: do not resolve DPP subqueries when AQE reads a native scan's partitioning #6268 avoided.
• When DPP prunes the only split, serializePartitions already keeps Spark's one empty partition, so the native block still runs exactly one partition.
• On Spark 3.5 and later, BatchScanExec never reports SinglePartition, so nothing changes there.

How are these changes tested?

CometIcebergNativeSuite "AQE DPP - V2 broadcast query stage creation order (SPARK-34637)" reproduces the failure on Spark 3.4 and covers the fix. This PR removes the Spark 3.4 skip that #7423 added to it. Spark 3.4 runs only through run-all-spark-profiles, and the Iceberg suites through run-iceberg-tests. The Spark 3.4 profile does not build offline in my environment, so I have not run it locally yet. On Spark 4.1, the suite's 25 DPP tests pass.

[andygrove]

Thanks @comphead. This is a tidy fix for the Spark 3.4 [scans] nightly failure.

I read DataSourceV2ScanExecBase and BatchScanExec at v3.4.3, v3.5.8 and v4.1.1. On 3.4, outputPartitioning is SinglePartition when partitions.length == 1, and partitions comes from the unfiltered inputPartitions. So reading it resolves no DPP subquery and the #6268 concern does not return. On 3.5 and later BatchScanExec never reports SinglePartition, so the new branch always falls through to UnknownPartitioning(0). The case where DPP prunes the only split is also covered, because serializePartitions keeps Spark's one empty partition in that case.

The originalPlan != null guard handles canonicalized instances. The SPARK-34637 test reproduces the failure and passes in CI with the Spark 3.4 profile. Looks good to me.

[comphead]

Thanks @andygrove for the review. 5fe99d2 drops the Spark 3.4 skip that #7423 added to the SPARK-34637 test, so the test exercises this fix on 3.4 again. The CometSubqueryBroadcastExec check stays limited to 3.5+, since native AQE DPP needs Spark 3.5.

I added run-iceberg-tests as well. The Iceberg suites have a Spark 3.4 target, the only version where the new branch applies, and they have not run on this PR yet.