dwsmith1983 opened a new pull request, #6270:
URL: https://github.com/apache/datafusion-comet/pull/6270
## Which issue does this PR close?
Closes #6264. Part of #6133 (the q64 row loss).
## Rationale for this change
`CometScanUtils.filterUnusedDynamicPruningExpressions` dropped a DPP filter
from a scan's canonical form when its subquery was still the adaptive
placeholder, not only when it had become `TrueLiteral`. AQE canonicalizes a
query stage from its exchange as it was before the stage optimizer rules ran,
which is before `CometPlanAdaptiveDynamicPruningFilters` converts the
placeholder. So an exchange above that stage saw a scan with no DPP filter. Two
such exchanges over scans of the same table with different DPP filters compared
equal, and AQE reused the first for both. One branch then read the other's
rows. TPC-DS q64 builds the same `store_sales` join for 1999 and 2000, which is
where #6133 loses its rows.
The extra stripping came with #4112 so that otherwise identical scans could
share a stage while their DPP was still unconverted. That reuse does not need
it. The quick `stageCache` lookup uses the exchange before optimization, but
`createQueryStages` checks the cache again with `newStage.plan.canonicalized`
after the stage rules have run, and by then the unused filter is `TrueLiteral`
and is dropped as in Spark.
## What changes are included in this PR?
- `filterUnusedDynamicPruningExpressions` drops only
`DynamicPruningExpression(TrueLiteral)`, matching `FileSourceScanExec`. It is
shared by `CometNativeScanExec`, `CometScanExec` and
`CometIcebergNativeScanExec`.
- `CometNativeScanExec.doCanonicalize` drops every DPP filter from
`originalPlan`. No DPP rule rewrites `originalPlan`, so its filters are a stale
copy. The scan's DPP identity is in its top-level `partitionFilters`. Without
this, the SPARK-32509 test ("unused DPP filter and exchange reuse") stops
reusing, because the stale placeholder in `originalPlan` keeps the scan apart
from its twin.
One reuse goes away, as in Spark: a parent exchange over two scan stages
whose DPP later becomes `TrueLiteral` is no longer shared, because its key was
fixed while the placeholder was still there. On TPC-DS at SF1 with AQE (and
`AQEPropagateEmptyRelation` excluded, so queries that return no rows on this
data keep their plan shape), the final plans of q14a, q14b, q23a, q23b, q24a
and q24b have the same number of `ReusedExchange` and `ReusedSubquery` nodes
before and after this change. q64 has one more `ReusedExchange` and keeps both
`store_sales` DPP scans, where main drops one of them.
## How are these changes tested?
New tests in `CometExecSuite` run a `UNION ALL` of one CTE joined to two
different dimension filters, with AQE on and coalescing off, and check the
answer against Spark. On Spark 3.5 and later, where these scans run AQE DPP in
Comet, they also check that each branch keeps its own DPP scan pruned to 1 and
3 partitions and that no `ReusedExchange` sits over a DPP scan:
- a broadcast over a sort-merge join of the DPP scan's shuffle stage, the
q64 shape. Main returns 100 rows where Spark returns 400.
- a shuffle over the DPP scan's aggregate stage. Main returns 7 rows on
Spark 3.5 and 21 on 4.1, where Spark returns 28.
- the dimension join in the scan's own stage, which passes on main and
guards the scan stage path.
`CometIcebergNativeSuite` gets the q64 shape for the Iceberg native scan.
Main returns 100 rows where Spark returns 400 on Spark 3.5 and 4.1, and a wrong
answer on 3.4. On 3.4 the Iceberg scan stays in Comet with Spark's DPP rule
planning its filter, so this path was exposed there too.
With the change, the new tests pass on Spark 3.4, 3.5, 4.0 and 4.1. On 3.5,
`CometExecSuite`, the DPP fallback suites, the TPC-DS plan stability suites and
`CometIcebergNativeSuite` pass. Reverting the helper change fails the new
tests. Reverting the `originalPlan` change fails the SPARK-32509 test.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
