This is an automated email from the ASF dual-hosted git repository.
jackietien pushed a commit to branch ty/InnerTimeJoin
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/ty/InnerTimeJoin by this push:
new e56bc757f5e support for TimePartitions in InnerTimeJoinNode
e56bc757f5e is described below
commit e56bc757f5e72af2954befc301b8b2560ba238f2
Author: JackieTien97 <[email protected]>
AuthorDate: Fri Jan 5 11:28:23 2024 +0800
support for TimePartitions in InnerTimeJoinNode
---
.../execution/fragment/FragmentInstanceContext.java | 15 ++++++++++++++-
.../plan/planner/LocalExecutionPlanContext.java | 16 ++++++++++++++++
.../plan/planner/LocalExecutionPlanner.java | 2 ++
.../plan/planner/OperatorTreeGenerator.java | 2 ++
.../plan/planner/distribution/SourceRewriter.java | 1 +
.../db/storageengine/dataregion/DataRegion.java | 7 ++++---
.../dataregion/IDataRegionForQuery.java | 3 ++-
.../storageengine/dataregion/VirtualDataRegion.java | 3 ++-
.../dataregion/tsfile/TsFileManager.java | 21 +++++++++++++++++++--
9 files changed, 62 insertions(+), 8 deletions(-)
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 921f7739214..bf39dec609a 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
@@ -71,6 +71,10 @@ public class FragmentInstanceContext extends QueryContext {
/** check if there is tmp file to be deleted. */
private boolean mayHaveTmpFile = false;
+ // null for all time partitions
+ // empty for zero time partitions
+ private List<Long> timePartitions;
+
private final AtomicLong startNanos = new AtomicLong();
private final AtomicLong endNanos = new AtomicLong();
@@ -339,7 +343,8 @@ public class FragmentInstanceContext extends QueryContext {
selectedDeviceIdSet.size() == 1 ?
selectedDeviceIdSet.iterator().next() : null,
this,
// time filter may be stateful, so we need to copy it
- globalTimeFilter != null ? globalTimeFilter.copy() : null);
+ globalTimeFilter != null ? globalTimeFilter.copy() : null,
+ timePartitions);
// used files should be added before mergeLock is unlocked, or they may
be deleted by
// running merge
@@ -528,4 +533,12 @@ public class FragmentInstanceContext extends QueryContext {
public boolean mayHaveTmpFile() {
return mayHaveTmpFile;
}
+
+ public Optional<List<Long>> getTimePartitions() {
+ return Optional.ofNullable(timePartitions);
+ }
+
+ public void setTimePartitions(List<Long> timePartitions) {
+ this.timePartitions = timePartitions;
+ }
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LocalExecutionPlanContext.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LocalExecutionPlanContext.java
index 7a6d5c57ade..3d9e635340d 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LocalExecutionPlanContext.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LocalExecutionPlanContext.java
@@ -44,9 +44,11 @@ import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
+import java.util.Optional;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
import static com.google.common.base.Preconditions.checkArgument;
import static java.util.Objects.requireNonNull;
@@ -82,6 +84,11 @@ public class LocalExecutionPlanContext {
public final DataNodeQueryContext dataNodeQueryContext;
+ // null for all time partitions
+ // empty for zero time partitions
+ // use AtomicReference not for thread-safe, just for updating same field in
different pipeline
+ private AtomicReference<List<Long>> timePartitions = new AtomicReference<>();
+
// for data region
public LocalExecutionPlanContext(
TypeProvider typeProvider,
@@ -112,6 +119,7 @@ public class LocalExecutionPlanContext {
this.driverContext =
parentContext.getDriverContext().createSubDriverContext(getNextPipelineId());
this.dataNodeQueryContext = parentContext.dataNodeQueryContext;
+ this.timePartitions = parentContext.timePartitions;
}
// for schema region
@@ -275,4 +283,12 @@ public class LocalExecutionPlanContext {
public Filter getGlobalTimeFilter() {
return driverContext.getFragmentInstanceContext().getGlobalTimeFilter();
}
+
+ public Optional<List<Long>> getTimePartitions() {
+ return Optional.ofNullable(timePartitions.get());
+ }
+
+ public void setTimePartitions(List<Long> timePartitions) {
+ this.timePartitions.set(timePartitions);
+ }
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LocalExecutionPlanner.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LocalExecutionPlanner.java
index 0b1fc6cecd7..f21e67b7323 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LocalExecutionPlanner.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LocalExecutionPlanner.java
@@ -79,6 +79,8 @@ public class LocalExecutionPlanner {
instanceContext.setSourcePaths(collectSourcePaths(context));
+ context.getTimePartitions().ifPresent(instanceContext::setTimePartitions);
+
// set maxBytes one SourceHandle can reserve after visiting the whole tree
context.setMaxBytesOneHandleCanReserve();
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 514c8e3ab10..d4b80556d52 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
@@ -2065,6 +2065,8 @@ public class OperatorTreeGenerator extends
PlanVisitor<Operator, LocalExecutionP
@Override
public Operator visitInnerTimeJoin(InnerTimeJoinNode node,
LocalExecutionPlanContext context) {
+ node.getTimePartitions().ifPresent(context::setTimePartitions);
+
List<Operator> children = dealWithConsumeAllChildrenPipelineBreaker(node,
context);
OperatorContext operatorContext =
context
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/distribution/SourceRewriter.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/distribution/SourceRewriter.java
index 460e4938480..6950ff95df4 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/distribution/SourceRewriter.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/distribution/SourceRewriter.java
@@ -729,6 +729,7 @@ public class SourceRewriter extends
SimplePlanNodeRewriter<DistributionPlanConte
context.queryContext.getQueryId().genPlanNodeId(),
node.getMergeOrder() == Ordering.ASC ? TIME_ASC : TIME_DESC,
node.getOutputColumnNames());
+
mergeSortNode.setChildren(children);
return Collections.singletonList(mergeSortNode);
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java
index 2819d0a4a00..f877b3f1c4a 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java
@@ -1708,12 +1708,13 @@ public class DataRegion implements IDataRegionForQuery {
List<PartialPath> pathList,
String singleDeviceId,
QueryContext context,
- Filter globalTimeFilter)
+ Filter globalTimeFilter,
+ List<Long> timePartitions)
throws QueryProcessException {
try {
List<TsFileResource> seqResources =
getFileResourceListForQuery(
- tsFileManager.getTsFileList(true),
+ tsFileManager.getTsFileList(true, timePartitions),
pathList,
singleDeviceId,
context,
@@ -1721,7 +1722,7 @@ public class DataRegion implements IDataRegionForQuery {
true);
List<TsFileResource> unseqResources =
getFileResourceListForQuery(
- tsFileManager.getTsFileList(false),
+ tsFileManager.getTsFileList(false, timePartitions),
pathList,
singleDeviceId,
context,
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/IDataRegionForQuery.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/IDataRegionForQuery.java
index 8de31c69b42..117cfa9b51e 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/IDataRegionForQuery.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/IDataRegionForQuery.java
@@ -39,7 +39,8 @@ public interface IDataRegionForQuery {
List<PartialPath> pathList,
String singleDeviceId,
QueryContext context,
- Filter globalTimeFilter)
+ Filter globalTimeFilter,
+ List<Long> timePartitions)
throws QueryProcessException;
/** Get TTL of this DataRegion */
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/VirtualDataRegion.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/VirtualDataRegion.java
index f8e07a520a8..5d4b6b1af52 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/VirtualDataRegion.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/VirtualDataRegion.java
@@ -57,7 +57,8 @@ public class VirtualDataRegion implements IDataRegionForQuery
{
List<PartialPath> pathList,
String singleDeviceId,
QueryContext context,
- Filter globalTimeFilter)
+ Filter globalTimeFilter,
+ List<Long> timePartitions)
throws QueryProcessException {
return EMPTY_QUERY_DATA_SOURCE;
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileManager.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileManager.java
index 69a3cff6901..3d85e88caac 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileManager.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileManager.java
@@ -58,14 +58,31 @@ public class TsFileManager {
}
public List<TsFileResource> getTsFileList(boolean sequence) {
+ return getTsFileList(sequence, null);
+ }
+
+ /**
+ * @param sequence true for sequence, false for unsequence
+ * @param timePartitions null for all time partitions, empty for zero time
partitions
+ */
+ public List<TsFileResource> getTsFileList(boolean sequence, List<Long>
timePartitions) {
// the iteration of ConcurrentSkipListMap is not concurrent secure
// so we must add read lock here
readLock();
try {
List<TsFileResource> allResources = new ArrayList<>();
Map<Long, TsFileResourceList> chosenMap = sequence ? sequenceFiles :
unsequenceFiles;
- for (Map.Entry<Long, TsFileResourceList> entry : chosenMap.entrySet()) {
- allResources.addAll(entry.getValue().getArrayList());
+ if (timePartitions == null) {
+ for (Map.Entry<Long, TsFileResourceList> entry : chosenMap.entrySet())
{
+ allResources.addAll(entry.getValue().getArrayList());
+ }
+ } else {
+ for (Long timePartitionId : timePartitions) {
+ TsFileResourceList tsFileResources = chosenMap.get(timePartitionId);
+ if (tsFileResources != null) {
+ allResources.addAll(tsFileResources.getArrayList());
+ }
+ }
}
return allResources;
} finally {
