This is an automated email from the ASF dual-hosted git repository. jackietien pushed a commit to branch stable-mpp in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 58e0565bd6d0e99ab605494e601aa58694dc76a7 Author: JackieTien97 <[email protected]> AuthorDate: Tue Apr 19 11:11:58 2022 +0800 add input location for TimeJoinNode --- .../db/mpp/sql/planner/plan/InputLocation.java | 39 ++++++++++++++++++++++ .../planner/plan/node/process/TimeJoinNode.java | 23 +++++++++++++ 2 files changed, 62 insertions(+) diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/plan/InputLocation.java b/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/plan/InputLocation.java new file mode 100644 index 0000000000..705d8cbace --- /dev/null +++ b/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/plan/InputLocation.java @@ -0,0 +1,39 @@ +/* + * 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.mpp.sql.planner.plan; + +public class InputLocation { + // which input tsblock + private final int tsBlockIndex; + // which value column of that tsblock + private final int valueColumnIndex; + + public InputLocation(int tsBlockIndex, int valueColumnIndex) { + this.tsBlockIndex = tsBlockIndex; + this.valueColumnIndex = valueColumnIndex; + } + + public int getTsBlockIndex() { + return tsBlockIndex; + } + + public int getValueColumnIndex() { + return valueColumnIndex; + } +} diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/plan/node/process/TimeJoinNode.java b/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/plan/node/process/TimeJoinNode.java index fdda60a311..484e49d8b4 100644 --- a/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/plan/node/process/TimeJoinNode.java +++ b/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/plan/node/process/TimeJoinNode.java @@ -21,6 +21,7 @@ package org.apache.iotdb.db.mpp.sql.planner.plan.node.process; import org.apache.iotdb.commons.utils.TestOnly; import org.apache.iotdb.db.mpp.common.header.ColumnHeader; import org.apache.iotdb.db.mpp.sql.planner.plan.IOutputPlanNode; +import org.apache.iotdb.db.mpp.sql.planner.plan.InputLocation; import org.apache.iotdb.db.mpp.sql.planner.plan.node.PlanNode; import org.apache.iotdb.db.mpp.sql.planner.plan.node.PlanNodeId; import org.apache.iotdb.db.mpp.sql.planner.plan.node.PlanNodeType; @@ -58,6 +59,16 @@ public class TimeJoinNode extends ProcessNode implements IOutputPlanNode { private List<ColumnHeader> columnHeaders = new ArrayList<>(); + // indicate each output column should use which value column of which input TsBlock + // size of locationList must be equal to the size of columnHeaders + private List<List<InputLocation>> locationList = new ArrayList<>(); + + // if overlapped[i] is true, it means that locationList.get(i).size() > 1 and input locations in + // locationList.get(i) are overlapped + // it will only happen when we do the load balance and more than one DataRegion is assigned to one + // time partition + private boolean[] overlapped; + public TimeJoinNode(PlanNodeId id, OrderBy mergeOrder) { super(id); this.mergeOrder = mergeOrder; @@ -125,6 +136,18 @@ public class TimeJoinNode extends ProcessNode implements IOutputPlanNode { } } + public List<List<InputLocation>> getLocationList() { + return locationList; + } + + public void setOverlapped(boolean[] overlapped) { + this.overlapped = overlapped; + } + + public boolean[] getOverlapped() { + return overlapped; + } + public static TimeJoinNode deserialize(ByteBuffer byteBuffer) { OrderBy orderBy = OrderBy.values()[ReadWriteIOUtils.readInt(byteBuffer)]; FilterNullPolicy filterNullPolicy =
