Reuse existing semi-join domains - #23112
Closed
pentschev wants to merge 67 commits into
Closed
Conversation
Add generic dynamic-planning support for join key prefilters in the streaming actor graph. The planner evaluates join type, key compatibility, size estimates, and configured selectivity thresholds to decide when a small side can build a bloom/key prefilter for the larger side before shuffle. The implementation supports prefix key selection for multi-key joins, records structured trace metadata and skip reasons, preserves the original full join after the row-reduction stage, and exposes conservative dynamic-planning options for enabling, sizing, and tracing the prefilter path.
Add a streaming optimizer pass that inserts generic derived key-domain semi joins before actor-graph lowering. The pass uses existing dynamic-planning scan statistics and join metadata to reduce large join inputs from selective domains while preserving the original full join for correctness. The optimizer handles simple selective-domain filters and constrained multi-key domains, including cases where a selective key on one side can narrow the domain used to prefilter a larger source. It skips unsupported join shapes, non-column keys, unselective simple domains, and non-inner joins. Wire the pass into streaming execution behind dynamic-planning options and add focused tests plus config coverage.
Teach the derived join-domain prefilter planner to estimate the source-scan cost needed to build candidate domains. Candidate selection now prefers cheaper domain producers before smaller estimated output rows, and rejects prefilters whose domain-build cost is too large relative to the target being reduced. Also prevent a single-source domain from being stacked onto a target source that is already below the filtered side of a semi join. This keeps the profitable Q9 domain choices while avoiding the Q8-shaped memory regression caused by adding a second high-cardinality source-only prefilter to an already-reduced lineitem scan. Extend trace metadata with estimated target, domain, and constraint costs, and add regression tests for both the Q9 profitability case and the Q8 stacked-prefilter case.
Reject boolean values for join_prefilter_max_key_columns instead of accepting them through Python's bool-is-int relationship. Add explicit config validation coverage so only None and positive integer limits remain valid.
Use a single typed helper for optional environment-value parsing and define the optional float and int converters in terms of it. This keeps the None/null handling in one place without changing accepted values.
Make join_prefilter_threshold the single dynamic-planning threshold for join prefilters, preserving the existing 0.5 default. Remove the obsolete bloom_filter_threshold option, its fallback/override tests, and the unused use_bloom_filter helper now replaced by _select_join_prefilter.
Allow numeric integer values such as 0 for join_prefilter_threshold and normalize the stored value to float during DynamicPlanningOptions validation. Reject booleans explicitly and add config coverage for the documented disable value.
Co-authored-by: Lawrence Mitchell <wence@gmx.li>
Clarify that join_prefilter_max_key_columns controls the size of the join-key prefix used by the prefilter, rather than selecting an arbitrary key subset.
Replace the hand-written JoinPrefilterDecision trace metadata mapping with dataclasses.asdict, so the trace output follows the dataclass fields without duplicating the field list.
Drop the always-true JoinPrefilterDecision.considered field since the presence of a decision already means the prefilter was considered. Inline skipped JoinPrefilterDecision construction at the return sites so the selector does not carry a helper that only forwards dataclass arguments.
Drop the no_join_keys branch from the join prefilter selector. Keyless cross joins are already unsupported by the selector, so cover that behavior directly instead of carrying a separate skip reason.
Replace the defensive mismatched-key skip reason with an assertion. A valid Join IR must provide the same number of left and right join keys, so reaching this state indicates malformed join metadata rather than a prefilter planning decision.
Remove the redundant build_side field from JoinPrefilterDecision. The bloom-filter build side is the inverse of filter_side, so task construction now derives that relationship from the side being filtered.
Drop apply_side_prepartitioned from join prefilter trace metadata. The prefilter does not make adaptive decisions from this value and the information is not consumed elsewhere, so keeping it in the filter trace adds noise without affecting planning.
Keep the dynamic-join reserved collective count at four, but remove the inaccurate strategy-allgather wording from the comment and error message. The four reserved IDs are the allgather, left shuffle, right shuffle, and bloom filter.
Clarify that ActorTracer extra metadata is for nested runtime decisions that do not have their own IR node but should still be logged with the parent actor trace.
Cast the join_prefilter_max_key_columns default to int | None so _make_default_factory infers the same optional type as _optional_int_converter. This avoids mypy narrowing the default to int and rejecting the converter.
Represent simple join keys directly as expr.Col nodes instead of duplicating their names and dtypes in a private wrapper. Rely on valid Join IR to bind each column to its input schema, while retaining explicit handling for non-column key expressions.
Rewrite join nodes with CachingVisitor and singledispatch, carry analysis through explicit visitor state, and compute row estimates and selectivity in post-order traversals. Replace target nodes with the shared DAG replacement helper, eliminating module-global caches and hand-written recursive traversal.
Restore the optional float environment converter used by the join-domain prefilter threshold after the join-prefilter branch merge removed the legacy definition. This keeps the documented numeric and null environment values valid and restores static-checking correctness.
Refresh row-count and selectivity analysis when bottom-up rewriting reconstructs a join subtree. This lets parent joins rank domains using newly inserted semi joins, preserving derived-filter propagation for Q5-like plans and avoiding harmful stacked filters for Q8-like plans. Add regression coverage for both behaviors.
Evaluate local row-ratio guards before identity checks and additional producer searches so unprofitable candidates avoid unnecessary subgraph traversals.
Build the selected domain and target semi join directly in the join rewrite, removing the single-use _make_target_filter wrapper.
Rename the key projection helper around its binding-aware purpose, pass the join-visible column expression explicitly, and assert that its dtype matches the producer-bound source column.
Move the logical domain-prefilter controls into a dedicated executor option group and run the rewrite independently of dynamic shuffle planning. This keeps static planning eligible for the same logical row reduction and gives the rewrite its own environment-variable namespace.
Exercise the domain-prefilter rewrite with nullable join keys for both null-equality modes. The regression verifies that inserted semi joins inherit the original join semantics and that optimized GPU execution matches Polars CPU results.
Reject unsupported join-domain prefilter option values during streaming executor construction so invalid configuration cannot fail later in logical optimization.
Use None as the join-domain prefilter disable sentinel, matching dynamic planning. Preserve enabled defaults, add top-level environment disabling, and cover option propagation and rewrite bypass behavior.
Shut down the default singleton created by the nullable-key execution test so later explicit engine fixtures can initialize. Register the join-domain options class in the Sphinx API and options references.
…n-domain-prefilter-guards
Discover selective key domains that are already attached to a semi join and
reuse them to prefilter another large join input. Follow exact column bindings
and shared source provenance, reject uncached source duplication, and retain
the original full join for final correctness.
For SF30K Q18, this inserts an early key-domain reduction on the second
lineitem input. The traced six-node run improved from 18.35s/10.80s to
11.51s/11.89s, and the untraced first iteration improved to 9.82s. This
materially reduces memory pressure, although the raw detail payload remains
large enough that a later hot iteration can still exhaust six nodes.
Q18 plan before, relevant subtree:
JOIN Inner (o_orderkey) (l_orderkey)
JOIN Semi (o_orderkey) (l_orderkey)
STREAMINGSCAN orders
PROJECTION selected_l_orderkey
FILTER sum_quantity > 300
GROUPBY lineitem by l_orderkey
CACHE lineitem (l_orderkey, l_quantity)
Q18 plan after, relevant subtree:
JOIN Inner (o_orderkey) (l_orderkey)
JOIN Semi (o_orderkey) (l_orderkey)
STREAMINGSCAN orders
PROJECTION selected_l_orderkey
FILTER sum_quantity > 300
GROUPBY lineitem by l_orderkey
JOIN Semi (l_orderkey) (l_orderkey)
CACHE lineitem (l_orderkey, l_quantity)
PROJECTION selected_l_orderkey
FILTER sum_quantity > 300
GROUPBY lineitem by l_orderkey
pentschev
added
0 - Blocked
|
Auto-sync is disabled for draft pull requests in this repository. Workflows must be run manually. Contributors can view more details about this message here. |
Contributor
Author
|
This PR is not necessary anymore. Changes in #23113 are sufficient for all queries to pass on 6xNVL4 nodes at SF30K. |
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
3 participants
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.
Discover selective key domains that are already attached to a semi-join and reuse them to prefilter another large join input. Follow exact column bindings and shared source provenance, reject uncached source duplication, and retain the original full join for final correctness.
For SF30K Q18, this inserts an early key-domain reduction on the second lineitem input. The traced 6xNVL4 run improved from 18.35s/10.80s to 11.51s/11.89s, and the untraced first iteration improved to 9.82s. This materially reduces memory pressure, although the raw detail payload remains large enough that a later hot iteration can still exhaust six nodes.
Q18 plan before, relevant subtree
Q18 plan after, relevant subtree
Note for reviewers: this PR builds on changes from #22997, the only novel change is 7f9d2b2 .