IGNITE-23799 : Calcite. Hash join.

docs/_docs/SQL/sql-calcite.adoc

@@ -371,10 +371,10 @@ SELECT /*+ ENFORCE_JOIN_ORDER */ T1.V1, T2.V1, T2.V2, T3.V1, T3.V2, T3.V3 FROM T
 SELECT t1.v1, t3.v2 FROM TBL1 t1 JOIN TBL3 t3 on t1.v3=t3.v3 WHERE t1.v2 in (SELECT /*+ ENFORCE_JOIN_ORDER */ t2.v2 FROM TBL2 t2 JOIN TBL3 t3 ON t2.v1=t3.v1)
 ----
 
-==== MERGE_JOIN, NL_JOIN, CNL_JOIN
-Forces certain join type: Merge, Nested Loop and Correlated Nested Loop respectively.
+==== MERGE_JOIN, NL_JOIN, CNL_JOIN, HASH_JOIN
+Forces certain join type: Merge, Nested Loop, Correlated Nested Loop and Hash Join respectively.
 
-Every of those has the negation like 'NO_INDEX': CNL_JOIN, NO_CNL_JOIN. The negation hint disables certain join type.
+Every of those has the negation like 'NO_INDEX': CNL_JOIN, NO_CNL_JOIN, NO_HASH_JOIN. The negation hint disables certain join type.
 
 ===== Parameters:
 * Empty. To force or disable certain join type for every join.
@@ -389,7 +389,7 @@ SELECT /*+ NL_JOIN(TBL3,TBL1) */ t4.v1, t2.v2 FROM TBL1 t4 JOIN TBL2 t2 on t1.v3
 
 SELECT t1.v1, t2.v2 FROM TBL2 t1 JOIN TBL1 t2 on t1.v3=t2.v3 WHERE t2.v3 in (SELECT /*+ NO_CNL_JOIN(TBL4) */ t3.v3 FROM TBL3 t3 JOIN TBL4 t4 on t3.v1=t4.v1)
 
-SELECT t4.v1, t2.v2 FROM TBL1 t4 JOIN TBL2 t2 on t1.v3=t2.v3 WHERE t2.v1 in (SELECT t3.v3 FROM TBL3 t3 JOIN TBL1 /*+ NL_JOIN */ t4 on t3.v2=t4.v2)
+SELECT t4.v1, t2.v2 FROM TBL1 t4 JOIN TBL2 t2 on t1.v3=t2.v3 WHERE t2.v1 in (SELECT t3.v3 FROM TBL3 t3 JOIN TBL1 /*+ HASH_JOIN */ t4 on t3.v2=t4.v2)
 ----
 
 ==== EXPAND_DISTINCT_AGG

modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/LogicalRelImplementor.java

@@ -18,6 +18,7 @@
 package org.apache.ignite.internal.processors.query.calcite.exec;
 
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Comparator;
 import java.util.List;
 import java.util.Objects;
@@ -37,6 +38,7 @@
 import org.apache.calcite.rel.type.RelDataTypeField;
 import org.apache.calcite.rex.RexLiteral;
 import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexUtil;
 import org.apache.calcite.util.ImmutableBitSet;
 import org.apache.calcite.util.mapping.IntPair;
 import org.apache.ignite.internal.processors.failure.FailureProcessor;
@@ -51,6 +53,7 @@
 import org.apache.ignite.internal.processors.query.calcite.exec.rel.CorrelatedNestedLoopJoinNode;
 import org.apache.ignite.internal.processors.query.calcite.exec.rel.FilterNode;
 import org.apache.ignite.internal.processors.query.calcite.exec.rel.HashAggregateNode;
+import org.apache.ignite.internal.processors.query.calcite.exec.rel.HashJoinNode;
 import org.apache.ignite.internal.processors.query.calcite.exec.rel.Inbox;
 import org.apache.ignite.internal.processors.query.calcite.exec.rel.IndexSpoolNode;
 import org.apache.ignite.internal.processors.query.calcite.exec.rel.IntersectNode;
@@ -77,9 +80,11 @@
 import org.apache.ignite.internal.processors.query.calcite.rel.IgniteExchange;
 import org.apache.ignite.internal.processors.query.calcite.rel.IgniteFilter;
 import org.apache.ignite.internal.processors.query.calcite.rel.IgniteHashIndexSpool;
+import org.apache.ignite.internal.processors.query.calcite.rel.IgniteHashJoin;
 import org.apache.ignite.internal.processors.query.calcite.rel.IgniteIndexBound;
 import org.apache.ignite.internal.processors.query.calcite.rel.IgniteIndexCount;
 import org.apache.ignite.internal.processors.query.calcite.rel.IgniteIndexScan;
+import org.apache.ignite.internal.processors.query.calcite.rel.IgniteJoinInfo;
 import org.apache.ignite.internal.processors.query.calcite.rel.IgniteLimit;
 import org.apache.ignite.internal.processors.query.calcite.rel.IgniteMergeJoin;
 import org.apache.ignite.internal.processors.query.calcite.rel.IgniteNestedLoopJoin;
@@ -268,6 +273,34 @@ public LogicalRelImplementor(
         return node;
     }
 
+    /** {@inheritDoc} */
+    @Override public Node<Row> visit(IgniteHashJoin rel) {
+        RelDataType outType = rel.getRowType();
+        RelDataType leftType = rel.getLeft().getRowType();
+        RelDataType rightType = rel.getRight().getRowType();
+        JoinRelType joinType = rel.getJoinType();
+
+        IgniteJoinInfo joinInfo = IgniteJoinInfo.of(rel);
+
+        RexNode nonEquiConditionExpression = RexUtil.composeConjunction(Commons.emptyCluster().getRexBuilder(),
+            rel.analyzeCondition().nonEquiConditions, true);
+
+        BiPredicate<Row, Row> nonEquiCondition = null;
+
+        if (nonEquiConditionExpression != null) {
+            RelDataType rowType = combinedRowType(ctx.getTypeFactory(), leftType, rightType);
+
+            nonEquiCondition = expressionFactory.biPredicate(rel.getCondition(), rowType);
+        }
+
+        Node<Row> node = HashJoinNode.create(ctx, outType, leftType, rightType, joinType, joinInfo,
+            nonEquiCondition);
+
+        node.register(Arrays.asList(visit(rel.getLeft()), visit(rel.getRight())));
+
+        return node;
+    }
+
     /** {@inheritDoc} */
     @Override public Node<Row> visit(IgniteCorrelatedNestedLoopJoin rel) {
         RelDataType outType = rel.getRowType();

modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/MappingRowHandler.java

@@ -36,6 +36,12 @@ public MappingRowHandler(RowHandler<Row> delegate, ImmutableBitSet requiredColum
         mapping = requiredColumns.toArray();
     }
 
+    /** */
+    public MappingRowHandler(RowHandler<Row> delegate, int[] requiredColumns) {
+        this.delegate = delegate;
+        mapping = requiredColumns;
+    }
+
     /** {@inheritDoc} */
     @Override public Object get(int field, Row row) {
         return delegate.get(mapping[field], row);

modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/RuntimeHashIndex.java

@@ -89,14 +89,7 @@ public Iterable<Row> scan(Supplier<Row> searchRow) {
      * IS NOT DISTINCT FROM condition).
      */
     private @Nullable GroupKey<Row> key(Row r) {
-        if (!allowNulls) {
-            for (int i = 0; i < keysRowHnd.columnCount(r); i++) {
-                if (keysRowHnd.get(i, r) == null)
-                    return null;
-            }
-        }
-
-        return new GroupKey<>(r, keysRowHnd);
+        return GroupKey.of(r, keysRowHnd, allowNulls);
     }
 
     /**

modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/agg/GroupKey.java

@@ -26,6 +26,7 @@
 import org.apache.ignite.binary.Binarylizable;
 import org.apache.ignite.internal.processors.query.calcite.exec.ArrayRowHandler;
 import org.apache.ignite.internal.processors.query.calcite.exec.RowHandler;
+import org.jetbrains.annotations.Nullable;
 
 /**
  *
@@ -38,7 +39,7 @@ public class GroupKey<Row> implements Binarylizable {
     private RowHandler<Row> hnd;
 
     /** */
-    public GroupKey(Row row, RowHandler<Row> hnd) {
+    private GroupKey(Row row, RowHandler<Row> hnd) {
         this.row = row;
         this.hnd = hnd;
     }
@@ -111,4 +112,24 @@ public RowHandler<Row> rowHandler() {
 
         return hashCode;
     }
+
+    /**
+     * @return Group key for provided row. Can be {@code null} if key fields of row contain NULL values and nulls are
+     * not allowed.
+     */
+    public static <Row> @Nullable GroupKey<Row> of(Row r, RowHandler<Row> hnd, boolean allowNulls) {
+        if (!allowNulls) {
+            for (int i = 0; i < hnd.columnCount(r); i++) {
+                if (hnd.get(i, r) == null)
+                    return null;
+            }
+        }
+
+        return new GroupKey<>(r, hnd);
+    }
+
+    /** */
+    public static <Row> @Nullable GroupKey<Row> of(Row r, RowHandler<Row> hnd) {
+        return of(r, hnd, true);
+    }
 }

modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/AbstractRightMaterializedJoinNode.java

@@ -0,0 +1,194 @@
+/*
+ * 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.ignite.internal.processors.query.calcite.exec.rel;
+
+import java.util.ArrayDeque;
+import java.util.Deque;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.ignite.internal.processors.query.calcite.exec.ExecutionContext;
+import org.apache.ignite.internal.util.typedef.F;
+import org.jetbrains.annotations.Nullable;
+
+/** Right-part materialized join node. Holds data from the right part locally. */
+public abstract class AbstractRightMaterializedJoinNode<Row> extends MemoryTrackingNode<Row> {
+    /** */
+    protected static final int HALF_BUF_SIZE = IN_BUFFER_SIZE >> 1;
+
+    /** Special flag which marks that all the rows are received. */
+    protected static final int NOT_WAITING = -1;
+
+    /** */
+    protected final Deque<Row> leftInBuf = new ArrayDeque<>(IN_BUFFER_SIZE);
+
+    /** */
+    protected boolean inLoop;
+
+    /** */
+    protected int requested;
+
+    /** */
+    protected int waitingLeft;
+
+    /** */
+    protected int waitingRight;
+
+    /** */
+    protected @Nullable Row left;
+
+    /** */
+    protected int processed;
+
+    /** */
+    protected AbstractRightMaterializedJoinNode(ExecutionContext<Row> ctx, RelDataType rowType) {
+        super(ctx, rowType);
+    }
+
+    /** */
+    protected abstract void join() throws Exception;
+
+    /** */
+    protected abstract void pushRight(Row row) throws Exception;
+
+    /** {@inheritDoc} */
+    @Override public void request(int rowsCnt) throws Exception {
+        assert !F.isEmpty(sources()) && sources().size() == 2;
+        assert rowsCnt > 0 && requested == 0;
+
+        requested = rowsCnt;
+
+        if (!inLoop)
+            context().execute(this::join0, this::onError);
+    }
+
+    /** {@inheritDoc} */
+    @Override protected void rewindInternal() {
+        requested = 0;
+        waitingLeft = 0;
+        waitingRight = 0;
+        left = null;
+
+        leftInBuf.clear();
+    }
+
+    /** {@inheritDoc} */
+    @Override protected Downstream<Row> requestDownstream(int idx) {
+        if (idx == 0) {
+            return new Downstream<>() {
+                @Override public void push(Row row) throws Exception {
+                    pushLeft(row);
+                }
+
+                @Override public void end() throws Exception {
+                    endLeft();
+                }
+
+                @Override public void onError(Throwable e) {
+                    AbstractRightMaterializedJoinNode.this.onError(e);
+                }
+            };
+        }
+        else if (idx == 1) {
+            return new Downstream<>() {
+                @Override public void push(Row row) throws Exception {
+                    pushRight(row);
+                }
+
+                @Override public void end() throws Exception {
+                    endRight();
+                }
+
+                @Override public void onError(Throwable e) {
+                    AbstractRightMaterializedJoinNode.this.onError(e);
+                }
+            };
+        }
+
+        throw new IndexOutOfBoundsException();
+    }
+
+    /** */
+    private void pushLeft(Row row) throws Exception {
+        assert downstream() != null;
+        assert waitingLeft > 0;
+
+        --waitingLeft;
+
+        leftInBuf.add(row);
+
+        join0();
+    }
+
+    /** */
+    private void endLeft() throws Exception {
+        assert downstream() != null;
+        assert waitingLeft > 0;
+
+        waitingLeft = NOT_WAITING;
+
+        join0();
+    }
+
+    /** */
+    private void endRight() throws Exception {
+        assert downstream() != null;
+        assert waitingRight > 0;
+
+        waitingRight = NOT_WAITING;
+
+        join0();
+    }
+
+    /** */
+    protected void tryToRequestInputs() throws Exception {
+        if (waitingLeft == 0 && leftInBuf.size() <= HALF_BUF_SIZE)
+            leftSource().request(waitingLeft = IN_BUFFER_SIZE - leftInBuf.size());
+
+        if (waitingRight == 0 && requested > 0)
+            rightSource().request(waitingRight = IN_BUFFER_SIZE);
+    }
+
+    /** */
+    protected Node<Row> leftSource() {
+        return sources().get(0);
+    }
+
+    /** */
+    protected Node<Row> rightSource() {
+        return sources().get(1);
+    }
+
+    /** */
+    private void join0() throws Exception {
+        checkState();
+
+        processed = 0;
+
+        join();
+    }
+
+    /** */
+    protected boolean rescheduleJoin() {
+        if (processed++ > IN_BUFFER_SIZE) {
+            context().execute(this::join0, this::onError);
+
+            return true;
+        }
+
+        return false;
+    }
+}

modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/AbstractSetOpNode.java

@@ -262,7 +262,7 @@ else if (type == AggregateType.MAP)
 
         /** */
         protected GroupKey<Row> key(Row row) {
-            return new GroupKey<>(row, hnd);
+            return GroupKey.of(row, hnd);
         }
 
         /** */

modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/HashAggregateNode.java

@@ -304,7 +304,7 @@ else if (type == AggregateType.MAP)
 
         /** */
         private GroupKey<Row> key(Row row) {
-            return new GroupKey<>(row, keyGrpRowHnd);
+            return GroupKey.of(row, keyGrpRowHnd);
         }
 
         /** */