Repository navigation
fix: keep a Spark shuffle between two Spark operators under cost-based pricing - #19
Merged
Merged
Conversation
…d pricing With spark.comet.exec.costBasedEngines.enabled, BoundaryFormats prices every shuffle format with EngineCostModel. Between a Spark producer and a Spark consumer the model can price Comet's columnar shuffle (rows to Arrow, Comet write and read, Arrow to rows) below a Spark shuffle once rows are wide enough, so a Comet columnar shuffle was planned where no Comet operator reads or writes it. In spark_etl load_sat_variant_warehouse_destinations_info (union and sort in Spark under a REPARTITION before a parquet write) that shuffle took 2960 s of encoding and writing against 197 s for the Spark shuffle of the same stage. A shuffle between two Spark operators is no longer offered the columnar format; the Spark shuffle, with the same hash function, is always a candidate there, so no co-partitioning mode loses its feasible choices. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Problem
With
spark.comet.exec.costBasedEngines.enabled=true(our prod and thrift config), shuffle formats are chosen byBoundaryFormatswithEngineCostModelprices. Between a Spark producer and a Spark consumer the model prices Comet's columnar shuffle as r2c + Comet write/read + constant + c2r, and for rows of ~15+ leaves that comes out below a Spark shuffle (~79 vs ~93 ns per leaf). So aCometColumnarExchangewas planned with no Comet operator on either side.Real case: spark_etl
platform.load_sat_variant_warehouse_destinations_info(Comet 1.1-8 vs vanilla, same input, 12× c4a-highmem-32-lssd): after the native SMJ a Hive UDF filter, union and sort run in Spark, thenREPARTITION→CometColumnarExchange→ C2R → parquet write. Encoding + writing that shuffle took 2960 s vs 197 s for the vanilla Spark shuffle, plus ~0.8 h more reading it back. The job is 1.12× vanilla task time overall.The narrow
revertRedundantColumnarShuffle(pair of JVM hash aggregates) does not apply on this path.Change
BoundaryFormats.optionsno longer offersColumnarShufflewhen the producer is Spark and the consumer is a Spark operator.SparkShuffle(same Spark hash) is always a candidate there, so co-partitioned decisions keep a feasible choice. Shuffles with a Comet producer or a Comet consumer are unchanged. The non-cost-based path (Comet shuffle only,revertRedundantColumnar) is unchanged.Tests
CostBasedEngineChoiceSuite: new tests (AQE on/off): union + sort in Spark → REPARTITION → INSERT with an artificially huge Spark shuffle price still plans a Spark shuffle; a native producer still writes a Comet shuffle. The first fails without the change (2 failed), passes with it; suite 64/64.GCP measurement on a reduced copy of the job follows in a comment.
🤖 Generated with Claude Code