You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by am...@apache.org on 2024/12/13 10:40:02 UTC
(ignite-3) branch main updated: IGNITE-23898 Sql. Calculate schema for concat two rows just once (#4882)
This is an automated email from the ASF dual-hosted git repository.
amashenkov pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
The following commit(s) were added to refs/heads/main by this push:
new f81f7b5aa4 IGNITE-23898 Sql. Calculate schema for concat two rows just once (#4882)
f81f7b5aa4 is described below
commit f81f7b5aa403b3a0ad873df7efd23e60e7b37aa2
Author: Andrew V. Mashenkov <AM...@users.noreply.github.com>
AuthorDate: Fri Dec 13 13:39:57 2024 +0300
IGNITE-23898 Sql. Calculate schema for concat two rows just once (#4882)
---
.../sql/engine/exec/LogicalRelImplementor.java | 3 +-
.../internal/sql/engine/exec/RowHandler.java | 6 +-
.../internal/sql/engine/exec/SqlRowHandler.java | 47 ++--
.../exec/rel/CorrelatedNestedLoopJoinNode.java | 24 +-
.../internal/sql/engine/exec/rel/HashJoinNode.java | 282 ++++++++++++---------
.../sql/engine/exec/rel/MergeJoinNode.java | 178 +++++++------
.../internal/sql/engine/exec/rel/ModifyNode.java | 25 +-
.../sql/engine/exec/rel/NestedLoopJoinNode.java | 147 ++++++-----
.../internal/sql/engine/exec/row/RowSchema.java | 34 ++-
.../sql/engine/exec/rel/ExecutionTest.java | 19 +-
.../sql/engine/exec/rel/HashJoinExecutionTest.java | 16 +-
.../engine/exec/rel/ModifyNodeExecutionTest.java | 23 +-
.../sql/engine/exec/row/RowSchemaTest.java | 54 ++++
.../sql/engine/exec/row/SqlRowHandlerTest.java | 11 +-
.../sql/engine/framework/ArrayRowHandler.java | 14 +-
15 files changed, 532 insertions(+), 351 deletions(-)
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/LogicalRelImplementor.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/LogicalRelImplementor.java
index 6e47942737..8952c5f07d 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/LogicalRelImplementor.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/LogicalRelImplementor.java
@@ -305,9 +305,10 @@ public class LogicalRelImplementor<RowT> implements IgniteRelVisitor<Node<RowT>>
: CNLJ_NOT_SUPPORTED_JOIN_ASSERTION_MSG;
RowFactory<RowT> rightRowFactory = ctx.rowHandler().factory(rightRowSchema);
+ RowFactory<RowT> outputRowFactory = ctx.rowHandler().factory(rowSchemaFromRelTypes(RelOptUtil.getFieldTypeList(rowType)));
Node<RowT> node = new CorrelatedNestedLoopJoinNode<>(ctx, cond, rel.getVariablesSet(),
- rel.getJoinType(), rightRowFactory);
+ rel.getJoinType(), rightRowFactory, outputRowFactory);
Node<RowT> leftInput = visit(rel.getLeft());
Node<RowT> rightInput = visit(rel.getRight());
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/RowHandler.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/RowHandler.java
index a7dbbb04f8..5381c233e5 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/RowHandler.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/RowHandler.java
@@ -43,9 +43,6 @@ public interface RowHandler<RowT> {
*/
boolean isNull(int field, RowT row);
- /** Concatenate two rows. */
- RowT concat(RowT left, RowT right);
-
/** Return column count contained in the incoming row. */
int columnCount(RowT row);
@@ -114,6 +111,9 @@ public interface RowHandler<RowT> {
* @return A new row with fields from the specified mapping.
*/
RowT map(RowT row, int[] mapping);
+
+ /** Concatenate two rows. The result row will satisfy the current factory's schema. */
+ RowT concat(RowT left, RowT right);
}
/**
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/SqlRowHandler.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/SqlRowHandler.java
index 83983df7b9..5212d0e0fb 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/SqlRowHandler.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/SqlRowHandler.java
@@ -25,7 +25,6 @@ import java.time.Instant;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.LocalTime;
-import java.util.List;
import java.util.UUID;
import org.apache.calcite.avatica.util.ByteString;
import org.apache.ignite.internal.binarytuple.BinaryTupleBuilder;
@@ -34,7 +33,6 @@ import org.apache.ignite.internal.lang.InternalTuple;
import org.apache.ignite.internal.schema.BinaryTuple;
import org.apache.ignite.internal.sql.engine.exec.SqlRowHandler.RowWrapper;
import org.apache.ignite.internal.sql.engine.exec.row.RowSchema;
-import org.apache.ignite.internal.sql.engine.exec.row.RowSchema.Builder;
import org.apache.ignite.internal.sql.engine.exec.row.RowSchemaTypes;
import org.apache.ignite.internal.sql.engine.exec.row.TypeSpec;
import org.apache.ignite.internal.sql.engine.util.Commons;
@@ -79,30 +77,6 @@ public class SqlRowHandler implements RowHandler<RowWrapper> {
return row.isNull(field);
}
- /** {@inheritDoc} */
- @Override
- public RowWrapper concat(RowWrapper left, RowWrapper right) {
- int leftLen = left.columnsCount();
- int rightLen = right.columnsCount();
- List<TypeSpec> leftTypes = left.rowSchema().fields();
- List<TypeSpec> rightTypes = right.rowSchema().fields();
-
- Object[] values = new Object[leftLen + rightLen];
- Builder schemaBuilder = RowSchema.builder();
-
- for (int i = 0; i < leftLen; i++) {
- values[i] = left.get(i);
- schemaBuilder.addField(leftTypes.get(i));
- }
-
- for (int i = 0; i < rightLen; i++) {
- values[leftLen + i] = right.get(i);
- schemaBuilder.addField(rightTypes.get(i));
- }
-
- return new ObjectsArrayRowWrapper(schemaBuilder.build(), values);
- }
-
@Override
public int columnCount(RowWrapper row) {
return row.columnsCount();
@@ -184,6 +158,27 @@ public class SqlRowHandler implements RowHandler<RowWrapper> {
return new ObjectsArrayRowWrapper(rowSchema, fields);
}
+
+ /** {@inheritDoc} */
+ @Override
+ public RowWrapper concat(RowWrapper left, RowWrapper right) {
+ int leftLen = left.columnsCount();
+ int rightLen = right.columnsCount();
+
+ assert leftLen + rightLen == schemaLen;
+
+ Object[] values = new Object[schemaLen];
+
+ for (int i = 0; i < leftLen; i++) {
+ values[i] = left.get(i);
+ }
+
+ for (int i = 0; i < rightLen; i++) {
+ values[leftLen + i] = right.get(i);
+ }
+
+ return new ObjectsArrayRowWrapper(rowSchema, values);
+ }
};
}
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/CorrelatedNestedLoopJoinNode.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/CorrelatedNestedLoopJoinNode.java
index c298cb553a..092ce38e93 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/CorrelatedNestedLoopJoinNode.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/CorrelatedNestedLoopJoinNode.java
@@ -29,7 +29,6 @@ import java.util.function.BiPredicate;
import org.apache.calcite.rel.core.CorrelationId;
import org.apache.calcite.rel.core.JoinRelType;
import org.apache.ignite.internal.sql.engine.exec.ExecutionContext;
-import org.apache.ignite.internal.sql.engine.exec.RowHandler;
import org.apache.ignite.internal.sql.engine.exec.RowHandler.RowFactory;
/**
@@ -43,7 +42,7 @@ public class CorrelatedNestedLoopJoinNode<RowT> extends AbstractNode<RowT> {
private final JoinRelType joinType;
- private final RowHandler<RowT> handler;
+ private final RowFactory<RowT> outputRowFactory;
private final int leftInBufferSize;
@@ -74,17 +73,23 @@ public class CorrelatedNestedLoopJoinNode<RowT> extends AbstractNode<RowT> {
}
/**
- * Constructor.
- * TODO Documentation https://issues.apache.org/jira/browse/IGNITE-15859
+ * Creates CorrelatedNestedLoopJoin node.
*
* @param ctx Execution context.
* @param cond Join expression.
* @param correlationIds Set of collections ids.
* @param joinType Join rel type.
* @param rightRowFactory Right row factory.
+ * @param outputRowFactory Output row factory.
*/
- public CorrelatedNestedLoopJoinNode(ExecutionContext<RowT> ctx, BiPredicate<RowT, RowT> cond,
- Set<CorrelationId> correlationIds, JoinRelType joinType, RowFactory<RowT> rightRowFactory) {
+ public CorrelatedNestedLoopJoinNode(
+ ExecutionContext<RowT> ctx,
+ BiPredicate<RowT, RowT> cond,
+ Set<CorrelationId> correlationIds,
+ JoinRelType joinType,
+ RowFactory<RowT> rightRowFactory,
+ RowFactory<RowT> outputRowFactory
+ ) {
super(ctx);
assert !nullOrEmpty(correlationIds);
@@ -92,13 +97,12 @@ public class CorrelatedNestedLoopJoinNode<RowT> extends AbstractNode<RowT> {
this.cond = cond;
this.correlationIds = new ArrayList<>(correlationIds);
this.joinType = joinType;
+ this.outputRowFactory = outputRowFactory;
leftInBufferSize = correlationIds.size();
rightInBufferSize = inBufSize;
rightEmptyRow = rightRowFactory.create();
-
- handler = ctx.rowHandler();
}
/** {@inheritDoc} */
@@ -375,7 +379,7 @@ public class CorrelatedNestedLoopJoinNode<RowT> extends AbstractNode<RowT> {
requested--;
- RowT row = handler.concat(left, right);
+ RowT row = outputRowFactory.concat(left, right);
downstream().push(row);
}
@@ -414,7 +418,7 @@ public class CorrelatedNestedLoopJoinNode<RowT> extends AbstractNode<RowT> {
while (requested > 0 && notMatchedIdx < leftInBuf.size()) {
requested--;
- downstream().push(handler.concat(leftInBuf.get(notMatchedIdx), rightEmptyRow));
+ downstream().push(outputRowFactory.concat(leftInBuf.get(notMatchedIdx), rightEmptyRow));
leftMatched.set(notMatchedIdx);
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/HashJoinNode.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/HashJoinNode.java
index c07258a9b6..72a9c88bac 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/HashJoinNode.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/HashJoinNode.java
@@ -24,7 +24,6 @@ import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.Iterator;
-import java.util.List;
import java.util.Map;
import java.util.NoSuchElementException;
import java.util.Objects;
@@ -32,64 +31,52 @@ import org.apache.calcite.plan.RelOptUtil;
import org.apache.calcite.rel.core.JoinInfo;
import org.apache.calcite.rel.core.JoinRelType;
import org.apache.calcite.rel.type.RelDataType;
-import org.apache.calcite.rel.type.RelDataTypeField;
-import org.apache.calcite.util.ImmutableIntList;
import org.apache.ignite.internal.sql.engine.exec.ExecutionContext;
import org.apache.ignite.internal.sql.engine.exec.RowHandler;
-import org.apache.ignite.internal.sql.engine.exec.RowHandler.RowBuilder;
import org.apache.ignite.internal.sql.engine.exec.RowHandler.RowFactory;
import org.apache.ignite.internal.sql.engine.exec.row.RowSchema;
/** HashJoin implementor. */
public abstract class HashJoinNode<RowT> extends AbstractRightMaterializedJoinNode<RowT> {
- final Map<RowWrapper<RowT>, TouchedCollection<RowT>> hashStore = new Object2ObjectOpenHashMap<>();
- protected final RowHandler<RowT> handler;
+ private static final int INITIAL_CAPACITY = 128;
- private final List<Integer> leftJoinPositions;
- private final List<Integer> rightJoinPositions;
+ /** All keys with null-fields are mapped to this object. */
+ private static final Key NULL_KEY = new Key();
- final boolean touchResults;
+ final Map<Key, TouchedCollection<RowT>> hashStore = new Object2ObjectOpenHashMap<>(INITIAL_CAPACITY);
- Iterator<RowT> rightIt = Collections.emptyIterator();
- private final RowSchema rightJoinRelatedRowSchema;
- private final RowSchema leftJoinRelatedRowSchema;
+ private final int[] leftJoinPositions;
+ private final int[] rightJoinPositions;
- private final RowBuilder<RowT> leftRowBuilder;
- private final RowBuilder<RowT> rightRowBuilder;
+ Iterator<RowT> rightIt = Collections.emptyIterator();
- private HashJoinNode(ExecutionContext<RowT> ctx, JoinInfo joinInfo, boolean touch,
- RelDataType leftRowType, RelDataType rightRowType) {
+ /** Output row factory. */
+ protected final RowFactory<RowT> outputRowFactory;
+ private final RowFactory<RowT> keyFactory;
+
+ /**
+ * Creates HashJoinNode.
+ *
+ * @param ctx Execution context.
+ * @param joinInfo Join info.
+ * @param outputRowFactory Output row factory.
+ * @param keyFactory Key factory.
+ */
+ private HashJoinNode(
+ ExecutionContext<RowT> ctx,
+ JoinInfo joinInfo,
+ RowFactory<RowT> outputRowFactory,
+ RowFactory<RowT> keyFactory
+ ) {
super(ctx);
- handler = ctx.rowHandler();
- touchResults = touch;
-
- leftJoinPositions = joinInfo.leftKeys.toIntegerList();
- rightJoinPositions = joinInfo.rightKeys.toIntegerList();
-
- assert leftJoinPositions.size() == rightJoinPositions.size();
-
- ImmutableIntList rightKeys = joinInfo.rightKeys;
- List<RelDataType> rightTypes = new ArrayList<>(rightKeys.size());
- List<RelDataTypeField> rightFields = rightRowType.getFieldList();
- for (int rightPos : rightKeys) {
- rightTypes.add(rightFields.get(rightPos).getType());
- }
- rightJoinRelatedRowSchema = rowSchemaFromRelTypes(rightTypes);
-
- ImmutableIntList leftKeys = joinInfo.leftKeys;
- List<RelDataType> leftTypes = new ArrayList<>(leftKeys.size());
- List<RelDataTypeField> leftFields = leftRowType.getFieldList();
- for (int leftPos : leftKeys) {
- leftTypes.add(leftFields.get(leftPos).getType());
- }
- leftJoinRelatedRowSchema = rowSchemaFromRelTypes(leftTypes);
-
- RowFactory<RowT> leftRowFactory = handler.factory(leftJoinRelatedRowSchema);
- leftRowBuilder = leftRowFactory.rowBuilder();
+ leftJoinPositions = joinInfo.leftKeys.toIntArray();
+ rightJoinPositions = joinInfo.rightKeys.toIntArray();
+ assert leftJoinPositions.length == rightJoinPositions.length;
+ assert leftJoinPositions.length == keyFactory.rowSchema().fields().size();
- RowFactory<RowT> rightRowFactory = handler.factory(rightJoinRelatedRowSchema);
- rightRowBuilder = rightRowFactory.rowBuilder();
+ this.keyFactory = keyFactory;
+ this.outputRowFactory = outputRowFactory;
}
@Override
@@ -104,39 +91,38 @@ public abstract class HashJoinNode<RowT> extends AbstractRightMaterializedJoinNo
/** Supplied algorithm implementation. */
public static <RowT> HashJoinNode<RowT> create(ExecutionContext<RowT> ctx, RelDataType outputRowType,
RelDataType leftRowType, RelDataType rightRowType, JoinRelType joinType, JoinInfo joinInfo) {
+ RowSchema leftRowSchema = rowSchemaFromRelTypes(RelOptUtil.getFieldTypeList(leftRowType));
+ RowSchema rightRowSchema = rowSchemaFromRelTypes(RelOptUtil.getFieldTypeList(rightRowType));
+ RowSchema outputRowSchema = rowSchemaFromRelTypes(RelOptUtil.getFieldTypeList(outputRowType));
+
+ RowFactory<RowT> keyFactory = ctx.rowHandler().factory(RowSchema.map(leftRowSchema, joinInfo.leftKeys.toIntArray()));
+ RowFactory<RowT> outputRowFactory = ctx.rowHandler().factory(outputRowSchema);
switch (joinType) {
case INNER:
- return new InnerHashJoin<>(ctx, joinInfo, leftRowType, rightRowType);
+ return new InnerHashJoin<>(ctx, joinInfo, outputRowFactory, keyFactory);
case LEFT: {
- RowSchema rightRowSchema = rowSchemaFromRelTypes(RelOptUtil.getFieldTypeList(rightRowType));
RowHandler.RowFactory<RowT> rightRowFactory = ctx.rowHandler().factory(rightRowSchema);
- return new LeftHashJoin<>(ctx, rightRowFactory, joinInfo, leftRowType, rightRowType);
+ return new LeftHashJoin<>(ctx, joinInfo, outputRowFactory, keyFactory, rightRowFactory);
}
-
case RIGHT: {
- RowSchema leftRowSchema = rowSchemaFromRelTypes(RelOptUtil.getFieldTypeList(leftRowType));
RowHandler.RowFactory<RowT> leftRowFactory = ctx.rowHandler().factory(leftRowSchema);
- return new RightHashJoin<>(ctx, leftRowFactory, joinInfo, leftRowType, rightRowType);
+ return new RightHashJoin<>(ctx, joinInfo, outputRowFactory, keyFactory, leftRowFactory);
}
-
case FULL: {
- RowSchema leftRowSchema = rowSchemaFromRelTypes(RelOptUtil.getFieldTypeList(leftRowType));
- RowSchema rightRowSchema = rowSchemaFromRelTypes(RelOptUtil.getFieldTypeList(rightRowType));
RowHandler.RowFactory<RowT> leftRowFactory = ctx.rowHandler().factory(leftRowSchema);
RowHandler.RowFactory<RowT> rightRowFactory = ctx.rowHandler().factory(rightRowSchema);
- return new FullOuterHashJoin<>(ctx, leftRowFactory, rightRowFactory, joinInfo, leftRowType, rightRowType);
+ return new FullOuterHashJoin<>(ctx, joinInfo, outputRowFactory, keyFactory, leftRowFactory, rightRowFactory);
}
-
case SEMI:
- return new SemiHashJoin<>(ctx, joinInfo, leftRowType, rightRowType);
+ return new SemiHashJoin<>(ctx, joinInfo, outputRowFactory, keyFactory);
case ANTI:
- return new AntiHashJoin<>(ctx, joinInfo, leftRowType, rightRowType);
+ return new AntiHashJoin<>(ctx, joinInfo, outputRowFactory, keyFactory);
default:
throw new IllegalStateException("Join type \"" + joinType + "\" is not supported yet");
@@ -144,13 +130,21 @@ public abstract class HashJoinNode<RowT> extends AbstractRightMaterializedJoinNo
}
private static class InnerHashJoin<RowT> extends HashJoinNode<RowT> {
+ /**
+ * Creates HashJoinNode for INNER JOIN operator.
+ *
+ * @param ctx Execution context.
+ * @param joinInfo Join info.
+ * @param outputRowFactory Output row factory.
+ * @param keyFactory Key factory.
+ */
private InnerHashJoin(
ExecutionContext<RowT> ctx,
JoinInfo joinInfo,
- RelDataType leftRowType,
- RelDataType rightRowType
+ RowFactory<RowT> outputRowFactory,
+ RowFactory<RowT> keyFactory
) {
- super(ctx, joinInfo, false, leftRowType, rightRowType);
+ super(ctx, joinInfo, outputRowFactory, keyFactory);
}
@Override
@@ -162,7 +156,7 @@ public abstract class HashJoinNode<RowT> extends AbstractRightMaterializedJoinNo
if (!rightIt.hasNext()) {
left = leftInBuf.remove();
- Collection<RowT> rightRows = lookup(left, touchResults);
+ Collection<RowT> rightRows = lookup(left);
rightIt = rightRows.iterator();
}
@@ -175,7 +169,7 @@ public abstract class HashJoinNode<RowT> extends AbstractRightMaterializedJoinNo
--requested;
- RowT row = handler.concat(left, right);
+ RowT row = outputRowFactory.concat(left, right);
downstream().push(row);
if (requested == 0) {
@@ -201,14 +195,23 @@ public abstract class HashJoinNode<RowT> extends AbstractRightMaterializedJoinNo
/** Right row factory. */
private final RowHandler.RowFactory<RowT> rightRowFactory;
+ /**
+ * Creates HashJoinNode for LEFT OUTER JOIN operator.
+ *
+ * @param ctx Execution context.
+ * @param joinInfo Join info.
+ * @param outputRowFactory Output row factory.
+ * @param keyFactory Key factory.
+ * @param rightRowFactory Right row factory.
+ */
private LeftHashJoin(
ExecutionContext<RowT> ctx,
- RowHandler.RowFactory<RowT> rightRowFactory,
JoinInfo joinInfo,
- RelDataType leftRowType,
- RelDataType rightRowType
+ RowFactory<RowT> outputRowFactory,
+ RowFactory<RowT> keyFactory,
+ RowFactory<RowT> rightRowFactory
) {
- super(ctx, joinInfo, false, leftRowType, rightRowType);
+ super(ctx, joinInfo, outputRowFactory, keyFactory);
this.rightRowFactory = rightRowFactory;
}
@@ -225,11 +228,11 @@ public abstract class HashJoinNode<RowT> extends AbstractRightMaterializedJoinNo
if (!rightIt.hasNext()) {
left = leftInBuf.remove();
- Collection<RowT> rightRows = lookup(left, touchResults);
+ Collection<RowT> rightRows = lookup(left);
if (rightRows.isEmpty()) {
requested--;
- downstream().push(handler.concat(left, rightRowFactory.create()));
+ downstream().push(outputRowFactory.concat(left, rightRowFactory.create()));
}
rightIt = rightRows.iterator();
@@ -243,7 +246,7 @@ public abstract class HashJoinNode<RowT> extends AbstractRightMaterializedJoinNo
--requested;
- RowT row = handler.concat(left, right);
+ RowT row = outputRowFactory.concat(left, right);
downstream().push(row);
if (requested == 0) {
@@ -269,14 +272,23 @@ public abstract class HashJoinNode<RowT> extends AbstractRightMaterializedJoinNo
/** Left row factory. */
private final RowHandler.RowFactory<RowT> leftRowFactory;
+ /**
+ * Creates HashJoinNode for RIGHT OUTER JOIN operator.
+ *
+ * @param ctx Execution context.
+ * @param joinInfo Join info.
+ * @param outputRowFactory Output row factory.
+ * @param keyFactory Key factory.
+ * @param leftRowFactory Left row factory.
+ */
private RightHashJoin(
ExecutionContext<RowT> ctx,
- RowHandler.RowFactory<RowT> leftRowFactory,
JoinInfo joinInfo,
- RelDataType leftRowType,
- RelDataType rightRowType
+ RowFactory<RowT> outputRowFactory,
+ RowFactory<RowT> keyFactory,
+ RowFactory<RowT> leftRowFactory
) {
- super(ctx, joinInfo, true, leftRowType, rightRowType);
+ super(ctx, joinInfo, outputRowFactory, keyFactory);
this.leftRowFactory = leftRowFactory;
}
@@ -292,7 +304,7 @@ public abstract class HashJoinNode<RowT> extends AbstractRightMaterializedJoinNo
if (!rightIt.hasNext()) {
left = leftInBuf.remove();
- Collection<RowT> rightRows = lookup(left, touchResults);
+ Collection<RowT> rightRows = lookup(left);
rightIt = rightRows.iterator();
}
@@ -305,7 +317,7 @@ public abstract class HashJoinNode<RowT> extends AbstractRightMaterializedJoinNo
--requested;
- RowT row = handler.concat(left, right);
+ RowT row = outputRowFactory.concat(left, right);
downstream().push(row);
if (requested == 0) {
@@ -335,7 +347,7 @@ public abstract class HashJoinNode<RowT> extends AbstractRightMaterializedJoinNo
while (rightIt.hasNext()) {
checkState();
RowT right = rightIt.next();
- RowT row = handler.concat(emptyLeft, right);
+ RowT row = outputRowFactory.concat(emptyLeft, right);
--requested;
downstream().push(row);
@@ -360,15 +372,25 @@ public abstract class HashJoinNode<RowT> extends AbstractRightMaterializedJoinNo
/** Right row factory. */
private final RowHandler.RowFactory<RowT> rightRowFactory;
+ /**
+ * Creates HashJoinNode for FULL OUTER JOIN operator.
+ *
+ * @param ctx Execution context.
+ * @param joinInfo Join info.
+ * @param outputRowFactory Output row factory.
+ * @param keyFactory Key factory.
+ * @param leftRowFactory Left row factory.
+ * @param rightRowFactory Right row factory.
+ */
private FullOuterHashJoin(
ExecutionContext<RowT> ctx,
- RowHandler.RowFactory<RowT> leftRowFactory,
- RowHandler.RowFactory<RowT> rightRowFactory,
JoinInfo joinInfo,
- RelDataType leftRowType,
- RelDataType rightRowType
+ RowFactory<RowT> outputRowFactory,
+ RowFactory<RowT> keyFactory,
+ RowFactory<RowT> leftRowFactory,
+ RowFactory<RowT> rightRowFactory
) {
- super(ctx, joinInfo, true, leftRowType, rightRowType);
+ super(ctx, joinInfo, outputRowFactory, keyFactory);
this.leftRowFactory = leftRowFactory;
this.rightRowFactory = rightRowFactory;
@@ -386,11 +408,11 @@ public abstract class HashJoinNode<RowT> extends AbstractRightMaterializedJoinNo
if (!rightIt.hasNext()) {
left = leftInBuf.remove();
- Collection<RowT> rightRows = lookup(left, touchResults);
+ Collection<RowT> rightRows = lookup(left);
if (rightRows.isEmpty()) {
requested--;
- downstream().push(handler.concat(left, rightRowFactory.create()));
+ downstream().push(outputRowFactory.concat(left, rightRowFactory.create()));
}
rightIt = rightRows.iterator();
@@ -404,7 +426,7 @@ public abstract class HashJoinNode<RowT> extends AbstractRightMaterializedJoinNo
--requested;
- RowT row = handler.concat(left, right);
+ RowT row = outputRowFactory.concat(left, right);
downstream().push(row);
if (requested == 0) {
@@ -435,7 +457,7 @@ public abstract class HashJoinNode<RowT> extends AbstractRightMaterializedJoinNo
while (rightIt.hasNext()) {
checkState();
RowT right = rightIt.next();
- RowT row = handler.concat(emptyLeft, right);
+ RowT row = outputRowFactory.concat(emptyLeft, right);
--requested;
downstream().push(row);
@@ -454,13 +476,21 @@ public abstract class HashJoinNode<RowT> extends AbstractRightMaterializedJoinNo
}
private static class SemiHashJoin<RowT> extends HashJoinNode<RowT> {
+ /**
+ * Creates HashJoinNode for SEMI JOIN operator.
+ *
+ * @param ctx Execution context.
+ * @param joinInfo Join info.
+ * @param outputRowFactory Output row factory.
+ * @param keyFactory Key factory.
+ */
private SemiHashJoin(
ExecutionContext<RowT> ctx,
JoinInfo joinInfo,
- RelDataType leftRowType,
- RelDataType rightRowType
+ RowFactory<RowT> outputRowFactory,
+ RowFactory<RowT> keyFactory
) {
- super(ctx, joinInfo, false, leftRowType, rightRowType);
+ super(ctx, joinInfo, outputRowFactory, keyFactory);
}
/** {@inheritDoc} */
@@ -474,7 +504,7 @@ public abstract class HashJoinNode<RowT> extends AbstractRightMaterializedJoinNo
left = leftInBuf.remove();
- Collection<RowT> rightRows = lookup(left, touchResults);
+ Collection<RowT> rightRows = lookup(left);
if (!rightRows.isEmpty()) {
requested--;
@@ -498,13 +528,21 @@ public abstract class HashJoinNode<RowT> extends AbstractRightMaterializedJoinNo
}
private static class AntiHashJoin<RowT> extends HashJoinNode<RowT> {
+ /**
+ * Creates HashJoinNode for ANTI JOIN operator.
+ *
+ * @param ctx Execution context.
+ * @param joinInfo Join info.
+ * @param outputRowFactory Output row factory.
+ * @param keyFactory Key factory.
+ */
private AntiHashJoin(
ExecutionContext<RowT> ctx,
JoinInfo joinInfo,
- RelDataType leftRowType,
- RelDataType rightRowType
+ RowFactory<RowT> outputRowFactory,
+ RowFactory<RowT> keyFactory
) {
- super(ctx, joinInfo, false, leftRowType, rightRowType);
+ super(ctx, joinInfo, outputRowFactory, keyFactory);
}
/** {@inheritDoc} */
@@ -518,7 +556,7 @@ public abstract class HashJoinNode<RowT> extends AbstractRightMaterializedJoinNo
left = leftInBuf.remove();
- Collection<RowT> rightRows = lookup(left, touchResults);
+ Collection<RowT> rightRows = lookup(left);
if (rightRows.isEmpty()) {
requested--;
@@ -541,36 +579,26 @@ public abstract class HashJoinNode<RowT> extends AbstractRightMaterializedJoinNo
}
}
- Collection<RowT> lookup(RowT row, boolean processTouched) {
- Collection<RowT> coll = Collections.emptyList();
-
- for (Integer entry : leftJoinPositions) {
- Object ent = handler.get(entry, row);
+ Collection<RowT> lookup(RowT row) {
+ Key row0 = extractKey(row, keyFactory, leftJoinPositions);
- if (ent == null) {
- leftRowBuilder.reset();
- return Collections.emptyList();
- }
-
- leftRowBuilder.addField(ent);
+ if (row0 == NULL_KEY) {
+ // Key with null field can't be compared with other keys.
+ return Collections.emptyList();
}
- RowWrapper<RowT> row0 = new RowWrapper<>(leftRowBuilder.buildAndReset(), handler, leftJoinPositions.size());
-
TouchedCollection<RowT> found = hashStore.get(row0);
if (found != null) {
- coll = found.items();
+ found.touched = true;
- if (processTouched) {
- found.touched = true;
- }
+ return found.items();
}
- return coll;
+ return Collections.emptyList();
}
- private static <RowT> Iterator<RowT> getUntouched(Map<RowWrapper<RowT>, TouchedCollection<RowT>> entries) {
+ private static <RowT> Iterator<RowT> getUntouched(Map<Key, TouchedCollection<RowT>> entries) {
return new Iterator<RowT>() {
private final Iterator<TouchedCollection<RowT>> it = entries.values().iterator();
private Iterator<RowT> innerIt = Collections.emptyIterator();
@@ -615,13 +643,9 @@ public abstract class HashJoinNode<RowT> extends AbstractRightMaterializedJoinNo
waitingRight--;
- for (Integer entry : rightJoinPositions) {
- Object ent = handler.get(entry, row);
- rightRowBuilder.addField(ent);
- }
+ Key key = extractKey(row, keyFactory, rightJoinPositions);
- RowWrapper<RowT> row0 = new RowWrapper<>(rightRowBuilder.buildAndReset(), handler, rightJoinPositions.size());
- TouchedCollection<RowT> raw = hashStore.computeIfAbsent(row0, k -> new TouchedCollection<>());
+ TouchedCollection<RowT> raw = hashStore.computeIfAbsent(key, k -> new TouchedCollection<>());
raw.add(row);
if (waitingRight == 0) {
@@ -629,7 +653,25 @@ public abstract class HashJoinNode<RowT> extends AbstractRightMaterializedJoinNo
}
}
- private static class RowWrapper<RowT> {
+ private Key extractKey(RowT row, RowFactory<RowT> keyFactory, int[] mapping) {
+ RowHandler<RowT> handler = keyFactory.handler();
+ RowT key = keyFactory.map(row, mapping);
+
+ for (int i = 0; i < mapping.length; i++) {
+ if (handler.get(i, key) == null) {
+ return NULL_KEY;
+ }
+ }
+
+ return new RowWrapper<>(key, handler, mapping.length);
+ }
+
+ /** Non-comparable key object. */
+ private static class Key {
+ }
+
+ /** Comparable key object. */
+ private static class RowWrapper<RowT> extends Key {
RowT row;
RowHandler<RowT> handler;
int itemsCount;
@@ -661,7 +703,7 @@ public abstract class HashJoinNode<RowT> extends AbstractRightMaterializedJoinNo
RowWrapper<RowT> row0 = (RowWrapper<RowT>) obj;
for (int i = 0; i < itemsCount; ++i) {
- Object input = handler.get(i, row0.row);
+ Object input = row0.handler.get(i, row0.row);
Object current = handler.get(i, row);
boolean comp = Objects.equals(input, current);
if (!comp) {
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/MergeJoinNode.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/MergeJoinNode.java
index 16182a88a0..9f1e757fa0 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/MergeJoinNode.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/MergeJoinNode.java
@@ -30,6 +30,7 @@ import org.apache.calcite.rel.core.JoinRelType;
import org.apache.calcite.rel.type.RelDataType;
import org.apache.ignite.internal.sql.engine.exec.ExecutionContext;
import org.apache.ignite.internal.sql.engine.exec.RowHandler;
+import org.apache.ignite.internal.sql.engine.exec.RowHandler.RowFactory;
import org.apache.ignite.internal.sql.engine.exec.row.RowSchema;
/**
@@ -42,7 +43,7 @@ public abstract class MergeJoinNode<RowT> extends AbstractNode<RowT> {
protected final Comparator<RowT> comp;
- protected final RowHandler<RowT> handler;
+ protected final RowFactory<RowT> outputRowFactory;
protected int requested;
@@ -57,17 +58,17 @@ public abstract class MergeJoinNode<RowT> extends AbstractNode<RowT> {
protected boolean inLoop;
/**
- * Constructor.
- * TODO Documentation https://issues.apache.org/jira/browse/IGNITE-15859
+ * Creates MergeJoinNode.
*
* @param ctx Execution context.
* @param comp Join expression.
+ * @param outputRowFactory Output row factory.
*/
- private MergeJoinNode(ExecutionContext<RowT> ctx, Comparator<RowT> comp) {
+ private MergeJoinNode(ExecutionContext<RowT> ctx, Comparator<RowT> comp, RowFactory<RowT> outputRowFactory) {
super(ctx);
this.comp = comp;
- handler = ctx.rowHandler();
+ this.outputRowFactory = outputRowFactory;
}
/** {@inheritDoc} */
@@ -209,43 +210,63 @@ public abstract class MergeJoinNode<RowT> extends AbstractNode<RowT> {
protected abstract void join() throws Exception;
/**
- * Create.
- * TODO Documentation https://issues.apache.org/jira/browse/IGNITE-15859
+ * Create MergeJoinNode for requested join operator type.
+ *
+ * @param ctx Execution context.
+ * @param leftRowType Row type of the left source.
+ * @param rightRowType Row type of the right source.
+ * @param joinType Join operator type.
+ * @param comp Join expression comparator.
*/
public static <RowT> MergeJoinNode<RowT> create(ExecutionContext<RowT> ctx, RelDataType leftRowType,
RelDataType rightRowType, JoinRelType joinType, Comparator<RowT> comp) {
+ RowSchema leftRowSchema = rowSchemaFromRelTypes(RelOptUtil.getFieldTypeList(leftRowType));
+ RowSchema rightRowSchema = rowSchemaFromRelTypes(RelOptUtil.getFieldTypeList(rightRowType));
+
switch (joinType) {
- case INNER:
- return new InnerJoin<>(ctx, comp);
+ case INNER: {
+ RowSchema outputSchema = RowSchema.concat(leftRowSchema, rightRowSchema);
+ RowFactory<RowT> outputRowFactory = ctx.rowHandler().factory(outputSchema);
+
+ return new InnerJoin<>(ctx, comp, outputRowFactory);
+ }
case LEFT: {
- RowSchema rightRowSchema = rowSchemaFromRelTypes(RelOptUtil.getFieldTypeList(rightRowType));
+ RowSchema outputSchema = RowSchema.concat(leftRowSchema, rightRowSchema);
+ RowFactory<RowT> outputRowFactory = ctx.rowHandler().factory(outputSchema);
RowHandler.RowFactory<RowT> rightRowFactory = ctx.rowHandler().factory(rightRowSchema);
- return new LeftJoin<>(ctx, comp, rightRowFactory);
+ return new LeftJoin<>(ctx, comp, outputRowFactory, rightRowFactory);
}
case RIGHT: {
- RowSchema leftRowSchema = rowSchemaFromRelTypes(RelOptUtil.getFieldTypeList(leftRowType));
+ RowSchema outputSchema = RowSchema.concat(leftRowSchema, rightRowSchema);
+ RowFactory<RowT> outputRowFactory = ctx.rowHandler().factory(outputSchema);
RowHandler.RowFactory<RowT> leftRowFactory = ctx.rowHandler().factory(leftRowSchema);
- return new RightJoin<>(ctx, comp, leftRowFactory);
+ return new RightJoin<>(ctx, comp, outputRowFactory, leftRowFactory);
}
case FULL: {
- RowSchema leftRowSchema = rowSchemaFromRelTypes(RelOptUtil.getFieldTypeList(leftRowType));
- RowSchema rightRowSchema = rowSchemaFromRelTypes(RelOptUtil.getFieldTypeList(rightRowType));
+ RowSchema outputSchema = RowSchema.concat(leftRowSchema, rightRowSchema);
+ RowFactory<RowT> outputRowFactory = ctx.rowHandler().factory(outputSchema);
RowHandler.RowFactory<RowT> leftRowFactory = ctx.rowHandler().factory(leftRowSchema);
RowHandler.RowFactory<RowT> rightRowFactory = ctx.rowHandler().factory(rightRowSchema);
- return new FullOuterJoin<>(ctx, comp, leftRowFactory, rightRowFactory);
+ return new FullOuterJoin<>(ctx, comp, outputRowFactory, leftRowFactory, rightRowFactory);
}
- case SEMI:
- return new SemiJoin<>(ctx, comp);
+ case SEMI: {
+ RowHandler.RowFactory<RowT> leftRowFactory = ctx.rowHandler().factory(leftRowSchema);
+
+ return new SemiJoin<>(ctx, comp, leftRowFactory);
+ }
- case ANTI:
- return new AntiJoin<>(ctx, comp);
+ case ANTI: {
+ RowHandler.RowFactory<RowT> leftRowFactory = ctx.rowHandler().factory(leftRowSchema);
+
+ return new AntiJoin<>(ctx, comp, leftRowFactory);
+ }
default:
throw new IllegalStateException("Join type \"" + joinType + "\" is not supported yet");
@@ -265,13 +286,14 @@ public abstract class MergeJoinNode<RowT> extends AbstractNode<RowT> {
private boolean drainMaterialization;
/**
- * Constructor.
+ * Creates MergeJoinNode for INNER JOIN operator.
*
- * @param ctx Execution context.
- * @param comp Join expression comparator.
+ * @param ctx Execution context.
+ * @param comp Join expression comparator.
+ * @param outputRowFactory Output row factory.
*/
- private InnerJoin(ExecutionContext<RowT> ctx, Comparator<RowT> comp) {
- super(ctx, comp);
+ private InnerJoin(ExecutionContext<RowT> ctx, Comparator<RowT> comp, RowFactory<RowT> outputRowFactory) {
+ super(ctx, comp, outputRowFactory);
}
/** {@inheritDoc} */
@@ -347,7 +369,7 @@ public abstract class MergeJoinNode<RowT> extends AbstractNode<RowT> {
}
}
- row = handler.concat(left, right);
+ row = outputRowFactory.concat(left, right);
if (rightMaterialization != null) {
rightMaterialization.add(right);
@@ -376,7 +398,7 @@ public abstract class MergeJoinNode<RowT> extends AbstractNode<RowT> {
continue;
}
- row = handler.concat(left, right);
+ row = outputRowFactory.concat(left, right);
}
requested--;
@@ -422,18 +444,20 @@ public abstract class MergeJoinNode<RowT> extends AbstractNode<RowT> {
private boolean matched;
/**
- * Constructor.
+ * Creates MergeJoinNode for LEFT OUTER JOIN operator.
*
- * @param ctx Execution context.
- * @param comp Join expression comparator.
- * @param rightRowFactory Right row factory.
+ * @param ctx Execution context.
+ * @param comp Join expression comparator.
+ * @param outputRowFactory Output row factory.
+ * @param rightRowFactory Row factory for the right source.
*/
private LeftJoin(
ExecutionContext<RowT> ctx,
Comparator<RowT> comp,
- RowHandler.RowFactory<RowT> rightRowFactory
+ RowFactory<RowT> outputRowFactory,
+ RowFactory<RowT> rightRowFactory
) {
- super(ctx, comp);
+ super(ctx, comp, outputRowFactory);
this.rightRowFactory = rightRowFactory;
}
@@ -485,7 +509,7 @@ public abstract class MergeJoinNode<RowT> extends AbstractNode<RowT> {
RowT row;
if (!drainMaterialization) {
if (right == null) {
- row = handler.concat(left, rightRowFactory.create());
+ row = outputRowFactory.concat(left, rightRowFactory.create());
requested--;
downstream().push(row);
@@ -499,7 +523,7 @@ public abstract class MergeJoinNode<RowT> extends AbstractNode<RowT> {
if (cmp < 0) {
if (!matched) {
- row = handler.concat(left, rightRowFactory.create());
+ row = outputRowFactory.concat(left, rightRowFactory.create());
requested--;
downstream().push(row);
@@ -533,7 +557,7 @@ public abstract class MergeJoinNode<RowT> extends AbstractNode<RowT> {
}
}
- row = handler.concat(left, right);
+ row = outputRowFactory.concat(left, right);
if (rightMaterialization != null) {
rightMaterialization.add(right);
@@ -562,7 +586,7 @@ public abstract class MergeJoinNode<RowT> extends AbstractNode<RowT> {
continue;
}
- row = handler.concat(left, right);
+ row = outputRowFactory.concat(left, right);
}
requested--;
@@ -606,19 +630,20 @@ public abstract class MergeJoinNode<RowT> extends AbstractNode<RowT> {
private boolean matched;
/**
- * Constructor.
- * TODO Documentation https://issues.apache.org/jira/browse/IGNITE-15859
+ * Creates MergeJoinNode for RIGHT OUTER JOIN operator.
*
- * @param ctx Execution context.
- * @param comp Join expression comparator.
- * @param leftRowFactory Left row factory.
+ * @param ctx Execution context.
+ * @param comp Join expression comparator.
+ * @param outputRowFactory Output row factory.
+ * @param leftRowFactory Row factory for the left source.
*/
private RightJoin(
ExecutionContext<RowT> ctx,
Comparator<RowT> comp,
+ RowHandler.RowFactory<RowT> outputRowFactory,
RowHandler.RowFactory<RowT> leftRowFactory
) {
- super(ctx, comp);
+ super(ctx, comp, outputRowFactory);
this.leftRowFactory = leftRowFactory;
}
@@ -671,7 +696,7 @@ public abstract class MergeJoinNode<RowT> extends AbstractNode<RowT> {
if (!drainMaterialization) {
if (left == null) {
if (!matched) {
- row = handler.concat(leftRowFactory.create(), right);
+ row = outputRowFactory.concat(leftRowFactory.create(), right);
requested--;
downstream().push(row);
@@ -695,7 +720,7 @@ public abstract class MergeJoinNode<RowT> extends AbstractNode<RowT> {
continue;
} else if (cmp > 0) {
if (!matched) {
- row = handler.concat(leftRowFactory.create(), right);
+ row = outputRowFactory.concat(leftRowFactory.create(), right);
requested--;
downstream().push(row);
@@ -720,7 +745,7 @@ public abstract class MergeJoinNode<RowT> extends AbstractNode<RowT> {
matched = true;
- row = handler.concat(left, right);
+ row = outputRowFactory.concat(left, right);
if (rightMaterialization != null) {
rightMaterialization.add(right);
@@ -757,7 +782,7 @@ public abstract class MergeJoinNode<RowT> extends AbstractNode<RowT> {
continue;
}
- row = handler.concat(left, right);
+ row = outputRowFactory.concat(left, right);
}
requested--;
@@ -807,17 +832,22 @@ public abstract class MergeJoinNode<RowT> extends AbstractNode<RowT> {
private boolean rightMatched;
/**
- * Constructor.
- * TODO Documentation https://issues.apache.org/jira/browse/IGNITE-15859
+ * Creates MergeJoinNode for FULL OUTER JOIN operator.
*
- * @param ctx Execution context.
- * @param comp Join expression comparator.
- * @param leftRowFactory Left row factory.
- * @param rightRowFactory Right row factory.
+ * @param ctx Execution context.
+ * @param comp Join expression comparator.
+ * @param outputRowFactory Output row factory.
+ * @param leftRowFactory Row factory for the left source.
+ * @param rightRowFactory Row factory for the right source.
*/
- private FullOuterJoin(ExecutionContext<RowT> ctx, Comparator<RowT> comp,
- RowHandler.RowFactory<RowT> leftRowFactory, RowHandler.RowFactory<RowT> rightRowFactory) {
- super(ctx, comp);
+ private FullOuterJoin(
+ ExecutionContext<RowT> ctx,
+ Comparator<RowT> comp,
+ RowHandler.RowFactory<RowT> outputRowFactory,
+ RowHandler.RowFactory<RowT> leftRowFactory,
+ RowHandler.RowFactory<RowT> rightRowFactory
+ ) {
+ super(ctx, comp, outputRowFactory);
this.leftRowFactory = leftRowFactory;
this.rightRowFactory = rightRowFactory;
@@ -874,7 +904,7 @@ public abstract class MergeJoinNode<RowT> extends AbstractNode<RowT> {
if (left == null || right == null) {
if (left == null && right != null) {
if (!rightMatched) {
- row = handler.concat(leftRowFactory.create(), right);
+ row = outputRowFactory.concat(leftRowFactory.create(), right);
requested--;
downstream().push(row);
@@ -887,7 +917,7 @@ public abstract class MergeJoinNode<RowT> extends AbstractNode<RowT> {
if (left != null && right == null) {
if (!leftMatched) {
- row = handler.concat(left, rightRowFactory.create());
+ row = outputRowFactory.concat(left, rightRowFactory.create());
requested--;
downstream().push(row);
@@ -905,7 +935,7 @@ public abstract class MergeJoinNode<RowT> extends AbstractNode<RowT> {
if (cmp < 0) {
if (!leftMatched) {
- row = handler.concat(left, rightRowFactory.create());
+ row = outputRowFactory.concat(left, rightRowFactory.create());
requested--;
downstream().push(row);
@@ -921,7 +951,7 @@ public abstract class MergeJoinNode<RowT> extends AbstractNode<RowT> {
continue;
} else if (cmp > 0) {
if (!rightMatched) {
- row = handler.concat(leftRowFactory.create(), right);
+ row = outputRowFactory.concat(leftRowFactory.create(), right);
requested--;
downstream().push(row);
@@ -947,7 +977,7 @@ public abstract class MergeJoinNode<RowT> extends AbstractNode<RowT> {
leftMatched = true;
rightMatched = true;
- row = handler.concat(left, right);
+ row = outputRowFactory.concat(left, right);
if (rightMaterialization != null) {
rightMaterialization.add(right);
@@ -986,7 +1016,7 @@ public abstract class MergeJoinNode<RowT> extends AbstractNode<RowT> {
leftMatched = true;
- row = handler.concat(left, right);
+ row = outputRowFactory.concat(left, right);
}
requested--;
@@ -1019,14 +1049,14 @@ public abstract class MergeJoinNode<RowT> extends AbstractNode<RowT> {
private RowT right;
/**
- * Constructor.
- * TODO Documentation https://issues.apache.org/jira/browse/IGNITE-15859
+ * Creates MergeJoinNode for SEMI JOIN operator.
*
- * @param ctx Execution context.
- * @param comp Join expression comparator.
+ * @param ctx Execution context.
+ * @param comp Join expression comparator.
+ * @param outputRowFactory Output row factory.
*/
- private SemiJoin(ExecutionContext<RowT> ctx, Comparator<RowT> comp) {
- super(ctx, comp);
+ private SemiJoin(ExecutionContext<RowT> ctx, Comparator<RowT> comp, RowFactory<RowT> outputRowFactory) {
+ super(ctx, comp, outputRowFactory);
}
/** {@inheritDoc} */
@@ -1098,14 +1128,14 @@ public abstract class MergeJoinNode<RowT> extends AbstractNode<RowT> {
private RowT right;
/**
- * Constructor.
- * TODO Documentation https://issues.apache.org/jira/browse/IGNITE-15859
+ * Creates MergeJoinNode for ANTI JOIN operator.
*
- * @param ctx Execution context.
- * @param comp Join expression comparator.
+ * @param ctx Execution context.
+ * @param comp Join expression comparator.
+ * @param outputRowFactory Output row factory.
*/
- private AntiJoin(ExecutionContext<RowT> ctx, Comparator<RowT> comp) {
- super(ctx, comp);
+ private AntiJoin(ExecutionContext<RowT> ctx, Comparator<RowT> comp, RowFactory<RowT> outputRowFactory) {
+ super(ctx, comp, outputRowFactory);
}
/** {@inheritDoc} */
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/ModifyNode.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/ModifyNode.java
index 6ec7bfff65..3e85e53431 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/ModifyNode.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/ModifyNode.java
@@ -34,8 +34,6 @@ import org.apache.ignite.internal.sql.engine.exec.RowHandler.RowFactory;
import org.apache.ignite.internal.sql.engine.exec.UpdatableTable;
import org.apache.ignite.internal.sql.engine.exec.mapping.ColocationGroup;
import org.apache.ignite.internal.sql.engine.exec.row.RowSchema;
-import org.apache.ignite.internal.sql.engine.exec.row.RowSchema.Builder;
-import org.apache.ignite.internal.sql.engine.exec.row.TypeSpec;
import org.apache.ignite.internal.sql.engine.schema.ColumnDescriptor;
import org.apache.ignite.internal.sql.engine.schema.IgniteTable;
import org.apache.ignite.internal.sql.engine.schema.TableDescriptor;
@@ -132,7 +130,8 @@ public class ModifyNode<RowT> extends AbstractNode<RowT> implements SingleNode<R
this.modifyOp = op;
this.updateColumns = updateColumns;
- int fullRowSize = inputRowFactory.rowSchema().fields().size();
+ RowSchema rowSchema = inputRowFactory.rowSchema();
+ int fullRowSize = rowSchema.fields().size();
this.mapping = mapping(table.descriptor(), updateColumns, fullRowSize);
// insert mapping actually can be required only for INSERT and MERGE operations
@@ -145,8 +144,8 @@ public class ModifyNode<RowT> extends AbstractNode<RowT> implements SingleNode<R
this.insertRowMapping = null;
}
- RowSchema mappedRowSchema = getMappedRowSchema(inputRowFactory, mapping);
- RowSchema mappedInsertRowSchema = getMappedRowSchema(inputRowFactory, insertRowMapping);
+ RowSchema mappedRowSchema = mapping != null ? RowSchema.map(rowSchema, mapping) : rowSchema;
+ RowSchema mappedInsertRowSchema = insertRowMapping != null ? RowSchema.map(rowSchema, insertRowMapping) : rowSchema;
this.mappedRowFactory = ctx.rowHandler().factory(mappedRowSchema);
this.mappedInsertRowFactory = ctx.rowHandler().factory(mappedInsertRowSchema);
@@ -484,20 +483,4 @@ public class ModifyNode<RowT> extends AbstractNode<RowT> implements SingleNode<R
return mapping;
}
-
- private RowSchema getMappedRowSchema(RowFactory<RowT> inputRowFactory, int[] mapping) {
- RowSchema schema;
- if (mapping != null) {
- List<TypeSpec> fields = inputRowFactory.rowSchema().fields();
- Builder builder = RowSchema.builder();
- for (int i : mapping) {
- TypeSpec typeSpec = fields.get(i);
- builder.addField(typeSpec);
- }
- schema = builder.build();
- } else {
- schema = inputRowFactory.rowSchema();
- }
- return schema;
- }
}
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/NestedLoopJoinNode.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/NestedLoopJoinNode.java
index a2807b1c5a..1050c4d047 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/NestedLoopJoinNode.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/NestedLoopJoinNode.java
@@ -27,7 +27,7 @@ import org.apache.calcite.plan.RelOptUtil;
import org.apache.calcite.rel.core.JoinRelType;
import org.apache.calcite.rel.type.RelDataType;
import org.apache.ignite.internal.sql.engine.exec.ExecutionContext;
-import org.apache.ignite.internal.sql.engine.exec.RowHandler;
+import org.apache.ignite.internal.sql.engine.exec.RowHandler.RowFactory;
import org.apache.ignite.internal.sql.engine.exec.row.RowSchema;
import org.jetbrains.annotations.Nullable;
@@ -38,22 +38,22 @@ import org.jetbrains.annotations.Nullable;
public abstract class NestedLoopJoinNode<RowT> extends AbstractRightMaterializedJoinNode<RowT> {
protected final BiPredicate<RowT, RowT> cond;
- protected final RowHandler<RowT> handler;
+ final RowFactory<RowT> outputRowFactory;
final List<RowT> rightMaterialized = new ArrayList<>(inBufSize);
/**
- * Constructor.
- * TODO Documentation https://issues.apache.org/jira/browse/IGNITE-15859
+ * Creates NestedLoopJoinNode.
*
* @param ctx Execution context.
* @param cond Join expression.
+ * @param outputRowFactory Output row factory.
*/
- NestedLoopJoinNode(ExecutionContext<RowT> ctx, BiPredicate<RowT, RowT> cond) {
+ NestedLoopJoinNode(ExecutionContext<RowT> ctx, BiPredicate<RowT, RowT> cond, RowFactory<RowT> outputRowFactory) {
super(ctx);
this.cond = cond;
- handler = ctx.rowHandler();
+ this.outputRowFactory = outputRowFactory;
}
/** {@inheritDoc} */
@@ -81,43 +81,57 @@ public abstract class NestedLoopJoinNode<RowT> extends AbstractRightMaterialized
}
/**
- * Create.
- * TODO Documentation https://issues.apache.org/jira/browse/IGNITE-15859
+ * Create NestedLoopJoinNode for requested join operator type.
+ *
+ * @param ctx Execution context.
+ * @param outputRowType Output row type.
+ * @param leftRowType Row type of the left source.
+ * @param rightRowType Row type of the right source.
+ * @param joinType Join operator type.
+ * @param cond Join condition predicate.
*/
- public static <RowT> NestedLoopJoinNode<RowT> create(ExecutionContext<RowT> ctx, RelDataType outputRowType,
- RelDataType leftRowType, RelDataType rightRowType, JoinRelType joinType, BiPredicate<RowT, RowT> cond) {
+ public static <RowT> NestedLoopJoinNode<RowT> create(
+ ExecutionContext<RowT> ctx,
+ RelDataType outputRowType,
+ RelDataType leftRowType,
+ RelDataType rightRowType,
+ JoinRelType joinType,
+ BiPredicate<RowT, RowT> cond
+ ) {
+ RowSchema leftRowSchema = rowSchemaFromRelTypes(RelOptUtil.getFieldTypeList(leftRowType));
+ RowSchema rightRowSchema = rowSchemaFromRelTypes(RelOptUtil.getFieldTypeList(rightRowType));
+ RowSchema outputSchema = rowSchemaFromRelTypes(RelOptUtil.getFieldTypeList(outputRowType));
+
+ RowFactory<RowT> outputRowFactory = ctx.rowHandler().factory(outputSchema);
+
switch (joinType) {
case INNER:
- return new InnerJoin<>(ctx, cond);
+ return new InnerJoin<>(ctx, cond, outputRowFactory);
case LEFT: {
- RowSchema rightRowSchema = rowSchemaFromRelTypes(RelOptUtil.getFieldTypeList(rightRowType));
- RowHandler.RowFactory<RowT> rightRowFactory = ctx.rowHandler().factory(rightRowSchema);
+ RowFactory<RowT> rightRowFactory = ctx.rowHandler().factory(rightRowSchema);
- return new LeftJoin<>(ctx, cond, rightRowFactory);
+ return new LeftJoin<>(ctx, cond, outputRowFactory, rightRowFactory);
}
case RIGHT: {
- RowSchema leftRowSchema = rowSchemaFromRelTypes(RelOptUtil.getFieldTypeList(leftRowType));
- RowHandler.RowFactory<RowT> leftRowFactory = ctx.rowHandler().factory(leftRowSchema);
+ RowFactory<RowT> leftRowFactory = ctx.rowHandler().factory(leftRowSchema);
- return new RightJoin<>(ctx, cond, leftRowFactory);
+ return new RightJoin<>(ctx, cond, outputRowFactory, leftRowFactory);
}
case FULL: {
- RowSchema leftRowSchema = rowSchemaFromRelTypes(RelOptUtil.getFieldTypeList(leftRowType));
- RowSchema rightRowSchema = rowSchemaFromRelTypes(RelOptUtil.getFieldTypeList(rightRowType));
- RowHandler.RowFactory<RowT> leftRowFactory = ctx.rowHandler().factory(leftRowSchema);
- RowHandler.RowFactory<RowT> rightRowFactory = ctx.rowHandler().factory(rightRowSchema);
+ RowFactory<RowT> leftRowFactory = ctx.rowHandler().factory(leftRowSchema);
+ RowFactory<RowT> rightRowFactory = ctx.rowHandler().factory(rightRowSchema);
- return new FullOuterJoin<>(ctx, cond, leftRowFactory, rightRowFactory);
+ return new FullOuterJoin<>(ctx, cond, outputRowFactory, leftRowFactory, rightRowFactory);
}
case SEMI:
- return new SemiJoin<>(ctx, cond);
+ return new SemiJoin<>(ctx, cond, outputRowFactory);
case ANTI:
- return new AntiJoin<>(ctx, cond);
+ return new AntiJoin<>(ctx, cond, outputRowFactory);
default:
throw new IllegalStateException("Join type \"" + joinType + "\" is not supported yet");
@@ -128,14 +142,14 @@ public abstract class NestedLoopJoinNode<RowT> extends AbstractRightMaterialized
private int rightIdx;
/**
- * Constructor.
- * TODO Documentation https://issues.apache.org/jira/browse/IGNITE-15859
+ * Creates NestedLoopJoinNode for INNER JOIN operator.
*
- * @param ctx Execution context.
+ * @param ctx Execution context.
* @param cond Join expression.
+ * @param outputRowFactory Output row factory.
*/
- private InnerJoin(ExecutionContext<RowT> ctx, BiPredicate<RowT, RowT> cond) {
- super(ctx, cond);
+ private InnerJoin(ExecutionContext<RowT> ctx, BiPredicate<RowT, RowT> cond, RowFactory<RowT> outputRowFactory) {
+ super(ctx, cond, outputRowFactory);
}
/** {@inheritDoc} */
@@ -164,7 +178,7 @@ public abstract class NestedLoopJoinNode<RowT> extends AbstractRightMaterialized
}
requested--;
- RowT row = handler.concat(left, rightMaterialized.get(rightIdx - 1));
+ RowT row = outputRowFactory.concat(left, rightMaterialized.get(rightIdx - 1));
downstream().push(row);
}
@@ -195,7 +209,7 @@ public abstract class NestedLoopJoinNode<RowT> extends AbstractRightMaterialized
private static class LeftJoin<RowT> extends NestedLoopJoinNode<RowT> {
/** Right row factory. */
- private final RowHandler.RowFactory<RowT> rightRowFactory;
+ private final RowFactory<RowT> rightRowFactory;
/** Whether current left row was matched or not. */
private boolean matched;
@@ -203,19 +217,20 @@ public abstract class NestedLoopJoinNode<RowT> extends AbstractRightMaterialized
private int rightIdx;
/**
- * Constructor.
- * TODO Documentation https://issues.apache.org/jira/browse/IGNITE-15859
+ * Creates NestedLoopJoinNode for LEFT OUTER JOIN operator.
*
- * @param ctx Execution context.
+ * @param ctx Execution context.
* @param cond Join expression.
+ * @param outputRowFactory Output row factory.
* @param rightRowFactory Right row factory.
*/
private LeftJoin(
ExecutionContext<RowT> ctx,
BiPredicate<RowT, RowT> cond,
- RowHandler.RowFactory<RowT> rightRowFactory
+ RowFactory<RowT> outputRowFactory,
+ RowFactory<RowT> rightRowFactory
) {
- super(ctx, cond);
+ super(ctx, cond, outputRowFactory);
this.rightRowFactory = rightRowFactory;
}
@@ -251,7 +266,7 @@ public abstract class NestedLoopJoinNode<RowT> extends AbstractRightMaterialized
requested--;
matched = true;
- RowT row = handler.concat(left, rightMaterialized.get(rightIdx - 1));
+ RowT row = outputRowFactory.concat(left, rightMaterialized.get(rightIdx - 1));
downstream().push(row);
}
@@ -262,7 +277,7 @@ public abstract class NestedLoopJoinNode<RowT> extends AbstractRightMaterialized
requested--;
wasPushed = true;
- downstream().push(handler.concat(left, rightRowFactory.create()));
+ downstream().push(outputRowFactory.concat(left, rightRowFactory.create()));
}
if (matched || wasPushed) {
@@ -293,7 +308,7 @@ public abstract class NestedLoopJoinNode<RowT> extends AbstractRightMaterialized
private static class RightJoin<RowT> extends NestedLoopJoinNode<RowT> {
/** Left row factory. */
- private final RowHandler.RowFactory<RowT> leftRowFactory;
+ private final RowFactory<RowT> leftRowFactory;
private @Nullable BitSet rightNotMatchedIndexes;
@@ -302,19 +317,20 @@ public abstract class NestedLoopJoinNode<RowT> extends AbstractRightMaterialized
private int rightIdx;
/**
- * Constructor.
- * TODO Documentation https://issues.apache.org/jira/browse/IGNITE-15859
+ * Creates NestedLoopJoinNode for RIGHT OUTER JOIN operator.
*
- * @param ctx Execution context.
+ * @param ctx Execution context.
* @param cond Join expression.
+ * @param outputRowFactory Output row factory.
* @param leftRowFactory Left row factory.
*/
private RightJoin(
ExecutionContext<RowT> ctx,
BiPredicate<RowT, RowT> cond,
- RowHandler.RowFactory<RowT> leftRowFactory
+ RowFactory<RowT> outputRowFactory,
+ RowFactory<RowT> leftRowFactory
) {
- super(ctx, cond);
+ super(ctx, cond, outputRowFactory);
this.leftRowFactory = leftRowFactory;
}
@@ -358,7 +374,7 @@ public abstract class NestedLoopJoinNode<RowT> extends AbstractRightMaterialized
requested--;
rightNotMatchedIndexes.clear(rightIdx - 1);
- RowT joined = handler.concat(left, right);
+ RowT joined = outputRowFactory.concat(left, right);
downstream().push(joined);
}
@@ -386,7 +402,7 @@ public abstract class NestedLoopJoinNode<RowT> extends AbstractRightMaterialized
break;
}
- RowT row = handler.concat(leftRowFactory.create(), rightMaterialized.get(lastPushedInd));
+ RowT row = outputRowFactory.concat(leftRowFactory.create(), rightMaterialized.get(lastPushedInd));
rightNotMatchedIndexes.clear(lastPushedInd);
@@ -420,10 +436,10 @@ public abstract class NestedLoopJoinNode<RowT> extends AbstractRightMaterialized
private static class FullOuterJoin<RowT> extends NestedLoopJoinNode<RowT> {
/** Left row factory. */
- private final RowHandler.RowFactory<RowT> leftRowFactory;
+ private final RowFactory<RowT> leftRowFactory;
/** Right row factory. */
- private final RowHandler.RowFactory<RowT> rightRowFactory;
+ private final RowFactory<RowT> rightRowFactory;
/** Whether current left row was matched or not. */
private boolean leftMatched;
@@ -435,21 +451,22 @@ public abstract class NestedLoopJoinNode<RowT> extends AbstractRightMaterialized
private int rightIdx;
/**
- * Constructor.
- * TODO Documentation https://issues.apache.org/jira/browse/IGNITE-15859
+ * Creates NestedLoopJoinNode for FULL OUTER JOIN operator.
*
- * @param ctx Execution context.
+ * @param ctx Execution context.
* @param cond Join expression.
+ * @param outputRowFactory Output row factory.
* @param leftRowFactory Left row factory.
* @param rightRowFactory Right row factory.
*/
private FullOuterJoin(
ExecutionContext<RowT> ctx,
BiPredicate<RowT, RowT> cond,
- RowHandler.RowFactory<RowT> leftRowFactory,
- RowHandler.RowFactory<RowT> rightRowFactory
+ RowFactory<RowT> outputRowFactory,
+ RowFactory<RowT> leftRowFactory,
+ RowFactory<RowT> rightRowFactory
) {
- super(ctx, cond);
+ super(ctx, cond, outputRowFactory);
this.leftRowFactory = leftRowFactory;
this.rightRowFactory = rightRowFactory;
@@ -498,7 +515,7 @@ public abstract class NestedLoopJoinNode<RowT> extends AbstractRightMaterialized
leftMatched = true;
rightNotMatchedIndexes.clear(rightIdx - 1);
- RowT joined = handler.concat(left, right);
+ RowT joined = outputRowFactory.concat(left, right);
downstream().push(joined);
}
@@ -509,7 +526,7 @@ public abstract class NestedLoopJoinNode<RowT> extends AbstractRightMaterialized
requested--;
wasPushed = true;
- downstream().push(handler.concat(left, rightRowFactory.create()));
+ downstream().push(outputRowFactory.concat(left, rightRowFactory.create()));
}
if (leftMatched || wasPushed) {
@@ -537,7 +554,7 @@ public abstract class NestedLoopJoinNode<RowT> extends AbstractRightMaterialized
break;
}
- RowT row = handler.concat(leftRowFactory.create(), rightMaterialized.get(lastPushedInd));
+ RowT row = outputRowFactory.concat(leftRowFactory.create(), rightMaterialized.get(lastPushedInd));
rightNotMatchedIndexes.clear(lastPushedInd);
@@ -573,14 +590,14 @@ public abstract class NestedLoopJoinNode<RowT> extends AbstractRightMaterialized
private int rightIdx;
/**
- * Constructor.
- * TODO Documentation https://issues.apache.org/jira/browse/IGNITE-15859
+ * Creates NestedLoopJoinNode for SEMI JOIN operator.
*
* @param ctx Execution context.
* @param cond Join expression.
+ * @param outputRowFactory Output row factory.
*/
- private SemiJoin(ExecutionContext<RowT> ctx, BiPredicate<RowT, RowT> cond) {
- super(ctx, cond);
+ private SemiJoin(ExecutionContext<RowT> ctx, BiPredicate<RowT, RowT> cond, RowFactory<RowT> outputRowFactory) {
+ super(ctx, cond, outputRowFactory);
}
/** {@inheritDoc} */
@@ -642,14 +659,14 @@ public abstract class NestedLoopJoinNode<RowT> extends AbstractRightMaterialized
private int rightIdx;
/**
- * Constructor.
- * TODO Documentation https://issues.apache.org/jira/browse/IGNITE-15859
+ * Creates NestedLoopJoinNode for ANTI JOIN operator.
*
* @param ctx Execution context.
* @param cond Join expression.
+ * @param outputRowFactory Output row factory.
*/
- private AntiJoin(ExecutionContext<RowT> ctx, BiPredicate<RowT, RowT> cond) {
- super(ctx, cond);
+ private AntiJoin(ExecutionContext<RowT> ctx, BiPredicate<RowT, RowT> cond, RowFactory<RowT> outputRowFactory) {
+ super(ctx, cond, outputRowFactory);
}
@Override
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/row/RowSchema.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/row/RowSchema.java
index 2a4d5298f4..1d3cbb693e 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/row/RowSchema.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/row/RowSchema.java
@@ -27,7 +27,6 @@ import org.apache.ignite.internal.type.NativeType;
* Row schema used by execution engine.
*/
public final class RowSchema {
-
private final List<TypeSpec> fields;
private RowSchema(List<TypeSpec> types) {
@@ -101,4 +100,37 @@ public final class RowSchema {
return new RowSchema(types);
}
}
+
+ /**
+ * Creates a new schema based on original schema and provided mapping.
+ */
+ public static RowSchema map(RowSchema schema, int[] mapping) {
+ assert mapping != null && mapping.length > 0;
+
+ List<TypeSpec> fields = schema.fields();
+ Builder builder = builder();
+
+ for (int i : mapping) {
+ TypeSpec typeSpec = fields.get(i);
+ builder.addField(typeSpec);
+ }
+
+ return builder.build();
+ }
+
+ /**
+ * Creates a new schema as a concatenation of the given schemas.
+ */
+ public static RowSchema concat(RowSchema left, RowSchema right) {
+ Builder builder = builder();
+
+ for (TypeSpec typeSpec : left.fields()) {
+ builder.addField(typeSpec);
+ }
+ for (TypeSpec typeSpec : right.fields()) {
+ builder.addField(typeSpec);
+ }
+
+ return builder.build();
+ }
}
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/rel/ExecutionTest.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/rel/ExecutionTest.java
index 24bc4fce4f..a7106cc0e9 100644
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/rel/ExecutionTest.java
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/rel/ExecutionTest.java
@@ -23,6 +23,7 @@ import static org.apache.calcite.rel.core.JoinRelType.FULL;
import static org.apache.calcite.rel.core.JoinRelType.INNER;
import static org.apache.calcite.rel.core.JoinRelType.LEFT;
import static org.apache.ignite.internal.sql.engine.util.Commons.getFieldFromBiRows;
+import static org.apache.ignite.internal.sql.engine.util.TypeUtils.rowSchemaFromRelTypes;
import static org.apache.ignite.internal.testframework.IgniteTestUtils.assertThrowsWithCause;
import static org.apache.ignite.internal.util.ArrayUtils.asList;
import static org.junit.jupiter.api.Assertions.assertArrayEquals;
@@ -34,6 +35,7 @@ import java.util.Arrays;
import java.util.List;
import java.util.Set;
import java.util.stream.Stream;
+import org.apache.calcite.plan.RelOptUtil;
import org.apache.calcite.rel.core.CorrelationId;
import org.apache.calcite.rel.core.JoinRelType;
import org.apache.calcite.rel.type.RelDataType;
@@ -301,14 +303,8 @@ public class ExecutionTest extends AbstractExecutionTest<Object[]> {
ScanNode<Object[]> left = new ScanNode<>(ctx, new TestTable(leftSize, rowType));
ScanNode<Object[]> right = new ScanNode<>(ctx, new TestTable(rightSize, rowType));
- RowSchema joinRowType = RowSchema.builder()
- .addField(NativeTypes.INT32)
- .addField(NativeTypes.STRING)
- .addField(NativeTypes.INT32)
- .addField(NativeTypes.INT32)
- .addField(NativeTypes.STRING)
- .addField(NativeTypes.INT32)
- .build();
+ RowSchema rowSchema = rowSchemaFromRelTypes(RelOptUtil.getFieldTypeList(rowType));
+ RowSchema outputRowSchema = RowSchema.concat(rowSchema, rowSchema);
RowHandler<Object[]> hnd = ctx.rowHandler();
@@ -317,7 +313,8 @@ public class ExecutionTest extends AbstractExecutionTest<Object[]> {
(r1, r2) -> getFieldFromBiRows(hnd, 0, r1, r2).equals(getFieldFromBiRows(hnd, 3, r1, r2)),
Set.of(new CorrelationId(0)),
joinType,
- hnd.factory(joinRowType)
+ hnd.factory(rowSchema),
+ hnd.factory(outputRowSchema)
);
IgniteTestUtils.setFieldValue(join, "rightInBufferSize", rightBufSize);
@@ -365,8 +362,8 @@ public class ExecutionTest extends AbstractExecutionTest<Object[]> {
MergeJoinNode<Object[]> join = MergeJoinNode.create(
ctx,
- null,
- null,
+ rowType,
+ rowType,
INNER,
(r1, r2) -> {
Object o1 = r1[0];
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/rel/HashJoinExecutionTest.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/rel/HashJoinExecutionTest.java
index 3412ce7853..b36193301c 100644
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/rel/HashJoinExecutionTest.java
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/rel/HashJoinExecutionTest.java
@@ -24,6 +24,7 @@ import static org.junit.jupiter.api.Assertions.assertEquals;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Comparator;
import java.util.List;
import org.apache.calcite.rel.core.JoinInfo;
import org.apache.calcite.rel.type.RelDataType;
@@ -34,7 +35,7 @@ import org.apache.ignite.internal.sql.engine.util.TypeUtils;
import org.apache.ignite.internal.type.NativeTypes;
import org.junit.jupiter.api.Test;
-/** Yash join execution tests. */
+/** Hash join execution tests. */
public class HashJoinExecutionTest extends AbstractJoinExecutionTest {
@Override
JoinAlgo joinAlgo() {
@@ -89,9 +90,9 @@ public class HashJoinExecutionTest extends AbstractJoinExecutionTest {
Object[][] expected = {
{0, "Igor", "Core"},
- {3, "Alexey", "Core"},
{1, "Roman", "SQL"},
- {2, "Ivan", null}
+ {2, "Ivan", null},
+ {3, "Alexey", "Core"}
};
assert2DimArrayEquals(expected, rows);
@@ -116,10 +117,10 @@ public class HashJoinExecutionTest extends AbstractJoinExecutionTest {
assertEquals(4, rowsAfterRewind.size());
Object[][] expectedAfterRewind = {
- {2, "Ivan", "QA"},
- {1, "Roman", null},
{0, "Igor", null},
- {3, "Alexey", null},
+ {1, "Roman", null},
+ {2, "Ivan", "QA"},
+ {3, "Alexey", null}
};
assert2DimArrayEquals(expectedAfterRewind, rowsAfterRewind);
@@ -128,8 +129,9 @@ public class HashJoinExecutionTest extends AbstractJoinExecutionTest {
static void assert2DimArrayEquals(Object[][] expected, ArrayList<Object[]> actual) {
assertEquals(expected.length, actual.size(), "expected length: " + expected.length + ", actual length: " + actual.size());
- int length = expected.length;
+ actual.sort(Comparator.comparing(r -> (int) r[0]));
+ int length = expected.length;
for (int i = 0; i < length; ++i) {
Object[] exp = expected[i];
Object[] act = actual.get(i);
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/rel/ModifyNodeExecutionTest.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/rel/ModifyNodeExecutionTest.java
index f849c9f196..e2bb80063d 100644
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/rel/ModifyNodeExecutionTest.java
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/rel/ModifyNodeExecutionTest.java
@@ -43,6 +43,7 @@ import java.util.stream.Stream;
import org.apache.calcite.rel.core.TableModify.Operation;
import org.apache.ignite.internal.sql.engine.exec.ExecutionContext;
import org.apache.ignite.internal.sql.engine.exec.RowHandler;
+import org.apache.ignite.internal.sql.engine.exec.RowHandler.RowBuilder;
import org.apache.ignite.internal.sql.engine.exec.RowHandler.RowFactory;
import org.apache.ignite.internal.sql.engine.exec.SqlRowHandler;
import org.apache.ignite.internal.sql.engine.exec.SqlRowHandler.RowWrapper;
@@ -344,6 +345,8 @@ public class ModifyNodeExecutionTest extends AbstractExecutionTest<RowWrapper> {
.addField(NativeTypes.INT32, true)
.build();
+ RowSchema mergeRowSchema = RowSchema.concat(RowSchema.concat(srcRowSchema, dstRowSchema), updateSchema);
+
Mockito.reset(updatableTable);
TableDescriptor tableDescriptor = createTableDescriptor(dstRowSchema);
@@ -384,8 +387,10 @@ public class ModifyNodeExecutionTest extends AbstractExecutionTest<RowWrapper> {
RowWrapper update = updateFactory.create(4);
RowWrapper noUpdate = updateFactory.create(new Object[]{null});
- RowWrapper mergeRow1 = rowHandler.concat(rowHandler.concat(srcRow1, dstRow1), noUpdate);
- RowWrapper mergeRow2 = rowHandler.concat(rowHandler.concat(srcRow2, dstRow2), update);
+ RowFactory<RowWrapper> mergeRowFactory = rowHandler.factory(mergeRowSchema);
+
+ RowWrapper mergeRow1 = concatRow(mergeRowFactory, srcRow1, dstRow1, noUpdate);
+ RowWrapper mergeRow2 = concatRow(mergeRowFactory, srcRow2, dstRow2, update);
Builder inputRowBuilder = RowSchema.builder();
srcFactory.rowSchema().fields().forEach(inputRowBuilder::addField);
@@ -434,6 +439,20 @@ public class ModifyNodeExecutionTest extends AbstractExecutionTest<RowWrapper> {
}
}
+ private static RowWrapper concatRow(RowFactory<RowWrapper> rowFactory, RowWrapper... rows) {
+ RowHandler<RowWrapper> handler = rowFactory.handler();
+ RowBuilder<RowWrapper> builder = rowFactory.rowBuilder();
+
+ for (RowWrapper row : rows) {
+ int cols = handler.columnCount(row);
+ for (int i = 0; i < cols; i++) {
+ builder.addField(handler.get(i, row));
+ }
+ }
+
+ return builder.build();
+ }
+
private static TableDescriptor createTableDescriptor(RowSchema rowSchema) {
List<ColumnDescriptor> columns = new ArrayList<>();
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/row/RowSchemaTest.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/row/RowSchemaTest.java
index 0754b60ee8..31355bb1d3 100644
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/row/RowSchemaTest.java
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/row/RowSchemaTest.java
@@ -72,5 +72,59 @@ public class RowSchemaTest {
assertNotEquals(s1, s2);
}
+
+ {
+ RowSchema s1 = RowSchema.builder()
+ .addField(new BaseTypeSpec(NativeTypes.INT32))
+ .addField(new BaseTypeSpec(NativeTypes.STRING))
+ .build();
+
+ RowSchema s2 = RowSchema.builder()
+ .addField(new BaseTypeSpec(NativeTypes.STRING))
+ .addField(new BaseTypeSpec(NativeTypes.INT32))
+ .build();
+
+ assertNotEquals(s1, s2);
+ }
+ }
+
+ /** Schema concatenation. */
+ @Test
+ public void testSchemaConcat() {
+ RowSchema s1 = RowSchema.builder()
+ .addField(new BaseTypeSpec(NativeTypes.STRING))
+ .build();
+
+ RowSchema s2 = RowSchema.builder()
+ .addField(new BaseTypeSpec(NativeTypes.INT32))
+ .build();
+
+ RowSchema expected = RowSchema.builder()
+ .addField(new BaseTypeSpec(NativeTypes.STRING))
+ .addField(new BaseTypeSpec(NativeTypes.INT32))
+ .build();
+
+ RowSchema concatenated = RowSchema.concat(s1, s2);
+
+ assertEquals(expected, concatenated);
+ }
+
+ /** Schema projection. */
+ @Test
+ public void testSchemaProjection() {
+ RowSchema s1 = RowSchema.builder()
+ .addField(new BaseTypeSpec(NativeTypes.INT32))
+ .addField(new BaseTypeSpec(NativeTypes.INT64))
+ .addField(new BaseTypeSpec(NativeTypes.STRING))
+ .build();
+
+ RowSchema expected = RowSchema.builder()
+ .addField(new BaseTypeSpec(NativeTypes.STRING))
+ .addField(new BaseTypeSpec(NativeTypes.INT32))
+ .build();
+
+ RowSchema projected = RowSchema.map(s1, new int[]{2, 0});
+
+ assertEquals(expected, projected);
}
}
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/row/SqlRowHandlerTest.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/row/SqlRowHandlerTest.java
index 547c634293..f2348239bb 100644
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/row/SqlRowHandlerTest.java
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/row/SqlRowHandlerTest.java
@@ -103,14 +103,11 @@ public class SqlRowHandlerTest extends IgniteAbstractTest {
public void testConcat(boolean leftTupleRequired, boolean rightTupleRequired) {
ConcatTestParameters params = new ConcatTestParameters(leftTupleRequired, rightTupleRequired);
- RowWrapper concatenated = handler.concat(params.left, params.right);
int leftLen = params.leftData.length;
int rightLen = params.rightData.length;
int totalElementsCount = leftLen + rightLen;
- assertThat(handler.columnCount(concatenated), equalTo(totalElementsCount));
-
// Build combined schema.
Builder builder = RowSchema.builder();
params.leftSchema.fields().forEach(builder::addField);
@@ -118,11 +115,17 @@ public class SqlRowHandlerTest extends IgniteAbstractTest {
RowSchema concatenatedSchema = builder.build();
+ RowFactory<RowWrapper> rowFactory = handler.factory(concatenatedSchema);
+ RowHandler<RowWrapper> handler = rowFactory.handler();
+
+ RowWrapper concatenated = rowFactory.concat(params.left, params.right);
+ assertThat(handler.columnCount(concatenated), equalTo(totalElementsCount));
+
// Serialize.
BinaryTuple tuple = handler.toBinaryTuple(concatenated);
// Wrap into row.
- RowWrapper result = handler.factory(concatenatedSchema).create(tuple);
+ RowWrapper result = rowFactory.create(tuple);
for (int i = 0; i < leftLen; i++) {
TypeSpec typeSpec = params.leftSchema.fields().get(i);
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/ArrayRowHandler.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/ArrayRowHandler.java
index 9cbe771cff..da8d6ebae3 100644
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/ArrayRowHandler.java
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/ArrayRowHandler.java
@@ -61,12 +61,6 @@ public class ArrayRowHandler implements RowHandler<Object[]> {
return row[field] == null;
}
- /** {@inheritDoc} */
- @Override
- public Object[] concat(Object[] left, Object[] right) {
- return ArrayUtils.concat(left, right);
- }
-
/** {@inheritDoc} */
@Override
public int columnCount(Object[] row) {
@@ -158,6 +152,14 @@ public class ArrayRowHandler implements RowHandler<Object[]> {
return newRow;
}
+
+ /** {@inheritDoc} */
+ @Override
+ public Object[] concat(Object[] left, Object[] right) {
+ assert left.length + right.length == schemaLen;
+
+ return ArrayUtils.concat(left, right);
+ }
};
}