Repository navigation
Conversation
github-actions
Bot
added
bug
sunchao
added
run-all-spark-profiles
|
I started this review before realizing the PR was a draft. Feel free to ignore if this isn't helpful. I checked this out and ran it locally, and I think the justification is stronger than what the description says. The description leads with decimal aggregate overflow, which is hard to pin down, because a deterministic hash still sends every row with a given key to one partition, so the per-group aggregation doesn't actually change. What does break is co-partitioning. I built that plan on main at Comet returns zero rows. Spark returns four. Default The encoding mismatch underneath is what you'd expect from reading the two implementations side by side. I should own part of this. I closed #3079 as not a bug, on the grounds that we don't need to allocate the same partitions as Spark. That was wrong, and the join above is why. Our own columnar path uses Spark's partitioner, so the native path has to match Spark in order to stay consistent with its sibling. I'll reopen it. Could the description lead with the join rather than the overflow argument? It's much easier for a reviewer to verify. And would you add it as a test? The six tests here are good, and I confirmed they're load-bearing, since reverting just the one line fails four of them. But they all assert plan shape or partition-id equality. None of them shows a wrong answer, which is the case that actually matters. A few smaller things. The new comment drops the link to #3079, so there's no longer a pointer from the code to the work that would lift the restriction. Could it reference #5994 instead, since that's the live issue for the Spark-compatible native encoding? Related, On that, the real fix looks small. Producing The Two things Eleven TPC-DS plans move from Last, the branch state. #5421 landed as |
|
#5421 is merged now, so the prerequisite here is met. Could you merge main and drop the carried commits? That would leave just One thing worth adding to the rationale: this fixes more than the aggregate-overflow case. |
b42be68 to
d2231af
Compare
sunchao
added
run-spark-4.1-tests
sunchao
left a comment
sunchao
left a comment
There was a problem hiding this comment.
Summary
- Prior state and problem: Native and JVM shuffles could assign matching wide-decimal keys to different partitions, silently dropping join rows.
- Design approach: Reject native multi-partition hashing when a key contains a decimal with precision greater than 18, using the existing fallback paths.
- Correctness / compatibility analysis: Checked Spark 3.4.3, 3.5.9, 4.0.4, 4.1.3 and 4.2.0 sources. The precision boundary matches Spark’s hashing implementation. Recursive checks cover nested decimal leaves. The single-partition exception matches the writer’s existing
SinglePartitionserialization. Aggregate-buffer repair remains in place. - Key design decisions: The change preserves native handling of decimal payloads, range keys and single-partition exchanges. Extending the existing predicate keeps the implementation simple. Cross-references connect the shuffle and SQL hash restrictions without adding an unnecessary abstraction.
- Implementation sketch: The routing guard is accompanied by partition-ID, mixed-input join, nested-key, aggregate-buffer and Celeborn regressions, updated fuzz expectations and plan snapshots, documentation, and a focused COUNT benchmark.
- Behavioral changes worth calling out:
autouses JVM columnar shuffle when eligible.nativeand Celeborn fall back to Spark, potentially restoring incompatible aggregates too. The conversion overhead is documented. The author’s measurements report increases of 9–33% for native mode and 17–25% for auto on the measured workload. I did not independently rerun those timings. - Suggested improvements: No introduced P1/P2 issues found within this review. No substantiated unresolved P1/P2 concerns remain from the existing discussion.
Reviewed the full base-relative diff at 8b5915ae9878238654f76e38a2fe1d72e600af9a against b8a2358cc0a54818f5ac6db85ce7e5a3a77e9692, including all four PR commits. The apparent timezone-documentation and nested-null-test removals are base-only commits after the branch point, not deletions introduced by this PR. Read the supplied discussions before forming conclusions. The PR remains non-draft.
Routed skills: review-comet-pr, review-comet-shuffle-pr, and review-comet-expression-pr.
Exact-head CI: run 36487974865 is associated with the reviewed SHA and tests synthetic merge 71a8847. At the final check, 35 checks passed, 11 were skipped and 12 remained running, with no failures. Shuffle suites passed across all five Spark profiles. Inspected logs confirm the new regressions passed on Spark 3.4, 3.5 and 4.1. Native tests, benchmark compilation, TPC-H and TPC-DS checks also passed. Spark SQL and several broader suites remain pending.
Local validation: A standalone diagnostic using the extracted native hash function confirmed the encoding mismatch and precision boundary. For wide-decimal 1, Spark-compatible encoding selects partition 4 of 7 while native encoding selects 6. Whitespace checks passed. No full local Comet build or end-to-end suite was run because this checkout has no prepared JVM/native artifacts. The diagnostic does not replace integration testing.
sunchao
left a comment
sunchao
left a comment
There was a problem hiding this comment.
Summary
- Prior state and problem: Native and JVM shuffles assigned matching wide-decimal keys to different partitions, allowing mixed-input joins to silently lose rows.
- Design approach: Reject native multi-partition hashing for decimal keys with precision greater than 18, including nested leaves, through the existing shuffle fallback paths.
- Correctness / compatibility analysis: Verified the hashing boundary against Spark 3.4.3, 3.5.9, 4.0.4, 4.1.3 and 4.2.0 sources. Recursive key checks cover nested decimals. The single-partition exception matches the writer’s
SinglePartitionserialization. Existing aggregate-buffer repair handles the newly triggered Spark fallback. - Key design decisions: Decimal payloads, range keys and single-partition exchanges retain native support. Extending the existing predicate keeps the change small. Cross-references connect the shuffle and SQL hash restrictions without adding an unnecessary abstraction.
- Implementation sketch: Adds partition-ID, mixed-input join, nested-key, aggregate-buffer and Celeborn regressions, updates fuzz expectations and plan snapshots, documents fallback behavior, and adds a focused
COUNTbenchmark. - Behavioral changes worth calling out:
autoselects JVM columnar shuffle when eligible.nativeand Celeborn use Spark fallback, potentially restoring incompatible aggregates too. Conversion overhead is documented. The author reports increases of 9–33% for native mode and 17–25% for auto in the measured workload. These timings were not independently rerun, and TPC-DS runtime impact remains unmeasured. - Suggested improvements: No introduced P1/P2 issues found within this review. No substantiated unresolved P1/P2 concerns remain from the existing discussion.
Reviewed full SHA 8b5915ae9878238654f76e38a2fe1d72e600af9a against base b8a2358cc0a54818f5ac6db85ce7e5a3a77e9692, covering all four PR commits and the full base-relative scope. The apparent timezone-documentation and nested-null-test removals correspond to base-only additions after the branch point. Read all supplied non-Copilot discussion and reviews. Live metadata confirms the PR remains non-draft at the requested head.
Routed skills: review-comet-pr, review-comet-shuffle-pr, and review-comet-expression-pr.
Exact-head CI: 48 checks passed, 13 skipped, none failed or pending. Run 36487974865 is associated with the reviewed SHA and tested synthetic merge 71a8847. Shuffle jobs passed across all five Spark profiles. Spark 4.1 SQL, native tests, benchmark compilation, and TPC-H/TPC-DS checks passed. Inspected logs confirm the new regressions executed successfully.
Local validation: Recompiled and ran a standalone diagnostic using the exact current native hash function. It confirmed precision-18 agreement and precision-19/38 encoding divergence, including positive, negative and boundary values. For wide-decimal 1, Spark-compatible encoding selects partition 4 of 7 while native encoding selects 6. Whitespace checks passed. No full local build or JVM integration suite was run because this checkout lacks prepared artifacts. The diagnostic supplements the CI evidence and does not replace integration testing.
apache#6005) Comet's native Murmur3 hashes a decimal with precision above 18 differently from Spark (apache#3079, apache#5994). Comet picks the shuffle of each exchange on its own, so the two inputs of a sort-merge join could be partitioned by different hash functions and matching keys meet in different partitions: rows were silently dropped (a decimal(38,10) join returned 108 of 2000 rows). Native shuffle now refuses hash partitioning over more than one partition whose keys contain such a decimal at any depth. In auto mode the shuffle becomes Comet's columnar shuffle when that applies; in native mode, or with Celeborn, a Spark shuffle. Wide decimals in the payload, range partitioning and single-partition shuffles stay native. BoundaryFormats offers a native shuffle only under the same condition, via CometShuffleExchangeExec.hasWideDecimalHashKey. Ports the PR's tests and approved TPC-DS plans, and adds a sort-merge join of a native-scan and a Spark-scan input on decimal(38,10) with AQE and boundary formats on and off. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
…VM shuffle A typed Dataset conversion moves the shuffle above it from Comet's columnar shuffle, which partitions with Spark's hash, to native shuffle. Native shuffle hashes decimals wider than 18 digits differently from Spark (apache#5994), so a join on such keys with an input that stays on columnar shuffle, for example one with an array<int> column the conversion declines, put matching keys in different partitions and returned 9 rows instead of 100. Such a shuffle now stays on columnar shuffle unless it has one partition, the same rule apache#6005 proposes for every native shuffle. Also declare the benchmark's row count as a Long, which the strict Spark 3.5 compile requires.
8b5915a to
f682d09
Compare
|
Rebased onto main 9866221 and pushed f682d09. The carried prerequisite commits are already absent; this remains the four-commit routing fix. The description leads with the mixed Parquet/JSON join, and the regression checks all four matching rows with AQE on and off. Code/docs retain the #5994 cross-reference, native/auto/Celeborn fallback behavior, recursive wide-decimal boundary and the single-partition exception. Current-head validation passed all 64 selected JDK 21 / Spark 4.1.3 tests: seven decimal tests and the full 57-test Celeborn planning suite. The exact-base Linux native artifact was verified and bundled first. Spotless and whitespace checks pass. The separate COUNT benchmark is retained to isolate shuffle fallback from the existing AVG benchmark’s wide-decimal support constraints. Current-head benchmark timings and TPC-DS query runtimes were not rerun, so the description now states that limit and removes the old timing table. Fresh hosted CI is pending. |
|
The fresh Spark 3.5 strict-warning job caught an implicit Int-to-Long conversion in the new benchmark constructor. Fixed with |
sunchao
left a comment
sunchao
left a comment
There was a problem hiding this comment.
Summary
- Prior state and problem: Native and JVM shuffles assigned matching wide-decimal keys to different partitions, allowing mixed-input joins to silently lose rows.
- Design approach: Reject native multi-partition hashing for decimal keys with precision greater than 18, including nested leaves, through the existing fallback paths.
- Correctness / compatibility analysis: Checked Spark 3.4.3, 3.5.9, 4.0.4, 4.1.3 and 4.2.0 sources. The precision boundary matches Spark’s decimal hashing. The single-partition exception matches the writer’s existing
SinglePartitionserialization. Aggregate-buffer repair preserves compatible execution when fallback occurs. - Key design decisions: Reusing the recursive type predicate and existing shuffle selection keeps the implementation small. Wide-decimal payloads, range keys and single-partition exchanges retain native eligibility.
- Implementation sketch: Adds partition-ID, mixed-input join, nested-key, aggregate-buffer and Celeborn regressions. Updates fuzz expectations, eleven plan snapshots and documentation, and adds a focused
COUNTbenchmark. - Behavioral changes worth calling out: Compared with
branch-1.1at992c806a7e38c2e88bd018aa5774164b0850e1fa, this intentionally changes affected shuffle routing.autouses JVM columnar shuffle when eligible, adding conversion overhead.nativeand Celeborn use Spark fallback, which can also restore incompatible aggregates. The documentation explains this tradeoff. - Suggested improvements: No introduced P1/P2 issues found within this review. No substantiated unresolved P1/P2 concerns remain from the existing discussion.
Reviewed the full 21-file, five-commit diff at f79e2ced39fbe30930a70f8370b1e506ed13a873 against ac9ae94d057013095fdf1b70e2d4bc5b24a58d90. Read the supplied reviews and discussions. Live metadata confirms the PR remains non-draft at the requested head.
Routed skills: review-comet-pr, review-comet-expression-pr and review-comet-shuffle-pr.
Exact-head CI: 53 checks succeeded, 14 skipped, none failed or pending. Run 37222542918 is associated with the reviewed SHA and tested synthetic merge f20fa8b7e351bb71fde38f9d9231634f4e4e850b. Inspected logs confirm the new decimal and Celeborn regressions passed across all five Spark profiles. Spark 4.1 SQL, native tests, strict Scala warnings, benchmark compilation and TPC-H/TPC-DS checks passed.
Local validation: Compiled and ran an isolated diagnostic using the current native hash function across 36 boundary values. It confirmed precision-18 agreement and precision-19/38 encoding divergence. Wide-decimal 1 selects partition 4 of 7 with Spark’s encoding versus 6 natively. Whitespace checks passed. No local full build or JVM integration suite was run because this checkout lacks prepared artifacts. Current-head benchmark timings and affected TPC-DS runtime costs remain unmeasured.
…che#6564) * feat: run the operators above a typed Dataset operation natively Typed Dataset operations (map, flatMap, mapPartitions, mapGroups, cogroup) pass JVM objects between their operators, so they stay on Spark, and today the operators above them stay on Spark too until the next shuffle. Every typed operation ends in SerializeFromObjectExec, whose output is ordinary rows. With the new spark.comet.convert.typedDataset.enabled, CometExecRule puts a CometSparkToColumnarExec above it, so a partial aggregate, a broadcast join or a native shuffle above the operation runs natively. Spark inserts no columnar transitions below a RowToColumnarTransition, so the rule adds them to the subtree under the conversion with Spark's own ApplyColumnarRulesAndInsertTransitions. Without them the typed operation would read its Comet child through Spark's interpreted columnar-to-row path. CometExecRule no longer tags a ColumnarToRowTransition as an unsupported operator, which it did to these transitions on its second pass under AQE. Off by default: it is 1.4-2.2x faster when an aggregate over many groups sits above the typed operation, and slower when the work above is cheap. Adds CometTypedDatasetSuite and CometTypedDatasetBenchmark. * fix: keep wide-decimal shuffles above a typed Dataset conversion on JVM shuffle A typed Dataset conversion moves the shuffle above it from Comet's columnar shuffle, which partitions with Spark's hash, to native shuffle. Native shuffle hashes decimals wider than 18 digits differently from Spark (apache#5994), so a join on such keys with an input that stays on columnar shuffle, for example one with an array<int> column the conversion declines, put matching keys in different partitions and returned 9 rows instead of 100. Such a shuffle now stays on columnar shuffle unless it has one partition, the same rule apache#6005 proposes for every native shuffle. Also declare the benchmark's row count as a Long, which the strict Spark 3.5 compile requires. * refactor: simplify the typed Dataset conversion's shuffle guard and tests - Use Spark's existsRecursively(DecimalType.isByteArrayDecimalType) for the wide-decimal check, fold the duplicated conversion cases in readsTypedDatasetConversion, and check the config and earlier native shuffle reasons before walking the plan. Add a TODO to drop the guard once apache#5994 lands. - Tests: assert the join's exchanges with checkCometExchange, drop conditions that cannot change the transition assertion, write the Parquet table once for both AQE settings, and take checkConverted's query by value. - Benchmark: import spark.implicits instead of declaring encoders. * fix: preserve typed Dataset limit short-circuiting * fix: leave typed Dataset output unconverted below any reader that stops early The limit guard covered only physical limits, so a mapPartitions function such as `_.take(1)`, or code reading Dataset.rdd, still read typed rows through an Arrow batch and ran the user function on rows that Spark never reaches. Walk down from each reader that can stop early (a limit, a top-k over sorted input, a MapPartitionsExec, and a DeserializeToObjectExec at the plan root) and stop at an operator that reads all of its input first, so a limit above an aggregate keeps the conversion. * fix: recognize Dataset.rdd plans whose deserializer Spark removed The guard for code reading Dataset.rdd looked for the DeserializeToObjectExec that Dataset.rdd puts at the root of the plan. When the Dataset ends in a typed operation such as map, Spark's EliminateSerialization drops that deserializer together with the operation's serializer, so the root is the operation itself, or a typed filter over it. The output of an earlier typed operation was then still converted, and map(f).filter(...).map(g).rdd.take(1) ran f on rows that Spark never reaches. The guard now takes any root that produces objects, under a filter or a project, as code reading Dataset.rdd. A Dataset's own plan ends in rows, so no other plan has such a root.
2 participants
Which issue does this PR close?
Related to #3079. Spark-compatible native decimal hashing is tracked in #5994.
Rationale for this change
Native hashing of decimal keys with precision greater than 18 differs from Spark's partition assignments. A join can mix native and JVM shuffles because Comet selects the implementation independently for each exchange. Matching keys then reach different partitions and rows silently disappear: the Parquet/JSON regression returns zero rows without this guard and the expected four with it, with AQE enabled or disabled.
What changes are included in this PR?
autouses Comet columnar shuffle when eligible;nativeand Celeborn use Spark shuffle.SinglePartition, so no decimal hashing occurs.collect_listandcollect_set, preserving native input scans. Add mixed-input join coverage and extend the existing Celeborn fallback test.How are these changes tested?
Rebased onto main
98662215d, preserving main's native-library reload regression and new shuffle/AQE coverage while keeping the wide-decimal routing tests in CometNativeShuffleSuite.At
f682d0902, the root Maven reactor passed all 64 selected tests on JDK 21 / Spark 4.1.3: seven decimal-routing tests plus all 57 Celeborn planning tests. Coverage includes the precision boundary, single-partition exception, unaffected payload/range keys, nested keys, aggregate buffers, and the mixed Parquet/JSON join with AQE enabled and disabled. Spotless and whitespace checks passed.Follow-up
f79e2ced3makes the benchmark row-count conversion explicit after the Spark 3.5 strict-warning job flagged numeric widening. The complete root reactortest-compile -Pspark-3.5 -Pstrict-warnings -DskipTestspasses on JDK 21 after that change.The native source and build-input trees match main exactly. Tests used the SHA-verified Linux native artifact 11309692999 from run 37219565655 at
98662215d, bundled before the JVM tests. No new native compilation was needed for this Scala-only rebase.The focused COUNT benchmark is retained to isolate shuffle fallback from decimal AVG support. Current-head timing measurements and TPC-DS runtimes were not rerun. Previous timing tables have been removed because they used an older revision and native artifact. The plan snapshots and fuzz expectation updates remain in the PR but were not rerun locally in this pass. Fresh hosted CI is pending.