CometNativeScanExec evaluates the inner (unrewritten) dynamic-pruning subquery from outputPartitioning: q5 fails with SubqueryAdaptiveBroadcastExec.execute(), q64 intermittently returns 0 rows at SF1000

[Neuw84]

Describe the bug

TPC-DS q64 returns 0 rows at scale factor 1000 with Comet native execution, where Spark returns 12,185. The same CometNativeScan feeding Spark's own operators (every spark.comet.exec.<operator>.enabled=false, so the scan output goes through ColumnarToRow) returns the correct 12,185 rows, and at SF10 every configuration agrees with Spark. So the scan's values are right when read through the ColumnVector accessors, and the rows are lost only when the scan's batches are consumed by native operators.

Where the rows go: with operators disabled one at a time, the loss sits in the join of the second cs_ui instance of the cross_sales CTE against the broadcast of store_sales (ss_sold_date_sk in 2000) ⋈ store_returns; before that join the intermediate row counts agree with Spark. Turning AQE off does not change the result. An unrelated third-party columnar operator set that consumes Comet's scan output through the Arrow C data interface loses the rows in exactly the same place, which suggests the difference is in the exported batches rather than in a particular operator.

Environment: Comet 1.0.0, Spark 4.1.3 (spark-4.1.3-bin-hadoop3), Java 25 (Corretto), Kubernetes (Spark operator), TPC-DS SF1000 parquet on S3 read through S3A. 8 executors × 13 cores, 18 GB heap + 16 GB overhead, spark.memory.offHeap.size=16g.

configuration	q64 rows
Spark (no Comet)	12,185
Comet scan + Comet native operators + Comet shuffle	0 (at 200 and 300 shuffle partitions)
Comet scan, Spark operators and shuffle (spark.comet.exec.*.enabled=false)	12,185
any of the above at SF10	12,185-equivalent, all agree

Steps to reproduce

1. Generate TPC-DS SF1000 as parquet (we used the tpcds-kit dsdgen, char/varchar columns stored as strings; SF10 does not reproduce).
2. Run q64 (the official TPC-DS text, parameters i_color in ('purple','burlywood','indian','spring','floral','medium'), i_current_price between 64 and 74, cs_ui.syear = 1999 / cs2.syear = 2000) with:

--conf spark.plugins=org.apache.spark.CometPlugin
--conf spark.comet.enabled=true --conf spark.comet.scan.enabled=true
--conf spark.comet.exec.enabled=true --conf spark.comet.exec.shuffle.enabled=true --conf spark.comet.exec.shuffle.mode=auto
--conf spark.shuffle.manager=org.apache.spark.sql.comet.execution.shuffle.CometShuffleManager
--conf spark.comet.explainFallback.enabled=true --conf spark.comet.cast.allowIncompatible=true
--conf spark.memory.offHeap.enabled=true --conf spark.memory.offHeap.size=16g
--conf spark.sql.shuffle.partitions=300

→ 0 rows.

3. Same, adding --conf spark.comet.exec.shuffle.enabled=false and spark.comet.exec.{project,filter,aggregate,sort,localLimit,globalLimit,takeOrderedAndProject,hashJoin,sortMergeJoin,broadcastHashJoin,broadcastExchange,expand,union,window,coalesce,collectLimit,explode,sample}.enabled=false, default shuffle manager → 12,185 rows, the plan still showing CometNativeScan for all 48 scans.

Expected behavior

12,185 rows, as Spark and as Comet's scan under Spark's operators.

Additional context

I can attach explain formatted output of both plans and the per-join intermediate row counts if useful. Related to the much older #74 (q64 among the queries once disabled for result differences), but here the scan alone is fine and the difference appears only at scale.

[Neuw84]

Update: the result is intermittent, not deterministic. Two further full runs today of the same two configurations that returned 0 rows this morning (Comet 1.0.0 end to end, and CometNativeScan under the third-party operators) both returned the correct 12,185 rows for q64 at SF1000, same image, same data, 300 shuffle partitions; a run yesterday of the end-to-end configuration was also correct. So the loss is run-dependent. q64's affected join reads store_sales through a dynamic-partition-pruning filter on ss_sold_date_sk; in the same environment we also see, on the third-party-operator configuration, q5 fail with SubqueryAdaptiveBroadcastExec does not support the execute() code path — a DPP subquery evaluated before AQE rewrote it. A timing-dependent evaluation of the scan's DPP InSubqueryExec (before the rewrite → exception; before the broadcast is ready → an empty set → every partition pruned → 0 rows) would explain both; we are capturing the caller stack of the q5 evaluation now and will add it here.

[Neuw84]

Caller stack captured for the q5 failure (Comet 1.0.0, Spark 4.1.3, CometNativeScan with third-party operators):

SparkUnsupportedOperationException: SubqueryAdaptiveBroadcastExec does not support the execute() code path.
  at org.apache.spark.sql.execution.SubqueryAdaptiveBroadcastExec.doExecute(SubqueryAdaptiveBroadcastExec.scala:44)
  at org.apache.spark.sql.execution.SparkPlan.getByteArrayRdd(SparkPlan.scala:378)
  at org.apache.spark.sql.execution.InSubqueryExec.updateResult(subquery.scala:134)
  at org.apache.spark.sql.comet.CometNativeScanExec.$anonfun$serializedPartitionData$1(CometNativeScanExec.scala:176)
  at org.apache.spark.sql.comet.CometNativeScanExec.serializedPartitionData$lzycompute(CometNativeScanExec.scala:171)
  at org.apache.spark.sql.comet.CometNativeScanExec.perPartitionData(CometNativeScanExec.scala:240)
  at org.apache.spark.sql.comet.CometNativeScanExec.outputPartitioning$lzycompute(CometNativeScanExec.scala:131)
  at org.apache.spark.sql.execution.exchange.ValidateRequirements$.validateInternal(ValidateRequirements.scala:48-50)
  ... (ValidateRequirements recursion, from AdaptiveSparkPlanExec.optimizeQueryStage)

So the mechanism is the one the code comment in serializedPartitionData describes: the inner scan.partitionFilters holds a separate InSubqueryExec instance for the dynamic-pruning filter, which Spark's expression walk does not see — so PlanAdaptiveDynamicPruningFilters rewrites the outer copy but the inner one is still the placeholder SubqueryAdaptiveBroadcastExec when outputPartitioning (reached by ValidateRequirements inside optimizeQueryStage) forces serializedPartitionData, which calls updateResult() on it → execute() → this exception. The same eager, unsynchronised evaluation of the inner copy is the natural explanation for the intermittent q64 zero rows: whatever the inner InSubqueryExec resolves to at that moment is what prunes the file partitions, and it is not the filter AQE ends up executing.

Why it shows with third-party operators and not in the pure Comet configuration is presumably the plan shape ValidateRequirements walks (our operators declare required distributions the walk checks down to the scan); the root is independent of them. A fix on Comet's side would be to make the inner copy follow the outer one (rewrite scan.partitionFilters when the wrapper's partitionFilters are rewritten, or evaluate only the wrapper's already-resolved InSubqueryExec values) and to keep outputPartitioning from evaluating runtime subqueries at all (originalPlan.outputPartitioning, or a count that does not depend on dynamic pruning).

[Neuw84]

One more data point from the same 1 TB TPC-DS setup (Comet 1.0.0, Spark 3.5, AQE on, scan-only configuration with every Comet operator disabled). With spark.comet.scan.impl=native_datafusion set explicitly instead of left at the default selection, q5 completes and its result matches Spark's (100 rows, equal checksum) in three of three runs tonight — under the default selection it failed on every run over two days with the SubqueryAdaptiveBroadcastExec placeholder trace above. q64 is unchanged: still 0 rows against Spark's 12,185 (the stale pruning value path), so the explicit setting changes the planner path q5 takes but not the underlying issue. Happy to run a patched build if it helps.