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

@@ -17,15 +17,17 @@
 package org.apache.ignite.internal.processors.query.calcite.exec;
 
 import java.util.ArrayList;
+import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.Iterator;
-import java.util.List;
+import java.util.Map;
 import java.util.function.Supplier;
 import org.apache.calcite.util.ImmutableBitSet;
+import org.apache.ignite.binary.BinaryObjectException;
+import org.apache.ignite.binary.BinaryReader;
+import org.apache.ignite.binary.BinaryWriter;
 import org.apache.ignite.internal.processors.query.calcite.exec.exp.agg.GroupKey;
-import org.apache.ignite.internal.util.typedef.F;
-import org.jetbrains.annotations.NotNull;
 import org.jetbrains.annotations.Nullable;
 
 /**
@@ -39,26 +41,51 @@ public class RuntimeHashIndex<Row> implements RuntimeIndex<Row> {
     private final RowHandler<Row> keysRowHnd;
 
     /** Rows. */
-    private final HashMap<GroupKey<Row>, List<Row>> rows;
+    private final Map<GroupKey<Row>, Collection<Row>> rows;
 
-    /** Allow NULL values. */
-    private final boolean allowNulls;
+    /** */
+    private final Supplier<Collection<Row>> collectionFactory;
 
-    /**
-     *
-     */
+    /** */
+    private final ImmutableBitSet nullsMatch;
+
+    /** Creates hash index with the default collection supplier. */
+    public RuntimeHashIndex(ExecutionContext<Row> ectx, ImmutableBitSet keys, ImmutableBitSet nullsMatch) {
+        this(ectx, keys.toArray(), nullsMatch, -1, null);
+    }
+
+    /** */
     public RuntimeHashIndex(
         ExecutionContext<Row> ectx,
-        ImmutableBitSet keys,
-        boolean allowNulls
+        int[] keys,
+        ImmutableBitSet nullsMatch,
+        int initCapacity,
+        @Nullable Supplier<Collection<Row>> collectionFactory
+    ) {
+        this(
+            ectx,
+            nullsMatch,
+            new MappingRowHandler<>(ectx.rowHandler(), keys),
+            initCapacity >= 0 ? new HashMap<>(initCapacity) : new HashMap<>(),
+            collectionFactory
+        );
+    }
+
+    /** Fields setting constructor. */
+    private RuntimeHashIndex(
+        ExecutionContext<Row> ectx,
+        ImmutableBitSet nullsMatch,
+        RowHandler<Row> keysRowHnd,
+        Map<GroupKey<Row>, Collection<Row>> rows,
+        @Nullable Supplier<Collection<Row>> collectionFactory
     ) {
         this.ectx = ectx;
-        this.allowNulls = allowNulls;
+        this.nullsMatch = nullsMatch;
 
-        assert !F.isEmpty(keys);
+        this.keysRowHnd = keysRowHnd;
+        this.rows = rows;
 
-        keysRowHnd = new MappingRowHandler<>(ectx.rowHandler(), keys);
-        rows = new HashMap<>();
+        this.collectionFactory = collectionFactory == null ? ArrayList::new : collectionFactory;
     }
 
     /** {@inheritDoc} */
@@ -68,7 +95,7 @@ public RuntimeHashIndex(
         if (key == null)
             return;
 
-        List<Row> eqRows = rows.computeIfAbsent(key, k -> new ArrayList<>());
+        Collection<Row> eqRows = rows.computeIfAbsent(key, k -> collectionFactory.get());
 
         eqRows.add(r);
     }
@@ -79,7 +106,12 @@ public RuntimeHashIndex(
     }
 
     /** */
-    public Iterable<Row> scan(Supplier<Row> searchRow) {
+    public Collection<Collection<Row>> rowSets() {
+        return Collections.unmodifiableCollection(rows.values());
+    }
+
+    /** */
+    public IndexScan scan(Supplier<Row> searchRow) {
         return new IndexScan(searchRow);
     }
 
@@ -89,40 +121,92 @@ 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 NullsCheckingGroupKey<>(r, keysRowHnd);
+    }
+
+    /** */
+    public RuntimeHashIndex<Row> remappedSearcher(int[] remappedKeys) {
+        return new RemappedSearcher<>(this, remappedKeys);
+    }
+
+    /** */
+    private class NullsCheckingGroupKey<Row> extends GroupKey<Row> {
+        /** */
+        private NullsCheckingGroupKey(Row row, RowHandler<Row> hnd) {
+            super(row, hnd);
+        }
+
+        /** {@inheritDoc} */
+        @Override protected boolean columnValuesEquals(int colIdx, Object v1, Object v2) {
+            if (v1 == null && v2 == null) {
+                if (nullsMatch.cardinality() == 0)
+                    return false;
+
+                return nullsMatch.get(colIdx);
             }
+
+            return super.columnValuesEquals(colIdx, v1, v2);
+        }
+
+        /** {@inheritDoc} */
+        @Override public void writeBinary(BinaryWriter writer) throws BinaryObjectException {
+            throw new UnsupportedOperationException("Serialization of row keys is not supported.");
+        }
+
+        /** {@inheritDoc} */
+        @Override public void readBinary(BinaryReader reader) throws BinaryObjectException {
+            throw new UnsupportedOperationException("Deserialization of row keys is not supported.");
+        }
+    }
+
+    /** */
+    private static class RemappedSearcher<Row> extends RuntimeHashIndex<Row> {
+        /** */
+        private final RuntimeHashIndex<Row> origin;
+
+        /** */
+        private RemappedSearcher(RuntimeHashIndex<Row> o, int[] remappedKeys) {
+            super(o.ectx, o.nullsMatch, new MappingRowHandler<>(o.ectx.rowHandler(), remappedKeys), o.rows,
+                o.collectionFactory);
+
+            origin = o;
         }
 
-        return new GroupKey<>(r, keysRowHnd);
+        /** {@inheritDoc} */
+        @Override public void push(Row r) {
+            origin.push(r);
+        }
     }
 
     /**
      *
      */
-    private class IndexScan implements Iterable<Row> {
+    public class IndexScan implements Iterable<Row> {
         /** Search row. */
         private final Supplier<Row> searchRow;
 
         /**
          * @param searchRow Search row.
          */
-        IndexScan(Supplier<Row> searchRow) {
+        private IndexScan(Supplier<Row> searchRow) {
             this.searchRow = searchRow;
         }
 
-        /** {@inheritDoc} */
-        @NotNull @Override public Iterator<Row> iterator() {
+        /**  */
+        public @Nullable Collection<Row> get() {
             GroupKey<Row> key = key(searchRow.get());
 
             if (key == null)
-                return Collections.emptyIterator();
+                return null;
 
-            List<Row> eqRows = rows.get(key);
+            return rows.get(key);
+        }
+
+        /** {@inheritDoc} */
+        @Override public Iterator<Row> iterator() {
+            Collection<Row> collection = get();
 
-            return eqRows == null ? Collections.emptyIterator() : eqRows.iterator();
+            return collection == null ? Collections.emptyIterator() : collection.iterator();
         }
     }
 }

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

@@ -95,13 +95,21 @@ public RowHandler<Row> rowHandler() {
             return false;
 
         for (int i = 0; i < colCnt; i++) {
-            if (!Objects.equals(hnd.get(i, row), other.hnd.get(i, other.row)))
+            Object o1 = hnd.get(i, row);
+            Object o2 = other.hnd.get(i, other.row);
+
+            if (!columnValuesEquals(i, o1, o2) || !Objects.equals(o1, o2))
                 return false;
         }
 
         return true;
     }
 
+    /** */
+    protected boolean columnValuesEquals(int colIdx, Object val1, Object val2) {
+        return true;
+    }
+
     /** {@inheritDoc} */
     @Override public int hashCode() {
         int hashCode = 0;