Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[multistage][bugfix] fix multi-semi-join issues #12038

Merged
merged 3 commits into from
Nov 22, 2023
Merged

[multistage][bugfix] fix multi-semi-join issues #12038

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
61547cf
swap rule order
Nov 21, 2023
eebeaaf
semi should only allow single join key
Nov 21, 2023
288e386
fix tests
Nov 22, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 5 additions & 1 deletion ...y-planner/src/main/java/org/apache/calcite/rel/rules/PinotJoinToDynamicBroadcastRule.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -142,9 +142,13 @@ public boolean matches(RelOptRuleCall call) {
RelNode right = join.getRight() instanceof HepRelVertex ? ((HepRelVertex) join.getRight()).getCurrentRel()
: join.getRight();
return left instanceof Exchange && right instanceof Exchange
// left side can be pushed as dynamic exchange
&& PinotRuleUtils.canPushDynamicBroadcastToLeaf(left.getInput(0))
// default enable dynamic broadcast for SEMI join unless other join strategy were specified
&& (!explicitOtherStrategy && join.getJoinType() == JoinRelType.SEMI && joinInfo.nonEquiConditions.isEmpty());
&& !explicitOtherStrategy
// condition for SEMI join
&& join.getJoinType() == JoinRelType.SEMI && joinInfo.nonEquiConditions.isEmpty()
&& joinInfo.leftKeys.size() == 1;
}

@Override
Expand Down
5 changes: 2 additions & 3 deletions pinot-query-planner/src/main/java/org/apache/calcite/rel/rules/PinotQueryRuleSets.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -50,14 +50,12 @@ private PinotQueryRuleSets() {
CoreRules.PROJECT_SET_OP_TRANSPOSE,

// aggregation and projection rules
CoreRules.AGGREGATE_PROJECT_MERGE, CoreRules.AGGREGATE_PROJECT_PULL_UP_CONSTANTS,
CoreRules.AGGREGATE_PROJECT_PULL_UP_CONSTANTS,
// push a projection past a filter or vice versa
CoreRules.PROJECT_FILTER_TRANSPOSE, CoreRules.FILTER_PROJECT_TRANSPOSE,
// push a projection to the children of a join
// push all expressions to handle the time indicator correctly
CoreRules.JOIN_CONDITION_PUSH,
// merge projections
CoreRules.PROJECT_MERGE,
// remove identity project
CoreRules.PROJECT_REMOVE,

Expand Down Expand Up @@ -102,6 +100,7 @@ private PinotQueryRuleSets() {

// The pruner rules run top-down to ensure Calcite restarts from root node after applying a transformation.
public static final Collection<RelOptRule> PRUNE_RULES = ImmutableList.of(
CoreRules.AGGREGATE_PROJECT_MERGE,
CoreRules.PROJECT_MERGE,
CoreRules.FILTER_MERGE,
CoreRules.AGGREGATE_REMOVE,
Expand Down
49 changes: 21 additions & 28 deletions pinot-query-planner/src/test/resources/queries/AggregatePlans.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,10 +31,9 @@
"\nLogicalProject(avg=[CAST(/(CASE(=($1, 0), null:DECIMAL(1000, 0), $0), $1)):DECIMAL(1000, 0)])",
"\n LogicalAggregate(group=[{}], agg#0=[$SUM0($0)], agg#1=[COUNT($1)])",
"\n PinotLogicalExchange(distribution=[hash])",
"\n LogicalAggregate(group=[{}], agg#0=[$SUM0($2)], agg#1=[COUNT()])",
"\n LogicalProject(col2=[$1], col3=[$2], col4=[$3])",
"\n LogicalFilter(condition=[AND(>=($2, 0), =($1, 'pink floyd'))])",
"\n LogicalTableScan(table=[[a]])",
"\n LogicalAggregate(group=[{}], agg#0=[$SUM0($3)], agg#1=[COUNT()])",
"\n LogicalFilter(condition=[AND(>=($2, 0), =($1, 'pink floyd'))])",
"\n LogicalTableScan(table=[[a]])",
"\n"
]
},
Expand All @@ -46,10 +45,9 @@
"\nLogicalProject(avg=[CAST(/(CASE(=($1, 0), null:DECIMAL(1000, 0), $0), $1)):DECIMAL(1000, 0)], sum=[CASE(=($1, 0), null:DECIMAL(1000, 0), $0)], max=[$2])",
"\n LogicalAggregate(group=[{}], agg#0=[$SUM0($0)], agg#1=[COUNT($1)], agg#2=[MAX($2)])",
"\n PinotLogicalExchange(distribution=[hash])",
"\n LogicalAggregate(group=[{}], agg#0=[$SUM0($2)], agg#1=[COUNT()], agg#2=[MAX($2)])",
"\n LogicalProject(col2=[$1], col3=[$2], col4=[$3])",
"\n LogicalFilter(condition=[AND(>=($2, 0), =($1, 'pink floyd'))])",
"\n LogicalTableScan(table=[[a]])",
"\n LogicalAggregate(group=[{}], agg#0=[$SUM0($3)], agg#1=[COUNT()], agg#2=[MAX($3)])",
"\n LogicalFilter(condition=[AND(>=($2, 0), =($1, 'pink floyd'))])",
"\n LogicalTableScan(table=[[a]])",
"\n"
]
},
Expand All @@ -61,10 +59,9 @@
"\nLogicalProject(avg=[/(CAST(CASE(=($1, 0), null:BIGINT, $0)):DOUBLE, $1)], count=[$1])",
"\n LogicalAggregate(group=[{}], agg#0=[$SUM0($0)], agg#1=[COUNT($1)])",
"\n PinotLogicalExchange(distribution=[hash])",
"\n LogicalAggregate(group=[{}], agg#0=[$SUM0($1)], agg#1=[COUNT()])",
"\n LogicalProject(col2=[$1], col3=[$2])",
"\n LogicalFilter(condition=[AND(>=($2, 0), =($1, 'pink floyd'))])",
"\n LogicalTableScan(table=[[a]])",
"\n LogicalAggregate(group=[{}], agg#0=[$SUM0($2)], agg#1=[COUNT()])",
"\n LogicalFilter(condition=[AND(>=($2, 0), =($1, 'pink floyd'))])",
"\n LogicalTableScan(table=[[a]])",
"\n"
]
},
Expand All @@ -89,10 +86,9 @@
"\nLogicalProject(EXPR$0=[CASE(=($1, 0), null:BIGINT, $0)], EXPR$1=[$1])",
"\n LogicalAggregate(group=[{}], agg#0=[$SUM0($0)], agg#1=[COUNT($1)])",
"\n PinotLogicalExchange(distribution=[hash])",
"\n LogicalAggregate(group=[{}], agg#0=[$SUM0($1)], agg#1=[COUNT()])",
"\n LogicalProject(col2=[$1], col3=[$2])",
"\n LogicalFilter(condition=[AND(>=($2, 0), =($1, 'a'))])",
"\n LogicalTableScan(table=[[a]])",
"\n LogicalAggregate(group=[{}], agg#0=[$SUM0($2)], agg#1=[COUNT()])",
"\n LogicalFilter(condition=[AND(>=($2, 0), =($1, 'a'))])",
"\n LogicalTableScan(table=[[a]])",
"\n"
]
},
Expand All @@ -104,10 +100,9 @@
"\nLogicalProject(sum=[CASE(=($1, 0), null:BIGINT, $0)], count=[$1])",
"\n LogicalAggregate(group=[{}], agg#0=[$SUM0($0)], agg#1=[COUNT($1)])",
"\n PinotLogicalExchange(distribution=[hash])",
"\n LogicalAggregate(group=[{}], agg#0=[$SUM0($1)], agg#1=[COUNT()])",
"\n LogicalProject(col2=[$1], col3=[$2])",
"\n LogicalFilter(condition=[AND(>=($2, 0), =($1, 'pink floyd'))])",
"\n LogicalTableScan(table=[[a]])",
"\n LogicalAggregate(group=[{}], agg#0=[$SUM0($2)], agg#1=[COUNT()])",
"\n LogicalFilter(condition=[AND(>=($2, 0), =($1, 'pink floyd'))])",
"\n LogicalTableScan(table=[[a]])",
"\n"
]
},
Expand All @@ -119,10 +114,9 @@
"\nLogicalProject(avg=[/(CAST(CASE(=($1, 0), null:BIGINT, $0)):DOUBLE, $1)], count=[$1])",
"\n LogicalAggregate(group=[{}], agg#0=[$SUM0($0)], agg#1=[COUNT($1)])",
"\n PinotLogicalExchange(distribution=[hash])",
"\n LogicalAggregate(group=[{}], agg#0=[$SUM0($1)], agg#1=[COUNT()])",
"\n LogicalProject(col2=[$1], col3=[$2])",
"\n LogicalFilter(condition=[AND(>=($2, 0), =($1, 'pink floyd'))])",
"\n LogicalTableScan(table=[[a]])",
"\n LogicalAggregate(group=[{}], agg#0=[$SUM0($2)], agg#1=[COUNT()])",
"\n LogicalFilter(condition=[AND(>=($2, 0), =($1, 'pink floyd'))])",
"\n LogicalTableScan(table=[[a]])",
"\n"
]
},
Expand All @@ -134,10 +128,9 @@
"\nLogicalProject(sum=[CASE(=($1, 0), null:BIGINT, $0)], count=[$1])",
"\n LogicalAggregate(group=[{}], agg#0=[$SUM0($0)], agg#1=[COUNT($1)])",
"\n PinotLogicalExchange(distribution=[hash])",
"\n LogicalAggregate(group=[{}], agg#0=[$SUM0($1)], agg#1=[COUNT()])",
"\n LogicalProject(col2=[$1], col3=[$2])",
"\n LogicalFilter(condition=[AND(>=($2, 0), =($1, 'pink floyd'))])",
"\n LogicalTableScan(table=[[a]])",
"\n LogicalAggregate(group=[{}], agg#0=[$SUM0($2)], agg#1=[COUNT()])",
"\n LogicalFilter(condition=[AND(>=($2, 0), =($1, 'pink floyd'))])",
"\n LogicalTableScan(table=[[a]])",
"\n"
]
}
Expand Down
36 changes: 18 additions & 18 deletions pinot-query-planner/src/test/resources/queries/ExplainPhysicalPlans.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -337,15 +337,15 @@
" └── [2]@localhost:1|[1] MAIL_RECEIVE(BROADCAST_DISTRIBUTED)\n",
" ├── [3]@localhost:1|[1] MAIL_SEND(BROADCAST_DISTRIBUTED)->{[2]@localhost:1|[0, 1],[2]@localhost:2|[2, 3]} (Subtree Omitted)\n",
" └── [3]@localhost:2|[0] MAIL_SEND(BROADCAST_DISTRIBUTED)->{[2]@localhost:1|[0, 1],[2]@localhost:2|[2, 3]}\n",
" └── [3]@localhost:2|[0] FILTER\n",
" └── [3]@localhost:2|[0] AGGREGATE_FINAL\n",
" └── [3]@localhost:2|[0] MAIL_RECEIVE(HASH_DISTRIBUTED)\n",
" ├── [4]@localhost:2|[2] MAIL_SEND(HASH_DISTRIBUTED)->{[3]@localhost:1|[1],[3]@localhost:2|[0]} (Subtree Omitted)\n",
" ├── [4]@localhost:2|[3] MAIL_SEND(HASH_DISTRIBUTED)->{[3]@localhost:1|[1],[3]@localhost:2|[0]} (Subtree Omitted)\n",
" ├── [4]@localhost:1|[0] MAIL_SEND(HASH_DISTRIBUTED)->{[3]@localhost:1|[1],[3]@localhost:2|[0]} (Subtree Omitted)\n",
" └── [4]@localhost:1|[1] MAIL_SEND(HASH_DISTRIBUTED)->{[3]@localhost:1|[1],[3]@localhost:2|[0]}\n",
" └── [4]@localhost:1|[1] AGGREGATE_LEAF\n",
" └── [4]@localhost:1|[1] PROJECT\n",
" └── [3]@localhost:2|[0] PROJECT\n",
" └── [3]@localhost:2|[0] FILTER\n",
" └── [3]@localhost:2|[0] AGGREGATE_FINAL\n",
" └── [3]@localhost:2|[0] MAIL_RECEIVE(HASH_DISTRIBUTED)\n",
" ├── [4]@localhost:2|[2] MAIL_SEND(HASH_DISTRIBUTED)->{[3]@localhost:1|[1],[3]@localhost:2|[0]} (Subtree Omitted)\n",
" ├── [4]@localhost:2|[3] MAIL_SEND(HASH_DISTRIBUTED)->{[3]@localhost:1|[1],[3]@localhost:2|[0]} (Subtree Omitted)\n",
" ├── [4]@localhost:1|[0] MAIL_SEND(HASH_DISTRIBUTED)->{[3]@localhost:1|[1],[3]@localhost:2|[0]} (Subtree Omitted)\n",
" └── [4]@localhost:1|[1] MAIL_SEND(HASH_DISTRIBUTED)->{[3]@localhost:1|[1],[3]@localhost:2|[0]}\n",
" └── [4]@localhost:1|[1] AGGREGATE_LEAF\n",
" └── [4]@localhost:1|[1] FILTER\n",
" └── [4]@localhost:1|[1] TABLE SCAN (b) null\n",
""
Expand Down Expand Up @@ -373,15 +373,15 @@
" ├── [3]@localhost:2|[3] MAIL_SEND(BROADCAST_DISTRIBUTED)->{[2]@localhost:1|[0, 1],[2]@localhost:2|[2, 3]} (Subtree Omitted)\n",
" ├── [3]@localhost:1|[0] MAIL_SEND(BROADCAST_DISTRIBUTED)->{[2]@localhost:1|[0, 1],[2]@localhost:2|[2, 3]} (Subtree Omitted)\n",
" └── [3]@localhost:1|[1] MAIL_SEND(BROADCAST_DISTRIBUTED)->{[2]@localhost:1|[0, 1],[2]@localhost:2|[2, 3]}\n",
" └── [3]@localhost:1|[1] FILTER\n",
" └── [3]@localhost:1|[1] AGGREGATE_FINAL\n",
" └── [3]@localhost:1|[1] MAIL_RECEIVE(HASH_DISTRIBUTED)\n",
" ├── [4]@localhost:2|[2] MAIL_SEND(HASH_DISTRIBUTED)[PARTITIONED]->{[3]@localhost:2|[2]} (Subtree Omitted)\n",
" ├── [4]@localhost:2|[3] MAIL_SEND(HASH_DISTRIBUTED)[PARTITIONED]->{[3]@localhost:2|[3]} (Subtree Omitted)\n",
" ├── [4]@localhost:1|[0] MAIL_SEND(HASH_DISTRIBUTED)[PARTITIONED]->{[3]@localhost:1|[0]} (Subtree Omitted)\n",
" └── [4]@localhost:1|[1] MAIL_SEND(HASH_DISTRIBUTED)[PARTITIONED]->{[3]@localhost:1|[1]}\n",
" └── [4]@localhost:1|[1] AGGREGATE_LEAF\n",
" └── [4]@localhost:1|[1] PROJECT\n",
" └── [3]@localhost:1|[1] PROJECT\n",
" └── [3]@localhost:1|[1] FILTER\n",
" └── [3]@localhost:1|[1] AGGREGATE_FINAL\n",
" └── [3]@localhost:1|[1] MAIL_RECEIVE(HASH_DISTRIBUTED)\n",
" ├── [4]@localhost:2|[2] MAIL_SEND(HASH_DISTRIBUTED)[PARTITIONED]->{[3]@localhost:2|[2]} (Subtree Omitted)\n",
" ├── [4]@localhost:2|[3] MAIL_SEND(HASH_DISTRIBUTED)[PARTITIONED]->{[3]@localhost:2|[3]} (Subtree Omitted)\n",
" ├── [4]@localhost:1|[0] MAIL_SEND(HASH_DISTRIBUTED)[PARTITIONED]->{[3]@localhost:1|[0]} (Subtree Omitted)\n",
" └── [4]@localhost:1|[1] MAIL_SEND(HASH_DISTRIBUTED)[PARTITIONED]->{[3]@localhost:1|[1]}\n",
" └── [4]@localhost:1|[1] AGGREGATE_LEAF\n",
" └── [4]@localhost:1|[1] FILTER\n",
" └── [4]@localhost:1|[1] TABLE SCAN (b) null\n",
""
Expand Down
Loading