andygrove opened a new issue, #2759: URL: https://github.com/apache/datafusion-ballista/issues/2759
> **This is a research task.** Spark AQE is the reference point for the problem, not for the solution. Ballista has different constraints than Spark (DataFusion operators, stage-at-a-time replanning, no external shuffle service, partition-count-sensitive operators), so we may not want to copy how Spark does this. The first deliverable is a write-up of whether and how to do it here, ideally with a measurement showing the gain, before any implementation. ## Problem The skew-join rule (#2714, `OptimizeSkewedJoinRule`) is a good port of Spark's `OptimizeSkewedJoin`, but its applicability is narrow compared with Spark's, and it is opt-in: - **It is skipped when join dynamic-filter pushdown is on.** DataFusion enables that by default, and the rule returns early if `enable_join_dynamic_filter_pushdown` is set, because partition-routed dynamic filters assume `hash(key) % partition_count` and shards break that. Users have to turn off `datafusion.optimizer.enable_dynamic_filter_pushdown` to get skew handling, giving up dynamic filters for the whole session. The rule is a silent no-op otherwise, only visible in debug logs. - **One partitioned join per stage.** Chains with a second join, final aggregates, windows or limits are skipped. Spark handles several joins in one stage by accepting extra shuffles under `forceOptimizeSkewedJoin`. - **Opt-in** (`ballista.planner.skew_join.enabled=false`). There is no stated criterion for when it would be safe to default on. ## Questions to answer - Can skewed shards coexist with dynamic filters, for example by disabling the partition-routed form of the filter only for the affected join, or by routing filters per shard? - Should the rule log or surface (metrics, plan annotation) when it declines because of dynamic filters, so users can tell it did not run? - Which multi-join and aggregate shapes can be split safely, and is it worth the complexity? The real question is how many production queries have skew in a stage the rule currently rejects. - What would justify enabling it by default? For instance, a no-op cost when no partition is skewed, plus a TPC-DS run with injected skew. - Spark only splits for sort-merge joins. Ballista also splits hash joins. Are there memory or build-side implications of replicating a build bucket across shards that Spark does not face? ## Related - #2714 (implementation), #1643 (skew from low-cardinality keys), #2700 (runtime filters) -- 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]
