Repository navigation
AQE reuses one exchange for scans with different dynamic pruning filters and drops rows #6264
Description
Describe the bug
With AQE on, two exchanges that differ only in the dynamic partition pruning filter of a scan below them can compare equal, and AQE replaces the second with a ReusedExchange of the first. The consumer of the second exchange then reads the other branch's rows, and those rows are silently lost from the result.
CometScanUtils.filterUnusedDynamicPruningExpressions drops DynamicPruningExpression(TrueLiteral) like Spark does. It also drops any pruning filter whose subquery is still the adaptive placeholder (SubqueryAdaptiveBroadcastExec or CometSubqueryAdaptiveBroadcastExec). Spark's FileSourceScanExec keeps that filter, and its canonical form includes the dimension side's plan, so IN (1999 dates) and IN (2000 dates) stay distinct.
AQE canonicalizes a query stage from its exchange as it was before the stage optimizer rules ran (ExchangeQueryStageExec._canonicalized). That is before CometPlanAdaptiveDynamicPruningFilters converts the placeholder. So any exchange above that stage sees the scan with no pruning filter at all. Two such parent exchanges, one over each scan, get the same canonical form, and the stage cache hands out the first one twice.
A coalesced AQEShuffleRead between the parent and the child stage hides the problem, because its partition specs carry per-partition data sizes that differ between the two branches. AQE adds no such read when coalescing is off, or when coalescing would not merge any partitions, which is what happens with large shuffles. This is probably why TPC-DS q64 returns 0 rows only sometimes, and only at large scale factors, in #6133: the two cross_sales references build the same store_sales join for 1999 and 2000. That link has not been confirmed at SF1000.
This is separate from the q5 failure in #6133, which comes from CometNativeScanExec.outputPartitioning running the placeholder subquery.
Steps to reproduce
Reproduced on main (605051a) with -Pspark-3.5 and -Pspark-4.1, with the whole plan running in Comet (native Parquet scan and Comet shuffle enabled).
spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "false")
spark.range(1000)
.selectExpr("cast(id % 10 as int) as store_id", "cast(id % 13 as int) as item", "cast(id as int) as amount")
.write.partitionBy("store_id").parquet("/tmp/fact")
spark.range(10)
.selectExpr("cast(id as int) as store_id",
"case when id = 1 then 'a' when id in (2, 3, 4) then 'b' else 'c' end as grp")
.write.parquet("/tmp/dim")
spark.read.parquet("/tmp/fact").createOrReplaceTempView("fact")
spark.read.parquet("/tmp/dim").createOrReplaceTempView("dim")
spark.sql("""
WITH x AS (SELECT store_id, item, sum(amount) AS s FROM fact GROUP BY store_id, item),
y AS (SELECT store_id, s % 7 AS b, count(*) AS c FROM x GROUP BY store_id, s % 7)
SELECT y.store_id, y.b, y.c FROM y JOIN dim d ON y.store_id = d.store_id WHERE d.grp = 'a'
UNION ALL
SELECT y.store_id, y.b, y.c FROM y JOIN dim d ON y.store_id = d.store_id WHERE d.grp = 'b'
""").collect().lengthSpark returns 28 rows. Comet returns 7 on Spark 3.5 and 21 on Spark 4.1, depending on which branch's stage is created first. The final plan has a ReusedExchange under the second branch pointing at the first branch's exchange, and only one of the two pruned scans runs.
The same thing happens when the parent is a broadcast, for example a broadcast of a sort merge join over the scan's shuffle stage, which is the q64 shape. With coalescing on, these small tables give correct results because the coalesced reads keep the two parents apart.
Expected behavior
Same rows as Spark. Two scans with different dynamic pruning filters should never share an exchange.
Additional context
The extra stripping was added in #4112 so that scans whose pruning filter is still unconverted could reuse each other. Matching Spark's rule and removing the stale filter copy inside CometNativeScanExec.originalPlan from its canonical form fixes the reproduction while keeping the SPARK-32509 reuse test passing. CometScanExec and CometIcebergNativeScanExec call the same helper.
Activity
dwsmith1983 commentedon Sep 27, 2026
I will take this as I found it doing something else so already have an idea on what to do.
Neuw84 commentedon Sep 29, 2026
A related symptom, possibly the same root cause: with the native scan, a dynamic partition pruning filter can be dropped from a scan, so the scan reads every partition. Results are still correct; the cost is the extra reading. I'm reporting it here first because it sits in the same area. It may deserve its own issue; I can open one.
Where it shows up. TPC-DS q23a/q23b at SF1000, Parquet with the fact tables partitioned by
*_sold_date_sk. Inside themax_store_salesscalar subquery, thestore_salesscan carriesdynamicpruningexpression(ss_sold_date_sk IN dynamicpruning#…)from thed_year IN (2000, 2001, 2002, 2003)date scan.store_salesscan, SF1000FileSourceScanExecCometNativeScanExec(Comet 1.0.0)SubqueryBroadcast dynamicpruning#…The other DPP scans in the same query (
frequent_ss_items, and thecatalog_salesandweb_salesscans) keep their subquery under Comet and prune like Spark.When it happens. It depends on the join shape:
customer, and the broadcast join withdate_dimthat the DPP filter reuses is above that join.customeris broadcast too, and the native scan prunes correctly.Minimal reproduction (SF1).
store_sales,catalog_salesandweb_saleswrittenpartitionBy(<x>_sold_date_sk); dimensions unpartitioned.spark.sql.autoBroadcastJoinThreshold=200000, socustomeris shuffled anddate_dimstays broadcast. AQE is on (default).spark.comet.scan.enabled=truewithspark.comet.scan.implleft as the default (the native scan).SubqueryBroadcast36aec83f36aec83fCometSortMergeJoin,CometBroadcastHashJoin,CometExchange)36aec83fThe 8 file ranges that are skipped are the
__HIVE_DEFAULT_PARTITION__ones, removed by the staticisnotnull(ss_sold_date_sk). The DPP filter itself prunes nothing.Relation to this issue. I could not tie it directly to
filterUnusedDynamicPruningExpressions. In 1.0.0 that function is only called fromCometNativeScanExec.doCanonicalize(and the scan here is not reused; it runs with its own metrics). What the plan shows is the DPP subquery missing from the native scan, so the scan's partition list never gets the runtime filter. If an unconvertedSubqueryAdaptiveBroadcastExecis being dropped here outside canonicalization as well, the fix for this issue may cover it; if not, a separate issue is probably right.dwsmith1983 commentedon Sep 29, 2026
@Neuw84 I am on my phone. Is it similar to SPARK-54554 ASF Jira ticket?
Neuw84 commentedon Sep 29, 2026
@dwsmith1983 I don't think it's the same. SPARK-54554 is about Spark's optimizer never inserting the DPP filter, because the pruning side comes from SHOW PARTITIONS and no plan gets one.
Here Spark does insert it. With Spark alone the same plan has dynamicpruningexpression(ss_sold_date_sk IN dynamicpruning#…) backed by a SubqueryBroadcast, and it prunes to 1.66 M rows at SF1. The filter only disappears once the scan becomes CometNativeScanExec, and the other DPP scans in the same query keep it. So it happens after DPP planning, in Comet's scan conversion (or its AQE handling), not in Spark's DPP rule.
dwsmith1983 commentedon Sep 29, 2026
It does not cover it, so a separate issue please. #6270 only changes the canonical copy of the scan that AQE compares for stage reuse. The filters the scan actually runs with never come from that copy.
My suspect is
CometPlanAdaptiveDynamicPruningFilters. When it looks for the broadcast to reuse, it searches the current stage and then the main query's plan. Spark'sPlanAdaptiveDynamicPruningFilterssearches the adaptive plan that owns the filter. In q23 thestore_salesandcustomerjoin is inside the scalar subquery, which has its own adaptive plan, so thedate_dimbroadcast may never be found and the filter becomesdynamicpruningexpression(true), the same result as Spark's no-reuse path. I have traced this in the source but not run it yet.Two things would confirm it: the
PartitionFiltersline of thatstore_salesscan in the final plan withspark.comet.enabled=falseat the same threshold, and the Spark version. I will write a repro from your SF1 settings once the issue is open.Neuw84 commentedon Sep 30, 2026
Confirming that #6270, applied together with #6268 on main
b58b2f3a(HEAD0f22d064), fixes this at SF1000.q64 now returns 12,185 rows, the same as vanilla Spark, with the same checksum. Test setup:
All 103 TPC-DS queries match Spark on row counts.
test: check that DPP scan stages on a Delta table are not reused acro…
fix: keep unconverted DPP filters in a scan's canonical form (apache#…
docs: address review feedback on the known issues page
docs: list known Parquet and Iceberg issues and add a data source aud…