ErikBPF opened a new issue, #6946: URL: https://github.com/apache/datafusion-comet/issues/6946
## Summary We investigated the q54/q68 performance observations in #6418, but could not reproduce the published distributed SF1000 environment with our available benchmark capacity. This issue records our smaller-scale results, their limits, and the artifacts needed to continue the investigation. It is a reproduction report, not a claim that #6418 is resolved. Related benchmark-hardening PR: #6897, **fix: preserve TPC-DS logical types in Python benchmarks** (currently draft). ## Reproduction limits The published results in #6308 use SF1000 on EKS/S3 with 32 executors. The documented configuration gives each executor 16 Spark cores but an 8-CPU Kubernetes request/limit: 512 task slots and 256 aggregate CPU quota. The newer comparison also changes Spark from 3.5.8 to 4.2.0 and uses pre-release Comet `branch-1.1@ee3f239`. Our available target was one host (14 physical cores / 28 SMT threads). We used SF100 and two isolated containers, each capped at eight logical CPUs, 64 GiB RAM and zero swap. Each container used four whole SMT cores, with disjoint CPU sets. Heavy runs were serialized through a shared host lock. This cannot reproduce the original distributed scheduling, S3 I/O, network shuffle or SF1000 plan/cardinality behavior. In the inspected upstream reports/discussions, we did not find original event logs, final AQE plans, per-iteration timings, query result counts, or a complete launch/registration record. Published two-iteration means establish an observation, but do not let us assess variability or identify the responsible execution path. ## Benchmark correctness finding The Python file-backed runner infers Parquet text columns as `STRING`, losing TPC-DS `CHAR` semantics. For example, padded `CHAR(50)` values can fail ordinary string comparisons against `'Men'` or `'shirts'`. Both vanilla Spark and Comet can therefore execute the same incorrectly registered workload; agreement between engines alone is insufficient. PR #6897 preserves the logical types through explicit registration using 24 TPC-DS schemas. It changes neither source data nor benchmark SQL and is not a native performance optimization. We have not established whether the original SF1000 run used the affected registration path with padded values, and the PR does not claim to explain or fix its slowdowns. ## Local Spark versus Comet comparison An earlier corrected-registration SF100 campaign used Spark 4.2 and Comet 1.1, with one warmup and five measured iterations per query/engine: | Query | Rows | Spark median | Comet median | Spark / Comet | | --- | ---: | ---: | ---: | ---: | | q54 | 35 | 8.660 s | 5.926 s | 1.46x | | q68 | 100 | 7.460 s | 5.401 s | 1.38x | Results matched exactly. These local observations did not reproduce the published slowdown. They do not invalidate the upstream observation or establish that registration caused it. ## Controlled configuration experiments We subsequently tested two hypotheses on a pinned main artifact with provenance `17c1413c8`, Spark 4.2 and JDK 17.0.20: 1. Native join dynamic filtering, off versus on, with batch size 8192. 2. Native batch size 8192 versus 32768, with join dynamic filtering off. Both retained Parquet row-filter pushdown off, JVM shuffle batch size 8192 and native shuffle direct-read enabled. Each hypothesis used one container and changed only its designated setting. | Experiment | Query | Baseline median | Candidate median | Median paired latency reduction | Paired bootstrap 95% interval | | --- | --- | ---: | ---: | ---: | --- | | Join filtering OFF → ON | q54 | 6.854 s | 6.731 s | +1.89% | [+0.99%, +4.36%] | | Join filtering OFF → ON | q68 | 7.859 s | 8.032 s | −2.25% | [−2.69%, −0.46%] | | Batch 8192 → 32768 | q54 | 6.852 s | 6.413 s | +6.49% | [+4.45%, +7.47%] | | Batch 8192 → 32768 | q68 | 7.785 s | 7.976 s | −2.71% | [−4.24%, −0.40%] | Positive reduction means faster. Reductions are medians of paired process ratios, not ratios of independently summarized arm medians. These are within-campaign Comet comparisons, not new Spark-versus-Comet comparisons; do not compare their absolute times against the earlier table to infer a version regression. Protocol and outcome: - Per hypothesis: two sessions, four paired blocks per query/session with balanced AB/BA order, fresh process per arm, one excluded warmup and five measurements. - Total: 64 timing processes, 320 measured iterations and 64 warmups. All 384 outputs passed exact schema/value/null/decimal/multiplicity/order checks against fresh Spark references, including offline replay. - Separate q68 LIMIT101 diagnostics verified that the unchanged measured LIMIT100 query did not split a tied ordering-key group. - Timing included query construction/planning through collection, excluding `explain()`, profiling, receipt serialization and oracle assertions. - Bootstrap: eight process pairs per query/hypothesis, 10,000 resamples, seed 4821. Iterations within a process were not treated as independent replications. - Both q68 regressions repeated in both sessions. Neither candidate passed the predeclared two-query adoption gate: at least 5% useful gain on one query, a positive confidence interval and replication, with no established regression on the other. - Memory gates passed; no OOM occurred. Candidate/baseline median JVM peak-RSS ratios were 0.928/0.895 for join filtering and 1.001/1.038 for batch size, q54/q68 respectively. - Join filtering pruned early/residual rows but did not reduce observed scan bytes. Larger batches reduced several batch counts but did not improve q68. We did not measure allocation counts. Existing host services were not CPU-isolated from the experiments. Memory includes startup/registration; sampled cgroup memory includes page cache and can miss short peaks. Dataset mounts were read-only and before/after metadata inventories matched; the entire dataset was not rehashed during this campaign. Raw local receipts, commands, plans, event logs and outputs have been retained. No production defaults were changed. ## Most useful next evidence The upstream q68 observation remains particularly interesting: Comet went from 1.417 s to 3.942 s while Spark improved from 2.806 s to 2.3785 s. By contrast, q54 Comet improved from 7.2835 s to 5.907 s but remained slower than Spark. These are different investigation targets, and the cross-version comparison does not isolate Comet from Spark/configuration changes. Later fixes #6547 and #6646 address stale AQE plans using JVM shuffle decoding despite native direct-read being enabled. This is consistent with a JVM/native bridge cost observed in our earlier local profiling, but is not attribution for the original SF1000 results. @andygrove, could you provide the following artifacts from the original benchmark, or capture them in a rerun? 1. Exact launch command, engine revisions, effective configuration and table-registration/data-generation details. 2. q54/q68 result counts and correctness receipts, including logical schemas and relevant filter/join cardinalities. 3. Final executed AQE plans and Spark event logs, especially actual shuffle-reader paths and stage/task metrics. 4. Individual iteration timings, warmup policy and execution order. That would let us distinguish a benchmark-semantics issue, an already-fixed shuffle path, a version-dependent plan change, and a remaining distributed performance problem before proposing another optimization. -- 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]
