Repository navigation
CometNativeScanExec evaluates the inner (unrewritten) dynamic-pruning subquery from outputPartitioning: q5 fails with SubqueryAdaptiveBroadcastExec.execute(), q64 intermittently returns 0 rows at SF1000 #6133
Description
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
- Generate TPC-DS SF1000 as parquet (we used the
tpcds-kitdsdgen,char/varcharcolumns stored as strings; SF10 does not reproduce). - 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.
- Same, adding
--conf spark.comet.exec.shuffle.enabled=falseandspark.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 showingCometNativeScanfor 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.
Activity
Neuw84 commentedon Sep 23, 2026
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
CometNativeScanunder 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 readsstore_salesthrough a dynamic-partition-pruning filter onss_sold_date_sk; in the same environment we also see, on the third-party-operator configuration, q5 fail withSubqueryAdaptiveBroadcastExec does not support the execute() code path— a DPP subquery evaluated before AQE rewrote it. A timing-dependent evaluation of the scan's DPPInSubqueryExec(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 commentedon Sep 23, 2026
Caller stack captured for the q5 failure (Comet 1.0.0, Spark 4.1.3,
CometNativeScanwith third-party operators):So the mechanism is the one the code comment in
serializedPartitionDatadescribes: the innerscan.partitionFiltersholds a separateInSubqueryExecinstance for the dynamic-pruning filter, which Spark's expression walk does not see — soPlanAdaptiveDynamicPruningFiltersrewrites the outer copy but the inner one is still the placeholderSubqueryAdaptiveBroadcastExecwhenoutputPartitioning(reached byValidateRequirementsinsideoptimizeQueryStage) forcesserializedPartitionData, which callsupdateResult()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 innerInSubqueryExecresolves 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
ValidateRequirementswalks (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 (rewritescan.partitionFilterswhen the wrapper'spartitionFiltersare rewritten, or evaluate only the wrapper's already-resolvedInSubqueryExecvalues) and to keepoutputPartitioningfrom evaluating runtime subqueries at all (originalPlan.outputPartitioning, or a count that does not depend on dynamic pruning).[-]TPC-DS q64 returns 0 rows at SF1000 with native execution; correct with Spark operators over the same CometNativeScan and at SF10[/-][+]CometNativeScanExec evaluates the inner (unrewritten) dynamic-pruning subquery from outputPartitioning: q5 fails with SubqueryAdaptiveBroadcastExec.execute(), q64 intermittently returns 0 rows at SF1000[/+]Neuw84 commentedon Sep 24, 2026
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_datafusionset 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 theSubqueryAdaptiveBroadcastExecplaceholder 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.dwsmith1983 commentedon Sep 27, 2026
The q64 loss looks like a different bug from the q5 failure. I filed it as #6264 with a small reproduction.
With AQE on, Comet canonicalizes a scan without its pruning filter while that filter is still the adaptive placeholder. A parent exchange over the 2000
store_salesstage can then compare equal to the one over 1999 and be reused. The secondcross_salesreference reads the other year's rows, and they are filtered out further up. The scan's canonical form is where the filter goes missing, so Spark operators overCometNativeScanare exposed too. It only happens when no coalesced shuffle read sits between the parent and the scan stage, which is normal for shuffles this large. Which stage AQE builds first changes between runs, which fits the intermittent results.Two checks on the SF1000 setup would confirm it:
spark.sql.exchange.reuse=false. If this is the cause, it should never return 0 rows.ReusedExchangeunder the secondcross_salesreference that points at the other year's broadcast ofstore_sales ⋈ store_returns, with only one of the twostore_salesscans still carrying a pruning filter.This path needs AQE. If q64 still returns 0 rows with
spark.sql.adaptive.enabled=false, something else is going on as well, and the plan from that run would help.Neuw84 commentedon Sep 27, 2026
Hi @dwsmith1983 !
Do you need some help with the tests? I can run both to cases.
dwsmith1983 commentedon Sep 27, 2026
Yes, that would help. On the SF1000 setup:
spark.sql.exchange.reuse=false. If the reuse problem is the cause, it should stop returning 0 rows.ReusedExchangeunder the secondcross_salesreference that points at the other year'sstore_salesjoin would confirm it.The fixes are up: #6268 for the q5 failure (Parquet and Iceberg native scans) and #6270 for the q64 row loss. A build of main with both applied, run at SF1000, is the check that matters. If q64 still comes back empty with AQE off, the plan from that run would help too.
Neuw84 commentedon Sep 28, 2026
Ran the checks at TPC-DS SF1000 on an 8-executor Spark 4.1.3 / Scala 2.13 cluster (JDK 25), 300 shuffle partitions, TPC-DS Parquet data partitioned by
ss_sold_date_sk. Each repeat is a separate application (fresh JVM / fresh AQE planning), since the loss is run-dependent. Row counts and checksums below are the runner's own (a checksum over the sorted result rounded to 10 significant digits; identical checksum = identical result).Check 1 — stock Comet 1.0.0, end-to-end, q64,
spark.sql.exchange.reuseon vs off (AQE on)All six returned 12,185 rows with the identical checksum, i.e. the 0-row loss did not reproduce in this batch — neither with reuse off (as expected) nor with reuse on. This is consistent with the intermittency already reported: some days the same image/data returns the correct result on every run.
Check 2 — final executed plan of a 0-row run
No 0-row run occurred in Check 1, so there is no empty-result plan to analyse. For reference, in the (correct, 12,185-row) reuse-on final plan the two
cross_salesreferences keep distinctstore_salesscans, each with its own dynamic-pruning filter:cross_sales:CometNativeScan store_sales…PartitionFilters: [isnotnull(ss_sold_date_sk), dynamicpruningexpression(ss_sold_date_sk IN dynamicpruning#1300)]→CometSubqueryBroadcast dynamicpruning#1300, [d_date_sk]cross_sales: a separateCometNativeScan store_sales…dynamicpruningexpression(ss_sold_date_sk IN dynamicpruning#1301)→CometSubqueryBroadcast dynamicpruning#1301, [d_date_sk]No
ReusedExchangecollapses the twostore_sales ⋈ store_returnsbranches; theReusedExchangenodes present are all benign shared dimension broadcasts (date_dim,customer_demographics,household_demographics,income_band, …) and the sharedstore_returnsshuffle. That is exactly the correct shape — distinct pruning filters, no cross-branch exchange reuse — which matches the correct row count. When a 0-row run is next captured, the plan to look at is the secondcross_salesscan'sdynamicpruning#id and whether it becomes aReusedExchangeof the first branch.Check 3 — Comet built from
main+ #6268 + #6270 (the one that matters)Built in-cluster from
apache/datafusion-comet:main=786bbc8fd630abfd0b29bdaffea6168f9f228863a88941e2ed7ba2bbc02bb687eaed11b8fa1663fa48708be48e3032a9602009e005aa1238316a2e68c2689fb1b9171493f389791f6db4397160536a76(6 commits over base; touchesCometScanUtils.scala,operators.scala)-Pspark-4.1 -Pscala-2.13, nativelibcomet.sobuilt-Ctarget-cpu=x86-64-v3.AQE on, exchange reuse on, default settings unless noted. Baselines are plain Spark on the same data.
Approximate wall times: Comet end-to-end q5 ≈ 36 s, q64 ≈ 69 s; third-party-operator config q5 ≈ 32 s, q64 ≈ 64 s (q64 AQE-off ≈ 108 s).
Results with the two fixes applied:
SubqueryAdaptiveBroadcastExec does not support the execute() code path. That failure did not recur.So with #6268 + #6270, q5 and q64 match Spark in both configurations, with AQE on and off. For q5 that is a clear change: under the default scan selection it failed on every run we made over two days. For q64 it is weaker evidence than it looks, because stock 1.0.0 did not lose rows in this batch either (Check 1), so these runs cannot tell the fix apart from a good day. We will keep running q64 on stock with reuse on and post the plan of the next 0-row run.
dwsmith1983 commentedon Sep 28, 2026
Thanks for running all three. q5 is the clear result: the failure under the default scan selection is gone in both configurations, and the checksums match Spark.
Agreed on q64: a clean batch on stock can't separate the fix from a good day. Your reuse-on plan has the shape #6270 keeps, with two separate
store_salesscans each carrying its own pruning filter and no reuse across the twostore_sales ⋈ store_returnsjoins. In a 0-row run on stock, I'd expect aReusedExchangein place of the second branch's join (or its broadcast) pointing at the first branch's, while the scan-level plan looks the same.Neuw84 commentedon Sep 28, 2026
I will do some more runs with those patches, will update the comment l!
comphead commentedon Sep 28, 2026
Thanks @Neuw84 and @dwsmith1983 this looks pretty critical.
dwsmith1983 commentedon Sep 28, 2026
The two fixes are up: #6268 for the q5 failure (Parquet and Iceberg native scans) and #6270 for the q64 row loss, filed as #6264. Both are waiting on CI approval. Neuw84's SF1000 run with both applied is in the comment above: q5 passes in both configurations, and q64 matches Spark on every run.
Neuw84 commentedon Sep 30, 2026
Running a full test now, will post the results in few hours.
q5 now completes. 23.1 s, 100 rows, the same checksum as vanilla Spark, with 77/77 operators on Comet (11x CometNativeScanExec). Before this fix, q5 failed at this scale with native scans.
The whole run passes. All 103 queries complete. Row counts match vanilla Spark on every query, and checksums match on all but q65, which has ties in its ORDER BY and differs the same way for every engine we compare.
Seems that this is fixed with those patches.
fix: do not resolve DPP subqueries when AQE reads a native scan's par…
fix: keep unconverted DPP filters in a scan's canonical form (apache#…