Neuw84 commented on issue #6133: URL: https://github.com/apache/datafusion-comet/issues/6133#issuecomment-5869860339
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.reuse` on vs off (AQE on) | exchange.reuse | run | rows | checksum | wall (s) | |---|---|---:|---|---:| | false | 1 | 12185 | bce69949 | 157.6 | | false | 2 | 12185 | bce69949 | 151.6 | | false | 3 | 12185 | bce69949 | 144.6 | | true (default) | 1 | 12185 | bce69949 | 66.1 | | true (default) | 2 | 12185 | bce69949 | 70.5 | | true (default) | 3 | 12185 | bce69949 | 70.0 | 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_sales` references keep **distinct** `store_sales` scans, each with its own dynamic-pruning filter: - first `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]` - second `cross_sales`: a **separate** `CometNativeScan store_sales` … `dynamicpruningexpression(ss_sold_date_sk IN dynamicpruning#1301)` → `CometSubqueryBroadcast dynamicpruning#1301, [d_date_sk]` No `ReusedExchange` collapses the two `store_sales ⋈ store_returns` branches; the `ReusedExchange` nodes present are all benign shared dimension broadcasts (`date_dim`, `customer_demographics`, `household_demographics`, `income_band`, …) and the shared `store_returns` shuffle. 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 second `cross_sales` scan's `dynamicpruning#` id and whether it becomes a `ReusedExchange` of the first branch. ### Check 3 — Comet built from `main` + #6268 + #6270 (the one that matters) Built in-cluster from `apache/datafusion-comet`: - base `main` = `786bbc8fd630abfd0b29bdaffea6168f9f228863` - #6268 head = `a88941e2ed7ba2bbc02bb687eaed11b8fa1663fa` - #6270 head = `48708be48e3032a9602009e005aa1238316a2e68` - merge commit = `c2689fb1b9171493f389791f6db4397160536a76` (6 commits over base; touches `CometScanUtils.scala`, `operators.scala`) - profile `-Pspark-4.1 -Pscala-2.13`, native `libcomet.so` built `-Ctarget-cpu=x86-64-v3`. AQE on, exchange reuse on, default settings unless noted. Baselines are plain Spark on the same data. | configuration | query | AQE | runs | rows (each) | checksum | matches Spark | |---|---|---|---|---:|---|---| | spark (baseline) | q5 | on | 1 | 100 | c7eda84b | — | | spark (baseline) | q64 | on | 1 | 12185 | bce69949 | — | | Comet end-to-end | q5 | on | 3 | 100 | c7eda84b | yes | | Comet end-to-end | q64 | on | 3 | 12185 | bce69949 | yes | | Comet end-to-end | q64 | off | 1 | 12185 | bce69949 | yes | | Comet scan under third-party operators | q5 | on | 3 | 100 | c7eda84b | yes | | Comet scan under third-party operators | q64 | on | 3 | 12185 | bce69949 | yes | | Comet scan under third-party operators | q64 | off | 1 | 12185 | bce69949 | yes | 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:** - **q64 returns 12,185 rows on every run** in both configurations, with AQE on and with AQE off — no row loss. - **q5 completes with 100 rows and a checksum equal to Spark's** in both configurations — including the "Comet scan under the third-party operators" configuration, which is where q5 previously failed with `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. -- 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]
