andygrove opened a new pull request, #2434: URL: https://github.com/apache/datafusion-ballista/pull/2434
# Which issue does this PR close? Closes #. Follows up on the q8 analysis in #2419: https://github.com/apache/datafusion-ballista/pull/2419#issuecomment-5591677739 (items 2 and 3). No issue filed yet, happy to open one if this direction looks right. # Rationale for this change TPC-H q8 at SF1000 takes 164s, and 131s of that is a single stage: `SortShuffleWriterExec Hash([l_partkey], 256)` over the raw lineitem scan. All 6B rows of lineitem are shuffled across the network in order to join against 1,332,379 rows of filtered `part`, which is one `p_partkey` i64 column, roughly 10.7 MB, about 12x under the 128 MiB broadcast budget. That shuffle should not exist. It exists because AQE decides the join before it can measure anything. The build side is `FilterExec(p_type = 'ECONOMY ANODIZED STEEL')` over `part`, and with no runtime statistics the planner falls back to `default_filter_selectivity` (20%). The resulting estimate sits over the broadcast budget, so `to_actual_join` picks `PartitionMode::Partitioned` and emits `JoinSelectionAction::Repartition`, which wraps **both** children in an `ExchangeExec`. The 131s is committed at that point. Worse, it is unrecoverable. Stage 0 finishes in 1.4s and reveals the true size, but by then `selection_state == Repartitioned`, so the only arms `to_actual_join` can still reach are Hash and SortMerge. There is no path back to `CollectLeft`. The same shape shows up in q9 s1 (58s), and the estimate/measurement gap is not specific to q8: any filtered dimension scan feeding a fact-table join hits it. # What changes are included in this PR? **1. Stage the build side before deciding (`JoinSelectionAction::StageBuildSide`).** A new action that shuffles only the prospective build side, on the join key, and leaves the probe side untouched. `selection_state` stays `Unknown`, so when `upstream_resolved()` goes true after that one cheap stage, `to_actual_join` runs again — this time against a resolved `ExchangeExec`, whose `partition_statistics` reports `Precision::Exact` row and byte counts read back from the shuffle's `PartitionStats`. For q8 that is 1.33M rows and a few MB, which promotes the join to `CollectLeft`. The probe side is then scanned once inside the join stage against a broadcast build side, and stages 1 and 2 (142s combined) collapse into roughly one 12s stage. `ExchangeExec::to_broadcast` shares the `stage_id` and `shuffle_partitions` `Arc`s, so the staged shuffle is *read* as a broadcast rather than re-run. `should_stage_build_side` gates this on three conditions, each necessary: - the build estimate is `Inexact`. An exact size over budget is a fact, not a guess, so measuring it again cannot flip the join and would only serialise two shuffles that could run concurrently. - the estimate is at most 32x over the byte budget. Past that, no measurement brings it back under. q8's `part` scan estimates ~10x over and measures three orders of magnitude under. - the probe side is at least 10x the build side in bytes. Staging serialises the two shuffles, so the deferral costs at most the *smaller* side's runtime; the ratio is what bounds the cost of being wrong. It fires at most once per join, guarded on neither child already being an `ExchangeExec` — once it fires, one child is, so the guard can never pass again. When the measured size then turns out to be too large to broadcast, the join falls through to `Repartition`, which now reuses an exchange already shuffling on the join key (`exchange_on`) instead of nesting a second stage boundary inside the first. Behind `ballista.optimizer.stage_build_side`, default `true`. The default is the part I would most like opinions on: on is what makes a benchmark run exercise it, but it is a real change to planner behaviour. **2. `broadcast_join_threshold_rows` becomes a fallback rather than a ceiling.** `supports_collect_by_thresholds` applied the row threshold *before* the schema-derived byte estimate and rejected on either, so a build side comfortably inside the byte budget could still be vetoed on row count alone. A 1M-row `i64` key column is 8 MB against a 128 MiB default budget, 16x under. The config already documents the row threshold as "a fallback when byte-size statistics are unavailable", and #2084's description says the same ("keeping DataFusion's row threshold as the absent-stats fallback"), but the code and its tests applied it as an additional ceiling. This makes the code match that stated intent: when a byte estimate can be formed it decides, and the row rule applies only when it cannot. `0` still disables the whole row-count path. To be clear about scope, this is a deliberate behaviour change, not a bug fix — `row_threshold_remains_a_ceiling` asserted the old semantics on purpose and is replaced here. It is also **not** what fixes q8 on its own: at pre-execution q8's build side has an `Inexact` `total_byte_size`, so the byte branch short-circuits and the row path is never reached. It matters for build sides whose byte statistics are absent, which the code comment notes is the common case rather than the exception (DataFusion discards `total_byte_size` on every join). **Tests.** Unit coverage for `should_stage_build_side` across all three conditions plus the disabled case; `to_actual_join` returning `StageBuildSide` for the q8 shape, `Repartition` when the config is off, and `Repartition` rather than a second staging once a child is an exchange; `exchange_on` reuse, non-reuse on a different key, and the plain-child case; and the reworked threshold tests. Not verified end to end on a cluster. The next step is an SF1000 run to confirm q8 actually drops, which I can do — treating this as a draft until then. ## Known follow-ups, not addressed here - The staged side is written as a hash shuffle into `target_partitions`, so the broadcast read pulls all of them per probe task. Correct, and cheap at 1.33M rows, but an unpartitioned write would be better. - The measured `total_byte_size` is compressed shuffle-file bytes, which understates the in-memory build size. Pre-existing: `hash_join_max_build_partition_bytes` already reads the same figure. # Are there any user-facing changes? No API changes. Two behaviour changes, both configurable: - New config key `ballista.optimizer.stage_build_side` (Boolean, default `true`). With it on, a plan may gain one extra stage boundary and lose a much larger one; stage counts and the plan shape for affected joins change. - `ballista.optimizer.broadcast_join_threshold_rows` no longer vetoes a build side whose byte estimate is inside `broadcast_join_threshold_bytes`. Anyone relying on it as a hard row cap will see more joins broadcast. Setting it to `0` still disables the row-count path entirely, as documented. Generated config docs regenerated via `./dev/update_config_docs.sh`. -- 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]
