You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ca...@apache.org on 2024/12/09 16:02:30 UTC
(iotdb) branch beyyes/fix_mergesort_exist_null updated: fix multi join key
This is an automated email from the ASF dual-hosted git repository.
caogaofei pushed a commit to branch beyyes/fix_mergesort_exist_null
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/beyyes/fix_mergesort_exist_null by this push:
new 62ff83b0e5f fix multi join key
62ff83b0e5f is described below
commit 62ff83b0e5f41a2ddfad7839a18da02b37d9b36f
Author: Beyyes <cg...@foxmail.com>
AuthorDate: Tue Dec 10 00:02:14 2024 +0800
fix multi join key
---
.../db/it/IoTDBMultiIDsWithAttributesTableIT.java | 3 +
.../merge/comparator/JoinKeyComparatorFactory.java | 12 ++
.../relational/AbstractMergeSortJoinOperator.java | 186 +++++++++++++--------
.../relational/MergeSortFullOuterJoinOperator.java | 76 +++------
.../relational/MergeSortInnerJoinOperator.java | 46 ++---
.../plan/planner/TableOperatorGenerator.java | 146 ++++++++--------
.../optimizations/PushPredicateIntoTableScan.java | 64 +++----
7 files changed, 269 insertions(+), 264 deletions(-)
diff --git a/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBMultiIDsWithAttributesTableIT.java b/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBMultiIDsWithAttributesTableIT.java
index 5956055065c..759adf9628d 100644
--- a/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBMultiIDsWithAttributesTableIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBMultiIDsWithAttributesTableIT.java
@@ -1864,6 +1864,9 @@ public class IoTDBMultiIDsWithAttributesTableIT {
tableResultSetEqualTest(sql, expectedHeader, retArray, DATABASE_NAME);
}
+ @Test
+ public void mergeSortJoinTest() {}
+
public static String[] buildHeaders(int length) {
String[] expectedHeader = new String[length];
for (int i = 0; i < length; i++) {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/join/merge/comparator/JoinKeyComparatorFactory.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/join/merge/comparator/JoinKeyComparatorFactory.java
index 995d568802c..c8f04978d1b 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/join/merge/comparator/JoinKeyComparatorFactory.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/join/merge/comparator/JoinKeyComparatorFactory.java
@@ -21,8 +21,20 @@ package org.apache.iotdb.db.queryengine.execution.operator.process.join.merge.co
import org.apache.tsfile.read.common.type.Type;
+import java.util.ArrayList;
+import java.util.List;
+
public class JoinKeyComparatorFactory {
+ public static List<JoinKeyComparator> getComparators(
+ List<Type> joinKeyTypes, boolean isAscending) {
+ List<JoinKeyComparator> comparators = new ArrayList<>(joinKeyTypes.size());
+ for (Type joinKeyType : joinKeyTypes) {
+ comparators.add(getComparator(joinKeyType, isAscending));
+ }
+ return comparators;
+ }
+
public static JoinKeyComparator getComparator(Type type, boolean isAscending) {
switch (type.getTypeEnum()) {
case INT32:
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/AbstractMergeSortJoinOperator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/AbstractMergeSortJoinOperator.java
index 5d78bb2f55d..7771691f5ae 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/AbstractMergeSortJoinOperator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/AbstractMergeSortJoinOperator.java
@@ -31,7 +31,6 @@ import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.read.common.block.TsBlock;
import org.apache.tsfile.read.common.block.TsBlockBuilder;
import org.apache.tsfile.read.common.block.column.RunLengthEncodedColumn;
-import org.apache.tsfile.read.common.type.Type;
import java.util.ArrayList;
import java.util.List;
@@ -46,24 +45,22 @@ public abstract class AbstractMergeSortJoinOperator extends AbstractOperator {
protected final Operator leftChild;
protected TsBlock leftBlock;
protected int leftIndex; // start index of leftTsBlock
- protected final int leftJoinKeyPosition;
+ protected final int[] leftJoinKeyPositions;
protected final int[] leftOutputSymbolIdx;
protected boolean rightFinished;
protected final Operator rightChild;
protected List<TsBlock> rightBlockList = new ArrayList<>();
- protected final int rightJoinKeyPosition;
+ protected final int[] rightJoinKeyPositions;
protected int rightBlockListIdx;
protected int rightIndex; // start index of rightTsBlock
protected final int[] rightOutputSymbolIdx;
protected TsBlock cachedNextRightBlock; // next candidate right block after rightBlockList
protected boolean rightConsumedUp = false; // if all data of right child are consumed up
- protected final JoinKeyComparator comparator;
+ protected final List<JoinKeyComparator> comparators;
protected final TsBlockBuilder resultBuilder;
- protected final Type joinKeyType;
-
protected final MemoryReservationManager memoryReservationManager;
protected long maxUsedMemory;
@@ -72,23 +69,21 @@ public abstract class AbstractMergeSortJoinOperator extends AbstractOperator {
protected AbstractMergeSortJoinOperator(
OperatorContext operatorContext,
Operator leftChild,
- int leftJoinKeyPosition,
+ int[] leftJoinKeyPositions,
int[] leftOutputSymbolIdx,
Operator rightChild,
- int rightJoinKeyPosition,
+ int[] rightJoinKeyPositions,
int[] rightOutputSymbolIdx,
- JoinKeyComparator comparator,
- List<TSDataType> dataTypes,
- Type joinKeyType) {
+ List<JoinKeyComparator> comparators,
+ List<TSDataType> dataTypes) {
this.operatorContext = operatorContext;
this.leftChild = leftChild;
- this.leftJoinKeyPosition = leftJoinKeyPosition;
+ this.leftJoinKeyPositions = leftJoinKeyPositions;
this.leftOutputSymbolIdx = leftOutputSymbolIdx;
this.rightChild = rightChild;
- this.rightJoinKeyPosition = rightJoinKeyPosition;
+ this.rightJoinKeyPositions = rightJoinKeyPositions;
this.rightOutputSymbolIdx = rightOutputSymbolIdx;
- this.comparator = comparator;
- this.joinKeyType = joinKeyType;
+ this.comparators = comparators;
this.memoryReservationManager =
operatorContext
@@ -192,32 +187,64 @@ public abstract class AbstractMergeSortJoinOperator extends AbstractOperator {
rightIndex = 0;
}
+ protected boolean currentLeftHasNullValue() {
+ for (int leftJoinKeyPosition : leftJoinKeyPositions) {
+ if (leftBlock.getColumn(leftJoinKeyPosition).isNull(leftIndex)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ protected boolean currentRightHasNullValue() {
+ for (int rightJoinKeyPosition : rightJoinKeyPositions) {
+ if (rightBlockList
+ .get(rightBlockListIdx)
+ .getColumn(rightJoinKeyPosition)
+ .isNull(rightIndex)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ // check if the last value of the right is less than left
+ // TODO introduce joinKey list, if join key size equals to 1, can return true in inner join
protected boolean allRightLessThanLeft() {
- // check if the last value of the right is less than left
- // TODO introduce joinKey list, if join key size equals to 1, can return true in inner join
- // situation
- return comparator
- .lessThan(
- rightBlockList.get(rightBlockList.size() - 1),
- rightJoinKeyPosition,
- rightBlockList.get(rightBlockList.size() - 1).getPositionCount() - 1,
- leftBlock,
- leftJoinKeyPosition,
- leftIndex)
- .orElse(false);
+ for (int i = 0; i < comparators.size(); i++) {
+ if (!comparators
+ .get(i)
+ .lessThan(
+ rightBlockList.get(rightBlockList.size() - 1),
+ rightJoinKeyPositions[i],
+ rightBlockList.get(rightBlockList.size() - 1).getPositionCount() - 1,
+ leftBlock,
+ leftJoinKeyPositions[i],
+ leftIndex)
+ .orElse(false)) {
+ return false;
+ }
+ }
+ return true;
}
+ // check if the last value of the left is less than right
protected boolean allLeftLessThanRight() {
- // check if the last value of the left is less than right
- return comparator
- .lessThan(
- leftBlock,
- leftJoinKeyPosition,
- leftBlock.getPositionCount() - 1,
- rightBlockList.get(rightBlockListIdx),
- rightJoinKeyPosition,
- rightIndex)
- .orElse(false);
+ for (int i = 0; i < comparators.size(); i++) {
+ if (!comparators
+ .get(i)
+ .lessThan(
+ leftBlock,
+ leftJoinKeyPositions[i],
+ leftBlock.getPositionCount() - 1,
+ rightBlockList.get(rightBlockListIdx),
+ rightJoinKeyPositions[i],
+ rightIndex)
+ .orElse(false)) {
+ return false;
+ }
+ }
+ return true;
}
/**
@@ -227,24 +254,16 @@ public abstract class AbstractMergeSortJoinOperator extends AbstractOperator {
* value.
*/
protected boolean currentRoundNeedStop() {
- if (comparator
- .lessThan(
- rightBlockList.get(0),
- rightJoinKeyPosition,
- rightBlockList.get(0).getPositionCount() - 1,
- rightBlockList.get(rightBlockListIdx),
- rightJoinKeyPosition,
- rightIndex)
- .orElse(false)
- || comparator
- .lessThan(
- rightBlockList.get(0),
- rightJoinKeyPosition,
- rightBlockList.get(0).getPositionCount() - 1,
- leftBlock,
- leftJoinKeyPosition,
- leftIndex)
- .orElse(false)) {
+ if (lessThan(
+ rightBlockList.get(0),
+ rightBlockList.get(0).getPositionCount() - 1,
+ rightBlockList.get(rightBlockListIdx),
+ rightIndex)
+ || lessThan(
+ rightBlockList.get(0),
+ rightBlockList.get(0).getPositionCount() - 1,
+ leftBlock,
+ leftIndex)) {
for (int i = 0; i < rightBlockListIdx; i++) {
long size = rightBlockList.get(i).getRetainedSizeInBytes();
usedMemory -= size;
@@ -328,15 +347,8 @@ public abstract class AbstractMergeSortJoinOperator extends AbstractOperator {
// if first value of block equals to last value of rightBlockList.get(0), append this block
// to
// rightBlockList
- if (comparator
- .equalsTo(
- block,
- rightJoinKeyPosition,
- 0,
- rightBlockList.get(0),
- rightJoinKeyPosition,
- rightBlockList.get(0).getPositionCount() - 1)
- .orElse(false)) {
+ if (equalsTo(
+ block, 0, rightBlockList.get(0), rightBlockList.get(0).getPositionCount() - 1)) {
addRightBlockWithMemoryReservation(block);
} else {
cachedNextRightBlock = block;
@@ -380,15 +392,7 @@ public abstract class AbstractMergeSortJoinOperator extends AbstractOperator {
int tmpBlockIdx = rightBlockListIdx;
int tmpIdx = rightIndex;
boolean hasMatched = false;
- while (comparator
- .equalsTo(
- leftBlock,
- leftJoinKeyPosition,
- leftIndex,
- rightBlockList.get(tmpBlockIdx),
- rightJoinKeyPosition,
- tmpIdx)
- .orElse(false)) {
+ while (equalsTo(leftBlock, leftIndex, rightBlockList.get(tmpBlockIdx), tmpIdx)) {
hasMatched = true;
recordsWhenDataMatches();
appendValueToResultWhenMatches(tmpBlockIdx, tmpIdx);
@@ -406,6 +410,42 @@ public abstract class AbstractMergeSortJoinOperator extends AbstractOperator {
return hasMatched;
}
+ protected boolean lessThan(TsBlock leftBlock, int lIndex, TsBlock rightBlock, int rIndex) {
+ for (int i = 0; i < lIndex; i++) {
+ if (!comparators
+ .get(i)
+ .lessThan(
+ leftBlock,
+ leftJoinKeyPositions[i],
+ lIndex,
+ rightBlock,
+ rightJoinKeyPositions[i],
+ rIndex)
+ .orElse(false)) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ protected boolean equalsTo(TsBlock leftBlock, int lIndex, TsBlock rightBlock, int rIndex) {
+ for (int i = 0; i < lIndex; i++) {
+ if (!comparators
+ .get(i)
+ .equalsTo(
+ leftBlock,
+ leftJoinKeyPositions[i],
+ lIndex,
+ rightBlock,
+ rightJoinKeyPositions[i],
+ rIndex)
+ .orElse(false)) {
+ return false;
+ }
+ }
+ return true;
+ }
+
protected void appendLeftBlockData(
int[] leftOutputSymbolIdx, TsBlockBuilder resultBuilder, TsBlock leftBlock, int leftIndex) {
for (int i = 0; i < leftOutputSymbolIdx.length; i++) {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/MergeSortFullOuterJoinOperator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/MergeSortFullOuterJoinOperator.java
index eaefbbcb0df..0e36c10927d 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/MergeSortFullOuterJoinOperator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/MergeSortFullOuterJoinOperator.java
@@ -28,7 +28,6 @@ import org.apache.tsfile.block.column.Column;
import org.apache.tsfile.block.column.ColumnBuilder;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.read.common.block.TsBlock;
-import org.apache.tsfile.read.common.type.Type;
import org.apache.tsfile.utils.RamUsageEstimator;
import java.util.List;
@@ -38,7 +37,7 @@ public class MergeSortFullOuterJoinOperator extends AbstractMergeSortJoinOperato
private static final long INSTANCE_SIZE =
RamUsageEstimator.shallowSizeOfInstance(MergeSortFullOuterJoinOperator.class);
- private final BiFunction<Column, Integer, TsBlock> updateLastMatchedRowFunction;
+ private final BiFunction<Column[], Integer, TsBlock> updateLastMatchedRowFunction;
// stores last row matched join criteria, only used in outer join
private TsBlock lastMatchedRightBlock = null;
@@ -46,26 +45,24 @@ public class MergeSortFullOuterJoinOperator extends AbstractMergeSortJoinOperato
public MergeSortFullOuterJoinOperator(
OperatorContext operatorContext,
Operator leftChild,
- int leftJoinKeyPosition,
+ int[] leftJoinKeyPositions,
int[] leftOutputSymbolIdx,
Operator rightChild,
- int rightJoinKeyPosition,
+ int[] rightJoinKeyPositions,
int[] rightOutputSymbolIdx,
- JoinKeyComparator joinKeyComparator,
+ List<JoinKeyComparator> joinKeyComparators,
List<TSDataType> dataTypes,
- Type joinKeyType,
- BiFunction<Column, Integer, TsBlock> updateLastMatchedRowFunction) {
+ BiFunction<Column[], Integer, TsBlock> updateLastMatchedRowFunction) {
super(
operatorContext,
leftChild,
- leftJoinKeyPosition,
+ leftJoinKeyPositions,
leftOutputSymbolIdx,
rightChild,
- rightJoinKeyPosition,
+ rightJoinKeyPositions,
rightOutputSymbolIdx,
- joinKeyComparator,
- dataTypes,
- joinKeyType);
+ joinKeyComparators,
+ dataTypes);
this.updateLastMatchedRowFunction = updateLastMatchedRowFunction;
}
@@ -113,38 +110,19 @@ public class MergeSortFullOuterJoinOperator extends AbstractMergeSortJoinOperato
}
// append all NULL join key values in right with empty left
- while (rightBlockList
- .get(rightBlockListIdx)
- .getColumn(rightJoinKeyPosition)
- .isNull(rightIndex)) {
+ while (currentRightHasNullValue()) {
appendOneRightRowWithEmptyLeft();
if (rightFinishedWithIncIndex()) {
return true;
}
}
// continue right < left, until right >= left
- while (comparator
- .lessThan(
- rightBlockList.get(rightBlockListIdx),
- rightJoinKeyPosition,
- rightIndex,
- leftBlock,
- leftJoinKeyPosition,
- leftIndex)
- .orElse(false)) {
+ while (lessThan(rightBlockList.get(rightBlockListIdx), rightIndex, leftBlock, leftIndex)) {
if (lastMatchedRightBlock == null) {
appendOneRightRowWithEmptyLeft();
} else {
// CurrentRight can only be greater than or equal to lastMatchedRight.
- if (comparator
- .lessThan(
- lastMatchedRightBlock,
- 0,
- 0,
- rightBlockList.get(rightBlockListIdx),
- rightJoinKeyPosition,
- rightIndex)
- .orElse(false)) {
+ if (lessThan(lastMatchedRightBlock, 0, rightBlockList.get(rightBlockListIdx), rightIndex)) {
appendOneRightRowWithEmptyLeft();
}
}
@@ -158,22 +136,14 @@ public class MergeSortFullOuterJoinOperator extends AbstractMergeSortJoinOperato
}
// skip all NULL values in left, because NULL value can not appear in the inner join result
- while (leftBlock.getColumn(leftJoinKeyPosition).isNull(leftIndex)) {
+ while (currentLeftHasNullValue()) {
appendOneLeftRowWithEmptyRight();
if (leftFinishedWithIncIndex()) {
return true;
}
}
// continue left < right, until left >= right
- while (comparator
- .lessThan(
- leftBlock,
- leftJoinKeyPosition,
- leftIndex,
- rightBlockList.get(rightBlockListIdx),
- rightJoinKeyPosition,
- rightIndex)
- .orElse(false)) {
+ while (lessThan(leftBlock, leftIndex, rightBlockList.get(rightBlockListIdx), rightIndex)) {
appendOneLeftRowWithEmptyRight();
if (leftFinishedWithIncIndex()) {
return true;
@@ -189,8 +159,11 @@ public class MergeSortFullOuterJoinOperator extends AbstractMergeSortJoinOperato
@Override
protected void recordsWhenDataMatches() {
- lastMatchedRightBlock =
- updateLastMatchedRowFunction.apply(leftBlock.getColumn(leftJoinKeyPosition), leftIndex);
+ Column[] columns = new Column[leftJoinKeyPositions.length];
+ for (int i = 0; i < leftJoinKeyPositions.length; i++) {
+ columns[i] = leftBlock.getColumn(leftJoinKeyPositions[i]);
+ }
+ lastMatchedRightBlock = updateLastMatchedRowFunction.apply(columns, leftIndex);
}
private void buildUseRemainingBlocks() {
@@ -212,15 +185,8 @@ public class MergeSortFullOuterJoinOperator extends AbstractMergeSortJoinOperato
// if current right value is null, the right row with empty left will be appended in the join
// result.
if (lastMatchedRightBlock == null
- || comparator
- .lessThan(
- lastMatchedRightBlock,
- 0,
- 0,
- rightBlockList.get(rightBlockListIdx),
- rightJoinKeyPosition,
- rightIndex)
- .orElse(true)) {
+ || lessThan(
+ lastMatchedRightBlock, 0, rightBlockList.get(rightBlockListIdx), rightIndex)) {
for (int i = 0; i < leftOutputSymbolIdx.length; i++) {
ColumnBuilder columnBuilder = resultBuilder.getColumnBuilder(i);
columnBuilder.appendNull();
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/MergeSortInnerJoinOperator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/MergeSortInnerJoinOperator.java
index cd98fa1726e..45ced1d82a9 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/MergeSortInnerJoinOperator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/MergeSortInnerJoinOperator.java
@@ -25,7 +25,6 @@ import org.apache.iotdb.db.queryengine.execution.operator.OperatorContext;
import org.apache.iotdb.db.queryengine.execution.operator.process.join.merge.comparator.JoinKeyComparator;
import org.apache.tsfile.enums.TSDataType;
-import org.apache.tsfile.read.common.type.Type;
import org.apache.tsfile.utils.RamUsageEstimator;
import java.util.List;
@@ -37,25 +36,23 @@ public class MergeSortInnerJoinOperator extends AbstractMergeSortJoinOperator {
public MergeSortInnerJoinOperator(
OperatorContext operatorContext,
Operator leftChild,
- int leftJoinKeyPosition,
+ int[] leftJoinKeyPositions,
int[] leftOutputSymbolIdx,
Operator rightChild,
- int rightJoinKeyPosition,
+ int[] rightJoinKeyPositions,
int[] rightOutputSymbolIdx,
- JoinKeyComparator joinKeyComparator,
- List<TSDataType> dataTypes,
- Type joinKeyType) {
+ List<JoinKeyComparator> joinKeyComparators,
+ List<TSDataType> dataTypes) {
super(
operatorContext,
leftChild,
- leftJoinKeyPosition,
+ leftJoinKeyPositions,
leftOutputSymbolIdx,
rightChild,
- rightJoinKeyPosition,
+ rightJoinKeyPositions,
rightOutputSymbolIdx,
- joinKeyComparator,
- dataTypes,
- joinKeyType);
+ joinKeyComparators,
+ dataTypes);
}
@Override
@@ -88,24 +85,13 @@ public class MergeSortInnerJoinOperator extends AbstractMergeSortJoinOperator {
}
// skip all NULL values in right, because NULL value can not appear in the inner join result
- while (rightBlockList
- .get(rightBlockListIdx)
- .getColumn(rightJoinKeyPosition)
- .isNull(rightIndex)) {
+ while (currentRightHasNullValue()) {
if (rightFinishedWithIncIndex()) {
return true;
}
}
// continue right < left, until right >= left
- while (comparator
- .lessThan(
- rightBlockList.get(rightBlockListIdx),
- rightJoinKeyPosition,
- rightIndex,
- leftBlock,
- leftJoinKeyPosition,
- leftIndex)
- .orElse(false)) {
+ while (lessThan(rightBlockList.get(rightBlockListIdx), rightIndex, leftBlock, leftIndex)) {
if (rightFinishedWithIncIndex()) {
return true;
}
@@ -115,21 +101,13 @@ public class MergeSortInnerJoinOperator extends AbstractMergeSortJoinOperator {
}
// skip all NULL values in left, because NULL value can not appear in the inner join result
- while (leftBlock.getColumn(leftJoinKeyPosition).isNull(leftIndex)) {
+ while (currentLeftHasNullValue()) {
if (leftFinishedWithIncIndex()) {
return true;
}
}
// continue left < right, until left >= right
- while (comparator
- .lessThan(
- leftBlock,
- leftJoinKeyPosition,
- leftIndex,
- rightBlockList.get(rightBlockListIdx),
- rightJoinKeyPosition,
- rightIndex)
- .orElse(false)) {
+ while (lessThan(leftBlock, leftIndex, rightBlockList.get(rightBlockListIdx), rightIndex)) {
if (leftFinishedWithIncIndex()) {
return true;
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/TableOperatorGenerator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/TableOperatorGenerator.java
index 7728bf7895a..bcc0c5414ea 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/TableOperatorGenerator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/TableOperatorGenerator.java
@@ -165,7 +165,6 @@ import org.apache.tsfile.read.common.type.BinaryType;
import org.apache.tsfile.read.common.type.BlobType;
import org.apache.tsfile.read.common.type.BooleanType;
import org.apache.tsfile.read.common.type.Type;
-import org.apache.tsfile.read.common.type.TypeEnum;
import org.apache.tsfile.read.filter.basic.Filter;
import org.apache.tsfile.utils.Binary;
import org.apache.tsfile.write.schema.IMeasurementSchema;
@@ -1267,24 +1266,34 @@ public class TableOperatorGenerator extends PlanVisitor<Operator, LocalExecution
dataTypes);
}
- Integer leftJoinKeyPosition = leftColumnNamesMap.get(node.getCriteria().get(0).getLeft());
- if (leftJoinKeyPosition == null) {
- throw new IllegalStateException("Left child of JoinNode doesn't contain left join key.");
+ int size = node.getCriteria().size();
+ int[] leftJoinKeyPositions = new int[size];
+ for (int i = 0; i < size; i++) {
+ Integer leftJoinKeyPosition = leftColumnNamesMap.get(node.getCriteria().get(i).getLeft());
+ if (leftJoinKeyPosition == null) {
+ throw new IllegalStateException("Left child of JoinNode doesn't contain left join key.");
+ }
+ leftJoinKeyPositions[i] = leftJoinKeyPosition;
}
- Integer rightJoinKeyPosition = rightColumnNamesMap.get(node.getCriteria().get(0).getRight());
- if (rightJoinKeyPosition == null) {
- throw new IllegalStateException("Right child of JoinNode doesn't contain right join key.");
+ List<Type> joinKeyTypes = new ArrayList<>(size);
+ int[] rightJoinKeyPositions = new int[size];
+ for (int i = 0; i < size; i++) {
+ Integer rightJoinKeyPosition = rightColumnNamesMap.get(node.getCriteria().get(0).getRight());
+ if (rightJoinKeyPosition == null) {
+ throw new IllegalStateException("Right child of JoinNode doesn't contain right join key.");
+ }
+ rightJoinKeyPositions[i] = rightJoinKeyPosition;
+
+ Type leftJoinKeyType =
+ context.getTypeProvider().getTableModelType(node.getCriteria().get(0).getLeft());
+ checkArgument(
+ leftJoinKeyType
+ == context.getTypeProvider().getTableModelType(node.getCriteria().get(0).getRight()),
+ "Join key type mismatch.");
+ joinKeyTypes.add(leftJoinKeyType);
}
- Type leftJoinKeyType =
- context.getTypeProvider().getTableModelType(node.getCriteria().get(0).getLeft());
-
- checkArgument(
- leftJoinKeyType
- == context.getTypeProvider().getTableModelType(node.getCriteria().get(0).getRight()),
- "Join key type mismatch.");
-
if (requireNonNull(node.getJoinType()) == JoinNode.JoinType.INNER) {
OperatorContext operatorContext =
context
@@ -1296,14 +1305,13 @@ public class TableOperatorGenerator extends PlanVisitor<Operator, LocalExecution
return new MergeSortInnerJoinOperator(
operatorContext,
leftChild,
- leftJoinKeyPosition,
+ leftJoinKeyPositions,
leftOutputSymbolIdx,
rightChild,
- rightJoinKeyPosition,
+ rightJoinKeyPositions,
rightOutputSymbolIdx,
- JoinKeyComparatorFactory.getComparator(leftJoinKeyType, true),
- dataTypes,
- leftJoinKeyType);
+ JoinKeyComparatorFactory.getComparators(joinKeyTypes, true),
+ dataTypes);
} else if (requireNonNull(node.getJoinType()) == JoinNode.JoinType.FULL) {
OperatorContext operatorContext =
context
@@ -1315,66 +1323,64 @@ public class TableOperatorGenerator extends PlanVisitor<Operator, LocalExecution
return new MergeSortFullOuterJoinOperator(
operatorContext,
leftChild,
- leftJoinKeyPosition,
+ leftJoinKeyPositions,
leftOutputSymbolIdx,
rightChild,
- rightJoinKeyPosition,
+ rightJoinKeyPositions,
rightOutputSymbolIdx,
- JoinKeyComparatorFactory.getComparator(leftJoinKeyType, true),
+ JoinKeyComparatorFactory.getComparators(joinKeyTypes, true),
dataTypes,
- leftJoinKeyType,
- buildUpdateLastRowFunction(leftJoinKeyType.getTypeEnum()));
+ buildUpdateLastRowFunction(joinKeyTypes));
}
throw new IllegalStateException("Unsupported join type: " + node.getJoinType());
}
- private BiFunction<Column, Integer, TsBlock> buildUpdateLastRowFunction(TypeEnum type) {
- switch (type) {
- case INT32:
- case DATE:
- return (column, rowIndex) ->
- new TsBlock(
- 1,
- TIME_COLUMN_TEMPLATE,
- new IntColumn(1, Optional.empty(), new int[] {column.getInt(rowIndex)}));
- case INT64:
- case TIMESTAMP:
- return (column, rowIndex) ->
- new TsBlock(
- 1,
- TIME_COLUMN_TEMPLATE,
- new LongColumn(1, Optional.empty(), new long[] {column.getLong(rowIndex)}));
- case FLOAT:
- return (column, rowIndex) ->
- new TsBlock(
- 1,
- TIME_COLUMN_TEMPLATE,
- new FloatColumn(1, Optional.empty(), new float[] {column.getFloat(rowIndex)}));
- case DOUBLE:
- return (column, rowIndex) ->
- new TsBlock(
- 1,
- TIME_COLUMN_TEMPLATE,
- new DoubleColumn(1, Optional.empty(), new double[] {column.getDouble(rowIndex)}));
- case BOOLEAN:
- return (column, rowIndex) ->
- new TsBlock(
- 1,
- TIME_COLUMN_TEMPLATE,
+ private BiFunction<Column[], Integer, TsBlock> buildUpdateLastRowFunction(
+ List<Type> joinKeyTypes) {
+ return (inputColumns, rowIndex) -> {
+ Column[] valueColumns = new Column[joinKeyTypes.size()];
+ for (int i = 0; i < joinKeyTypes.size(); i++) {
+ Type joinKeyType = joinKeyTypes.get(i);
+ switch (joinKeyType.getTypeEnum()) {
+ case INT32:
+ case DATE:
+ valueColumns[i] =
+ new IntColumn(1, Optional.empty(), new int[] {inputColumns[i].getInt(rowIndex)});
+ break;
+ case INT64:
+ case TIMESTAMP:
+ valueColumns[i] =
+ new LongColumn(1, Optional.empty(), new long[] {inputColumns[i].getLong(rowIndex)});
+ break;
+ case FLOAT:
+ valueColumns[i] =
+ new FloatColumn(
+ 1, Optional.empty(), new float[] {inputColumns[i].getFloat(rowIndex)});
+ break;
+ case DOUBLE:
+ valueColumns[i] =
+ new DoubleColumn(
+ 1, Optional.empty(), new double[] {inputColumns[i].getDouble(rowIndex)});
+ break;
+ case BOOLEAN:
+ valueColumns[i] =
new BooleanColumn(
- 1, Optional.empty(), new boolean[] {column.getBoolean(rowIndex)}));
- case STRING:
- case TEXT:
- case BLOB:
- return (column, rowIndex) ->
- new TsBlock(
- 1,
- TIME_COLUMN_TEMPLATE,
- new BinaryColumn(1, Optional.empty(), new Binary[] {column.getBinary(rowIndex)}));
- default:
- throw new UnsupportedOperationException("Unsupported data type: " + type);
- }
+ 1, Optional.empty(), new boolean[] {inputColumns[i].getBoolean(rowIndex)});
+ break;
+ case STRING:
+ case TEXT:
+ case BLOB:
+ valueColumns[i] =
+ new BinaryColumn(
+ 1, Optional.empty(), new Binary[] {valueColumns[i].getBinary(rowIndex)});
+ break;
+ default:
+ throw new UnsupportedOperationException("Unsupported data type: " + joinKeyType);
+ }
+ }
+ return new TsBlock(1, TIME_COLUMN_TEMPLATE, valueColumns);
+ };
}
@Override
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/PushPredicateIntoTableScan.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/PushPredicateIntoTableScan.java
index b08503c68f1..4597ad94ee0 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/PushPredicateIntoTableScan.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/PushPredicateIntoTableScan.java
@@ -20,7 +20,6 @@
package org.apache.iotdb.db.queryengine.plan.relational.planner.optimizations;
import org.apache.iotdb.common.rpc.thrift.TTimePartitionSlot;
-import org.apache.iotdb.commons.conf.IoTDBConstant;
import org.apache.iotdb.commons.partition.DataPartition;
import org.apache.iotdb.commons.partition.DataPartitionQueryParam;
import org.apache.iotdb.db.conf.IoTDBConfig;
@@ -42,6 +41,7 @@ import org.apache.iotdb.db.queryengine.plan.relational.metadata.Metadata;
import org.apache.iotdb.db.queryengine.plan.relational.planner.Assignments;
import org.apache.iotdb.db.queryengine.plan.relational.planner.EqualityInference;
import org.apache.iotdb.db.queryengine.plan.relational.planner.OrderingScheme;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.SortOrder;
import org.apache.iotdb.db.queryengine.plan.relational.planner.Symbol;
import org.apache.iotdb.db.queryengine.plan.relational.planner.SymbolAllocator;
import org.apache.iotdb.db.queryengine.plan.relational.planner.ir.ReplaceSymbolInExpression;
@@ -67,6 +67,7 @@ import javax.annotation.Nullable;
import java.util.ArrayList;
import java.util.Collections;
+import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
@@ -585,7 +586,7 @@ public class PushPredicateIntoTableScan implements PlanOptimizer {
}
// todo: Remove this check after supporting join on multiple columns.
- checkArgument(equiJoinClauses.size() <= 1, "Only support Join on one column for now.");
+ // checkArgument(equiJoinClauses.size() <= 1, "Only support Join on one column for now.");
if (!equiJoinClauses.isEmpty() && hasFilter) {
equiJoinClauses.clear();
joinFilterBuilder.add(lastEquiJoinConjunct);
@@ -658,33 +659,9 @@ public class PushPredicateIntoTableScan implements PlanOptimizer {
node.isSpillable());
}
- // sort the left and right child of join node if it is not a cross join
+ // using MergeSortJoinNode, sort the left and right child of join node
if (!((JoinNode) output).isCrossJoin()) {
- JoinNode.EquiJoinClause joinCriteria = ((JoinNode) output).getCriteria().get(0);
- OrderingScheme leftOrderingScheme =
- new OrderingScheme(
- Collections.singletonList(joinCriteria.getLeft()),
- Collections.singletonMap(joinCriteria.getLeft(), ASC_NULLS_LAST));
- OrderingScheme rightOrderingScheme =
- new OrderingScheme(
- Collections.singletonList(joinCriteria.getRight()),
- Collections.singletonMap(joinCriteria.getRight(), ASC_NULLS_LAST));
- SortNode leftSortNode =
- new SortNode(
- queryId.genPlanNodeId(),
- ((JoinNode) output).getLeftChild(),
- leftOrderingScheme,
- false,
- false);
- SortNode rightSortNode =
- new SortNode(
- queryId.genPlanNodeId(),
- ((JoinNode) output).getRightChild(),
- rightOrderingScheme,
- false,
- false);
- ((JoinNode) output).setLeftChild(leftSortNode);
- ((JoinNode) output).setRightChild(rightSortNode);
+ appendSortNodeForMergeSortJoin((JoinNode) output);
}
if (!postJoinPredicate.equals(TRUE_LITERAL)) {
@@ -725,10 +702,7 @@ public class PushPredicateIntoTableScan implements PlanOptimizer {
// This is temporary workaround.
Expression left = equality.getLeft();
Expression right = equality.getRight();
- return (left instanceof SymbolReference
- && ((SymbolReference) left).getName().contains(IoTDBConstant.TIME))
- && (right instanceof SymbolReference
- && ((SymbolReference) right).getName().contains(IoTDBConstant.TIME));
+ return (left instanceof SymbolReference && right instanceof SymbolReference);
}
private Symbol symbolForExpression(Expression expression) {
@@ -740,6 +714,32 @@ public class PushPredicateIntoTableScan implements PlanOptimizer {
return symbolAllocator.newSymbol(expression, analysis.getType(expression));
}
+ private void appendSortNodeForMergeSortJoin(JoinNode joinNode) {
+ // TODO(beyyes) sort with the order: ID, Attr, Measurement
+
+ int size = joinNode.getCriteria().size();
+ List<Symbol> leftOrderBy = new ArrayList<>(size);
+ List<Symbol> rightOrderBy = new ArrayList<>(size);
+ Map<Symbol, SortOrder> leftOrderings = new HashMap<>(size);
+ Map<Symbol, SortOrder> rightOrderings = new HashMap<>(size);
+ for (JoinNode.EquiJoinClause equiJoinClause : joinNode.getCriteria()) {
+ leftOrderBy.add(equiJoinClause.getLeft());
+ leftOrderings.put(equiJoinClause.getLeft(), ASC_NULLS_LAST);
+ rightOrderBy.add(equiJoinClause.getRight());
+ rightOrderings.put(equiJoinClause.getRight(), ASC_NULLS_LAST);
+ }
+ OrderingScheme leftOrderingScheme = new OrderingScheme(leftOrderBy, leftOrderings);
+ OrderingScheme rightOrderingScheme = new OrderingScheme(rightOrderBy, rightOrderings);
+ SortNode leftSortNode =
+ new SortNode(
+ queryId.genPlanNodeId(), joinNode.getLeftChild(), leftOrderingScheme, false, false);
+ SortNode rightSortNode =
+ new SortNode(
+ queryId.genPlanNodeId(), joinNode.getRightChild(), rightOrderingScheme, false, false);
+ joinNode.setLeftChild(leftSortNode);
+ joinNode.setRightChild(rightSortNode);
+ }
+
@Override
public PlanNode visitInsertTablet(InsertTabletNode node, RewriteContext context) {
return node;