This is an automated email from the ASF dual-hosted git repository.
jackietien pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 6b17e71bbd9 Add join operator implementation, fix the problem in join
logical planner and distributed planner
6b17e71bbd9 is described below
commit 6b17e71bbd920f5112779c6dcd1bb717b2e7ceb0
Author: Beyyes <[email protected]>
AuthorDate: Mon Sep 23 12:06:13 2024 +0800
Add join operator implementation, fix the problem in join logical planner
and distributed planner
---
.../org/apache/iotdb/db/it/utils/TestUtils.java | 1 +
.../db/it/IoTDBMultiIDsWithAttributesTableIT.java | 119 ++++++
.../fragment/FragmentInstanceContext.java | 8 +-
.../source/relational/InnerJoinOperator.java | 407 +++++++++++++++++++++
.../relational/ColumnTransformerBuilder.java | 10 +-
.../db/queryengine/plan/analyze/TypeProvider.java | 4 +
.../plan/planner/OperatorTreeGenerator.java | 2 +-
.../plan/planner/TableOperatorGenerator.java | 50 ++-
.../plan/planner/plan/node/PlanGraphPrinter.java | 3 +
.../plan/planner/plan/node/PlanNodeType.java | 4 +
.../planner/plan/node/process/ExchangeNode.java | 12 +
.../plan/relational/analyzer/Analysis.java | 4 +
.../plan/relational/planner/RelationPlanner.java | 10 +-
.../planner/distribute/AddExchangeNodes.java | 1 +
.../distribute/TableDistributedPlanner.java | 11 +-
...AddTableScanColumnsToTypeProviderOptimizer.java | 66 ++++
.../planner/iterative/rule/InlineProjections.java | 11 +-
.../planner/iterative/rule/MergeLimitWithSort.java | 24 +-
.../iterative/rule/PruneJoinChildrenColumns.java | 77 ++++
.../iterative/rule/PruneTableScanColumns.java | 18 +-
.../relational/planner/iterative/rule/Util.java | 21 +-
.../plan/relational/planner/node/JoinNode.java | 90 ++++-
.../optimizations/DistributedOptimizeFactory.java | 56 +--
.../optimizations/LogicalOptimizeFactory.java | 163 ++++-----
.../optimizations/PushPredicateIntoTableScan.java | 155 ++++----
.../planner/optimizations/SortElimination.java | 21 +-
.../plan/relational/analyzer/JoinTest.java | 92 +++--
27 files changed, 1134 insertions(+), 306 deletions(-)
diff --git
a/integration-test/src/test/java/org/apache/iotdb/db/it/utils/TestUtils.java
b/integration-test/src/test/java/org/apache/iotdb/db/it/utils/TestUtils.java
index b4d8a521bd4..d95f5f60e89 100644
--- a/integration-test/src/test/java/org/apache/iotdb/db/it/utils/TestUtils.java
+++ b/integration-test/src/test/java/org/apache/iotdb/db/it/utils/TestUtils.java
@@ -253,6 +253,7 @@ public class TestUtils {
builder.append(resultSet.getString(i)).append(",");
}
assertEquals(expectedRetArray[cnt], builder.toString());
+ // System.out.println(String.format("\"%s\",",
builder.toString()));
cnt++;
}
assertEquals(expectedRetArray.length, cnt);
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 75cc1855bd6..348aebc6b5b 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
@@ -389,6 +389,7 @@ public class IoTDBMultiIDsWithAttributesTableIT {
DATABASE_NAME);
}
+ // ========== SubQuery Test =========
@Test
public void subQueryTest1() {
String[] expectedHeader = new String[] {"time", "level", "device",
"add_num"};
@@ -410,4 +411,122 @@ public class IoTDBMultiIDsWithAttributesTableIT {
retArray,
DATABASE_NAME);
}
+
+ // ========== Join Test =========
+ // no filter
+ @Test
+ public void innerJoinTest1() {
+ String[] expectedHeader =
+ new String[] {"time", "device", "level", "num", "device", "attr2",
"num", "str"};
+ String[] retArray =
+ new String[] {
+ "1970-01-01T00:00:00.000Z,d2,l1,3,d1,d,3,coconut,",
+ "1970-01-01T00:00:00.000Z,d2,l1,3,d2,c,3,coconut,",
+ "1970-01-01T00:00:00.020Z,d1,l2,2,d1,zz,2,pineapple,",
+ "1970-01-01T00:00:00.020Z,d1,l2,2,d2,null,2,pineapple,",
+ "1970-01-01T00:00:00.020Z,d2,l2,2,d1,zz,2,pineapple,",
+ "1970-01-01T00:00:00.020Z,d2,l2,2,d2,null,2,pineapple,"
+ };
+
+ // join on
+ String sql =
+ "SELECT t1.time as time, t1.device, t1.level, t1.num, t2.device,
t2.attr2, t2.num, t2.str\n"
+ + "FROM table0 t1 JOIN table0 t2 ON t1.time = t2.time \n"
+ + "ORDER BY t1.time, t1.device, t2.device OFFSET 2 LIMIT 6";
+ tableResultSetEqualTest(sql, expectedHeader, retArray, DATABASE_NAME);
+
+ // implicit join
+ sql =
+ "SELECT t1.time as time, t1.device, t1.level, t1.num, t2.device,
t2.attr2, t2.num, t2.str\n"
+ + "FROM table0 t1, table0 t2 WHERE t1.time = t2.time \n"
+ + "ORDER BY t1.time, t1.device, t2.device OFFSET 2 LIMIT 6";
+ tableResultSetEqualTest(sql, expectedHeader, retArray, DATABASE_NAME);
+
+ // join using
+ sql =
+ "SELECT time, t1.device, t1.level, t1.num, t2.device, t2.attr2,
t2.num, t2.str\n"
+ + "FROM table0 t1 JOIN table0 t2 USING(time)\n"
+ + "ORDER BY time, t1.device, t2.device OFFSET 2 LIMIT 6";
+ tableResultSetEqualTest(sql, expectedHeader, retArray, DATABASE_NAME);
+ }
+
+ // has filter
+ @Test
+ public void innerJoinTest2() {
+ String[] expectedHeader =
+ new String[] {"time", "device", "level", "t1_num_add", "device",
"attr2", "num", "str"};
+ String[] retArray =
+ new String[] {
+ "1970-01-01T00:00:00.080Z,d1,l4,10,d1,null,9,apple,",
+ "1970-01-01T00:00:00.080Z,d1,l4,10,d2,null,9,apple,",
+ "1970-01-01T00:00:00.080Z,d2,l4,10,d1,null,9,apple,",
+ "1970-01-01T00:00:00.080Z,d2,l4,10,d2,null,9,apple,",
+ "1971-01-01T00:00:00.100Z,d1,l2,11,d1,zz,10,pumelo,",
+ "1971-01-01T00:00:00.100Z,d1,l2,11,d2,null,10,pumelo,",
+ "1971-01-01T00:00:00.100Z,d2,l2,11,d1,zz,10,pumelo,",
+ "1971-01-01T00:00:00.100Z,d2,l2,11,d2,null,10,pumelo,",
+ "1971-01-01T00:00:00.500Z,d1,l3,5,d1,a,4,peach,",
+ "1971-01-01T00:00:00.500Z,d1,l3,5,d2,null,4,peach,",
+ "1971-01-01T00:00:00.500Z,d2,l3,5,d1,a,4,peach,",
+ "1971-01-01T00:00:00.500Z,d2,l3,5,d2,null,4,peach,",
+ "1971-01-01T00:00:01.000Z,d1,l4,6,d1,null,5,orange,",
+ "1971-01-01T00:00:01.000Z,d1,l4,6,d2,null,5,orange,",
+ "1971-01-01T00:00:01.000Z,d2,l4,6,d1,null,5,orange,",
+ "1971-01-01T00:00:01.000Z,d2,l4,6,d2,null,5,orange,",
+ };
+
+ // join on
+ String sql =
+ "SELECT t2.time,t1.device, t1.level, t1_num_add, t2.device, t2.attr2,
t2.num, t2.str\n"
+ + "FROM (SELECT *,num+1 as t1_num_add FROM table0 WHERE TIME>=80
AND level!='l1' AND cast(num as double)>0) t1 \n"
+ + "JOIN (SELECT * FROM table0 WHERE TIME<=31536001000 AND
floatNum<1000 AND device in ('d1','d2')) t2 \n"
+ + "ON t1.time = t2.time \n"
+ + "ORDER BY t1.time, t1.device, t2.device LIMIT 20";
+ tableResultSetEqualTest(sql, expectedHeader, retArray, DATABASE_NAME);
+
+ sql =
+ "SELECT t2.time,t1.device, t1.level, t1_num_add, t2.device, t2.attr2,
t2.num, t2.str\n"
+ + "FROM (SELECT *,num+1 as t1_num_add FROM table0) t1 \n"
+ + "JOIN (SELECT * FROM table0) t2 ON t1.time = t2.time \n"
+ + "WHERE t1.TIME>=80 AND cast(t1.num as double)>0 AND
t1.level!='l1' \n"
+ + "AND t2.time<=31536001000 AND t2.floatNum<1000 AND t2.device in
('d1','d2')\n"
+ + "ORDER BY t1.time, t1.device, t2.device LIMIT 20";
+ tableResultSetEqualTest(sql, expectedHeader, retArray, DATABASE_NAME);
+
+ sql =
+ "SELECT t2.time,t1.device, t1.level, t1_num_add, t2.device, t2.attr2,
t2.num, t2.str\n"
+ + "FROM (SELECT *,num+1 as t1_num_add FROM table0 WHERE time>=80)
t1 \n"
+ + "JOIN (SELECT * FROM table0 WHERE floatNum<1000) t2 ON t1.time =
t2.time \n"
+ + "WHERE cast(t1.num as double)>0 AND t1.level!='l1' \n"
+ + "AND t2.time<=31536001000 AND t2.device in ('d1','d2')\n"
+ + "ORDER BY t1.time, t1.device, t2.device LIMIT 20";
+ tableResultSetEqualTest(sql, expectedHeader, retArray, DATABASE_NAME);
+
+ // implicit join
+ sql =
+ "SELECT t2.time,t1.device, t1.level, t1_num_add, t2.device, t2.attr2,
t2.num, t2.str\n"
+ + "FROM (SELECT *,num+1 as t1_num_add FROM table0 WHERE TIME>=80
AND level!='l1' AND cast(num as double)>0) t1, \n"
+ + " (SELECT * FROM table0 WHERE TIME<=31536001000 AND
floatNum<1000 AND device in ('d1','d2')) t2 \n"
+ + "WHERE t1.time = t2.time \n"
+ + "ORDER BY t1.time, t1.device, t2.device LIMIT 20";
+ tableResultSetEqualTest(sql, expectedHeader, retArray, DATABASE_NAME);
+
+ sql =
+ "SELECT t2.time,t1.device, t1.level, t1_num_add, t2.device, t2.attr2,
t2.num, t2.str\n"
+ + "FROM (SELECT *,num+1 as t1_num_add FROM table0) t1, \n"
+ + " (SELECT * FROM table0) t2 \n"
+ + "WHERE t1.time=t2.time AND t1.TIME>=80 AND cast(t1.num as
double)>0 AND t1.level!='l1' \n"
+ + "AND t2.time<=31536001000 AND t2.floatNum<1000 AND t2.device in
('d1','d2')\n"
+ + "ORDER BY t1.time, t1.device, t2.device LIMIT 20";
+ tableResultSetEqualTest(sql, expectedHeader, retArray, DATABASE_NAME);
+
+ sql =
+ "SELECT t2.time,t1.device, t1.level, t1_num_add, t2.device, t2.attr2,
t2.num, t2.str\n"
+ + "FROM (SELECT *,num+1 as t1_num_add FROM table0 WHERE time>=80)
t1, \n"
+ + " (SELECT * FROM table0 WHERE floatNum<1000) t2 \n"
+ + "WHERE t1.time=t2.time AND cast(t1.num as double)>0 AND
t1.level!='l1' \n"
+ + "AND t2.time<=31536001000 AND t2.device in ('d1','d2')\n"
+ + "ORDER BY t1.time, t1.device, t2.device LIMIT 20";
+ tableResultSetEqualTest(sql, expectedHeader, retArray, DATABASE_NAME);
+ }
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/fragment/FragmentInstanceContext.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/fragment/FragmentInstanceContext.java
index 9519ecb0212..a2a3021a51b 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/fragment/FragmentInstanceContext.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/fragment/FragmentInstanceContext.java
@@ -45,6 +45,7 @@ import
org.apache.iotdb.mpp.rpc.thrift.TFetchFragmentInstanceStatisticsResp;
import org.apache.tsfile.file.metadata.IDeviceID;
import org.apache.tsfile.read.filter.basic.Filter;
+import org.apache.tsfile.read.filter.factory.FilterFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -405,8 +406,11 @@ public class FragmentInstanceContext extends QueryContext {
if (globalTimeFilter == null) {
globalTimeFilter = timeFilter;
} else {
- throw new IllegalStateException(
- "globalTimeFilter in FragmentInstanceContext should only be set once
in Table Model!");
+ // In join case, there may exist more than one table and time filter,
join criteria only
+ // support and condition, so use and to connect these two filters
+ globalTimeFilter = FilterFactory.and(globalTimeFilter, timeFilter);
+ // throw new IllegalStateException(
+ // "globalTimeFilter in FragmentInstanceContext should only be set
once in Table Model!");
}
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/InnerJoinOperator.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/InnerJoinOperator.java
new file mode 100644
index 00000000000..c086666ece5
--- /dev/null
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/InnerJoinOperator.java
@@ -0,0 +1,407 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.queryengine.execution.operator.source.relational;
+
+import org.apache.iotdb.db.queryengine.execution.MemoryEstimationHelper;
+import org.apache.iotdb.db.queryengine.execution.operator.Operator;
+import org.apache.iotdb.db.queryengine.execution.operator.OperatorContext;
+import
org.apache.iotdb.db.queryengine.execution.operator.process.ProcessOperator;
+import
org.apache.iotdb.db.queryengine.execution.operator.process.join.merge.TimeComparator;
+import
org.apache.iotdb.db.queryengine.plan.planner.memory.MemoryReservationManager;
+
+import com.google.common.util.concurrent.ListenableFuture;
+import org.apache.tsfile.block.column.Column;
+import org.apache.tsfile.block.column.ColumnBuilder;
+import org.apache.tsfile.common.conf.TSFileDescriptor;
+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.utils.RamUsageEstimator;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+import static com.google.common.util.concurrent.Futures.successfulAsList;
+import static
org.apache.iotdb.db.queryengine.execution.operator.source.relational.TableScanOperator.TIME_COLUMN_TEMPLATE;
+
+public class InnerJoinOperator implements ProcessOperator {
+ private static final long INSTANCE_SIZE =
+ RamUsageEstimator.shallowSizeOfInstance(InnerJoinOperator.class);
+
+ private final OperatorContext operatorContext;
+
+ private final Operator leftChild;
+ private TsBlock leftBlock;
+ private int leftIndex; // start index of leftTsBlock
+ private static final int TIME_COLUMN_POSITION = 0;
+ private final int[] leftOutputSymbolIdx;
+
+ private final Operator rightChild;
+ private final List<TsBlock> rightBlockList = new ArrayList<>();
+ private int rightBlockListIdx;
+ private int rightIndex; // start index of rightTsBlock
+ private final int[] rightOutputSymbolIdx;
+ private TsBlock cachedNextRightBlock;
+ private boolean hasCachedNextRightBlock;
+
+ private final TimeComparator comparator;
+ private final TsBlockBuilder resultBuilder;
+
+ private final long maxReturnSize =
+ TSFileDescriptor.getInstance().getConfig().getMaxTsBlockSizeInBytes();
+
+ protected MemoryReservationManager memoryReservationManager;
+
+ public InnerJoinOperator(
+ OperatorContext operatorContext,
+ Operator leftChild,
+ int[] leftOutputSymbolIdx,
+ Operator rightChild,
+ int[] rightOutputSymbolIdx,
+ TimeComparator timeComparator,
+ List<TSDataType> dataTypes) {
+ this.operatorContext = operatorContext;
+ this.leftChild = leftChild;
+ this.leftOutputSymbolIdx = leftOutputSymbolIdx;
+ this.rightChild = rightChild;
+ this.rightOutputSymbolIdx = rightOutputSymbolIdx;
+
+ this.comparator = timeComparator;
+ this.resultBuilder = new TsBlockBuilder(dataTypes);
+
+ this.memoryReservationManager =
+ operatorContext
+ .getDriverContext()
+ .getFragmentInstanceContext()
+ .getMemoryReservationContext();
+ }
+
+ @Override
+ public ListenableFuture<?> isBlocked() {
+ ListenableFuture<?> leftBlocked = leftChild.isBlocked();
+ ListenableFuture<?> rightBlocked = rightChild.isBlocked();
+ if (leftBlocked.isDone()) {
+ return rightBlocked;
+ } else if (rightBlocked.isDone()) {
+ return leftBlocked;
+ } else {
+ return successfulAsList(leftBlocked, rightBlocked);
+ }
+ }
+
+ @Override
+ public boolean hasNext() throws Exception {
+ return (leftBlockNotEmpty() || leftChild.hasNextWithTimer())
+ && (rightBlockNotEmpty() || rightChild.hasNextWithTimer());
+ }
+
+ @Override
+ public TsBlock next() throws Exception {
+ long maxRuntime =
operatorContext.getMaxRunTime().roundTo(TimeUnit.NANOSECONDS);
+ long start = System.nanoTime();
+ // prepare leftBlock and rightBlockList with cachedNextRightBlock
+ if (!prepareInput(start, maxRuntime)) {
+ return null;
+ }
+
+ // all the rightTsBlock is less than leftTsBlock, just skip right
+ if (comparator.lessThan(getRightEndTime(), getCurrentLeftTime())) {
+ for (int i = 1; i < rightBlockList.size(); i++) {
+ memoryReservationManager.releaseMemoryCumulatively(
+ rightBlockList.get(i).getRetainedSizeInBytes());
+ }
+ rightBlockList.clear();
+ rightBlockListIdx = 0;
+ rightIndex = 0;
+ return null;
+ }
+
+ // all the leftTsBlock is less than rightTsBlock, just skip left
+ else if (comparator.lessThan(getLeftEndTime(), getCurrentRightTime())) {
+ leftBlock = null;
+ leftIndex = 0;
+ return null;
+ }
+
+ long leftProbeTime = getCurrentLeftTime();
+ while (!resultBuilder.isFull()) {
+
+ // all right block time is not matched
+ if (!comparator.canContinueInclusive(leftProbeTime, getRightEndTime())) {
+ for (int i = 1; i < rightBlockList.size(); i++) {
+ memoryReservationManager.releaseMemoryCumulatively(
+ rightBlockList.get(i).getRetainedSizeInBytes());
+ }
+ rightBlockList.clear();
+ rightBlockListIdx = 0;
+ rightIndex = 0;
+ break;
+ }
+
+ appendResult(leftProbeTime);
+
+ leftIndex++;
+
+ if (leftIndex >= leftBlock.getPositionCount()) {
+ leftBlock = null;
+ leftIndex = 0;
+ break;
+ }
+
+ leftProbeTime = getCurrentLeftTime();
+ }
+
+ if (resultBuilder.isEmpty()) {
+ return null;
+ }
+
+ Column[] valueColumns = new
Column[resultBuilder.getValueColumnBuilders().length];
+ for (int i = 0; i < valueColumns.length; ++i) {
+ valueColumns[i] = resultBuilder.getValueColumnBuilders()[i].build();
+ if (valueColumns[i].getPositionCount() !=
resultBuilder.getPositionCount()) {
+ throw new IllegalStateException(
+ String.format(
+ "Declared positions (%s) does not match column %s's number of
entries (%s)",
+ resultBuilder.getPositionCount(), i,
valueColumns[i].getPositionCount()));
+ }
+ }
+
+ TsBlock result =
+ TsBlock.wrapBlocksWithoutCopy(
+ this.resultBuilder.getPositionCount(),
+ new RunLengthEncodedColumn(TIME_COLUMN_TEMPLATE,
this.resultBuilder.getPositionCount()),
+ valueColumns);
+ resultBuilder.reset();
+ return result;
+ }
+
+ private boolean prepareInput(long start, long maxRuntime) throws Exception {
+ if ((leftBlock == null || leftBlock.getPositionCount() == leftIndex)
+ && leftChild.hasNextWithTimer()) {
+ leftBlock = leftChild.nextWithTimer();
+ leftIndex = 0;
+ }
+
+ if (rightBlockList.isEmpty()) {
+ if (hasCachedNextRightBlock && cachedNextRightBlock != null) {
+ rightBlockList.add(cachedNextRightBlock);
+ hasCachedNextRightBlock = false;
+ cachedNextRightBlock = null;
+ tryCachedNextRightTsBlock();
+ } else if (rightChild.hasNextWithTimer()) {
+ TsBlock block = rightChild.nextWithTimer();
+ if (block != null) {
+ rightBlockList.add(block);
+ tryCachedNextRightTsBlock();
+ }
+ } else {
+ hasCachedNextRightBlock = true;
+ cachedNextRightBlock = null;
+ }
+ } else {
+ if (!hasCachedNextRightBlock) {
+ tryCachedNextRightTsBlock();
+ }
+ }
+
+ return leftBlockNotEmpty() && rightBlockNotEmpty() &&
hasCachedNextRightBlock;
+ }
+
+ private void tryCachedNextRightTsBlock() throws Exception {
+ if (rightChild.hasNextWithTimer()) {
+ TsBlock block = rightChild.nextWithTimer();
+ if (block != null) {
+ if (block.getColumn(TIME_COLUMN_POSITION).getLong(0) ==
getRightEndTime()) {
+
memoryReservationManager.reserveMemoryCumulatively(block.getRetainedSizeInBytes());
+ rightBlockList.add(block);
+ } else {
+ hasCachedNextRightBlock = true;
+ cachedNextRightBlock = block;
+ }
+ }
+ } else {
+ hasCachedNextRightBlock = true;
+ cachedNextRightBlock = null;
+ }
+ }
+
+ private long getCurrentLeftTime() {
+ return leftBlock.getColumn(TIME_COLUMN_POSITION).getLong(leftIndex);
+ }
+
+ private long getLeftEndTime() {
+ return
leftBlock.getColumn(TIME_COLUMN_POSITION).getLong(leftBlock.getPositionCount()
- 1);
+ }
+
+ private long getCurrentRightTime() {
+ return rightBlockList
+ .get(rightBlockListIdx)
+ .getColumn(TIME_COLUMN_POSITION)
+ .getLong(rightIndex);
+ }
+
+ private long getRightTime(int idx1, int idx2) {
+ return
rightBlockList.get(idx1).getColumn(TIME_COLUMN_POSITION).getLong(idx2);
+ }
+
+ private long getRightEndTime() {
+ TsBlock lastRightTsBlock = rightBlockList.get(rightBlockList.size() - 1);
+ return lastRightTsBlock
+ .getColumn(TIME_COLUMN_POSITION)
+ .getLong(lastRightTsBlock.getPositionCount() - 1);
+ }
+
+ private void appendResult(long leftTime) {
+
+ while (comparator.lessThan(getCurrentRightTime(), leftTime)) {
+ rightIndex++;
+
+ if (rightIndex >=
rightBlockList.get(rightBlockListIdx).getPositionCount()) {
+ rightBlockListIdx++;
+ rightIndex = 0;
+ }
+
+ if (rightBlockListIdx >= rightBlockList.size()) {
+ rightBlockListIdx = 0;
+ rightIndex = 0;
+ return;
+ }
+ }
+
+ int tmpBlockIdx = rightBlockListIdx, tmpIdx = rightIndex;
+ while (leftTime == getRightTime(tmpBlockIdx, tmpIdx)) {
+ appendValueToResult(tmpBlockIdx, tmpIdx);
+
+ resultBuilder.declarePosition();
+
+ tmpIdx++;
+ if (tmpIdx >= rightBlockList.get(tmpBlockIdx).getPositionCount()) {
+ tmpIdx = 0;
+ tmpBlockIdx++;
+ }
+
+ if (tmpBlockIdx >= rightBlockList.size()) {
+ break;
+ }
+ }
+ }
+
+ private boolean leftBlockNotEmpty() {
+ return leftBlock != null && leftIndex < leftBlock.getPositionCount();
+ }
+
+ private boolean rightBlockNotEmpty() {
+ return !rightBlockList.isEmpty()
+ && rightBlockListIdx < rightBlockList.size()
+ && rightIndex <
rightBlockList.get(rightBlockListIdx).getPositionCount();
+ }
+
+ private void appendValueToResult(int tmpRightBlockListIdx, int
tmpRightIndex) {
+ for (int i = 0; i < leftOutputSymbolIdx.length; i++) {
+ ColumnBuilder columnBuilder = resultBuilder.getColumnBuilder(i);
+ if (leftBlock.getColumn(leftOutputSymbolIdx[i]).isNull(leftIndex)) {
+ columnBuilder.appendNull();
+ } else {
+ columnBuilder.write(leftBlock.getColumn(leftOutputSymbolIdx[i]),
leftIndex);
+ }
+ }
+
+ for (int i = 0; i < rightOutputSymbolIdx.length; i++) {
+ ColumnBuilder columnBuilder =
resultBuilder.getColumnBuilder(leftOutputSymbolIdx.length + i);
+
+ if (rightBlockList
+ .get(tmpRightBlockListIdx)
+ .getColumn(rightOutputSymbolIdx[i])
+ .isNull(tmpRightIndex)) {
+ columnBuilder.appendNull();
+ } else {
+ columnBuilder.write(
+
rightBlockList.get(tmpRightBlockListIdx).getColumn(rightOutputSymbolIdx[i]),
+ tmpRightIndex);
+ }
+ }
+ }
+
+ @Override
+ public boolean isFinished() throws Exception {
+ return !leftBlockNotEmpty()
+ && leftChild.isFinished()
+ && !rightBlockNotEmpty()
+ && rightChild.isFinished();
+ }
+
+ @Override
+ public void close() throws Exception {
+ if (leftChild != null) {
+ leftChild.close();
+ }
+ if (rightChild != null) {
+ rightChild.close();
+ }
+
+ if (!rightBlockList.isEmpty()) {
+ for (TsBlock block : rightBlockList) {
+
memoryReservationManager.reserveMemoryCumulatively(block.getRetainedSizeInBytes());
+ }
+ }
+ }
+
+ @Override
+ public OperatorContext getOperatorContext() {
+ return operatorContext;
+ }
+
+ @Override
+ public long calculateMaxPeekMemory() {
+ return Math.max(
+ Math.max(
+ leftChild.calculateMaxPeekMemoryWithCounter(),
+ rightChild.calculateMaxPeekMemoryWithCounter()),
+ calculateRetainedSizeAfterCallingNext() + calculateMaxReturnSize());
+ }
+
+ @Override
+ public long calculateMaxReturnSize() {
+ return maxReturnSize;
+ }
+
+ @Override
+ public long calculateRetainedSizeAfterCallingNext() {
+ // leftTsBlock + leftChild.RetainedSizeAfterCallingNext + rightTsBlock +
+ // rightChild.RetainedSizeAfterCallingNext
+ return leftChild.calculateMaxReturnSize()
+ + leftChild.calculateRetainedSizeAfterCallingNext()
+ + rightChild.calculateMaxReturnSize()
+ + rightChild.calculateRetainedSizeAfterCallingNext();
+ }
+
+ @Override
+ public long ramBytesUsed() {
+ return INSTANCE_SIZE
+ + MemoryEstimationHelper.getEstimatedSizeOfAccountableObject(leftChild)
+ +
MemoryEstimationHelper.getEstimatedSizeOfAccountableObject(rightChild)
+ + RamUsageEstimator.sizeOf(leftOutputSymbolIdx)
+ + RamUsageEstimator.sizeOf(rightOutputSymbolIdx)
+ +
MemoryEstimationHelper.getEstimatedSizeOfAccountableObject(operatorContext)
+ + resultBuilder.getRetainedSizeInBytes();
+ }
+}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/relational/ColumnTransformerBuilder.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/relational/ColumnTransformerBuilder.java
index d2b02a73ea0..f6b8099928f 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/relational/ColumnTransformerBuilder.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/relational/ColumnTransformerBuilder.java
@@ -1281,6 +1281,11 @@ public class ColumnTransformerBuilder
return res;
}
+ @Override
+ protected ColumnTransformer visitCoalesceExpression(CoalesceExpression node,
Context context) {
+ throw new
UnsupportedOperationException(String.format(UNSUPPORTED_EXPRESSION, node));
+ }
+
@Override
protected ColumnTransformer visitSimpleCaseExpression(
SimpleCaseExpression node, Context context) {
@@ -1308,11 +1313,6 @@ public class ColumnTransformerBuilder
throw new
UnsupportedOperationException(String.format(UNSUPPORTED_EXPRESSION, node));
}
- @Override
- protected ColumnTransformer visitCoalesceExpression(CoalesceExpression node,
Context context) {
- throw new
UnsupportedOperationException(String.format(UNSUPPORTED_EXPRESSION, node));
- }
-
public static boolean isLongLiteral(Expression expression) {
return expression instanceof LongLiteral;
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/TypeProvider.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/TypeProvider.java
index fc6c56ae599..d599e702d42 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/TypeProvider.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/TypeProvider.java
@@ -138,6 +138,10 @@ public class TypeProvider {
return Collections.unmodifiableMap(tableModelTypes);
}
+ public void setTableModelTypes(Map<Symbol, Type> tableModelTypes) {
+ this.tableModelTypes = tableModelTypes;
+ }
+
public void serialize(ByteBuffer byteBuffer) {
ReadWriteIOUtils.write(treeModelTypeMap.size(), byteBuffer);
for (Map.Entry<String, TSDataType> entry : treeModelTypeMap.entrySet()) {
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/OperatorTreeGenerator.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/OperatorTreeGenerator.java
index 93c6b6a1f3b..cf9a3dbba65 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/OperatorTreeGenerator.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/OperatorTreeGenerator.java
@@ -327,7 +327,7 @@ public class OperatorTreeGenerator extends
PlanVisitor<Operator, LocalExecutionP
private static final DataNodeSchemaCache DATA_NODE_SCHEMA_CACHE =
DataNodeSchemaCache.getInstance();
- private static final TimeComparator ASC_TIME_COMPARATOR = new
AscTimeComparator();
+ public static final TimeComparator ASC_TIME_COMPARATOR = new
AscTimeComparator();
private static final TimeComparator DESC_TIME_COMPARATOR = new
DescTimeComparator();
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 d2c5205f581..754cde57912 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
@@ -49,6 +49,7 @@ import
org.apache.iotdb.db.queryengine.execution.operator.schema.source.SchemaSo
import
org.apache.iotdb.db.queryengine.execution.operator.sink.IdentitySinkOperator;
import
org.apache.iotdb.db.queryengine.execution.operator.source.AlignedSeriesScanOperator;
import
org.apache.iotdb.db.queryengine.execution.operator.source.ExchangeOperator;
+import
org.apache.iotdb.db.queryengine.execution.operator.source.relational.InnerJoinOperator;
import
org.apache.iotdb.db.queryengine.execution.operator.source.relational.TableScanOperator;
import
org.apache.iotdb.db.queryengine.execution.relational.ColumnTransformerBuilder;
import org.apache.iotdb.db.queryengine.plan.analyze.TypeProvider;
@@ -117,6 +118,7 @@ import static
org.apache.iotdb.db.queryengine.common.DataNodeEndPoints.isSameNod
import static
org.apache.iotdb.db.queryengine.execution.operator.process.join.merge.MergeSortComparator.getComparatorForTable;
import static
org.apache.iotdb.db.queryengine.execution.operator.source.relational.TableScanOperator.constructAlignedPath;
import static
org.apache.iotdb.db.queryengine.plan.analyze.PredicateUtils.convertPredicateToFilter;
+import static
org.apache.iotdb.db.queryengine.plan.planner.OperatorTreeGenerator.ASC_TIME_COMPARATOR;
import static
org.apache.iotdb.db.queryengine.plan.relational.type.InternalTypeManager.getTSDataType;
/** This Visitor is responsible for transferring Table PlanNode Tree to Table
Operator Tree. */
@@ -162,10 +164,16 @@ public class TableOperatorGenerator extends
PlanVisitor<Operator, LocalExecution
sinkHandle.setMaxBytesCanReserve(context.getMaxBytesOneHandleCanReserve());
context.getDriverContext().setSink(sinkHandle);
- Operator child = node.getChildren().get(0).accept(this, context);
- List<Operator> children = new ArrayList<>(1);
- children.add(child);
- return new IdentitySinkOperator(operatorContext, children,
downStreamChannelIndex, sinkHandle);
+ if (node.getChildren().size() == 1) {
+ Operator child = node.getChildren().get(0).accept(this, context);
+ List<Operator> children = new ArrayList<>(1);
+ children.add(child);
+ return new IdentitySinkOperator(
+ operatorContext, children, downStreamChannelIndex, sinkHandle);
+ } else {
+ throw new IllegalStateException(
+ "IdentitySinkNode should only have one child in table model.");
+ }
}
@Override
@@ -742,7 +750,39 @@ public class TableOperatorGenerator extends
PlanVisitor<Operator, LocalExecution
@Override
public Operator visitJoin(JoinNode node, LocalExecutionPlanContext context) {
- throw new IllegalStateException("JoinOperator is not implemented
currently.");
+ OperatorContext operatorContext =
+ context
+ .getDriverContext()
+ .addOperatorContext(
+ context.getNextOperatorId(), node.getPlanNodeId(),
JoinNode.class.getSimpleName());
+ List<TSDataType> dataTypes = getOutputColumnTypes(node,
context.getTypeProvider());
+
+ Operator leftChild = node.getLeftChild().accept(this, context);
+ Operator rightChild = node.getRightChild().accept(this, context);
+
+ int[] leftOutputSymbolIdx = new int[node.getLeftOutputSymbols().size()];
+ for (int i = 0; i < leftOutputSymbolIdx.length; i++) {
+ leftOutputSymbolIdx[i] =
+
node.getLeftChild().getOutputSymbols().indexOf(node.getLeftOutputSymbols().get(i));
+ }
+ int[] rightOutputSymbolIdx = new int[node.getRightOutputSymbols().size()];
+ for (int i = 0; i < rightOutputSymbolIdx.length; i++) {
+ rightOutputSymbolIdx[i] =
+
node.getRightChild().getOutputSymbols().indexOf(node.getRightOutputSymbols().get(i));
+ }
+
+ if (requireNonNull(node.getJoinType()) == JoinNode.JoinType.INNER) {
+ return new InnerJoinOperator(
+ operatorContext,
+ leftChild,
+ leftOutputSymbolIdx,
+ rightChild,
+ rightOutputSymbolIdx,
+ ASC_TIME_COMPARATOR,
+ dataTypes);
+ }
+
+ throw new IllegalStateException("Unsupported join type: " +
node.getJoinType());
}
@Override
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanGraphPrinter.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanGraphPrinter.java
index b78cd78940a..f33f8e93029 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanGraphPrinter.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanGraphPrinter.java
@@ -613,6 +613,9 @@ public class PlanGraphPrinter extends
PlanVisitor<List<String>, PlanGraphPrinter
boxValue.add(String.format("OutputSymbols: %s", node.getOutputSymbols()));
boxValue.add(String.format("DeviceEntriesSize: %s",
node.getDeviceEntries().size()));
boxValue.add(String.format("ScanOrder: %s", node.getScanOrder()));
+ if (node.getTimePredicate().isPresent()) {
+ boxValue.add(String.format("TimePredicate: %s",
node.getTimePredicate().get()));
+ }
if (node.getPushDownPredicate() != null) {
boxValue.add(String.format("PushDownPredicate: %s",
node.getPushDownPredicate()));
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanNodeType.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanNodeType.java
index 95a53e9189a..26d0d4c82ea 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanNodeType.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanNodeType.java
@@ -250,6 +250,7 @@ public enum PlanNodeType {
TABLE_TOPK_NODE((short) 1008),
TABLE_COLLECT_NODE((short) 1009),
TABLE_STREAM_SORT_NODE((short) 1010),
+ TABLE_JOIN_NODE((short) 1011),
RELATIONAL_INSERT_TABLET((short) 2000),
RELATIONAL_INSERT_ROW((short) 2001),
@@ -563,6 +564,9 @@ public enum PlanNodeType {
case 1010:
return
org.apache.iotdb.db.queryengine.plan.relational.planner.node.StreamSortNode
.deserialize(buffer);
+ case 1011:
+ return
org.apache.iotdb.db.queryengine.plan.relational.planner.node.JoinNode.deserialize(
+ buffer);
case 2000:
return RelationalInsertTabletNode.deserialize(buffer);
case 2001:
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/process/ExchangeNode.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/process/ExchangeNode.java
index a8cfa04c410..da3baeb46e6 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/process/ExchangeNode.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/process/ExchangeNode.java
@@ -25,6 +25,7 @@ import
org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeId;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeType;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanVisitor;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.Symbol;
import org.apache.tsfile.utils.ReadWriteIOUtils;
@@ -45,6 +46,8 @@ public class ExchangeNode extends SingleChildProcessNode {
private List<String> outputColumnNames = new ArrayList<>();
+ private List<Symbol> outputSymbols = null;
+
/** Exchange needs to know which child of IdentitySinkNode/ShuffleSinkNode
it matches */
private int indexOfUpstreamSinkHandle = 0;
@@ -84,6 +87,15 @@ public class ExchangeNode extends SingleChildProcessNode {
this.outputColumnNames = outputColumnNames;
}
+ @Override
+ public List<Symbol> getOutputSymbols() {
+ return outputSymbols;
+ }
+
+ public void setOutputSymbols(List<Symbol> outputSymbols) {
+ this.outputSymbols = outputSymbols;
+ }
+
public void setUpstream(TEndPoint endPoint, FragmentInstanceId instanceId,
PlanNodeId nodeId) {
this.upstreamEndpoint = endPoint;
this.upstreamInstanceId = instanceId;
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/Analysis.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/Analysis.java
index bb1678f6598..a9ddeeea37e 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/Analysis.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/Analysis.java
@@ -490,6 +490,10 @@ public class Analysis implements IAnalysis {
return joins.get(NodeRef.of(join));
}
+ public boolean hasJoinNode() {
+ return !joinUsing.isEmpty() || !joins.isEmpty();
+ }
+
public void recordSubqueries(Node node, ExpressionAnalysis
expressionAnalysis) {
SubqueryAnalysis subqueries =
this.subQueries.computeIfAbsent(NodeRef.of(node), key -> new
SubqueryAnalysis());
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/RelationPlanner.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/RelationPlanner.java
index 67648fe254a..37bbc69c060 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/RelationPlanner.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/RelationPlanner.java
@@ -37,7 +37,6 @@ import
org.apache.iotdb.db.queryengine.plan.relational.planner.node.ProjectNode;
import
org.apache.iotdb.db.queryengine.plan.relational.planner.node.TableScanNode;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.AliasedRelation;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.AstVisitor;
-import
org.apache.iotdb.db.queryengine.plan.relational.sql.ast.CoalesceExpression;
import
org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ComparisonExpression;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.Except;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.Expression;
@@ -331,10 +330,11 @@ public class RelationPlanner extends
AstVisitor<RelationPlan, Void> {
outputs.add(output);
queryContext.getTypeProvider().putTableModelType(output, LongType.INT64);
assignments.put(
- output,
- new CoalesceExpression(
- leftJoinColumns.get(column).toSymbolReference(),
- rightJoinColumns.get(column).toSymbolReference()));
+ output, leftJoinColumns.get(column).toSymbolReference()
+ // new CoalesceExpression(
+ // leftJoinColumns.get(column).toSymbolReference(),
+ // rightJoinColumns.get(column).toSymbolReference())
+ );
}
for (int field : joinAnalysis.getOtherLeftFields()) {
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/AddExchangeNodes.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/AddExchangeNodes.java
index 0334ffe864f..a9a27d12b31 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/AddExchangeNodes.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/AddExchangeNodes.java
@@ -77,6 +77,7 @@ public class AddExchangeNodes
if (!region.equals(context.mostUsedRegion)) {
ExchangeNode exchangeNode = new
ExchangeNode(queryContext.getQueryId().genPlanNodeId());
exchangeNode.addChild(rewriteNode);
+ exchangeNode.setOutputSymbols(rewriteNode.getOutputSymbols());
newNode.addChild(exchangeNode);
context.hasExchangeNode = true;
} else {
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableDistributedPlanner.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableDistributedPlanner.java
index eb3c9839ed7..976d5ad2de9 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableDistributedPlanner.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableDistributedPlanner.java
@@ -210,10 +210,15 @@ public class TableDistributedPlanner {
if (child instanceof ExchangeNode) {
ExchangeNode exchangeNode = (ExchangeNode) child;
+ // IdentitySinkNode identitySinkNode =
+ // regionNodeMap.computeIfAbsent(
+ //
+ //
context.getNodeDistribution(exchangeNode.getChild().getPlanNodeId()).getRegion(),
+ // k -> new
IdentitySinkNode(mppQueryContext.getQueryId().genPlanNodeId()));
+
+ // In table model, each ExchangeNode matches only one IdentitySinkNode
IdentitySinkNode identitySinkNode =
- regionNodeMap.computeIfAbsent(
-
context.getNodeDistribution(exchangeNode.getChild().getPlanNodeId()).getRegion(),
- k -> new
IdentitySinkNode(mppQueryContext.getQueryId().genPlanNodeId()));
+ new IdentitySinkNode(mppQueryContext.getQueryId().genPlanNodeId());
identitySinkNode.addChild(exchangeNode.getChild());
identitySinkNode.addDownStreamChannelLocation(
new
DownStreamChannelLocation(exchangeNode.getPlanNodeId().toString()));
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/iterative/rule/AddTableScanColumnsToTypeProviderOptimizer.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/iterative/rule/AddTableScanColumnsToTypeProviderOptimizer.java
new file mode 100644
index 00000000000..b23a2fe346a
--- /dev/null
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/iterative/rule/AddTableScanColumnsToTypeProviderOptimizer.java
@@ -0,0 +1,66 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.queryengine.plan.relational.planner.iterative.rule;
+
+import org.apache.iotdb.db.queryengine.common.MPPQueryContext;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanVisitor;
+import
org.apache.iotdb.db.queryengine.plan.relational.planner.node.TableScanNode;
+import
org.apache.iotdb.db.queryengine.plan.relational.planner.optimizations.PlanOptimizer;
+
+/**
+ * <b>Optimization phase:</b> Logical plan planning.
+ *
+ * <p>Only when exist JoinNode need execute this optimize rule before
+ * TableModelTypeProviderExtractor.
+ */
+public class AddTableScanColumnsToTypeProviderOptimizer implements
PlanOptimizer {
+
+ @Override
+ public PlanNode optimize(PlanNode plan, PlanOptimizer.Context context) {
+ if (!context.getAnalysis().hasJoinNode()) {
+ return plan;
+ }
+
+ return plan.accept(new Rewriter(context.getQueryContext()), null);
+ }
+
+ private static class Rewriter extends PlanVisitor<PlanNode, Void> {
+
+ private final MPPQueryContext queryContext;
+
+ public Rewriter(MPPQueryContext queryContext) {
+ this.queryContext = queryContext;
+ }
+
+ @Override
+ public PlanNode visitPlan(PlanNode node, Void context) {
+ node.getChildren().forEach(child -> child.accept(this, context));
+ return node;
+ }
+
+ @Override
+ public PlanNode visitTableScan(TableScanNode node, Void context) {
+ node.getAssignments()
+ .forEach((k, v) ->
queryContext.getTypeProvider().putTableModelType(k, v.getType()));
+ return node;
+ }
+ }
+}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/iterative/rule/InlineProjections.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/iterative/rule/InlineProjections.java
index 0731acb5f63..5fe62438c41 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/iterative/rule/InlineProjections.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/iterative/rule/InlineProjections.java
@@ -32,6 +32,7 @@ import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Sets;
+import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
@@ -98,13 +99,13 @@ public class InlineProjections implements Rule<ProjectNode>
{
return Optional.empty();
}
- // inline the expressions
+ // inline the expressions; use LinkedHashMap to keep order
Assignments assignments = child.getAssignments().filter(targets::contains);
Map<Symbol, Expression> parentAssignments =
- parent.getAssignments().entrySet().stream()
- .collect(
- Collectors.toMap(
- Map.Entry::getKey, entry ->
inlineReferences(entry.getValue(), assignments)));
+ new LinkedHashMap<>(parent.getAssignments().getMap().size());
+ for (Map.Entry<Symbol, Expression> entry :
parent.getAssignments().getMap().entrySet()) {
+ parentAssignments.put(entry.getKey(), inlineReferences(entry.getValue(),
assignments));
+ }
// Synthesize identity assignments for the inputs of expressions that were
inlined
// to place in the child projection.
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/iterative/rule/MergeLimitWithSort.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/iterative/rule/MergeLimitWithSort.java
index 92962a0e05d..90948d96a41 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/iterative/rule/MergeLimitWithSort.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/iterative/rule/MergeLimitWithSort.java
@@ -1,16 +1,22 @@
/*
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
*
- * http://www.apache.org/licenses/LICENSE-2.0
+ * http://www.apache.org/licenses/LICENSE-2.0
*
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
*/
+
package org.apache.iotdb.db.queryengine.plan.relational.planner.iterative.rule;
import org.apache.iotdb.db.queryengine.plan.relational.planner.iterative.Rule;
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/iterative/rule/PruneJoinChildrenColumns.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/iterative/rule/PruneJoinChildrenColumns.java
new file mode 100644
index 00000000000..66023182067
--- /dev/null
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/iterative/rule/PruneJoinChildrenColumns.java
@@ -0,0 +1,77 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iotdb.db.queryengine.plan.relational.planner.iterative.rule;
+
+import org.apache.iotdb.db.queryengine.plan.relational.planner.Symbol;
+import
org.apache.iotdb.db.queryengine.plan.relational.planner.SymbolsExtractor;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.iterative.Rule;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.JoinNode;
+import org.apache.iotdb.db.queryengine.plan.relational.utils.matching.Captures;
+import org.apache.iotdb.db.queryengine.plan.relational.utils.matching.Pattern;
+
+import com.google.common.collect.ImmutableSet;
+
+import java.util.Set;
+
+import static
org.apache.iotdb.db.queryengine.plan.relational.planner.iterative.rule.Util.restrictChildOutputs;
+import static
org.apache.iotdb.db.queryengine.plan.relational.planner.node.Patterns.join;
+
+/**
+ * Joins support output symbol selection, so make any project-off of child
columns explicit in
+ * project nodes.
+ */
+public class PruneJoinChildrenColumns implements Rule<JoinNode> {
+ private static final Pattern<JoinNode> PATTERN = join();
+
+ @Override
+ public Pattern<JoinNode> getPattern() {
+ return PATTERN;
+ }
+
+ @Override
+ public Result apply(JoinNode joinNode, Captures captures, Context context) {
+ Set<Symbol> globallyUsableInputs =
+ ImmutableSet.<Symbol>builder()
+ .addAll(joinNode.getOutputSymbols())
+ .addAll(
+
joinNode.getFilter().map(SymbolsExtractor::extractUnique).orElse(ImmutableSet.of()))
+ .build();
+
+ Set<Symbol> leftUsableInputs =
+ ImmutableSet.<Symbol>builder()
+ .addAll(globallyUsableInputs)
+ .addAll(
+
joinNode.getCriteria().stream().map(JoinNode.EquiJoinClause::getLeft).iterator())
+ //
.addAll(joinNode.getLeftHashSymbol().map(ImmutableSet::of).orElse(ImmutableSet.of()))
+ .build();
+
+ Set<Symbol> rightUsableInputs =
+ ImmutableSet.<Symbol>builder()
+ .addAll(globallyUsableInputs)
+ .addAll(
+
joinNode.getCriteria().stream().map(JoinNode.EquiJoinClause::getRight).iterator())
+ //
.addAll(joinNode.getRightHashSymbol().map(ImmutableSet::of).orElse(ImmutableSet.of()))
+ .build();
+
+ return restrictChildOutputs(
+ context.getIdAllocator(), joinNode, leftUsableInputs,
rightUsableInputs)
+ .map(Result::ofPlanNode)
+ .orElse(Result.empty());
+ }
+}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/iterative/rule/PruneTableScanColumns.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/iterative/rule/PruneTableScanColumns.java
index 8c033882112..2bd37ea1a20 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/iterative/rule/PruneTableScanColumns.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/iterative/rule/PruneTableScanColumns.java
@@ -13,8 +13,6 @@
*/
package org.apache.iotdb.db.queryengine.plan.relational.planner.iterative.rule;
-import org.apache.iotdb.db.queryengine.common.SessionInfo;
-import org.apache.iotdb.db.queryengine.plan.analyze.TypeProvider;
import org.apache.iotdb.db.queryengine.plan.expression.leaf.TimestampOperand;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNode;
import org.apache.iotdb.db.queryengine.plan.relational.metadata.ColumnSchema;
@@ -25,7 +23,7 @@ import
org.apache.iotdb.db.queryengine.plan.relational.planner.node.AggregationT
import
org.apache.iotdb.db.queryengine.plan.relational.planner.node.TableScanNode;
import java.util.ArrayList;
-import java.util.HashMap;
+import java.util.LinkedHashMap;
import java.util.List;
import java.util.Locale;
import java.util.Map;
@@ -47,23 +45,15 @@ public class PruneTableScanColumns extends
ProjectOffPushDownRule<TableScanNode>
@Override
protected Optional<PlanNode> pushDownProjectOff(
Context context, TableScanNode node, Set<Symbol> referencedOutputs) {
- SessionInfo sessionInfo = context.getSessionInfo();
- TypeProvider types = context.getSymbolAllocator().getTypes();
-
- return pruneColumns(metadata, types, sessionInfo, node, referencedOutputs);
+ return pruneColumns(node, referencedOutputs);
}
- public static Optional<PlanNode> pruneColumns(
- Metadata metadata,
- TypeProvider types,
- SessionInfo sessionInfo,
- TableScanNode node,
- Set<Symbol> referencedOutputs) {
+ public static Optional<PlanNode> pruneColumns(TableScanNode node,
Set<Symbol> referencedOutputs) {
if (node instanceof AggregationTableScanNode) {
return Optional.empty();
}
List<Symbol> newOutputs = new ArrayList<>();
- Map<Symbol, ColumnSchema> newAssignments = new HashMap<>();
+ Map<Symbol, ColumnSchema> newAssignments = new LinkedHashMap<>();
for (Symbol symbol : node.getOutputSymbols()) {
if (referencedOutputs.contains(symbol)) {
newOutputs.add(symbol);
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/iterative/rule/Util.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/iterative/rule/Util.java
index e9bcf582773..7447508ab1e 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/iterative/rule/Util.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/iterative/rule/Util.java
@@ -1,15 +1,20 @@
/*
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
*/
package org.apache.iotdb.db.queryengine.plan.relational.planner.iterative.rule;
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/JoinNode.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/JoinNode.java
index 9ceaed6d723..0775075a685 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/JoinNode.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/JoinNode.java
@@ -20,6 +20,7 @@ package
org.apache.iotdb.db.queryengine.plan.relational.planner.node;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeId;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeType;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanVisitor;
import
org.apache.iotdb.db.queryengine.plan.planner.plan.node.process.TwoChildProcessNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.Symbol;
@@ -29,10 +30,12 @@ import
org.apache.iotdb.db.queryengine.plan.relational.sql.ast.NullLiteral;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableSet;
+import org.apache.tsfile.utils.ReadWriteIOUtils;
import java.io.DataOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
+import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.Optional;
@@ -125,6 +128,26 @@ public class JoinNode extends TwoChildProcessNode {
equiJoinClause));
}
+ // only used for deserialize
+ public JoinNode(
+ PlanNodeId id,
+ JoinType joinType,
+ List<EquiJoinClause> criteria,
+ List<Symbol> leftOutputSymbols,
+ List<Symbol> rightOutputSymbols) {
+ super(id);
+ requireNonNull(joinType, "type is null");
+ requireNonNull(criteria, "criteria is null");
+
+ this.leftOutputSymbols = leftOutputSymbols;
+ this.rightOutputSymbols = rightOutputSymbols;
+ this.filter = Optional.empty();
+ this.spillable = Optional.empty();
+
+ this.joinType = joinType;
+ this.criteria = criteria;
+ }
+
@Override
public <R, C> R accept(PlanVisitor<R, C> visitor, C context) {
return visitor.visitJoin(this, context);
@@ -177,10 +200,73 @@ public class JoinNode extends TwoChildProcessNode {
}
@Override
- protected void serializeAttributes(ByteBuffer byteBuffer) {}
+ protected void serializeAttributes(ByteBuffer byteBuffer) {
+ PlanNodeType.TABLE_JOIN_NODE.serialize(byteBuffer);
+
+ ReadWriteIOUtils.write(joinType.ordinal(), byteBuffer);
+
+ ReadWriteIOUtils.write(criteria.size(), byteBuffer);
+ for (EquiJoinClause equiJoinClause : criteria) {
+ Symbol.serialize(equiJoinClause.getLeft(), byteBuffer);
+ Symbol.serialize(equiJoinClause.getRight(), byteBuffer);
+ }
+
+ ReadWriteIOUtils.write(leftOutputSymbols.size(), byteBuffer);
+ for (Symbol leftOutputSymbol : leftOutputSymbols) {
+ Symbol.serialize(leftOutputSymbol, byteBuffer);
+ }
+ ReadWriteIOUtils.write(rightOutputSymbols.size(), byteBuffer);
+ for (Symbol rightOutputSymbol : rightOutputSymbols) {
+ Symbol.serialize(rightOutputSymbol, byteBuffer);
+ }
+ }
@Override
- protected void serializeAttributes(DataOutputStream stream) throws
IOException {}
+ protected void serializeAttributes(DataOutputStream stream) throws
IOException {
+ PlanNodeType.TABLE_JOIN_NODE.serialize(stream);
+
+ ReadWriteIOUtils.write(joinType.ordinal(), stream);
+
+ ReadWriteIOUtils.write(criteria.size(), stream);
+ for (EquiJoinClause equiJoinClause : criteria) {
+ Symbol.serialize(equiJoinClause.getLeft(), stream);
+ Symbol.serialize(equiJoinClause.getRight(), stream);
+ }
+
+ ReadWriteIOUtils.write(leftOutputSymbols.size(), stream);
+ for (Symbol leftOutputSymbol : leftOutputSymbols) {
+ Symbol.serialize(leftOutputSymbol, stream);
+ }
+ ReadWriteIOUtils.write(rightOutputSymbols.size(), stream);
+ for (Symbol rightOutputSymbol : rightOutputSymbols) {
+ Symbol.serialize(rightOutputSymbol, stream);
+ }
+ }
+
+ public static JoinNode deserialize(ByteBuffer byteBuffer) {
+ JoinType joinType =
JoinType.values()[ReadWriteIOUtils.readInt(byteBuffer)];
+ int size = ReadWriteIOUtils.readInt(byteBuffer);
+ List<EquiJoinClause> criteria = new ArrayList<>(size);
+ while (size-- > 0) {
+ criteria.add(
+ new EquiJoinClause(Symbol.deserialize(byteBuffer),
Symbol.deserialize(byteBuffer)));
+ }
+
+ size = ReadWriteIOUtils.readInt(byteBuffer);
+ List<Symbol> leftOutputSymbols = new ArrayList<>(size);
+ while (size-- > 0) {
+ leftOutputSymbols.add(Symbol.deserialize(byteBuffer));
+ }
+
+ size = ReadWriteIOUtils.readInt(byteBuffer);
+ List<Symbol> rightOutputSymbols = new ArrayList<>(size);
+ while (size-- > 0) {
+ rightOutputSymbols.add(Symbol.deserialize(byteBuffer));
+ }
+
+ PlanNodeId planNodeId = PlanNodeId.deserialize(byteBuffer);
+ return new JoinNode(planNodeId, joinType, criteria, leftOutputSymbols,
rightOutputSymbols);
+ }
public JoinType getJoinType() {
return joinType;
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/DistributedOptimizeFactory.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/DistributedOptimizeFactory.java
index 7c165318406..0f46a8887d1 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/DistributedOptimizeFactory.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/DistributedOptimizeFactory.java
@@ -1,15 +1,20 @@
/*
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
*/
package org.apache.iotdb.db.queryengine.plan.relational.planner.optimizations;
@@ -31,23 +36,24 @@ public class DistributedOptimizeFactory {
private final List<PlanOptimizer> planOptimizers;
public DistributedOptimizeFactory(PlannerContext plannerContext) {
- IterativeOptimizer topKOptimizer =
- new IterativeOptimizer(
- plannerContext,
- new RuleStatsRecorder(),
- ImmutableSet.of(
- new MergeLimitWithMergeSort(), new
MergeLimitOverProjectWithMergeSort()));
-
- PlanOptimizer sortElimination = new SortElimination();
-
- IterativeOptimizer limitElimination =
- new IterativeOptimizer(
- plannerContext,
- new RuleStatsRecorder(),
- ImmutableSet.of(
- new EliminateLimitWithTableScan(), new
EliminateLimitProjectWithTableScan()));
-
- this.planOptimizers = ImmutableList.of(topKOptimizer, sortElimination,
limitElimination);
+ RuleStatsRecorder ruleStats = new RuleStatsRecorder();
+
+ this.planOptimizers =
+ ImmutableList.of(
+ // transfer Limit+Sort to TopK
+ new IterativeOptimizer(
+ plannerContext,
+ ruleStats,
+ ImmutableSet.of(
+ new MergeLimitWithMergeSort(), new
MergeLimitOverProjectWithMergeSort())),
+ // eliminate unnecessary SortNode
+ new SortElimination(),
+ // other optimize rules
+ new IterativeOptimizer(
+ plannerContext,
+ ruleStats,
+ ImmutableSet.of(
+ new EliminateLimitWithTableScan(), new
EliminateLimitProjectWithTableScan())));
}
public List<PlanOptimizer> getPlanOptimizers() {
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/LogicalOptimizeFactory.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/LogicalOptimizeFactory.java
index 629027105d6..43ce9adf50c 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/LogicalOptimizeFactory.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/LogicalOptimizeFactory.java
@@ -24,6 +24,7 @@ import
org.apache.iotdb.db.queryengine.plan.relational.planner.PlannerContext;
import
org.apache.iotdb.db.queryengine.plan.relational.planner.iterative.IterativeOptimizer;
import org.apache.iotdb.db.queryengine.plan.relational.planner.iterative.Rule;
import
org.apache.iotdb.db.queryengine.plan.relational.planner.iterative.RuleStatsRecorder;
+import
org.apache.iotdb.db.queryengine.plan.relational.planner.iterative.rule.AddTableScanColumnsToTypeProviderOptimizer;
import
org.apache.iotdb.db.queryengine.plan.relational.planner.iterative.rule.CanonicalizeExpressions;
import
org.apache.iotdb.db.queryengine.plan.relational.planner.iterative.rule.InlineProjections;
import
org.apache.iotdb.db.queryengine.plan.relational.planner.iterative.rule.MergeFilters;
@@ -33,6 +34,7 @@ import
org.apache.iotdb.db.queryengine.plan.relational.planner.iterative.rule.Me
import
org.apache.iotdb.db.queryengine.plan.relational.planner.iterative.rule.PruneAggregationColumns;
import
org.apache.iotdb.db.queryengine.plan.relational.planner.iterative.rule.PruneAggregationSourceColumns;
import
org.apache.iotdb.db.queryengine.plan.relational.planner.iterative.rule.PruneFilterColumns;
+import
org.apache.iotdb.db.queryengine.plan.relational.planner.iterative.rule.PruneJoinChildrenColumns;
import
org.apache.iotdb.db.queryengine.plan.relational.planner.iterative.rule.PruneJoinColumns;
import
org.apache.iotdb.db.queryengine.plan.relational.planner.iterative.rule.PruneLimitColumns;
import
org.apache.iotdb.db.queryengine.plan.relational.planner.iterative.rule.PruneOffsetColumns;
@@ -63,9 +65,6 @@ public class LogicalOptimizeFactory {
Metadata metadata = plannerContext.getMetadata();
RuleStatsRecorder ruleStats = new RuleStatsRecorder();
- PlanOptimizer pushPredicateIntoTableScanOptimizer = new
PushPredicateIntoTableScan();
- PlanOptimizer transformSortToStreamSortOptimizer = new
TransformSortToStreamSort();
-
Set<Rule<?>> columnPruningRules =
ImmutableSet.of(
new PruneAggregationColumns(),
@@ -80,7 +79,8 @@ public class LogicalOptimizeFactory {
new PruneSortColumns(),
new PruneTableScanColumns(plannerContext.getMetadata()),
new PruneTopKColumns(),
- new PruneJoinColumns());
+ new PruneJoinColumns(),
+ new PruneJoinChildrenColumns());
IterativeOptimizer columnPruningOptimizer =
new IterativeOptimizer(plannerContext, ruleStats, columnPruningRules);
@@ -125,95 +125,84 @@ public class LogicalOptimizeFactory {
Set<Rule<?>> limitPushdownRules =
ImmutableSet.of(new PushLimitThroughOffset(), new
PushLimitThroughProject());
- IterativeOptimizer limitPushdownOptimizer =
- new IterativeOptimizer(plannerContext, ruleStats, limitPushdownRules);
-
- PlanOptimizer unAliasSymbolReferences =
- new UnaliasSymbolReferences(plannerContext.getMetadata());
-
- PlanOptimizer transformAggregationToStreamableOptimizer =
- new TransformAggregationToStreamable();
- PlanOptimizer pushAggregationIntoTableScanOptimizer = new
PushAggregationIntoTableScan();
+ ImmutableList.Builder<PlanOptimizer> optimizerBuilder =
ImmutableList.builder();
- PlanOptimizer pushLimitOffsetIntoTableScanOptimizer = new
PushLimitOffsetIntoTableScan();
-
- IterativeOptimizer topKOptimizer =
+ optimizerBuilder.add(
+ new IterativeOptimizer(
+ plannerContext,
+ ruleStats,
+ ImmutableSet.<Rule<?>>builder()
+ .addAll(columnPruningRules)
+ // .addAll(projectionPushdownRules).
+ // addAll(newUnwrapRowSubscript().rules()).
+ // addAll(new PushCastIntoRow().rules())
+ .addAll(
+ ImmutableSet.of(
+ new MergeFilters(),
+ new InlineProjections(plannerContext),
+ new RemoveRedundantIdentityProjections(),
+ new MergeLimits(),
+ new RemoveTrivialFilters()
+ // new RemoveRedundantLimit(),
+ // new RemoveRedundantOffset(),
+ // new RemoveRedundantSort(),
+ // new
RemoveRedundantSortBelowLimitWithTies(),
+ // new RemoveRedundantTopN(),
+ // new
RemoveRedundantDistinctLimit(),
+ // new
ReplaceRedundantJoinWithSource(),
+ // new RemoveRedundantJoin(),
+ // new
ReplaceRedundantJoinWithProject(),
+ // new RemoveRedundantExists(),
+ // new RemoveRedundantWindow(),
+ // new
SingleDistinctAggregationToGroupBy(),
+ // new MergeLimitWithDistinct(),
+ // new
PruneCountAggregationOverScalar(metadata),
+ // new
SimplifyCountOverConstant(plannerContext),
+ // new
+ // PreAggregateCaseAggregations(plannerContext,
typeAnalyzer)))
+ ))
+ .build()),
+ // MergeUnion and related projection pruning rules must run before
limit pushdown rules,
+ // otherwise
+ // an intermediate limit node will prevent unions from being merged
later on
+ new IterativeOptimizer(
+ plannerContext,
+ ruleStats,
+ ImmutableSet.<Rule<?>>builder()
+ // .addAll(projectionPushdownRules)
+ .addAll(columnPruningRules)
+ .addAll(limitPushdownRules)
+ .addAll(
+ ImmutableSet.of(
+ // new MergeUnion(),
+ // new RemoveEmptyUnionBranches(),
+ new MergeFilters(),
+ new RemoveTrivialFilters(),
+ new MergeLimits(),
+ new InlineProjections(plannerContext),
+ new RemoveRedundantIdentityProjections()))
+ .build()),
+ simplifyOptimizer,
+ new UnaliasSymbolReferences(plannerContext.getMetadata()),
+ columnPruningOptimizer,
+ inlineProjectionLimitFiltersOptimizer,
+ new PushPredicateIntoTableScan(),
+ // redo columnPrune and inlineProjections after
pushPredicateIntoTableScan
+ columnPruningOptimizer,
+ inlineProjectionLimitFiltersOptimizer,
+ new IterativeOptimizer(plannerContext, ruleStats, limitPushdownRules),
+ new PushLimitOffsetIntoTableScan(),
+ new TransformAggregationToStreamable(),
+ new PushAggregationIntoTableScan(),
+ new TransformSortToStreamSort(),
new IterativeOptimizer(
plannerContext,
ruleStats,
- ImmutableSet.of(new MergeLimitWithSort(), new
MergeLimitOverProjectWithSort()));
+ ImmutableSet.of(new MergeLimitWithSort(), new
MergeLimitOverProjectWithSort())),
+ new AddTableScanColumnsToTypeProviderOptimizer());
- this.planOptimizers =
- ImmutableList.of(
- new IterativeOptimizer(
- plannerContext,
- ruleStats,
- ImmutableSet.<Rule<?>>builder()
- .addAll(columnPruningRules)
- // .addAll(projectionPushdownRules)
- // .addAll(new
UnwrapRowSubscript().rules())
- // .addAll(new
PushCastIntoRow().rules())
- .addAll(
- ImmutableSet.of(
- new MergeFilters(),
- new InlineProjections(plannerContext),
- new RemoveRedundantIdentityProjections(),
- new MergeLimits(),
- new RemoveTrivialFilters()
- // new
RemoveRedundantLimit(),
- // new
RemoveRedundantOffset(),
- // new
RemoveRedundantSort(),
- // new
RemoveRedundantSortBelowLimitWithTies(),
- // new
RemoveRedundantTopN(),
- // new
RemoveRedundantDistinctLimit(),
- // new
ReplaceRedundantJoinWithSource(),
- // new
RemoveRedundantJoin(),
- // new
ReplaceRedundantJoinWithProject(),
- // new
RemoveRedundantExists(),
- // new
RemoveRedundantWindow(),
- // new
SingleDistinctAggregationToGroupBy(),
- // new
MergeLimitWithDistinct(),
- // new
PruneCountAggregationOverScalar(metadata),
- // new
SimplifyCountOverConstant(plannerContext),
- // new
- // PreAggregateCaseAggregations(plannerContext,
typeAnalyzer)))
- ))
- .build()),
- // MergeUnion and related projection pruning rules must run before
limit pushdown rules,
- // otherwise
- // an intermediate limit node will prevent unions from being
merged later on
- new IterativeOptimizer(
- plannerContext,
- ruleStats,
- ImmutableSet.<Rule<?>>builder()
- // .addAll(projectionPushdownRules)
- .addAll(columnPruningRules)
- .addAll(limitPushdownRules)
- .addAll(
- ImmutableSet.of(
- // new MergeUnion(),
- // new
RemoveEmptyUnionBranches(),
- new MergeFilters(),
- new RemoveTrivialFilters(),
- new MergeLimits(),
- new InlineProjections(plannerContext),
- new RemoveRedundantIdentityProjections()))
- .build()),
- simplifyOptimizer,
- unAliasSymbolReferences,
- columnPruningOptimizer,
- inlineProjectionLimitFiltersOptimizer,
- pushPredicateIntoTableScanOptimizer,
- // redo columnPrune and inlineProjections after
pushPredicateIntoTableScan
- columnPruningOptimizer,
- inlineProjectionLimitFiltersOptimizer,
- limitPushdownOptimizer,
- pushLimitOffsetIntoTableScanOptimizer,
- transformAggregationToStreamableOptimizer,
- pushAggregationIntoTableScanOptimizer,
- transformSortToStreamSortOptimizer,
- topKOptimizer);
+ this.planOptimizers = optimizerBuilder.build();
}
public List<PlanOptimizer> getPlanOptimizers() {
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 2edf99178ce..241e035a033 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
@@ -294,12 +294,69 @@ public class PushPredicateIntoTableScan implements
PlanOptimizer {
}
@Override
- public PlanNode visitTableScan(TableScanNode node, RewriteContext context)
{
- if (!TRUE_LITERAL.equals(context.inheritedPredicate)) {
- return combineFilterAndScan(node, context.inheritedPredicate);
+ public PlanNode visitTableScan(TableScanNode tableScanNode, RewriteContext
context) {
+ // columnSymbols in TableScanNode may be added suffix in Join
situation(such as self join),
+ // in which we need add a new ProjectNode above TableScanNode.
+ boolean hasSuffixInScanNodeColumns = false;
+ for (Map.Entry<Symbol, ColumnSchema> entry :
tableScanNode.getAssignments().entrySet()) {
+ Symbol columnSymbol = entry.getKey();
+ ColumnSchema columnSchema = entry.getValue();
+ if (!columnSymbol.getName().equals(columnSchema.getName())) {
+ hasSuffixInScanNodeColumns = true;
+ break;
+ }
+ }
+
+ Map<Symbol, Expression> newProjectAssignments = null;
+ if (hasSuffixInScanNodeColumns) {
+ newProjectAssignments = getProjectAssignments(tableScanNode, context);
+ }
+
+ // no predicate, just scan all matched deviceEntries
+ if (TRUE_LITERAL.equals(context.inheritedPredicate)) {
+ getDeviceEntriesWithDataPartitions(tableScanNode,
Collections.emptyList());
+ return hasSuffixInScanNodeColumns
+ ? new ProjectNode(
+ queryId.genPlanNodeId(), tableScanNode, new
Assignments(newProjectAssignments))
+ : tableScanNode;
+ }
+
+ // has predicate, deal with split predicate
+ PlanNode result = combineFilterAndScan(tableScanNode,
context.inheritedPredicate);
+ return hasSuffixInScanNodeColumns
+ ? new ProjectNode(queryId.genPlanNodeId(), result, new
Assignments(newProjectAssignments))
+ : result;
+ }
+
+ private Map<Symbol, Expression> getProjectAssignments(
+ TableScanNode tableScanNode, RewriteContext context) {
+ context.inheritedPredicate =
+ ReplaceSymbolInExpression.transform(
+ context.inheritedPredicate, tableScanNode.getAssignments());
+
+ int size = tableScanNode.getOutputSymbols().size();
+ Map<Symbol, Expression> projectAssignments = new LinkedHashMap<>(size);
+ List<Symbol> newTableScanSymbols = new ArrayList<>(size);
+ Map<Symbol, ColumnSchema> newTableScanAssignments = new
LinkedHashMap<>(size);
+ for (Symbol originalSymbol : tableScanNode.getOutputSymbols()) {
+ ColumnSchema columnSchema =
tableScanNode.getAssignments().get(originalSymbol);
+
+ Symbol realSymbol = Symbol.of(columnSchema.getName());
+ newTableScanSymbols.add(realSymbol);
+ newTableScanAssignments.put(realSymbol, columnSchema);
+ projectAssignments.put(originalSymbol, new
SymbolReference(columnSchema.getName()));
+ queryContext.getTypeProvider().putTableModelType(originalSymbol,
columnSchema.getType());
+ Map<Symbol, Integer> idAndAttributeIndexMap =
tableScanNode.getIdAndAttributeIndexMap();
+ if (idAndAttributeIndexMap.containsKey(originalSymbol)) {
+ Integer idx = idAndAttributeIndexMap.get(originalSymbol);
+ idAndAttributeIndexMap.remove(originalSymbol);
+ idAndAttributeIndexMap.put(realSymbol, idx);
+ }
}
- return tableMetadataIndexScan(node, Collections.emptyList());
+ tableScanNode.setOutputSymbols(newTableScanSymbols);
+ tableScanNode.setAssignments(newTableScanAssignments);
+ return projectAssignments;
}
public PlanNode combineFilterAndScan(TableScanNode tableScanNode,
Expression predicate) {
@@ -332,21 +389,20 @@ public class PushPredicateIntoTableScan implements
PlanOptimizer {
}
// do index scan after expressionCanPushDown is processed
- PlanNode resultNode =
- tableMetadataIndexScan(tableScanNode,
splitExpression.getMetadataExpressions());
+ getDeviceEntriesWithDataPartitions(tableScanNode,
splitExpression.getMetadataExpressions());
// exist expressions can not push down to scan operator
if (!splitExpression.getExpressionsCannotPushDown().isEmpty()) {
List<Expression> expressions =
splitExpression.getExpressionsCannotPushDown();
return new FilterNode(
queryId.genPlanNodeId(),
- resultNode,
+ tableScanNode,
expressions.size() == 1
? expressions.get(0)
: new LogicalExpression(LogicalExpression.Operator.AND,
expressions));
}
- return resultNode;
+ return tableScanNode;
}
private SplitExpression splitPredicate(TableScanNode node, Expression
predicate) {
@@ -397,72 +453,6 @@ public class PushPredicateIntoTableScan implements
PlanOptimizer {
metadataExpressions, expressionsCanPushDown,
expressionsCannotPushDown);
}
- /** Get deviceEntries and DataPartition used in TableScan. */
- private PlanNode tableMetadataIndexScan(
- TableScanNode tableScanNode, List<Expression> metadataExpressions) {
-
- ProjectNode newProjectNode =
- addProjectNodeIfColumnRenamed(tableScanNode, metadataExpressions);
-
- getDeviceEntriesWithDataPartitions(tableScanNode, metadataExpressions);
-
- return newProjectNode != null ? newProjectNode : tableScanNode;
- }
-
- private ProjectNode addProjectNodeIfColumnRenamed(
- TableScanNode tableScanNode, List<Expression> metadataExpressions) {
-
- // for join operator, columnSymbols in TableScanNode may be renamed in
Join situation,
- // in this situation we need add a new ProjectNode above TableScanNode.
- boolean hasColumnRenamed = false;
- for (Map.Entry<Symbol, ColumnSchema> entry :
tableScanNode.getAssignments().entrySet()) {
- Symbol columnSymbol = entry.getKey();
- ColumnSchema columnSchema = entry.getValue();
- if (!columnSymbol.getName().equals(columnSchema.getName())) {
- hasColumnRenamed = true;
- break;
- }
- }
-
- if (!hasColumnRenamed) {
- return null;
- }
-
- metadataExpressions.replaceAll(
- expression ->
- ReplaceSymbolInExpression.transform(expression,
tableScanNode.getAssignments()));
- if (tableScanNode.getPushDownPredicate() != null) {
- ReplaceSymbolInExpression.transform(
- tableScanNode.getPushDownPredicate(),
tableScanNode.getAssignments());
- }
-
- int size = tableScanNode.getOutputSymbols().size();
- List<Symbol> newTableScanSymbols = new ArrayList<>(size);
- Map<Symbol, ColumnSchema> newTableScanAssignments = new
LinkedHashMap<>(size);
- Map<Symbol, Expression> projectAssignments = new LinkedHashMap<>(size);
- for (Map.Entry<Symbol, ColumnSchema> entry :
tableScanNode.getAssignments().entrySet()) {
- Symbol originalSymbol = entry.getKey();
- ColumnSchema columnSchema = entry.getValue();
-
- Symbol realSymbol = Symbol.of(columnSchema.getName());
- newTableScanSymbols.add(realSymbol);
- newTableScanAssignments.put(realSymbol, columnSchema);
- projectAssignments.put(originalSymbol, new
SymbolReference(columnSchema.getName()));
- queryContext.getTypeProvider().putTableModelType(originalSymbol,
columnSchema.getType());
- Map<Symbol, Integer> idAndAttributeIndexMap =
tableScanNode.getIdAndAttributeIndexMap();
- if (idAndAttributeIndexMap.containsKey(originalSymbol)) {
- Integer idx = idAndAttributeIndexMap.get(originalSymbol);
- idAndAttributeIndexMap.remove(originalSymbol);
- idAndAttributeIndexMap.put(realSymbol, idx);
- }
- }
-
- tableScanNode.setOutputSymbols(newTableScanSymbols);
- tableScanNode.setAssignments(newTableScanAssignments);
- return new ProjectNode(
- queryId.genPlanNodeId(), tableScanNode, new
Assignments(projectAssignments));
- }
-
private void getDeviceEntriesWithDataPartitions(
TableScanNode tableScanNode, List<Expression> metadataExpressions) {
@@ -673,23 +663,28 @@ public class PushPredicateIntoTableScan implements
PlanOptimizer {
newJoinFilter,
node.isSpillable());
}
- Symbol timeSymbol = Symbol.of("time");
- OrderingScheme orderingScheme =
+
+ 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(timeSymbol),
- Collections.singletonMap(timeSymbol, ASC_NULLS_LAST));
+ Collections.singletonList(joinCriteria.getRight()),
+ Collections.singletonMap(joinCriteria.getRight(),
ASC_NULLS_LAST));
SortNode leftSortNode =
new SortNode(
queryId.genPlanNodeId(),
((JoinNode) output).getLeftChild(),
- orderingScheme,
+ leftOrderingScheme,
false,
false);
SortNode rightSortNode =
new SortNode(
queryId.genPlanNodeId(),
((JoinNode) output).getRightChild(),
- orderingScheme,
+ rightOrderingScheme,
false,
false);
((JoinNode) output).setLeftChild(leftSortNode);
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/SortElimination.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/SortElimination.java
index e85d7623592..da83b773124 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/SortElimination.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/SortElimination.java
@@ -1,15 +1,20 @@
/*
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
*/
package org.apache.iotdb.db.queryengine.plan.relational.planner.optimizations;
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/JoinTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/JoinTest.java
index 3f17f59a519..fdcf49a9142 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/JoinTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/JoinTest.java
@@ -78,28 +78,25 @@ public class JoinTest {
@Test
public void innerJoinTest1() {
// join on
- // assertInnerJoinTest1(
- // "SELECT t1.time, t1.tag1, t1.tag2, t1.attr2, t1.s1, t1.s2,"
- // + "t2.tag1, t2.tag3, t2.attr2, t2.s1, t2.s3 "
- // + "FROM table1 t1 JOIN table1 t2 ON t1.time = t2.time
OFFSET 3 LIMIT 6",
- // false);
- //
- // // implicit join
- // assertInnerJoinTest1(
- // "SELECT t1.time, t1.tag1, t1.tag2, t1.attr2, t1.s1, t1.s2,"
- // + "t2.tag1, t2.tag3, t2.attr2, t2.s1, t2.s3 "
- // + "FROM table1 t1, table1 t2 WHERE t1.time = t2.time
OFFSET 3 LIMIT 6",
- // false);
+ assertInnerJoinTest1(
+ "SELECT t1.time, t1.tag1, t1.tag2, t1.attr2, t1.s1, t1.s2,"
+ + "t2.tag1, t2.tag3, t2.attr2, t2.s1, t2.s3 "
+ + "FROM table1 t1 JOIN table1 t2 ON t1.time = t2.time OFFSET 3
LIMIT 6");
+
+ // implicit join
+ assertInnerJoinTest1(
+ "SELECT t1.time, t1.tag1, t1.tag2, t1.attr2, t1.s1, t1.s2,"
+ + "t2.tag1, t2.tag3, t2.attr2, t2.s1, t2.s3 "
+ + "FROM table1 t1, table1 t2 WHERE t1.time = t2.time OFFSET 3
LIMIT 6");
// join using
assertInnerJoinTest1(
"SELECT time, t1.tag1, t1.tag2, t1.attr2, t1.s1, t1.s2,"
+ "t2.tag1, t2.tag3, t2.attr2, t2.s1, t2.s3 "
- + "FROM table1 t1 JOIN table1 t2 USING(time) OFFSET 3 LIMIT 6",
- true);
+ + "FROM table1 t1 JOIN table1 t2 USING(time) OFFSET 3 LIMIT 6");
}
- private void assertInnerJoinTest1(String sql, boolean joinUsing) {
+ private void assertInnerJoinTest1(String sql) {
analysis = analyzeSQL(sql, TEST_MATADATA, QUERY_CONTEXT);
logicalQueryPlan =
new TableLogicalPlanner(QUERY_CONTEXT, TEST_MATADATA, SESSION_INFO,
DEFAULT_WARNING)
@@ -107,23 +104,10 @@ public class JoinTest {
// LogicalPlan: `Output-Offset-Limit-Join-(Left +
Right)-Sort-(Project)-TableScan`
logicalPlanNode = logicalQueryPlan.getRootNode();
- if (joinUsing) {
- assertNodeMatches(
- logicalPlanNode,
- OutputNode.class,
- OffsetNode.class,
- ProjectNode.class,
- LimitNode.class,
- JoinNode.class);
- } else {
- assertNodeMatches(
- logicalPlanNode, OutputNode.class, OffsetNode.class,
LimitNode.class, JoinNode.class);
- }
+ assertNodeMatches(
+ logicalPlanNode, OutputNode.class, OffsetNode.class, LimitNode.class,
JoinNode.class);
- joinNode =
- joinUsing
- ? (JoinNode) getChildrenNode(logicalPlanNode, 4)
- : (JoinNode) getChildrenNode(logicalPlanNode, 3);
+ joinNode = (JoinNode) getChildrenNode(logicalPlanNode, 3);
List<JoinNode.EquiJoinClause> joinCriteria =
Collections.singletonList(
new JoinNode.EquiJoinClause(Symbol.of("time"),
Symbol.of("time_0")));
@@ -163,27 +147,31 @@ public class JoinTest {
* │ └──TableScanNode-123
* └──ExchangeNode-175:
[SourceAddress:192.0.10.1/test_query.3.0/177]
*
- * IdentitySinkNode-176
- * ├──SortNode-116
- * │ └──TableScanNode-112
- * └──SortNode-129
- * └──ProjectNode-125
- * └──TableScanNode-122
+ * IdentitySinkNode-201
+ * └──SortNode-141
+ * └──TableScanNode-137
*
- * IdentitySinkNode-177
- * ├──SortNode-118
- * │ └──TableScanNode-114
- * └──SortNode-131
- * └──ProjectNode-127
- * └──TableScanNode-124
+ * IdentitySinkNode-201
+ * └──SortNode-141
+ * └──TableScanNode-137
+ *
+ * IdentitySinkNode-203
+ * └──SortNode-154
+ * └──ProjectNode-150
+ * └──TableScanNode-147
+ *
+ * IdentitySinkNode-203
+ * └──SortNode-154
+ * └──ProjectNode-150
+ * └──TableScanNode-147
*/
distributedQueryPlan = new TableDistributedPlanner(analysis,
logicalQueryPlan).plan();
- assertEquals(3, distributedQueryPlan.getFragments().size());
+ assertEquals(5, distributedQueryPlan.getFragments().size());
IdentitySinkNode identitySinkNode =
(IdentitySinkNode)
distributedQueryPlan.getFragments().get(0).getPlanNodeTree();
outputNode = (OutputNode) getChildrenNode(identitySinkNode, 1);
- assertTrue(getChildrenNode(outputNode, joinUsing ? 4 : 3) instanceof
JoinNode);
- joinNode = (JoinNode) getChildrenNode(outputNode, joinUsing ? 4 : 3);
+ assertTrue(getChildrenNode(outputNode, 3) instanceof JoinNode);
+ joinNode = (JoinNode) getChildrenNode(outputNode, 3);
assertTrue(joinNode.getLeftChild() instanceof MergeSortNode);
MergeSortNode mergeSortNode = (MergeSortNode) joinNode.getLeftChild();
assertMergeSortNode(mergeSortNode);
@@ -193,7 +181,17 @@ public class JoinTest {
identitySinkNode =
(IdentitySinkNode)
distributedQueryPlan.getFragments().get(1).getPlanNodeTree();
- tableScanNode = (TableScanNode)
getChildrenNode(identitySinkNode.getChildren().get(1), 2);
+ assertTrue(getChildrenNode(identitySinkNode, 1) instanceof SortNode);
+ assertTrue(getChildrenNode(identitySinkNode, 2) instanceof TableScanNode);
+ tableScanNode = (TableScanNode) getChildrenNode(identitySinkNode, 2);
+ assertTableScan(tableScanNode, SHENZHEN_DEVICE_ENTRIES, Ordering.ASC, 0,
0, true, "");
+
+ identitySinkNode =
+ (IdentitySinkNode)
distributedQueryPlan.getFragments().get(3).getPlanNodeTree();
+ assertTrue(getChildrenNode(identitySinkNode, 1) instanceof SortNode);
+ assertTrue(getChildrenNode(identitySinkNode, 2) instanceof ProjectNode);
+ assertTrue(getChildrenNode(identitySinkNode, 3) instanceof TableScanNode);
+ tableScanNode = (TableScanNode) getChildrenNode(identitySinkNode, 3);
assertTableScan(tableScanNode, SHENZHEN_DEVICE_ENTRIES, Ordering.ASC, 0,
0, true, "");
}
