You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by mi...@apache.org on 2024/09/05 21:28:12 UTC
(impala) 01/02: IMPALA-13043: Implement Join Capability to the Calcite Planner
This is an automated email from the ASF dual-hosted git repository.
michaelsmith pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit e3583cb200881d098f300fe35982b00ab53a9239
Author: Steve Carlin <sc...@cloudera.com>
AuthorDate: Wed Apr 3 19:07:23 2024 -0700
IMPALA-13043: Implement Join Capability to the Calcite Planner
This commit adds the ability to handle joins in the Calcite planner.
Some items worth noting:
There is extra handling in the ImpalaJoinRel class to deal with outer
joins. The AnalyzedTupleIsNullExpr object is needed for processing
which derives from TupleIsNullExpr. Normally, expressions are created
in the CreateExprVisitor, but the join requires that the TupleIsNullExpr
object is wrapped around the expressions retrieved from the inputs.
The execution engine requires separation of the equijoin conditions and
the non-equijoin conditions. Furthermore, the equijoin conditions are
BinaryCompPredicates instead of normal FunctionCallExprs, so the
AnalyzedBinaryCompExpr class had to be created.
Special processing needed to be coded for runtime filter generators.
The conditions needed to be added to the value transfer graph in order
to enable the Impala planner logic to create and push these generators.
The join also required some rules to be added to the optimizer. If a join
is done through the "ON" clause, Calcite is able to place the join
condition directly in the Join RelNode. However, if it is in the "WHERE"
clause, Calcite creates a Filter RelNode and creates a cross join RelNode
object. Therefore, in order to handle "WHERE" joins, we need to implement
the rules in the optimizer.
Change-Id: I5db097577907d79877f52feff2922000af074ecd
Reviewed-on: http://gerrit.cloudera.org:8080/21239
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Reviewed-by: Michael Smith <mi...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
.../calcite/functions/AnalyzedBinaryCompExpr.java | 67 +++
.../functions/AnalyzedFunctionCallExpr.java | 9 +-
.../impala/calcite/functions/FunctionResolver.java | 2 -
.../impala/calcite/functions/RexCallConverter.java | 32 +-
.../calcite/functions/RexLiteralConverter.java | 7 +-
.../calcite/rel/node/ConvertToImpalaRelRules.java | 14 +
.../impala/calcite/rel/node/ImpalaAggRel.java | 4 +-
.../impala/calcite/rel/node/ImpalaJoinRel.java | 465 +++++++++++++++++++++
.../impala/calcite/rel/node/ImpalaPlanRel.java | 1 +
.../impala/calcite/rel/node/ImpalaProjectRel.java | 2 +-
.../impala/calcite/rel/node/ImpalaSortRel.java | 2 +-
.../impala/calcite/rel/node/NodeCreationUtils.java | 4 +-
.../impala/calcite/rel/node/NodeWithExprs.java | 18 +
.../calcite/rel/phys/ImpalaHashJoinNode.java | 54 +++
.../calcite/rel/phys/ImpalaNestedLoopJoinNode.java | 53 +++
.../impala/calcite/service/CalciteOptimizer.java | 99 +++--
.../impala/calcite/type/ImpalaTypeConverter.java | 5 +
.../queries/QueryTest/calcite.test | 106 +++++
18 files changed, 897 insertions(+), 47 deletions(-)
diff --git a/java/calcite-planner/src/main/java/org/apache/impala/calcite/functions/AnalyzedBinaryCompExpr.java b/java/calcite-planner/src/main/java/org/apache/impala/calcite/functions/AnalyzedBinaryCompExpr.java
new file mode 100644
index 000000000..4fa06af6e
--- /dev/null
+++ b/java/calcite-planner/src/main/java/org/apache/impala/calcite/functions/AnalyzedBinaryCompExpr.java
@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.impala.calcite.functions;
+
+import org.apache.impala.analysis.Analyzer;
+import org.apache.impala.analysis.Expr;
+import org.apache.impala.analysis.BinaryPredicate;
+import org.apache.impala.catalog.Function;
+import org.apache.impala.catalog.Type;
+import org.apache.impala.common.AnalysisException;
+
+/**
+ * A BinaryPredicate that is always in analyzed state
+ */
+public class AnalyzedBinaryCompExpr extends BinaryPredicate {
+
+ // Need to save the function because it is known at constructor time. The
+ // resetAnalyzeState() method can be called at various points which could
+ // set the fn_ member to null. So we save the function in the savedFunction_
+ // variable so it can be properly set in analyzeImpl()
+ private final Function savedFunction_;
+
+ public AnalyzedBinaryCompExpr(Function fn, Operator op, Expr e1, Expr e2) {
+ super(op, e1, e2);
+ this.savedFunction_ = fn;
+ this.type_ = Type.BOOLEAN;
+ }
+
+ public AnalyzedBinaryCompExpr(AnalyzedBinaryCompExpr other) {
+ super(other);
+ this.savedFunction_ = other.savedFunction_;
+ }
+
+ @Override
+ public Expr clone() {
+ return new AnalyzedBinaryCompExpr(this);
+ }
+
+ @Override
+ protected void analyzeImpl(Analyzer analyzer) throws AnalysisException {
+ this.computeSelectivity();
+ this.fn_ = savedFunction_;
+ this.type_ = Type.BOOLEAN;
+ }
+
+ @Override
+ protected float computeEvalCost() {
+ // TODO: IMPALA-13098: need to implement
+ return 1;
+ }
+}
diff --git a/java/calcite-planner/src/main/java/org/apache/impala/calcite/functions/AnalyzedFunctionCallExpr.java b/java/calcite-planner/src/main/java/org/apache/impala/calcite/functions/AnalyzedFunctionCallExpr.java
index f40d31e91..2325cb6e5 100644
--- a/java/calcite-planner/src/main/java/org/apache/impala/calcite/functions/AnalyzedFunctionCallExpr.java
+++ b/java/calcite-planner/src/main/java/org/apache/impala/calcite/functions/AnalyzedFunctionCallExpr.java
@@ -17,8 +17,6 @@
package org.apache.impala.calcite.functions;
-import org.apache.calcite.rex.RexCall;
-import org.apache.calcite.sql.SqlKind;
import org.apache.impala.analysis.Analyzer;
import org.apache.impala.analysis.Expr;
import org.apache.impala.analysis.FunctionCallExpr;
@@ -26,7 +24,6 @@ import org.apache.impala.analysis.FunctionParams;
import org.apache.impala.catalog.Function;
import org.apache.impala.catalog.Type;
import org.apache.impala.common.AnalysisException;
-import org.apache.impala.common.ImpalaException;
import java.util.List;
@@ -48,15 +45,13 @@ public class AnalyzedFunctionCallExpr extends FunctionCallExpr {
private final Function savedFunction_;
// c'tor that takes a list of Exprs that eventually get converted to FunctionParams
- public AnalyzedFunctionCallExpr(Function fn, List<Expr> params,
- RexCall rexCall, Type retType) {
+ public AnalyzedFunctionCallExpr(Function fn, List<Expr> params, Type retType) {
super(fn.getFunctionName(), params);
this.savedFunction_ = fn;
this.type_ = retType;
}
- // c'tor which does not depend on Calcite's RexCall but is used when Impala's
- // FunctionParams are created or there is some modifications to it
+ // c'tor when FunctionParams are known
public AnalyzedFunctionCallExpr(Function fn, FunctionParams funcParams,
Type retType) {
super(fn.getFunctionName(), funcParams);
diff --git a/java/calcite-planner/src/main/java/org/apache/impala/calcite/functions/FunctionResolver.java b/java/calcite-planner/src/main/java/org/apache/impala/calcite/functions/FunctionResolver.java
index 3101e94d3..e87062303 100644
--- a/java/calcite-planner/src/main/java/org/apache/impala/calcite/functions/FunctionResolver.java
+++ b/java/calcite-planner/src/main/java/org/apache/impala/calcite/functions/FunctionResolver.java
@@ -24,11 +24,9 @@ import org.apache.calcite.sql.SqlKind;
import org.apache.impala.analysis.FunctionName;
import org.apache.impala.calcite.type.ImpalaTypeConverter;
import org.apache.impala.catalog.BuiltinsDb;
-import org.apache.impala.catalog.Db;
import org.apache.impala.catalog.Function;
import org.apache.impala.catalog.Type;
-import java.util.HashMap;
import java.util.List;
import java.util.Map;
diff --git a/java/calcite-planner/src/main/java/org/apache/impala/calcite/functions/RexCallConverter.java b/java/calcite-planner/src/main/java/org/apache/impala/calcite/functions/RexCallConverter.java
index 0a91fe514..11f513d11 100644
--- a/java/calcite-planner/src/main/java/org/apache/impala/calcite/functions/RexCallConverter.java
+++ b/java/calcite-planner/src/main/java/org/apache/impala/calcite/functions/RexCallConverter.java
@@ -18,22 +18,24 @@
package org.apache.impala.calcite.functions;
import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Lists;
import org.apache.calcite.rel.type.RelDataType;
import org.apache.calcite.rex.RexBuilder;
import org.apache.calcite.rex.RexCall;
import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.impala.analysis.BinaryPredicate;
import org.apache.impala.analysis.Expr;
import org.apache.impala.calcite.type.ImpalaTypeConverter;
import org.apache.impala.catalog.Function;
import org.apache.impala.catalog.Type;
-import org.apache.impala.common.AnalysisException;
-import org.apache.impala.common.ImpalaException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.List;
+import java.util.Map;
/**
* Static Helper class that returns Exprs for RexCall nodes.
@@ -42,6 +44,17 @@ public class RexCallConverter {
protected static final Logger LOG =
LoggerFactory.getLogger(RexCallConverter.class.getName());
+ public static Map<SqlKind, BinaryPredicate.Operator> BINARY_OP_MAP =
+ ImmutableMap.<SqlKind, BinaryPredicate.Operator> builder()
+ .put(SqlKind.EQUALS, BinaryPredicate.Operator.EQ)
+ .put(SqlKind.NOT_EQUALS, BinaryPredicate.Operator.NE)
+ .put(SqlKind.GREATER_THAN, BinaryPredicate.Operator.GT)
+ .put(SqlKind.GREATER_THAN_OR_EQUAL, BinaryPredicate.Operator.GE)
+ .put(SqlKind.LESS_THAN, BinaryPredicate.Operator.LT)
+ .put(SqlKind.LESS_THAN_OR_EQUAL, BinaryPredicate.Operator.LE)
+ .put(SqlKind.IS_DISTINCT_FROM, BinaryPredicate.Operator.DISTINCT_FROM)
+ .put(SqlKind.IS_NOT_DISTINCT_FROM, BinaryPredicate.Operator.NOT_DISTINCT)
+ .build();
/*
* Returns the Impala Expr object for RexCallConverter.
*/
@@ -54,7 +67,12 @@ public class RexCallConverter {
Type impalaRetType = ImpalaTypeConverter.createImpalaType(fn.getReturnType(),
rexCall.getType().getPrecision(), rexCall.getType().getScale());
- return new AnalyzedFunctionCallExpr(fn, params, rexCall, impalaRetType);
+ if (rexCall.isA(SqlKind.BINARY_COMPARISON)) {
+ return createBinaryCompExpr(fn, params, rexCall.getOperator().getKind(),
+ impalaRetType);
+ }
+
+ return new AnalyzedFunctionCallExpr(fn, params, impalaRetType);
}
private static Function getFunction(RexCall call) {
@@ -65,4 +83,12 @@ public class RexCallConverter {
+ "with args " + argTypes + " and return type " + call.getType());
return fn;
}
+
+ private static Expr createBinaryCompExpr(Function fn, List<Expr> params,
+ SqlKind sqlKind, Type retType) {
+ Preconditions.checkArgument(params.size() == 2);
+ BinaryPredicate.Operator op = BINARY_OP_MAP.get(sqlKind);
+ Preconditions.checkNotNull(op, "Unknown Calcite op: " + sqlKind);
+ return new AnalyzedBinaryCompExpr(fn, op, params.get(0), params.get(1));
+ }
}
diff --git a/java/calcite-planner/src/main/java/org/apache/impala/calcite/functions/RexLiteralConverter.java b/java/calcite-planner/src/main/java/org/apache/impala/calcite/functions/RexLiteralConverter.java
index 960e5a21e..626b2af8b 100644
--- a/java/calcite-planner/src/main/java/org/apache/impala/calcite/functions/RexLiteralConverter.java
+++ b/java/calcite-planner/src/main/java/org/apache/impala/calcite/functions/RexLiteralConverter.java
@@ -17,6 +17,7 @@
package org.apache.impala.calcite.functions;
+import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.Lists;
import org.apache.calcite.rel.type.RelDataType;
@@ -34,13 +35,9 @@ import org.apache.impala.catalog.Function;
import org.apache.impala.catalog.PrimitiveType;
import org.apache.impala.catalog.ScalarType;
import org.apache.impala.catalog.Type;
-import org.apache.impala.common.AnalysisException;
-import org.apache.impala.common.ImpalaException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import com.google.common.base.Preconditions;
-
import java.math.BigDecimal;
import java.util.List;
@@ -113,6 +110,6 @@ public class RexLiteralConverter {
List<Expr> argList =
Lists.newArrayList(new StringLiteral(timestamp, Type.STRING, false));
Function castFunc = FunctionResolver.getFunction("casttotimestamp", typeNames);
- return new AnalyzedFunctionCallExpr(castFunc, argList, null, Type.TIMESTAMP);
+ return new AnalyzedFunctionCallExpr(castFunc, argList, Type.TIMESTAMP);
}
}
diff --git a/java/calcite-planner/src/main/java/org/apache/impala/calcite/rel/node/ConvertToImpalaRelRules.java b/java/calcite-planner/src/main/java/org/apache/impala/calcite/rel/node/ConvertToImpalaRelRules.java
index b3c18d29d..349489e90 100644
--- a/java/calcite-planner/src/main/java/org/apache/impala/calcite/rel/node/ConvertToImpalaRelRules.java
+++ b/java/calcite-planner/src/main/java/org/apache/impala/calcite/rel/node/ConvertToImpalaRelRules.java
@@ -21,6 +21,7 @@ import org.apache.calcite.plan.RelOptRule;
import org.apache.calcite.plan.RelOptRuleCall;
import org.apache.calcite.rel.logical.LogicalAggregate;
import org.apache.calcite.rel.logical.LogicalFilter;
+import org.apache.calcite.rel.logical.LogicalJoin;
import org.apache.calcite.rel.logical.LogicalProject;
import org.apache.calcite.rel.logical.LogicalSort;
import org.apache.calcite.rel.logical.LogicalTableScan;
@@ -123,4 +124,17 @@ public class ConvertToImpalaRelRules {
}
}
+ public static class ImpalaJoinRule extends RelOptRule {
+
+ public ImpalaJoinRule() {
+ super(operand(LogicalJoin.class, any()));
+ }
+
+ @Override
+ public void onMatch(RelOptRuleCall call) {
+ final LogicalJoin join = call.rel(0);
+ call.transformTo(new ImpalaJoinRel(join));
+ }
+ }
+
}
diff --git a/java/calcite-planner/src/main/java/org/apache/impala/calcite/rel/node/ImpalaAggRel.java b/java/calcite-planner/src/main/java/org/apache/impala/calcite/rel/node/ImpalaAggRel.java
index f2809ec26..3f4e613a8 100644
--- a/java/calcite-planner/src/main/java/org/apache/impala/calcite/rel/node/ImpalaAggRel.java
+++ b/java/calcite-planner/src/main/java/org/apache/impala/calcite/rel/node/ImpalaAggRel.java
@@ -143,7 +143,7 @@ public class ImpalaAggRel extends Aggregate
aggNode.init(simplifiedAnalyzer);
simplifiedAnalyzer.clearUnassignedConjuncts();
- return new NodeWithExprs(aggNode, outputExprs);
+ return new NodeWithExprs(aggNode, outputExprs, inputWithExprs);
}
private NodeWithExprs getChildPlanNode(ParentPlanRelContext context
@@ -326,7 +326,7 @@ public class ImpalaAggRel extends Aggregate
cardinalityCheckNode.init(ctx.getRootAnalyzer());
- return new NodeWithExprs(cardinalityCheckNode, outputExprs);
+ return new NodeWithExprs(cardinalityCheckNode, outputExprs, inputNodeWithExprs);
}
public Aggregate copy(RelTraitSet relTraitSet, RelNode relNode,
diff --git a/java/calcite-planner/src/main/java/org/apache/impala/calcite/rel/node/ImpalaJoinRel.java b/java/calcite-planner/src/main/java/org/apache/impala/calcite/rel/node/ImpalaJoinRel.java
new file mode 100644
index 000000000..4c90c9bc7
--- /dev/null
+++ b/java/calcite-planner/src/main/java/org/apache/impala/calcite/rel/node/ImpalaJoinRel.java
@@ -0,0 +1,465 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.impala.calcite.rel.node;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Lists;
+import org.apache.calcite.plan.RelOptCluster;
+import org.apache.calcite.plan.RelTraitSet;
+import org.apache.calcite.plan.RelOptUtil;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.core.Join;
+import org.apache.calcite.rel.core.JoinRelType;
+import org.apache.calcite.rex.RexCall;
+import org.apache.calcite.rex.RexInputRef;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexVisitorImpl;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.fun.SqlStdOperatorTable;
+import org.apache.impala.analysis.Analyzer;
+import org.apache.impala.analysis.BinaryPredicate;
+import org.apache.impala.analysis.Expr;
+import org.apache.impala.analysis.JoinOperator;
+import org.apache.impala.analysis.SlotRef;
+import org.apache.impala.analysis.TupleId;
+import org.apache.impala.analysis.TupleIsNullPredicate;
+import org.apache.impala.calcite.functions.AnalyzedFunctionCallExpr;
+import org.apache.impala.calcite.functions.AnalyzedNullLiteral;
+import org.apache.impala.calcite.functions.FunctionResolver;
+import org.apache.impala.calcite.rel.phys.ImpalaHashJoinNode;
+import org.apache.impala.calcite.rel.phys.ImpalaNestedLoopJoinNode;
+import org.apache.impala.calcite.rel.util.CreateExprVisitor;
+import org.apache.impala.calcite.rel.util.ExprConjunctsConverter;
+import org.apache.impala.calcite.type.ImpalaTypeConverter;
+import org.apache.impala.catalog.Function;
+import org.apache.impala.catalog.Type;
+import org.apache.impala.common.AnalysisException;
+import org.apache.impala.common.ImpalaException;
+import org.apache.impala.planner.JoinNode;
+import org.apache.impala.planner.PlanNode;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * ImpalaJoinRel
+ */
+public class ImpalaJoinRel extends Join
+ implements ImpalaPlanRel {
+
+ protected static final Logger LOG =
+ LoggerFactory.getLogger(ImpalaJoinRel.class.getName());
+
+ public ImpalaJoinRel(Join join) {
+ super(join.getCluster(), join.getTraitSet(), join.getLeft(), join.getRight(),
+ join.getCondition(), join.getJoinType(), new HashSet<>());
+ }
+
+ public ImpalaJoinRel(RelOptCluster cluster, RelTraitSet relTraitSet,
+ RelNode leftInput, RelNode rightInput, RexNode condition, JoinRelType joinType) {
+ super(cluster, relTraitSet, leftInput, rightInput, condition, joinType,
+ new HashSet<>());
+ }
+
+ @Override
+ public ImpalaJoinRel copy(RelTraitSet traitSet, RexNode conditionExpr, RelNode left,
+ RelNode right, JoinRelType joinType, boolean semiJoinDone) {
+ return new ImpalaJoinRel(getCluster(), traitSet, left, right, conditionExpr,
+ joinType);
+ }
+
+ @Override
+ public NodeWithExprs getPlanNode(ParentPlanRelContext context) throws ImpalaException {
+
+ NodeWithExprs leftInput = getChildPlanNode(getInput(0), context);
+ NodeWithExprs rightInput = getChildPlanNode(getInput(1), context);
+
+ Analyzer analyzer = context.ctx_.getRootAnalyzer();
+ JoinOperator joinOp = getImpalaJoinOp();
+
+ List<BinaryPredicate> equiJoinConjuncts = new ArrayList<>();
+ List<Expr> nonEquiJoinConjuncts = new ArrayList<>();
+
+ // IMPALA-13176: TODO: Impala allows forcing hints for the distribution mode
+ // - e.g force broadcast or hash partition join. However, we are not
+ // currently supporting hints from the new planner.
+ JoinNode.DistributionMode distMode = JoinNode.DistributionMode.NONE;
+
+ // get the Exprs from both sides. If it's an outer join, some of the
+ // Exprs will be Null wrapped (see comment on method)
+ List<Expr> outputExprs = new ArrayList<>();
+ outputExprs.addAll(getOutputExprs(analyzer, leftInput, true, joinOp));
+ outputExprs.addAll(getOutputExprs(analyzer, rightInput, false, joinOp));
+
+ // Populate equijoin and non-equijoin conditions
+ if (!getCondition().isAlwaysTrue()) {
+ List<ConjunctInfo> conjunctInfos = getConditionConjuncts(getCondition(),
+ leftInput, rightInput, analyzer);
+ for (ConjunctInfo conjunctInfo : conjunctInfos) {
+ if (conjunctInfo.isEquiJoin_) {
+ equiJoinConjuncts.add((BinaryPredicate) conjunctInfo.conjunct_);
+ } else {
+ nonEquiJoinConjuncts.add(conjunctInfo.conjunct_);
+ }
+ }
+ }
+
+ // Populate filter condition
+ List<Expr> filterConjuncts = new ArrayList<>();
+ if (context.filterCondition_ != null) {
+ ExprConjunctsConverter converter = new ExprConjunctsConverter(
+ context.filterCondition_, outputExprs, getCluster().getRexBuilder(), analyzer);
+ filterConjuncts.addAll(converter.getImpalaConjuncts());
+ }
+
+ // Create Impala plan node
+ PlanNode joinNode = equiJoinConjuncts.size() == 0
+ ? new ImpalaNestedLoopJoinNode(context.ctx_.getNextNodeId(), leftInput.planNode_,
+ rightInput.planNode_, false /* not a straight join */, distMode, joinOp,
+ nonEquiJoinConjuncts, filterConjuncts, analyzer)
+ : new ImpalaHashJoinNode(context.ctx_.getNextNodeId(), leftInput.planNode_,
+ rightInput.planNode_, false /* not a straight join */, distMode, joinOp,
+ equiJoinConjuncts, nonEquiJoinConjuncts, filterConjuncts, analyzer);
+
+ if (equiJoinConjuncts.size() > 0) {
+ // register the equi and non-equi join conjuncts with the analyzer such that
+ // value transfer graph creation can consume it
+ List<Expr> equiJoinExprs = new ArrayList<Expr>(equiJoinConjuncts);
+ analyzer.registerConjuncts(getJoinConjunctListToRegister(equiJoinExprs));
+ }
+ analyzer.registerConjuncts(getJoinConjunctListToRegister(nonEquiJoinConjuncts));
+
+ return new NodeWithExprs(joinNode, outputExprs);
+ }
+
+ private NodeWithExprs getChildPlanNode(RelNode relInput,
+ ParentPlanRelContext context) throws ImpalaException {
+ ImpalaPlanRel inputPlanRel = (ImpalaPlanRel) relInput;
+ ParentPlanRelContext.Builder builder =
+ new ParentPlanRelContext.Builder(context, this);
+ builder.setFilterCondition(null);
+ return inputPlanRel.getPlanNode(builder.build());
+ }
+
+
+ /**
+ * Checks and adds null wrapping for expressions fed into the null producing side of an
+ * outer join (hence, both sides for a Full Outer Join).
+ * The null wrapping checks if the tupleId is null and will produce a null row on that
+ * side of the join.
+ * In the case of anti joins and semi joins, no columns are created on one side of the
+ * join.
+ */
+ private List<Expr> getOutputExprs(Analyzer analyzer, NodeWithExprs input,
+ boolean isLeftInput, JoinOperator joinOp) throws ImpalaException {
+
+ if (isLeftInput) {
+ // No expressions needed on the right side on a semi or anti join.
+ if (joinOp == JoinOperator.RIGHT_SEMI_JOIN ||
+ joinOp == JoinOperator.RIGHT_ANTI_JOIN) {
+ return new ArrayList<>();
+ }
+ // No null wrapping is needed if it's not an outer join.
+ if ((joinOp != JoinOperator.RIGHT_OUTER_JOIN &&
+ joinOp != JoinOperator.FULL_OUTER_JOIN)) {
+ return input.outputExprs_;
+ }
+ } else {
+ // No expressions needed on the left side on a semi or anti join.
+ if (joinOp == JoinOperator.LEFT_SEMI_JOIN ||
+ joinOp == JoinOperator.LEFT_ANTI_JOIN) {
+ return new ArrayList<>();
+ }
+ // No null wrapping is needed if it's not an outer join.
+ if ((joinOp != JoinOperator.LEFT_OUTER_JOIN &&
+ joinOp != JoinOperator.FULL_OUTER_JOIN)) {
+ return input.outputExprs_;
+ }
+ }
+
+ // Apply null wrapping if necessary.
+ List<Expr> wrappedExprs = new ArrayList<>();
+ for (Expr expr : input.outputExprs_) {
+ Expr exprToAdd = TupleIsNullPredicate.requiresNullWrapping(expr, analyzer)
+ ? createIfTupleIsNullPredicate(analyzer, expr, input.planNode_.getTupleIds())
+ : expr;
+
+ wrappedExprs.add(exprToAdd);
+ }
+ return wrappedExprs;
+ }
+
+
+ /**
+ * Returns a new conditional expr 'IF(TupleIsNull(tids), NULL, expr)' to
+ * make an input expr nullable. This is especially useful in cases where the Calcite
+ * planner generates a literal TRUE and later does a IS_NULL($x) or IS_NOT_NULL($x)
+ * check on this column - this happens for NOT IN, NOT EXISTS queries where the planner
+ * generates a Left Outer Join and checks the nullability of the column being output
+ * from the right side of the LOJ. Since the literal TRUE is a non-null value coming
+ * into the join but after the join becomes nullable, we add this function to ensure
+ * that happens. Without adding this function the direct translation would be
+ * 'TRUE IS NULL' which is incorrect.
+ */
+ private static Expr createIfTupleIsNullPredicate(Analyzer analyzer, Expr expr,
+ List<TupleId> tupleIds) throws ImpalaException {
+ List<Expr> tmpArgs = new ArrayList<>();
+ TupleIsNullPredicate tupleIsNullExpr = new TupleIsNullPredicate(tupleIds);
+ tupleIsNullExpr.analyze(analyzer);
+ tmpArgs.add(tupleIsNullExpr);
+ // null type needs to be cast to appropriate target type before thrift serialization
+ AnalyzedNullLiteral nullLiteral = new AnalyzedNullLiteral(expr.getType());
+ nullLiteral.analyze(analyzer);
+ tmpArgs.add(nullLiteral);
+ tmpArgs.add(expr);
+ List<Type> typeNames = ImmutableList.of(Type.BOOLEAN, expr.getType(), expr.getType());
+ Function conditionalFunc = FunctionResolver.getFunction("if",
+ ImpalaTypeConverter.getRelDataTypesForArgs(typeNames));
+ Preconditions.checkNotNull(conditionalFunc,
+ "Could not create IF function for arg types %s and return type %s",
+ typeNames, expr.getType());
+ Expr retExpr = new AnalyzedFunctionCallExpr(conditionalFunc, tmpArgs, expr.getType());
+ retExpr.analyze(analyzer);
+ return retExpr;
+ }
+
+ private JoinOperator getImpalaJoinOp() throws ImpalaException {
+ switch (getJoinType()) {
+ case INNER:
+ return JoinOperator.INNER_JOIN;
+ case FULL:
+ return JoinOperator.FULL_OUTER_JOIN;
+ case LEFT:
+ return JoinOperator.LEFT_OUTER_JOIN;
+ case RIGHT:
+ return JoinOperator.RIGHT_OUTER_JOIN;
+ case SEMI:
+ // Calcite SEMI is always a LEFT_SEMI right now.
+ return JoinOperator.LEFT_SEMI_JOIN;
+ case ANTI:
+ return JoinOperator.LEFT_ANTI_JOIN;
+ }
+ throw new AnalysisException("Unsupported join type: " + getJoinType());
+ }
+
+ /**
+ * Get all the join conjuncts that need registering.
+ */
+ public List<Expr> getJoinConjunctListToRegister(List<Expr> joinConjuncts) {
+ List<Expr> conjunctsToRegister = new ArrayList<>();
+ for (Expr joinConjunct : joinConjuncts) {
+ conjunctsToRegister.addAll(getJoinConjunctToRegister(joinConjunct));
+ }
+ return conjunctsToRegister;
+ }
+
+ /**
+ * Given a join conjunct, derive the join conjunct that needs to be registered
+ * with the analyzer.
+ *
+ * The expression produced here will get registered in the Impala value transfer
+ * graph. This value transfer graph is then used to calculate the runtime filter
+ * generators that can be created.
+ *
+ * There are two major differences between the runtime filter generators generated
+ * within the original planner versus the ones generated within this Calcite planner.
+ *
+ * 1) The original planner only seems to generate one runtime filter generator for each
+ * join conjunct. It traces down the SlotRef used on the probe side down to the
+ * scan node. This planner, however, makes liberal use of the value transfer
+ * graph. Once it goes down the probe side, it can apply the runtime filter to
+ * either side of join conjuncts below the current join node.
+ *
+ * 2) Because the original planner only generates one runtime filter generator off of
+ * the given SlotRef, the data type always matches. There is never any casting
+ * adjustment that needs to be done. However, if a lower conjunct within the Calcite
+ * planner is comparing two expressions with different SlotRef types, there will need to
+ * be a casting adjustment. This has not been coded into Impala so this will cause
+ * impalad to crash.
+ *
+ * An example query where the casting adjustment is needed within the Impala test
+ * framework is:
+ *
+ * select straight_join count(*) from functional.alltypessmall a
+ * inner join
+ * (select straight_join t2.* from functional.alltypessmall t1
+ * inner join functional.alltypessmall t2 on t1.tinyint_col = t2.smallint_col) b
+ * on a.int_col = b.smallint_col
+ * inner join
+ * (select distinct tinyint_col from functional.alltypessmall) c
+ * on a.int_col = c.tinyint_col
+ *
+ * Ultimately, it would be better to code the casting adjustment within Impala.
+ * However, this has not been coded in Impala yet, making this a TODO:
+ * Because it hasn't been coded yet within Impala, we do not allow a runtime filter
+ * generator to be generated on SlotRefs that don't match the top level SlotRef.
+ *
+ * The value transfer graph is used to show expressions that are similar. So if we
+ * detect a mismatch of SlotRef types, the full conjunct cannot be entered into
+ * the value transfer graph which would show equivalentce of the SlotRefs. We do still
+ * need to put in the left side of the conjunct into the value transfer graph to show
+ * its existence.
+ *
+ * TL:DR; We only register the full conjunct if the SlotRef types match on both sides
+ * of the join conjunct expression.
+ */
+ public List<Expr> getJoinConjunctToRegister(Expr joinConjunct) {
+ Preconditions.checkState(joinConjunct.getChildren().size() == 2);
+ List<SlotRef> leftSideSlotRefs = new ArrayList<>();
+ List<SlotRef> rightSideSlotRefs = new ArrayList<>();
+ joinConjunct.getChild(0).collect(SlotRef.class, leftSideSlotRefs);
+ joinConjunct.getChild(1).collect(SlotRef.class, rightSideSlotRefs);
+
+ if (leftSideSlotRefs.size() == 0) {
+ return Lists.newArrayList();
+ }
+
+ if (rightSideSlotRefs.size() == 0) {
+ return Lists.newArrayList(joinConjunct.getChild(0));
+ }
+
+ // If there is more than one slotRef on either side of the conjunct, we play it
+ // safe and don't put in the SlotRef equivalency for the full conjunct. Break
+ // the conjunct into a list of its left side and right side.
+ if (leftSideSlotRefs.size() > 1 || rightSideSlotRefs.size() > 1) {
+ return Lists.newArrayList(joinConjunct.getChild(0), joinConjunct.getChild(1));
+ }
+
+ if (leftSideSlotRefs.get(0).getType() != rightSideSlotRefs.get(0).getType()) {
+ return Lists.newArrayList(joinConjunct.getChild(0), joinConjunct.getChild(1));
+ }
+
+ return Lists.newArrayList(joinConjunct);
+ }
+
+ private List<Expr> getAllInputExprs(NodeWithExprs leftInput, NodeWithExprs rightInput) {
+ List<Expr> allExprs = new ArrayList<>();
+ allExprs.addAll(leftInput.outputExprs_);
+ allExprs.addAll(rightInput.outputExprs_);
+ return allExprs;
+ }
+
+ /**
+ * Canonicalize the equijoin condition such that it is
+ * represented as =($M, $N) where M < N. The Impala backend
+ * expects this. If the join condition has expressions
+ * e.g CAST($5, INT) + $4 = CAST($3, INT) + $1
+ * This method will traverse the left and right sides of the
+ * condition and identify the minimum RexInputRef index
+ * on each side.
+ * If the left's min index is >= the number of expressions
+ * on the left side, it will swap the two sides. In the case
+ * that there is no input ref on the left side, it will check
+ * the right side index to see if the expression needs swapping.
+ */
+ private RexNode getCanonical(RexNode conjunct, int numLHSExprs) {
+ if (!conjunct.isA(SqlKind.EQUALS)) {
+ return conjunct;
+ }
+
+ boolean swapExprs = false;
+ MinIndexVisitor visitor = new MinIndexVisitor();
+ RexNode lhsRexCall = ((RexCall) conjunct).getOperands().get(0);
+ RexNode rhsRexCall = ((RexCall) conjunct).getOperands().get(1);
+ // visit the left and right child
+ lhsRexCall.accept(visitor);
+ int minIndex = visitor.getMinIndex();
+ if (minIndex != Integer.MAX_VALUE) {
+ if (minIndex >= numLHSExprs) {
+ swapExprs = true;
+ }
+ } else {
+ visitor.reset();
+ // didn't find an input ref on the left side, check the right side.
+ rhsRexCall.accept(visitor);
+ minIndex = visitor.getMinIndex();
+ Preconditions.checkState(minIndex != Integer.MAX_VALUE);
+ if (minIndex < numLHSExprs) {
+ swapExprs = true;
+ }
+ }
+
+ if (swapExprs) {
+ // swap the left and right children
+ conjunct = getCluster().getRexBuilder().
+ makeCall(SqlStdOperatorTable.EQUALS, rhsRexCall, lhsRexCall);
+ }
+ return conjunct;
+ }
+
+ /**
+ * Break up the given RexNode condition into its AND components and return a list
+ * of the Expr conjuncts (and whether it is an EQUALS conjunct).
+ */
+ private List<ConjunctInfo> getConditionConjuncts(RexNode condition,
+ NodeWithExprs leftInput, NodeWithExprs rightInput, Analyzer analyzer) {
+
+ List<ConjunctInfo> conjunctInfos = new ArrayList<>();
+ CreateExprVisitor visitor = new CreateExprVisitor(getCluster().getRexBuilder(),
+ getAllInputExprs(leftInput, rightInput), analyzer);
+
+ // break up the conjunctions into each AND component
+ List<RexNode> conjuncts = RelOptUtil.conjunctions(getCondition());
+ // convert the conjuncts to Impala Expr
+ for (RexNode conj : conjuncts) {
+ // get a canonicalized representation
+ conj = getCanonical(conj, leftInput.outputExprs_.size());
+ Expr impalaConjunct = conj.accept(visitor);
+ conjunctInfos.add(new ConjunctInfo(impalaConjunct, conj.isA(SqlKind.EQUALS)));
+ }
+ return conjunctInfos;
+ }
+
+ private static class ConjunctInfo {
+ public final Expr conjunct_;
+ public final boolean isEquiJoin_;
+
+ public ConjunctInfo(Expr conjunct, boolean isEquiJoin) {
+ this.conjunct_ = conjunct;
+ this.isEquiJoin_ = isEquiJoin;
+ }
+ }
+
+ private static class MinIndexVisitor extends RexVisitorImpl<Void> {
+ private int minIndex = Integer.MAX_VALUE;
+ public MinIndexVisitor() {
+ super(true);
+ }
+ @Override
+ public Void visitInputRef(RexInputRef inputRef) {
+ minIndex = Math.min(inputRef.getIndex(), minIndex);
+ return null;
+ }
+ public int getMinIndex() {
+ return minIndex;
+ }
+ public void reset() {
+ minIndex = Integer.MAX_VALUE;
+ }
+ }
+
+ @Override
+ public RelNodeType relNodeType() {
+ return RelNodeType.JOIN;
+ }
+}
diff --git a/java/calcite-planner/src/main/java/org/apache/impala/calcite/rel/node/ImpalaPlanRel.java b/java/calcite-planner/src/main/java/org/apache/impala/calcite/rel/node/ImpalaPlanRel.java
index ed1ca9f05..065117805 100644
--- a/java/calcite-planner/src/main/java/org/apache/impala/calcite/rel/node/ImpalaPlanRel.java
+++ b/java/calcite-planner/src/main/java/org/apache/impala/calcite/rel/node/ImpalaPlanRel.java
@@ -36,6 +36,7 @@ public interface ImpalaPlanRel {
AGGREGATE,
FILTER,
HDFSSCAN,
+ JOIN,
PROJECT,
SORT,
UNION,
diff --git a/java/calcite-planner/src/main/java/org/apache/impala/calcite/rel/node/ImpalaProjectRel.java b/java/calcite-planner/src/main/java/org/apache/impala/calcite/rel/node/ImpalaProjectRel.java
index d8f571351..53e5550d9 100644
--- a/java/calcite-planner/src/main/java/org/apache/impala/calcite/rel/node/ImpalaProjectRel.java
+++ b/java/calcite-planner/src/main/java/org/apache/impala/calcite/rel/node/ImpalaProjectRel.java
@@ -76,7 +76,7 @@ public class ImpalaProjectRel extends Project
// There is no Impala Plan Node mapped to Project, so we just return the child
// PlanNode. However, the outputExprs change with the Project.
- return new NodeWithExprs(inputWithExprs.planNode_, outputExprs);
+ return new NodeWithExprs(inputWithExprs.planNode_, outputExprs, inputWithExprs);
}
/**
diff --git a/java/calcite-planner/src/main/java/org/apache/impala/calcite/rel/node/ImpalaSortRel.java b/java/calcite-planner/src/main/java/org/apache/impala/calcite/rel/node/ImpalaSortRel.java
index e4546306f..9808a521b 100644
--- a/java/calcite-planner/src/main/java/org/apache/impala/calcite/rel/node/ImpalaSortRel.java
+++ b/java/calcite-planner/src/main/java/org/apache/impala/calcite/rel/node/ImpalaSortRel.java
@@ -142,7 +142,7 @@ public class ImpalaSortRel extends Sort
inputNodeWithExprs.planNode_, sortInfo, limit_, offset_, limit_ != -1,
context.ctx_.getRootAnalyzer());
- NodeWithExprs retNode = new NodeWithExprs(sortNode, outputExprs);
+ NodeWithExprs retNode = new NodeWithExprs(sortNode, outputExprs, inputNodeWithExprs);
// If there is a filter condition, a SelectNode will get added on top
// of the retNode.
diff --git a/java/calcite-planner/src/main/java/org/apache/impala/calcite/rel/node/NodeCreationUtils.java b/java/calcite-planner/src/main/java/org/apache/impala/calcite/rel/node/NodeCreationUtils.java
index ee5c5ee45..879a3b317 100644
--- a/java/calcite-planner/src/main/java/org/apache/impala/calcite/rel/node/NodeCreationUtils.java
+++ b/java/calcite-planner/src/main/java/org/apache/impala/calcite/rel/node/NodeCreationUtils.java
@@ -54,7 +54,7 @@ public class NodeCreationUtils {
SelectNode selectNode =
SelectNode.createFromCalcite(nodeId, nodeWithExprs.planNode_, filterConjuncts);
selectNode.init(analyzer);
- return new NodeWithExprs(selectNode, nodeWithExprs.outputExprs_);
+ return new NodeWithExprs(selectNode, nodeWithExprs);
}
public static NodeWithExprs createEmptySetPlanNode(PlanNodeId nodeId,
@@ -87,7 +87,7 @@ public class NodeCreationUtils {
unionNode.init(analyzer);
- return new NodeWithExprs(unionNode, outputExprs);
+ return new NodeWithExprs(unionNode, outputExprs, childrenPlanNodes);
}
public static List<Expr> createOutputExprs(List<SlotDescriptor> slotDescs) {
diff --git a/java/calcite-planner/src/main/java/org/apache/impala/calcite/rel/node/NodeWithExprs.java b/java/calcite-planner/src/main/java/org/apache/impala/calcite/rel/node/NodeWithExprs.java
index 88f724479..1c8ac67ac 100644
--- a/java/calcite-planner/src/main/java/org/apache/impala/calcite/rel/node/NodeWithExprs.java
+++ b/java/calcite-planner/src/main/java/org/apache/impala/calcite/rel/node/NodeWithExprs.java
@@ -36,4 +36,22 @@ public class NodeWithExprs {
this.planNode_ = planNode;
this.outputExprs_ = outputExprs;
}
+
+ public NodeWithExprs(PlanNode planNode, List<Expr> outputExprs,
+ List<NodeWithExprs> childrenNodeWithExprs) {
+ this.planNode_ = planNode;
+ this.outputExprs_ = outputExprs;
+ }
+
+ public NodeWithExprs(PlanNode planNode, List<Expr> outputExprs,
+ NodeWithExprs childNodeWithExprs) {
+ this.planNode_ = planNode;
+ this.outputExprs_ = outputExprs;
+ }
+
+ public NodeWithExprs(PlanNode planNode, NodeWithExprs childNodeWithExprs) {
+
+ this.planNode_ = planNode;
+ this.outputExprs_ = childNodeWithExprs.outputExprs_;
+ }
}
diff --git a/java/calcite-planner/src/main/java/org/apache/impala/calcite/rel/phys/ImpalaHashJoinNode.java b/java/calcite-planner/src/main/java/org/apache/impala/calcite/rel/phys/ImpalaHashJoinNode.java
new file mode 100644
index 000000000..8b8363081
--- /dev/null
+++ b/java/calcite-planner/src/main/java/org/apache/impala/calcite/rel/phys/ImpalaHashJoinNode.java
@@ -0,0 +1,54 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.impala.calcite.rel.phys;
+
+import org.apache.impala.analysis.Analyzer;
+import org.apache.impala.analysis.BinaryPredicate;
+import org.apache.impala.analysis.Expr;
+import org.apache.impala.analysis.JoinOperator;
+import org.apache.impala.common.ImpalaException;
+import org.apache.impala.planner.HashJoinNode;
+import org.apache.impala.planner.PlanNode;
+import org.apache.impala.planner.PlanNodeId;
+
+import java.util.List;
+
+/**
+ * ImpalaHashJoinNode: Derived class of HashJoinNode. This is needed
+ * because the HashJoinNode tracks certain conjuncts, but this has already
+ * been handled by Calcite so we need to override them with the Calcite
+ * conjuncts.
+ */
+public class ImpalaHashJoinNode extends HashJoinNode {
+
+ public ImpalaHashJoinNode(PlanNodeId id, PlanNode leftInput, PlanNode rightInput,
+ boolean isStraightJoin, DistributionMode distMode, JoinOperator joinOp,
+ List<BinaryPredicate> equiJoinConjuncts, List<Expr> nonEquijoinConjuncts,
+ List<Expr> filterConjuncts, Analyzer analyzer) throws ImpalaException {
+ super(leftInput, rightInput, isStraightJoin, distMode, joinOp, equiJoinConjuncts,
+ nonEquijoinConjuncts);
+ setId(id);
+ init(analyzer);
+ this.conjuncts_ = filterConjuncts;
+ }
+
+ @Override
+ public void assignConjuncts(Analyzer analyzer) {
+ }
+
+}
diff --git a/java/calcite-planner/src/main/java/org/apache/impala/calcite/rel/phys/ImpalaNestedLoopJoinNode.java b/java/calcite-planner/src/main/java/org/apache/impala/calcite/rel/phys/ImpalaNestedLoopJoinNode.java
new file mode 100644
index 000000000..b177878fb
--- /dev/null
+++ b/java/calcite-planner/src/main/java/org/apache/impala/calcite/rel/phys/ImpalaNestedLoopJoinNode.java
@@ -0,0 +1,53 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.impala.calcite.rel.phys;
+
+import com.google.common.base.Preconditions;
+import org.apache.impala.analysis.Analyzer;
+import org.apache.impala.analysis.Expr;
+import org.apache.impala.analysis.JoinOperator;
+import org.apache.impala.common.ImpalaException;
+import org.apache.impala.planner.NestedLoopJoinNode;
+import org.apache.impala.planner.PlanNode;
+import org.apache.impala.planner.PlanNodeId;
+
+import java.util.List;
+
+/**
+ * ImpalaNestedLoopJoinNode: Derived class of NestedLoopJoinNode. This is needed
+ * because the HashJoinNode tracks certain conjuncts, but this has already
+ * been handled by Calcite so we need to override them with the Calcite
+ * conjuncts.
+ */
+public class ImpalaNestedLoopJoinNode extends NestedLoopJoinNode {
+
+ public ImpalaNestedLoopJoinNode(PlanNodeId id, PlanNode leftInput, PlanNode rightInput,
+ boolean isStraightJoin, DistributionMode distMode, JoinOperator joinOp,
+ List<Expr> joinConjuncts,
+ List<Expr> filterConjuncts, Analyzer analyzer) throws ImpalaException {
+ super(leftInput, rightInput, isStraightJoin, distMode, joinOp, joinConjuncts);
+ setId(id);
+ init(analyzer);
+ this.conjuncts_ = filterConjuncts;
+ }
+
+ @Override
+ public void assignConjuncts(Analyzer analyzer) {
+ }
+
+}
diff --git a/java/calcite-planner/src/main/java/org/apache/impala/calcite/service/CalciteOptimizer.java b/java/calcite-planner/src/main/java/org/apache/impala/calcite/service/CalciteOptimizer.java
index b39c6cad7..f42780430 100644
--- a/java/calcite-planner/src/main/java/org/apache/impala/calcite/service/CalciteOptimizer.java
+++ b/java/calcite-planner/src/main/java/org/apache/impala/calcite/service/CalciteOptimizer.java
@@ -20,16 +20,19 @@ package org.apache.impala.calcite.service;
import com.google.common.collect.ImmutableList;
import org.apache.calcite.plan.RelOptCostImpl;
import org.apache.calcite.plan.RelOptUtil;
+import org.apache.calcite.plan.hep.HepMatchOrder;
import org.apache.calcite.plan.hep.HepPlanner;
+import org.apache.calcite.plan.hep.HepProgram;
import org.apache.calcite.plan.hep.HepProgramBuilder;
import org.apache.calcite.rel.RelNode;
-import org.apache.calcite.rel.metadata.JaninoRelMetadataProvider;
+import org.apache.calcite.rel.core.RelFactories;
+import org.apache.calcite.rel.rules.CoreRules;
import org.apache.calcite.sql.SqlExplainFormat;
import org.apache.calcite.sql.SqlExplainLevel;
+import org.apache.calcite.tools.RelBuilder;
import org.apache.impala.calcite.rel.node.ConvertToImpalaRelRules;
import org.apache.impala.calcite.rel.node.ImpalaPlanRel;
import org.apache.impala.common.ImpalaException;
-import org.apache.impala.common.InternalException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -50,30 +53,78 @@ public class CalciteOptimizer implements CompilerStep {
}
public ImpalaPlanRel optimize(RelNode logPlan) throws ImpalaException {
+ RelBuilder relBuilder = RelFactories.LOGICAL_BUILDER.create(logPlan.getCluster(),
+ validator_.getCatalogReader());
+
+ RelNode optimizedPlan = runJoinProgram(relBuilder, logPlan);
+
+ ImpalaPlanRel finalOptimizedPlan =
+ runImpalaConvertProgram(relBuilder, optimizedPlan);
+
+ return finalOptimizedPlan;
+ }
+
+ /**
+ * Run the rules specifically for join ordering.
+ *
+ */
+ public RelNode runJoinProgram(RelBuilder relBuilder,
+ RelNode plan) throws ImpalaException {
+
HepProgramBuilder builder = new HepProgramBuilder();
- // rules to convert Calcite nodes into ImpalaPlanRel nodes
- builder.addRuleCollection(
- ImmutableList.of(
- new ConvertToImpalaRelRules.ImpalaAggRule(),
- new ConvertToImpalaRelRules.ImpalaScanRule(),
- new ConvertToImpalaRelRules.ImpalaFilterRule(),
- new ConvertToImpalaRelRules.ImpalaSortRule(),
- new ConvertToImpalaRelRules.ImpalaValuesRule(),
- new ConvertToImpalaRelRules.ImpalaUnionRule(),
- new ConvertToImpalaRelRules.ImpalaProjectRule()));
-
- HepPlanner planner = new HepPlanner(builder.build(),
- logPlan.getCluster().getPlanner().getContext(),
- false, null, RelOptCostImpl.FACTORY);
- logPlan.getCluster().setMetadataProvider(JaninoRelMetadataProvider.DEFAULT);
- planner.setRoot(logPlan);
- RelNode optimizedPlan = planner.findBestExp();
- if (!(optimizedPlan instanceof ImpalaPlanRel)) {
- throw new InternalException("Could not generate Impala RelNode plan. Plan " +
- "is \n" + getDebugString(optimizedPlan));
- }
- return (ImpalaPlanRel) optimizedPlan;
+ builder.addMatchOrder(HepMatchOrder.BOTTOM_UP);
+
+ // Merge the filter nodes into the Join. Also include
+ // The filter/project transpose in case the Filter
+ // exists above the Project in the RelNode so it can
+ // then be merged into the Join. The idea is to place
+ // joins next to each other if possible for the join
+ // optimization step.
+ builder.addRuleCollection(ImmutableList.of(
+ CoreRules.FILTER_INTO_JOIN,
+ CoreRules.FILTER_PROJECT_TRANSPOSE
+ ));
+
+ // Join rules work in a two step process. The first step
+ // is to merge all adjacent joins into one big "multijoin"
+ // RelNode (the JOIN_TO_MULTIJOIN rule). Then the
+ // MULTI_JOIN_OPTIMIZE rule is used to determine the join
+ // ordering.
+ builder.addMatchOrder(HepMatchOrder.BOTTOM_UP);
+ builder.addRuleInstance(CoreRules.JOIN_TO_MULTI_JOIN);
+ builder.addRuleInstance(CoreRules.MULTI_JOIN_OPTIMIZE);
+
+ return runProgram(plan, builder.build());
+ }
+
+ public ImpalaPlanRel runImpalaConvertProgram(RelBuilder relBuilder,
+ RelNode plan) throws ImpalaException {
+
+ HepProgramBuilder builder = new HepProgramBuilder();
+
+ builder.addMatchOrder(HepMatchOrder.BOTTOM_UP);
+ builder.addRuleCollection(ImmutableList.of(
+ new ConvertToImpalaRelRules.ImpalaScanRule(),
+ new ConvertToImpalaRelRules.ImpalaSortRule(),
+ new ConvertToImpalaRelRules.ImpalaProjectRule(),
+ new ConvertToImpalaRelRules.ImpalaAggRule(),
+ new ConvertToImpalaRelRules.ImpalaJoinRule(),
+ new ConvertToImpalaRelRules.ImpalaFilterRule(),
+ new ConvertToImpalaRelRules.ImpalaUnionRule(),
+ new ConvertToImpalaRelRules.ImpalaValuesRule()
+ ));
+
+ return (ImpalaPlanRel) runProgram(plan, builder.build());
+ }
+
+ private RelNode runProgram(RelNode currentNode, HepProgram program) {
+ HepPlanner planner = new HepPlanner(program,
+ currentNode.getCluster().getPlanner().getContext(), true, null,
+ RelOptCostImpl.FACTORY);
+ planner.setRoot(currentNode);
+
+ return planner.findBestExp();
}
public String getDebugString(Object optimizedPlan) {
diff --git a/java/calcite-planner/src/main/java/org/apache/impala/calcite/type/ImpalaTypeConverter.java b/java/calcite-planner/src/main/java/org/apache/impala/calcite/type/ImpalaTypeConverter.java
index 0f6573a68..dc559a0fc 100644
--- a/java/calcite-planner/src/main/java/org/apache/impala/calcite/type/ImpalaTypeConverter.java
+++ b/java/calcite-planner/src/main/java/org/apache/impala/calcite/type/ImpalaTypeConverter.java
@@ -336,4 +336,9 @@ public class ImpalaTypeConverter {
Type normalizedImpalaType = getImpalaType(primitiveType);
return impalaToCalciteMap.get(normalizedImpalaType);
}
+
+ // helper function to handle translation of lists.
+ public static List<RelDataType> getRelDataTypesForArgs(List<Type> impalaTypes) {
+ return Lists.transform(impalaTypes, ImpalaTypeConverter::getRelDataType);
+ }
}
diff --git a/testdata/workloads/functional-query/queries/QueryTest/calcite.test b/testdata/workloads/functional-query/queries/QueryTest/calcite.test
index c3b1d42d8..f55856bc7 100644
--- a/testdata/workloads/functional-query/queries/QueryTest/calcite.test
+++ b/testdata/workloads/functional-query/queries/QueryTest/calcite.test
@@ -381,3 +381,109 @@ NULL,10,40
---- TYPES
int, bigint, bigint
====
+---- QUERY
+# having test
+select id, sum(bigint_col) from functional.alltypestiny group by id having sum(bigint_col) > cast(5 as bigint) order by id;
+---- RESULTS
+1,10
+3,10
+5,10
+7,10
+---- TYPES
+int, bigint
+====
+---- QUERY
+# join test
+select a.id, a.bigint_col, b.bigint_col
+from functional.alltypestiny a
+join functional.alltypessmall b
+on (a.id = b.id) order by a.id;
+---- RESULTS
+0,0,0
+1,10,10
+2,0,20
+3,10,30
+4,0,40
+5,10,50
+6,0,60
+7,10,70
+---- TYPES
+int, bigint, bigint
+====
+---- QUERY
+# join inequality test
+select a.id, b.id, a.bigint_col, b.bigint_col
+from functional.alltypestiny a
+join functional.alltypessmall b
+on (b.id < a.id)
+where b.id = 0
+order by a.id;
+---- RESULTS
+1,0,10,0
+2,0,0,0
+3,0,10,0
+4,0,0,0
+5,0,10,0
+6,0,0,0
+7,0,10,0
+---- TYPES
+int, int, bigint, bigint
+====
+---- QUERY
+# cross join test
+select a.id, b.id, a.bigint_col, b.bigint_col
+from functional.alltypestiny a, functional.alltypessmall b
+where b.id < 2
+order by a.id, b.id;
+---- RESULTS
+0,0,0,0
+0,1,0,10
+1,0,10,0
+1,1,10,10
+2,0,0,0
+2,1,0,10
+3,0,10,0
+3,1,10,10
+4,0,0,0
+4,1,0,10
+5,0,10,0
+5,1,10,10
+6,0,0,0
+6,1,0,10
+7,0,10,0
+7,1,10,10
+---- TYPES
+int, int, bigint, bigint
+====
+---- QUERY
+# left outer join test
+select a.id, a.bigint_col, b.bigint_col
+from functional.alltypessmall a
+left outer join functional.alltypestiny b
+on (a.id = b.id)
+where a.id < 20
+order by a.id
+---- RESULTS
+0,0,0
+1,10,10
+2,20,0
+3,30,10
+4,40,0
+5,50,10
+6,60,0
+7,70,10
+8,80,NULL
+9,90,NULL
+10,0,NULL
+11,10,NULL
+12,20,NULL
+13,30,NULL
+14,40,NULL
+15,50,NULL
+16,60,NULL
+17,70,NULL
+18,80,NULL
+19,90,NULL
+---- TYPES
+int, bigint, bigint
+====