You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ro...@apache.org on 2024/10/09 10:22:40 UTC
(iotdb) branch master updated: Pipe: Introduce TablePattern for table model (#13695)
This is an automated email from the ASF dual-hosted git repository.
rong 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 360ca21fe3c Pipe: Introduce TablePattern for table model (#13695)
360ca21fe3c is described below
commit 360ca21fe3cebb538bf759f8fb0b65fb33b669cd
Author: Steve Yurong Su <ro...@apache.org>
AuthorDate: Wed Oct 9 18:22:34 2024 +0800
Pipe: Introduce TablePattern for table model (#13695)
---
...FormatIT.java => IoTDBTreePatternFormatIT.java} | 2 +-
.../protocol/IoTDBConfigRegionAirGapConnector.java | 2 +-
.../protocol/IoTDBConfigRegionConnector.java | 2 +-
.../pipe/event/PipeConfigRegionSnapshotEvent.java | 28 +++++--
.../pipe/event/PipeConfigRegionWritePlanEvent.java | 15 ++--
.../PipeConfigPhysicalPlanPatternParseVisitor.java | 52 ++++++------
.../receiver/protocol/IoTDBConfigNodeReceiver.java | 6 +-
...eConfigPhysicalPlanPatternParseVisitorTest.java | 6 +-
.../agent/task/connection/PipeEventCollector.java | 7 +-
.../airgap/IoTDBSchemaRegionAirGapConnector.java | 2 +-
.../thrift/sync/IoTDBSchemaRegionConnector.java | 2 +-
.../db/pipe/event/UserDefinedEnrichedEvent.java | 11 ++-
.../event/common/heartbeat/PipeHeartbeatEvent.java | 10 ++-
.../schema/PipeSchemaRegionSnapshotEvent.java | 22 +++--
.../schema/PipeSchemaRegionWritePlanEvent.java | 21 +++--
.../tablet/PipeInsertNodeTabletInsertionEvent.java | 23 +++---
.../common/tablet/PipeRawTabletInsertionEvent.java | 42 +++++++---
.../tablet/TabletInsertionDataContainer.java | 16 ++--
.../event/common/terminate/PipeTerminateEvent.java | 8 +-
.../common/tsfile/PipeTsFileInsertionEvent.java | 21 +++--
.../common/tsfile/TsFileInsertionPointCounter.java | 6 +-
.../container/TsFileInsertionDataContainer.java | 6 +-
.../TsFileInsertionDataContainerProvider.java | 14 ++--
.../query/TsFileInsertionQueryDataContainer.java | 8 +-
.../scan/TsFileInsertionScanDataContainer.java | 4 +-
.../db/pipe/event/realtime/PipeRealtimeEvent.java | 30 +++++--
.../event/realtime/PipeRealtimeEventFactory.java | 6 +-
.../dataregion/IoTDBDataRegionExtractor.java | 38 +++++----
.../PipeHistoricalDataRegionTsFileExtractor.java | 20 +++--
.../realtime/PipeRealtimeDataRegionExtractor.java | 57 +++++++++----
.../realtime/assigner/PipeDataRegionAssigner.java | 6 +-
.../realtime/epoch/TsFileEpochManager.java | 6 +-
.../matcher/CachedSchemaPatternMatcher.java | 9 ++-
.../schemaregion/PipePlanPatternParseVisitor.java | 52 ++++++------
.../protocol/thrift/IoTDBDataNodeReceiver.java | 6 +-
...peStatementDataTypeConvertExecutionVisitor.java | 4 +-
.../visitor/PipeStatementPatternParseVisitor.java | 26 +++---
.../PipeStatementPatternParseVisitorTest.java | 6 +-
.../pipe/event/PipeTabletInsertionEventTest.java | 26 +++---
.../event/TsFileInsertionDataContainerTest.java | 48 +++++------
.../extractor/PipePlanPatternParseVisitorTest.java | 6 +-
.../pattern/CachedSchemaPatternMatcherTest.java | 16 ++--
...ePatternTest.java => IoTDBTreePatternTest.java} | 24 +++---
...PatternTest.java => PrefixTreePatternTest.java} | 24 +++---
.../config/constant/PipeExtractorConstant.java | 4 +
...IoTDBPipePattern.java => IoTDBTreePattern.java} | 12 +--
...efixPipePattern.java => PrefixTreePattern.java} | 4 +-
.../pipe/datastructure/pattern/TablePattern.java | 94 ++++++++++++++++++++++
.../pattern/{PipePattern.java => TreePattern.java} | 28 +++----
.../iotdb/commons/pipe/event/EnrichedEvent.java | 48 +++++++----
.../commons/pipe/event/PipeSnapshotEvent.java | 15 +++-
.../commons/pipe/event/PipeWritePlanEvent.java | 15 +++-
.../commons/pipe/event/ProgressReportEvent.java | 13 +--
.../extractor/IoTDBNonDataRegionExtractor.java | 40 ++++++---
54 files changed, 655 insertions(+), 364 deletions(-)
diff --git a/integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBPipePatternFormatIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBTreePatternFormatIT.java
similarity index 99%
rename from integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBPipePatternFormatIT.java
rename to integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBTreePatternFormatIT.java
index 1d80e175344..ea6f99c75b9 100644
--- a/integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBPipePatternFormatIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBTreePatternFormatIT.java
@@ -41,7 +41,7 @@ import java.util.Set;
@RunWith(IoTDBTestRunner.class)
@Category({MultiClusterIT2AutoCreateSchema.class})
-public class IoTDBPipePatternFormatIT extends AbstractPipeDualAutoIT {
+public class IoTDBTreePatternFormatIT extends AbstractPipeDualAutoIT {
@Test
public void testPrefixPattern() throws Exception {
final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/connector/protocol/IoTDBConfigRegionAirGapConnector.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/connector/protocol/IoTDBConfigRegionAirGapConnector.java
index e615616e2ba..a03bda94050 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/connector/protocol/IoTDBConfigRegionAirGapConnector.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/connector/protocol/IoTDBConfigRegionAirGapConnector.java
@@ -211,7 +211,7 @@ public class IoTDBConfigRegionAirGapConnector extends IoTDBAirGapConnector {
socket,
PipeTransferConfigSnapshotSealReq.toTPipeTransferBytes(
// The pattern is surely Non-null
- pipeConfigRegionSnapshotEvent.getPatternString(),
+ pipeConfigRegionSnapshotEvent.getTreePatternString(),
snapshot.getName(),
snapshot.length(),
Objects.nonNull(templateFile) ? templateFile.getName() : null,
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/connector/protocol/IoTDBConfigRegionConnector.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/connector/protocol/IoTDBConfigRegionConnector.java
index 307151df706..d768050f806 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/connector/protocol/IoTDBConfigRegionConnector.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/connector/protocol/IoTDBConfigRegionConnector.java
@@ -217,7 +217,7 @@ public class IoTDBConfigRegionConnector extends IoTDBSslSyncConnector {
compressIfNeeded(
PipeTransferConfigSnapshotSealReq.toTPipeTransferReq(
// The pattern is surely Non-null
- snapshotEvent.getPatternString(),
+ snapshotEvent.getTreePatternString(),
snapshotFile.getName(),
snapshotFile.length(),
Objects.nonNull(templateFile) ? templateFile.getName() : null,
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/event/PipeConfigRegionSnapshotEvent.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/event/PipeConfigRegionSnapshotEvent.java
index 8e17cb9b515..8705d577942 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/event/PipeConfigRegionSnapshotEvent.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/event/PipeConfigRegionSnapshotEvent.java
@@ -20,7 +20,8 @@
package org.apache.iotdb.confignode.manager.pipe.event;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta;
-import org.apache.iotdb.commons.pipe.datastructure.pattern.PipePattern;
+import org.apache.iotdb.commons.pipe.datastructure.pattern.TablePattern;
+import org.apache.iotdb.commons.pipe.datastructure.pattern.TreePattern;
import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
import org.apache.iotdb.commons.pipe.event.PipeSnapshotEvent;
import org.apache.iotdb.confignode.consensus.request.ConfigPhysicalPlanType;
@@ -90,7 +91,7 @@ public class PipeConfigRegionSnapshotEvent extends PipeSnapshotEvent {
public PipeConfigRegionSnapshotEvent(
final String snapshotPath, final String templateFilePath, final CNSnapshotFileType type) {
- this(snapshotPath, templateFilePath, type, null, 0, null, null);
+ this(snapshotPath, templateFilePath, type, null, 0, null, null, null);
}
public PipeConfigRegionSnapshotEvent(
@@ -100,8 +101,15 @@ public class PipeConfigRegionSnapshotEvent extends PipeSnapshotEvent {
final String pipeName,
final long creationTime,
final PipeTaskMeta pipeTaskMeta,
- final PipePattern pattern) {
- super(pipeName, creationTime, pipeTaskMeta, pattern, PipeConfigNodeResourceManager.snapshot());
+ final TreePattern treePattern,
+ final TablePattern tablePattern) {
+ super(
+ pipeName,
+ creationTime,
+ pipeTaskMeta,
+ treePattern,
+ tablePattern,
+ PipeConfigNodeResourceManager.snapshot());
this.snapshotPath = snapshotPath;
this.templateFilePath = Objects.nonNull(templateFilePath) ? templateFilePath : "";
this.fileType = type;
@@ -160,11 +168,19 @@ public class PipeConfigRegionSnapshotEvent extends PipeSnapshotEvent {
final String pipeName,
final long creationTime,
final PipeTaskMeta pipeTaskMeta,
- final PipePattern pattern,
+ final TreePattern treePattern,
+ final TablePattern tablePattern,
final long startTime,
final long endTime) {
return new PipeConfigRegionSnapshotEvent(
- snapshotPath, templateFilePath, fileType, pipeName, creationTime, pipeTaskMeta, pattern);
+ snapshotPath,
+ templateFilePath,
+ fileType,
+ pipeName,
+ creationTime,
+ pipeTaskMeta,
+ treePattern,
+ tablePattern);
}
@Override
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/event/PipeConfigRegionWritePlanEvent.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/event/PipeConfigRegionWritePlanEvent.java
index 1515724c1e5..241c54c1f01 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/event/PipeConfigRegionWritePlanEvent.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/event/PipeConfigRegionWritePlanEvent.java
@@ -20,7 +20,8 @@
package org.apache.iotdb.confignode.manager.pipe.event;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta;
-import org.apache.iotdb.commons.pipe.datastructure.pattern.PipePattern;
+import org.apache.iotdb.commons.pipe.datastructure.pattern.TablePattern;
+import org.apache.iotdb.commons.pipe.datastructure.pattern.TreePattern;
import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
import org.apache.iotdb.commons.pipe.event.PipeWritePlanEvent;
import org.apache.iotdb.confignode.consensus.request.ConfigPhysicalPlan;
@@ -41,7 +42,7 @@ public class PipeConfigRegionWritePlanEvent extends PipeWritePlanEvent {
public PipeConfigRegionWritePlanEvent(
final ConfigPhysicalPlan configPhysicalPlan, final boolean isGeneratedByPipe) {
- this(configPhysicalPlan, null, 0, null, null, isGeneratedByPipe);
+ this(configPhysicalPlan, null, 0, null, null, null, isGeneratedByPipe);
}
public PipeConfigRegionWritePlanEvent(
@@ -49,9 +50,10 @@ public class PipeConfigRegionWritePlanEvent extends PipeWritePlanEvent {
final String pipeName,
final long creationTime,
final PipeTaskMeta pipeTaskMeta,
- final PipePattern pattern,
+ final TreePattern treePattern,
+ final TablePattern tablePattern,
final boolean isGeneratedByPipe) {
- super(pipeName, creationTime, pipeTaskMeta, pattern, isGeneratedByPipe);
+ super(pipeName, creationTime, pipeTaskMeta, treePattern, tablePattern, isGeneratedByPipe);
this.configPhysicalPlan = configPhysicalPlan;
}
@@ -64,11 +66,12 @@ public class PipeConfigRegionWritePlanEvent extends PipeWritePlanEvent {
final String pipeName,
final long creationTime,
final PipeTaskMeta pipeTaskMeta,
- final PipePattern pattern,
+ final TreePattern treePattern,
+ final TablePattern tablePattern,
final long startTime,
final long endTime) {
return new PipeConfigRegionWritePlanEvent(
- configPhysicalPlan, pipeName, creationTime, pipeTaskMeta, pattern, false);
+ configPhysicalPlan, pipeName, creationTime, pipeTaskMeta, treePattern, tablePattern, false);
}
@Override
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/extractor/PipeConfigPhysicalPlanPatternParseVisitor.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/extractor/PipeConfigPhysicalPlanPatternParseVisitor.java
index 3d0602875bc..b1ed53faea4 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/extractor/PipeConfigPhysicalPlanPatternParseVisitor.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/extractor/PipeConfigPhysicalPlanPatternParseVisitor.java
@@ -21,7 +21,7 @@ package org.apache.iotdb.confignode.manager.pipe.extractor;
import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.commons.path.PathPatternTree;
-import org.apache.iotdb.commons.pipe.datastructure.pattern.IoTDBPipePattern;
+import org.apache.iotdb.commons.pipe.datastructure.pattern.IoTDBTreePattern;
import org.apache.iotdb.confignode.consensus.request.ConfigPhysicalPlan;
import org.apache.iotdb.confignode.consensus.request.ConfigPhysicalPlanVisitor;
import org.apache.iotdb.confignode.consensus.request.write.auth.AuthorPlan;
@@ -56,12 +56,12 @@ import java.util.stream.Stream;
/**
* The {@link PipeConfigPhysicalPlanPatternParseVisitor} will transform the schema {@link
- * ConfigPhysicalPlan}s using {@link IoTDBPipePattern}. Rule:
+ * ConfigPhysicalPlan}s using {@link IoTDBTreePattern}. Rule:
*
* <p>1. All patterns in the output {@link ConfigPhysicalPlan} will be the intersection of the
- * original {@link ConfigPhysicalPlan}'s patterns and the given {@link IoTDBPipePattern}.
+ * original {@link ConfigPhysicalPlan}'s patterns and the given {@link IoTDBTreePattern}.
*
- * <p>2. If a pattern does not intersect with the {@link IoTDBPipePattern}, it's dropped.
+ * <p>2. If a pattern does not intersect with the {@link IoTDBTreePattern}, it's dropped.
*
* <p>3. If all the patterns in the {@link ConfigPhysicalPlan} is dropped, the {@link
* ConfigPhysicalPlan} is dropped.
@@ -70,13 +70,13 @@ import java.util.stream.Stream;
* one is used in the {@link PipeConfigRegionWritePlanEvent} in {@link ConfigRegionListeningQueue}.
*/
public class PipeConfigPhysicalPlanPatternParseVisitor
- extends ConfigPhysicalPlanVisitor<Optional<ConfigPhysicalPlan>, IoTDBPipePattern> {
+ extends ConfigPhysicalPlanVisitor<Optional<ConfigPhysicalPlan>, IoTDBTreePattern> {
private static final Logger LOGGER =
LoggerFactory.getLogger(PipeConfigPhysicalPlanPatternParseVisitor.class);
@Override
public Optional<ConfigPhysicalPlan> visitPlan(
- final ConfigPhysicalPlan plan, final IoTDBPipePattern pattern) {
+ final ConfigPhysicalPlan plan, final IoTDBTreePattern pattern) {
return Optional.of(plan);
}
@@ -88,7 +88,7 @@ public class PipeConfigPhysicalPlanPatternParseVisitor
// Other matches using "matchPrefixPath" are with the same principle.
@Override
public Optional<ConfigPhysicalPlan> visitCreateDatabase(
- final DatabaseSchemaPlan createDatabasePlan, final IoTDBPipePattern pattern) {
+ final DatabaseSchemaPlan createDatabasePlan, final IoTDBTreePattern pattern) {
return pattern.matchPrefixPath(createDatabasePlan.getSchema().getName())
? Optional.of(createDatabasePlan)
: Optional.empty();
@@ -96,7 +96,7 @@ public class PipeConfigPhysicalPlanPatternParseVisitor
@Override
public Optional<ConfigPhysicalPlan> visitAlterDatabase(
- final DatabaseSchemaPlan alterDatabasePlan, final IoTDBPipePattern pattern) {
+ final DatabaseSchemaPlan alterDatabasePlan, final IoTDBTreePattern pattern) {
return pattern.matchPrefixPath(alterDatabasePlan.getSchema().getName())
? Optional.of(alterDatabasePlan)
: Optional.empty();
@@ -104,7 +104,7 @@ public class PipeConfigPhysicalPlanPatternParseVisitor
@Override
public Optional<ConfigPhysicalPlan> visitDeleteDatabase(
- final DeleteDatabasePlan deleteDatabasePlan, final IoTDBPipePattern pattern) {
+ final DeleteDatabasePlan deleteDatabasePlan, final IoTDBTreePattern pattern) {
return pattern.matchPrefixPath(deleteDatabasePlan.getName())
? Optional.of(deleteDatabasePlan)
: Optional.empty();
@@ -112,7 +112,7 @@ public class PipeConfigPhysicalPlanPatternParseVisitor
@Override
public Optional<ConfigPhysicalPlan> visitCreateSchemaTemplate(
- final CreateSchemaTemplatePlan createSchemaTemplatePlan, final IoTDBPipePattern pattern) {
+ final CreateSchemaTemplatePlan createSchemaTemplatePlan, final IoTDBTreePattern pattern) {
// This is a deserialized template and can be arbitrarily altered
final Template template = createSchemaTemplatePlan.getTemplate();
template.getSchemaMap().keySet().removeIf(measurement -> !pattern.matchTailNode(measurement));
@@ -124,7 +124,7 @@ public class PipeConfigPhysicalPlanPatternParseVisitor
@Override
public Optional<ConfigPhysicalPlan> visitCommitSetSchemaTemplate(
final CommitSetSchemaTemplatePlan commitSetSchemaTemplatePlan,
- final IoTDBPipePattern pattern) {
+ final IoTDBTreePattern pattern) {
return pattern.matchPrefixPath(commitSetSchemaTemplatePlan.getPath())
? Optional.of(commitSetSchemaTemplatePlan)
: Optional.empty();
@@ -133,7 +133,7 @@ public class PipeConfigPhysicalPlanPatternParseVisitor
@Override
public Optional<ConfigPhysicalPlan> visitPipeUnsetSchemaTemplate(
final PipeUnsetSchemaTemplatePlan pipeUnsetSchemaTemplatePlan,
- final IoTDBPipePattern pattern) {
+ final IoTDBTreePattern pattern) {
return pattern.matchPrefixPath(pipeUnsetSchemaTemplatePlan.getPath())
? Optional.of(pipeUnsetSchemaTemplatePlan)
: Optional.empty();
@@ -141,7 +141,7 @@ public class PipeConfigPhysicalPlanPatternParseVisitor
@Override
public Optional<ConfigPhysicalPlan> visitExtendSchemaTemplate(
- final ExtendSchemaTemplatePlan extendSchemaTemplatePlan, final IoTDBPipePattern pattern) {
+ final ExtendSchemaTemplatePlan extendSchemaTemplatePlan, final IoTDBTreePattern pattern) {
final TemplateExtendInfo extendInfo = extendSchemaTemplatePlan.getTemplateExtendInfo();
final int[] filteredIndexes =
IntStream.range(0, extendInfo.getMeasurements().size())
@@ -152,41 +152,41 @@ public class PipeConfigPhysicalPlanPatternParseVisitor
new ExtendSchemaTemplatePlan(
new TemplateExtendInfo(
extendInfo.getTemplateName(),
- IoTDBPipePattern.applyIndexesOnList(
+ IoTDBTreePattern.applyIndexesOnList(
filteredIndexes, extendInfo.getMeasurements()),
- IoTDBPipePattern.applyIndexesOnList(filteredIndexes, extendInfo.getDataTypes()),
- IoTDBPipePattern.applyIndexesOnList(filteredIndexes, extendInfo.getEncodings()),
- IoTDBPipePattern.applyIndexesOnList(
+ IoTDBTreePattern.applyIndexesOnList(filteredIndexes, extendInfo.getDataTypes()),
+ IoTDBTreePattern.applyIndexesOnList(filteredIndexes, extendInfo.getEncodings()),
+ IoTDBTreePattern.applyIndexesOnList(
filteredIndexes, extendInfo.getCompressors()))))
: Optional.empty();
}
@Override
public Optional<ConfigPhysicalPlan> visitGrantUser(
- final AuthorPlan grantUserPlan, final IoTDBPipePattern pattern) {
+ final AuthorPlan grantUserPlan, final IoTDBTreePattern pattern) {
return visitPathRelatedAuthorPlan(grantUserPlan, pattern);
}
@Override
public Optional<ConfigPhysicalPlan> visitRevokeUser(
- final AuthorPlan revokeUserPlan, final IoTDBPipePattern pattern) {
+ final AuthorPlan revokeUserPlan, final IoTDBTreePattern pattern) {
return visitPathRelatedAuthorPlan(revokeUserPlan, pattern);
}
@Override
public Optional<ConfigPhysicalPlan> visitGrantRole(
- final AuthorPlan revokeUserPlan, final IoTDBPipePattern pattern) {
+ final AuthorPlan revokeUserPlan, final IoTDBTreePattern pattern) {
return visitPathRelatedAuthorPlan(revokeUserPlan, pattern);
}
@Override
public Optional<ConfigPhysicalPlan> visitRevokeRole(
- final AuthorPlan revokeUserPlan, final IoTDBPipePattern pattern) {
+ final AuthorPlan revokeUserPlan, final IoTDBTreePattern pattern) {
return visitPathRelatedAuthorPlan(revokeUserPlan, pattern);
}
private Optional<ConfigPhysicalPlan> visitPathRelatedAuthorPlan(
- final AuthorPlan pathRelatedAuthorPlan, final IoTDBPipePattern pattern) {
+ final AuthorPlan pathRelatedAuthorPlan, final IoTDBTreePattern pattern) {
final List<PartialPath> intersectedPaths =
pathRelatedAuthorPlan.getNodeNameList().stream()
.map(pattern::getIntersection)
@@ -208,7 +208,7 @@ public class PipeConfigPhysicalPlanPatternParseVisitor
@Override
public Optional<ConfigPhysicalPlan> visitPipeDeleteTimeSeries(
- final PipeDeleteTimeSeriesPlan pipeDeleteTimeSeriesPlan, final IoTDBPipePattern pattern) {
+ final PipeDeleteTimeSeriesPlan pipeDeleteTimeSeriesPlan, final IoTDBTreePattern pattern) {
try {
final PathPatternTree intersectedTree =
pattern.getIntersection(
@@ -226,7 +226,7 @@ public class PipeConfigPhysicalPlanPatternParseVisitor
@Override
public Optional<ConfigPhysicalPlan> visitPipeDeleteLogicalView(
- final PipeDeleteLogicalViewPlan pipeDeleteLogicalViewPlan, final IoTDBPipePattern pattern) {
+ final PipeDeleteLogicalViewPlan pipeDeleteLogicalViewPlan, final IoTDBTreePattern pattern) {
try {
final PathPatternTree intersectedTree =
pattern.getIntersection(
@@ -244,7 +244,7 @@ public class PipeConfigPhysicalPlanPatternParseVisitor
@Override
public Optional<ConfigPhysicalPlan> visitPipeDeactivateTemplate(
- final PipeDeactivateTemplatePlan pipeDeactivateTemplatePlan, final IoTDBPipePattern pattern) {
+ final PipeDeactivateTemplatePlan pipeDeactivateTemplatePlan, final IoTDBTreePattern pattern) {
final Map<PartialPath, List<Template>> newTemplateSetInfo =
pipeDeactivateTemplatePlan.getTemplateSetInfo().entrySet().stream()
.flatMap(
@@ -267,7 +267,7 @@ public class PipeConfigPhysicalPlanPatternParseVisitor
@Override
public Optional<ConfigPhysicalPlan> visitTTL(
- final SetTTLPlan setTTLPlan, final IoTDBPipePattern pattern) {
+ final SetTTLPlan setTTLPlan, final IoTDBTreePattern pattern) {
final PartialPath partialPath = new PartialPath(setTTLPlan.getPathPattern());
final List<PartialPath> intersectionList =
pattern.matchPrefixPath(partialPath.getFullPath())
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/receiver/protocol/IoTDBConfigNodeReceiver.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/receiver/protocol/IoTDBConfigNodeReceiver.java
index 8466f5fb16a..43f17bcfb6d 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/receiver/protocol/IoTDBConfigNodeReceiver.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/receiver/protocol/IoTDBConfigNodeReceiver.java
@@ -26,7 +26,7 @@ import org.apache.iotdb.commons.pipe.connector.payload.thrift.request.PipeReques
import org.apache.iotdb.commons.pipe.connector.payload.thrift.request.PipeTransferCompressedReq;
import org.apache.iotdb.commons.pipe.connector.payload.thrift.request.PipeTransferFileSealReqV1;
import org.apache.iotdb.commons.pipe.connector.payload.thrift.request.PipeTransferFileSealReqV2;
-import org.apache.iotdb.commons.pipe.datastructure.pattern.IoTDBPipePattern;
+import org.apache.iotdb.commons.pipe.datastructure.pattern.IoTDBTreePattern;
import org.apache.iotdb.commons.pipe.receiver.IoTDBFileReceiver;
import org.apache.iotdb.commons.pipe.receiver.PipeReceiverStatusHandler;
import org.apache.iotdb.commons.schema.ttl.TTLCache;
@@ -354,8 +354,8 @@ public class IoTDBConfigNodeReceiver extends IoTDBFileReceiver {
final Set<ConfigPhysicalPlanType> executionTypes =
PipeConfigRegionSnapshotEvent.getConfigPhysicalPlanTypeSet(
parameters.get(ColumnHeaderConstant.TYPE));
- final IoTDBPipePattern pattern =
- new IoTDBPipePattern(parameters.get(ColumnHeaderConstant.PATH_PATTERN));
+ final IoTDBTreePattern pattern =
+ new IoTDBTreePattern(parameters.get(ColumnHeaderConstant.PATH_PATTERN));
final List<TSStatus> results = new ArrayList<>();
while (generator.hasNext()) {
IoTDBConfigRegionExtractor.PATTERN_PARSE_VISITOR
diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/pipe/extractor/PipeConfigPhysicalPlanPatternParseVisitorTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/pipe/extractor/PipeConfigPhysicalPlanPatternParseVisitorTest.java
index 16bf1e344ff..2114f7b521f 100644
--- a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/pipe/extractor/PipeConfigPhysicalPlanPatternParseVisitorTest.java
+++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/pipe/extractor/PipeConfigPhysicalPlanPatternParseVisitorTest.java
@@ -22,7 +22,7 @@ package org.apache.iotdb.confignode.manager.pipe.extractor;
import org.apache.iotdb.commons.exception.IllegalPathException;
import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.commons.path.PathPatternTree;
-import org.apache.iotdb.commons.pipe.datastructure.pattern.IoTDBPipePattern;
+import org.apache.iotdb.commons.pipe.datastructure.pattern.IoTDBTreePattern;
import org.apache.iotdb.confignode.consensus.request.ConfigPhysicalPlanType;
import org.apache.iotdb.confignode.consensus.request.write.auth.AuthorPlan;
import org.apache.iotdb.confignode.consensus.request.write.database.DatabaseSchemaPlan;
@@ -54,8 +54,8 @@ import java.util.List;
public class PipeConfigPhysicalPlanPatternParseVisitorTest {
- private final IoTDBPipePattern prefixPathPattern = new IoTDBPipePattern("root.db.device.**");
- private final IoTDBPipePattern fullPathPattern = new IoTDBPipePattern("root.db.device.s1");
+ private final IoTDBTreePattern prefixPathPattern = new IoTDBTreePattern("root.db.device.**");
+ private final IoTDBTreePattern fullPathPattern = new IoTDBTreePattern("root.db.device.s1");
@Test
public void testCreateDatabase() {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/connection/PipeEventCollector.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/connection/PipeEventCollector.java
index 7b78febf428..5ff7a87bc2c 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/connection/PipeEventCollector.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/connection/PipeEventCollector.java
@@ -21,7 +21,7 @@ package org.apache.iotdb.db.pipe.agent.task.connection;
import org.apache.iotdb.commons.pipe.agent.task.connection.UnboundedBlockingPendingQueue;
import org.apache.iotdb.commons.pipe.agent.task.progress.PipeEventCommitManager;
-import org.apache.iotdb.commons.pipe.datastructure.pattern.IoTDBPipePattern;
+import org.apache.iotdb.commons.pipe.datastructure.pattern.IoTDBTreePattern;
import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
import org.apache.iotdb.commons.pipe.event.ProgressReportEvent;
import org.apache.iotdb.db.pipe.agent.PipeDataNodeAgent;
@@ -145,7 +145,7 @@ public class PipeEventCollector implements EventCollector {
// Only used by events containing delete data node, no need to bind progress index here since
// delete data event does not have progress index currently
IoTDBSchemaRegionExtractor.PATTERN_PARSE_VISITOR
- .process(deleteDataEvent.getPlanNode(), (IoTDBPipePattern) deleteDataEvent.getPipePattern())
+ .process(deleteDataEvent.getPlanNode(), (IoTDBTreePattern) deleteDataEvent.getTreePattern())
.map(
planNode ->
new PipeSchemaRegionWritePlanEvent(
@@ -153,7 +153,8 @@ public class PipeEventCollector implements EventCollector {
deleteDataEvent.getPipeName(),
deleteDataEvent.getCreationTime(),
deleteDataEvent.getPipeTaskMeta(),
- deleteDataEvent.getPipePattern(),
+ deleteDataEvent.getTreePattern(),
+ deleteDataEvent.getTablePattern(),
deleteDataEvent.isGeneratedByPipe()))
.ifPresent(
event -> {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/airgap/IoTDBSchemaRegionAirGapConnector.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/airgap/IoTDBSchemaRegionAirGapConnector.java
index 342807a5dfb..49eced7dee8 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/airgap/IoTDBSchemaRegionAirGapConnector.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/airgap/IoTDBSchemaRegionAirGapConnector.java
@@ -119,7 +119,7 @@ public class IoTDBSchemaRegionAirGapConnector extends IoTDBDataNodeAirGapConnect
socket,
PipeTransferSchemaSnapshotSealReq.toTPipeTransferBytes(
// The pattern is surely Non-null
- pipeSchemaRegionSnapshotEvent.getPatternString(),
+ pipeSchemaRegionSnapshotEvent.getTreePatternString(),
mtreeSnapshotFile.getName(),
mtreeSnapshotFile.length(),
Objects.nonNull(tagLogSnapshotFile) ? tagLogSnapshotFile.getName() : null,
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/sync/IoTDBSchemaRegionConnector.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/sync/IoTDBSchemaRegionConnector.java
index f70e18c0651..776215e3b57 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/sync/IoTDBSchemaRegionConnector.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/sync/IoTDBSchemaRegionConnector.java
@@ -115,7 +115,7 @@ public class IoTDBSchemaRegionConnector extends IoTDBDataNodeSyncConnector {
compressIfNeeded(
PipeTransferSchemaSnapshotSealReq.toTPipeTransferReq(
// The pattern is surely Non-null
- snapshotEvent.getPatternString(),
+ snapshotEvent.getTreePatternString(),
mTreeSnapshotFile.getName(),
mTreeSnapshotFile.length(),
Objects.nonNull(tagLogSnapshotFile) ? tagLogSnapshotFile.getName() : null,
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/UserDefinedEnrichedEvent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/UserDefinedEnrichedEvent.java
index 6a9b312e61b..ea299f568d1 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/UserDefinedEnrichedEvent.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/UserDefinedEnrichedEvent.java
@@ -21,7 +21,8 @@ package org.apache.iotdb.db.pipe.event;
import org.apache.iotdb.commons.consensus.index.ProgressIndex;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta;
-import org.apache.iotdb.commons.pipe.datastructure.pattern.PipePattern;
+import org.apache.iotdb.commons.pipe.datastructure.pattern.TablePattern;
+import org.apache.iotdb.commons.pipe.datastructure.pattern.TreePattern;
import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
import org.apache.iotdb.pipe.api.event.Event;
import org.apache.iotdb.pipe.api.event.UserDefinedEvent;
@@ -45,7 +46,8 @@ public class UserDefinedEnrichedEvent extends EnrichedEvent {
enrichedEvent.getPipeName(),
enrichedEvent.getCreationTime(),
enrichedEvent.getPipeTaskMeta(),
- enrichedEvent.getPipePattern(),
+ enrichedEvent.getTreePattern(),
+ enrichedEvent.getTablePattern(),
enrichedEvent.getStartTime(),
enrichedEvent.getEndTime());
this.userDefinedEvent = userDefinedEvent;
@@ -76,11 +78,12 @@ public class UserDefinedEnrichedEvent extends EnrichedEvent {
final String pipeName,
final long creationTime,
final PipeTaskMeta pipeTaskMeta,
- final PipePattern pattern,
+ final TreePattern treePattern,
+ final TablePattern tablePattern,
final long startTime,
final long endTime) {
return enrichedEvent.shallowCopySelfAndBindPipeTaskMetaForProgressReport(
- pipeName, creationTime, pipeTaskMeta, pattern, startTime, endTime);
+ pipeName, creationTime, pipeTaskMeta, treePattern, tablePattern, startTime, endTime);
}
@Override
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/heartbeat/PipeHeartbeatEvent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/heartbeat/PipeHeartbeatEvent.java
index 5a72be19bc8..c99f192d190 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/heartbeat/PipeHeartbeatEvent.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/heartbeat/PipeHeartbeatEvent.java
@@ -23,7 +23,8 @@ import org.apache.iotdb.commons.consensus.index.ProgressIndex;
import org.apache.iotdb.commons.consensus.index.impl.MinimumProgressIndex;
import org.apache.iotdb.commons.pipe.agent.task.connection.UnboundedBlockingPendingQueue;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta;
-import org.apache.iotdb.commons.pipe.datastructure.pattern.PipePattern;
+import org.apache.iotdb.commons.pipe.datastructure.pattern.TablePattern;
+import org.apache.iotdb.commons.pipe.datastructure.pattern.TreePattern;
import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
import org.apache.iotdb.db.pipe.metric.PipeHeartbeatEventMetrics;
import org.apache.iotdb.db.utils.DateTimeUtils;
@@ -59,7 +60,7 @@ public class PipeHeartbeatEvent extends EnrichedEvent {
private final boolean shouldPrintMessage;
public PipeHeartbeatEvent(final String dataRegionId, final boolean shouldPrintMessage) {
- super(null, 0, null, null, Long.MIN_VALUE, Long.MAX_VALUE);
+ super(null, 0, null, null, null, Long.MIN_VALUE, Long.MAX_VALUE);
this.dataRegionId = dataRegionId;
this.shouldPrintMessage = shouldPrintMessage;
}
@@ -71,7 +72,7 @@ public class PipeHeartbeatEvent extends EnrichedEvent {
final String dataRegionId,
final long timePublished,
final boolean shouldPrintMessage) {
- super(pipeName, creationTime, pipeTaskMeta, null, Long.MIN_VALUE, Long.MAX_VALUE);
+ super(pipeName, creationTime, pipeTaskMeta, null, null, Long.MIN_VALUE, Long.MAX_VALUE);
this.dataRegionId = dataRegionId;
this.timePublished = timePublished;
this.shouldPrintMessage = shouldPrintMessage;
@@ -102,7 +103,8 @@ public class PipeHeartbeatEvent extends EnrichedEvent {
final String pipeName,
final long creationTime,
final PipeTaskMeta pipeTaskMeta,
- final PipePattern pattern,
+ final TreePattern treePattern,
+ final TablePattern tablePattern,
final long startTime,
final long endTime) {
// Should record PipeTaskMeta, for sometimes HeartbeatEvents should report exceptions.
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/schema/PipeSchemaRegionSnapshotEvent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/schema/PipeSchemaRegionSnapshotEvent.java
index 0d3b9540f28..39f6aa30403 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/schema/PipeSchemaRegionSnapshotEvent.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/schema/PipeSchemaRegionSnapshotEvent.java
@@ -20,7 +20,8 @@
package org.apache.iotdb.db.pipe.event.common.schema;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta;
-import org.apache.iotdb.commons.pipe.datastructure.pattern.PipePattern;
+import org.apache.iotdb.commons.pipe.datastructure.pattern.TablePattern;
+import org.apache.iotdb.commons.pipe.datastructure.pattern.TreePattern;
import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
import org.apache.iotdb.commons.pipe.event.PipeSnapshotEvent;
import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager;
@@ -78,7 +79,7 @@ public class PipeSchemaRegionSnapshotEvent extends PipeSnapshotEvent {
public PipeSchemaRegionSnapshotEvent(
final String mTreeSnapshotPath, final String tagLogSnapshotPath, final String databaseName) {
- this(mTreeSnapshotPath, tagLogSnapshotPath, databaseName, null, 0, null, null);
+ this(mTreeSnapshotPath, tagLogSnapshotPath, databaseName, null, 0, null, null, null);
}
public PipeSchemaRegionSnapshotEvent(
@@ -88,8 +89,15 @@ public class PipeSchemaRegionSnapshotEvent extends PipeSnapshotEvent {
final String pipeName,
final long creationTime,
final PipeTaskMeta pipeTaskMeta,
- final PipePattern pattern) {
- super(pipeName, creationTime, pipeTaskMeta, pattern, PipeDataNodeResourceManager.snapshot());
+ final TreePattern treePattern,
+ final TablePattern tablePattern) {
+ super(
+ pipeName,
+ creationTime,
+ pipeTaskMeta,
+ treePattern,
+ tablePattern,
+ PipeDataNodeResourceManager.snapshot());
this.mTreeSnapshotPath = mTreeSnapshotPath;
this.tagLogSnapshotPath = Objects.nonNull(tagLogSnapshotPath) ? tagLogSnapshotPath : "";
this.databaseName = databaseName;
@@ -148,7 +156,8 @@ public class PipeSchemaRegionSnapshotEvent extends PipeSnapshotEvent {
final String pipeName,
final long creationTime,
final PipeTaskMeta pipeTaskMeta,
- final PipePattern pattern,
+ final TreePattern treePattern,
+ final TablePattern tablePattern,
final long startTime,
final long endTime) {
return new PipeSchemaRegionSnapshotEvent(
@@ -158,7 +167,8 @@ public class PipeSchemaRegionSnapshotEvent extends PipeSnapshotEvent {
pipeName,
creationTime,
pipeTaskMeta,
- pattern);
+ treePattern,
+ tablePattern);
}
@Override
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/schema/PipeSchemaRegionWritePlanEvent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/schema/PipeSchemaRegionWritePlanEvent.java
index 435b38d61c6..3499291a2f6 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/schema/PipeSchemaRegionWritePlanEvent.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/schema/PipeSchemaRegionWritePlanEvent.java
@@ -20,7 +20,8 @@
package org.apache.iotdb.db.pipe.event.common.schema;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta;
-import org.apache.iotdb.commons.pipe.datastructure.pattern.PipePattern;
+import org.apache.iotdb.commons.pipe.datastructure.pattern.TablePattern;
+import org.apache.iotdb.commons.pipe.datastructure.pattern.TreePattern;
import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
import org.apache.iotdb.commons.pipe.event.PipeWritePlanEvent;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNode;
@@ -40,7 +41,7 @@ public class PipeSchemaRegionWritePlanEvent extends PipeWritePlanEvent {
}
public PipeSchemaRegionWritePlanEvent(final PlanNode planNode, final boolean isGeneratedByPipe) {
- this(planNode, null, 0, null, null, isGeneratedByPipe);
+ this(planNode, null, 0, null, null, null, isGeneratedByPipe);
}
public PipeSchemaRegionWritePlanEvent(
@@ -48,9 +49,10 @@ public class PipeSchemaRegionWritePlanEvent extends PipeWritePlanEvent {
final String pipeName,
final long creationTime,
final PipeTaskMeta pipeTaskMeta,
- final PipePattern pattern,
+ final TreePattern treePattern,
+ final TablePattern tablePattern,
final boolean isGeneratedByPipe) {
- super(pipeName, creationTime, pipeTaskMeta, pattern, isGeneratedByPipe);
+ super(pipeName, creationTime, pipeTaskMeta, treePattern, tablePattern, isGeneratedByPipe);
this.planNode = planNode;
}
@@ -63,11 +65,18 @@ public class PipeSchemaRegionWritePlanEvent extends PipeWritePlanEvent {
final String pipeName,
final long creationTime,
final PipeTaskMeta pipeTaskMeta,
- final PipePattern pattern,
+ final TreePattern treePattern,
+ final TablePattern tablePattern,
final long startTime,
final long endTime) {
return new PipeSchemaRegionWritePlanEvent(
- planNode, pipeName, creationTime, pipeTaskMeta, pattern, isGeneratedByPipe);
+ planNode,
+ pipeName,
+ creationTime,
+ pipeTaskMeta,
+ treePattern,
+ tablePattern,
+ isGeneratedByPipe);
}
@Override
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeInsertNodeTabletInsertionEvent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeInsertNodeTabletInsertionEvent.java
index b72665a78a6..9f719642da3 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeInsertNodeTabletInsertionEvent.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeInsertNodeTabletInsertionEvent.java
@@ -23,7 +23,8 @@ import org.apache.iotdb.commons.consensus.index.ProgressIndex;
import org.apache.iotdb.commons.consensus.index.impl.MinimumProgressIndex;
import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta;
-import org.apache.iotdb.commons.pipe.datastructure.pattern.PipePattern;
+import org.apache.iotdb.commons.pipe.datastructure.pattern.TablePattern;
+import org.apache.iotdb.commons.pipe.datastructure.pattern.TreePattern;
import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertNode;
@@ -82,6 +83,7 @@ public class PipeInsertNodeTabletInsertionEvent extends EnrichedEvent
0,
null,
null,
+ null,
Long.MIN_VALUE,
Long.MAX_VALUE);
}
@@ -95,10 +97,11 @@ public class PipeInsertNodeTabletInsertionEvent extends EnrichedEvent
final String pipeName,
final long creationTime,
final PipeTaskMeta pipeTaskMeta,
- final PipePattern pattern,
+ final TreePattern treePattern,
+ final TablePattern tablePattern,
final long startTime,
final long endTime) {
- super(pipeName, creationTime, pipeTaskMeta, pattern, startTime, endTime);
+ super(pipeName, creationTime, pipeTaskMeta, treePattern, tablePattern, startTime, endTime);
this.walEntryHandler = walEntryHandler;
// Record device path here so there's no need to get it from InsertNode cache later.
this.devicePath = devicePath;
@@ -179,7 +182,8 @@ public class PipeInsertNodeTabletInsertionEvent extends EnrichedEvent
final String pipeName,
final long creationTime,
final PipeTaskMeta pipeTaskMeta,
- final PipePattern pattern,
+ final TreePattern treePattern,
+ final TablePattern tablePattern,
final long startTime,
final long endTime) {
return new PipeInsertNodeTabletInsertionEvent(
@@ -191,7 +195,8 @@ public class PipeInsertNodeTabletInsertionEvent extends EnrichedEvent
pipeName,
creationTime,
pipeTaskMeta,
- pattern,
+ treePattern,
+ tablePattern,
startTime,
endTime);
}
@@ -256,7 +261,7 @@ public class PipeInsertNodeTabletInsertionEvent extends EnrichedEvent
if (insertNode instanceof InsertRowNode || insertNode instanceof InsertTabletNode) {
final PartialPath devicePartialPath = insertNode.getTargetPath();
return Objects.isNull(devicePartialPath)
- || pipePattern.mayOverlapWithDevice(devicePartialPath.getIDeviceIDAsFullDevice());
+ || treePattern.mayOverlapWithDevice(devicePartialPath.getIDeviceIDAsFullDevice());
}
if (insertNode instanceof InsertRowsNode) {
@@ -265,7 +270,7 @@ public class PipeInsertNodeTabletInsertionEvent extends EnrichedEvent
.anyMatch(
insertRowNode ->
Objects.isNull(insertRowNode.getTargetPath())
- || pipePattern.mayOverlapWithDevice(
+ || treePattern.mayOverlapWithDevice(
insertRowNode.getTargetPath().getIDeviceIDAsFullDevice()));
}
@@ -328,12 +333,12 @@ public class PipeInsertNodeTabletInsertionEvent extends EnrichedEvent
case INSERT_TABLET:
case RELATIONAL_INSERT_TABLET:
dataContainers.add(
- new TabletInsertionDataContainer(pipeTaskMeta, this, node, pipePattern));
+ new TabletInsertionDataContainer(pipeTaskMeta, this, node, treePattern));
break;
case INSERT_ROWS:
for (final InsertRowNode insertRowNode : ((InsertRowsNode) node).getInsertRowNodeList()) {
dataContainers.add(
- new TabletInsertionDataContainer(pipeTaskMeta, this, insertRowNode, pipePattern));
+ new TabletInsertionDataContainer(pipeTaskMeta, this, insertRowNode, treePattern));
}
break;
default:
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeRawTabletInsertionEvent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeRawTabletInsertionEvent.java
index 1b79b9f27d5..013ccb2579e 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeRawTabletInsertionEvent.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeRawTabletInsertionEvent.java
@@ -22,7 +22,8 @@ package org.apache.iotdb.db.pipe.event.common.tablet;
import org.apache.iotdb.commons.consensus.index.ProgressIndex;
import org.apache.iotdb.commons.consensus.index.impl.MinimumProgressIndex;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta;
-import org.apache.iotdb.commons.pipe.datastructure.pattern.PipePattern;
+import org.apache.iotdb.commons.pipe.datastructure.pattern.TablePattern;
+import org.apache.iotdb.commons.pipe.datastructure.pattern.TreePattern;
import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
import org.apache.iotdb.commons.utils.TestOnly;
import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent;
@@ -62,10 +63,11 @@ public class PipeRawTabletInsertionEvent extends EnrichedEvent implements Tablet
final String pipeName,
final long creationTime,
final PipeTaskMeta pipeTaskMeta,
- final PipePattern pattern,
+ final TreePattern treePattern,
+ final TablePattern tablePattern,
final long startTime,
final long endTime) {
- super(pipeName, creationTime, pipeTaskMeta, pattern, startTime, endTime);
+ super(pipeName, creationTime, pipeTaskMeta, treePattern, tablePattern, startTime, endTime);
this.tablet = Objects.requireNonNull(tablet);
this.isAligned = isAligned;
this.sourceEvent = sourceEvent;
@@ -89,25 +91,37 @@ public class PipeRawTabletInsertionEvent extends EnrichedEvent implements Tablet
creationTime,
pipeTaskMeta,
null,
+ null,
Long.MIN_VALUE,
Long.MAX_VALUE);
}
@TestOnly
public PipeRawTabletInsertionEvent(final Tablet tablet, final boolean isAligned) {
- this(tablet, isAligned, null, false, null, 0, null, null, Long.MIN_VALUE, Long.MAX_VALUE);
+ this(tablet, isAligned, null, false, null, 0, null, null, null, Long.MIN_VALUE, Long.MAX_VALUE);
}
@TestOnly
public PipeRawTabletInsertionEvent(
- final Tablet tablet, final boolean isAligned, final PipePattern pattern) {
- this(tablet, isAligned, null, false, null, 0, null, pattern, Long.MIN_VALUE, Long.MAX_VALUE);
+ final Tablet tablet, final boolean isAligned, final TreePattern treePattern) {
+ this(
+ tablet,
+ isAligned,
+ null,
+ false,
+ null,
+ 0,
+ null,
+ treePattern,
+ null,
+ Long.MIN_VALUE,
+ Long.MAX_VALUE);
}
@TestOnly
public PipeRawTabletInsertionEvent(
final Tablet tablet, final long startTime, final long endTime) {
- this(tablet, false, null, false, null, 0, null, null, startTime, endTime);
+ this(tablet, false, null, false, null, 0, null, null, null, startTime, endTime);
}
@Override
@@ -169,7 +183,8 @@ public class PipeRawTabletInsertionEvent extends EnrichedEvent implements Tablet
final String pipeName,
final long creationTime,
final PipeTaskMeta pipeTaskMeta,
- final PipePattern pattern,
+ final TreePattern treePattern,
+ final TablePattern tablePattern,
final long startTime,
final long endTime) {
return new PipeRawTabletInsertionEvent(
@@ -180,7 +195,8 @@ public class PipeRawTabletInsertionEvent extends EnrichedEvent implements Tablet
pipeName,
creationTime,
pipeTaskMeta,
- pattern,
+ treePattern,
+ tablePattern,
startTime,
endTime);
}
@@ -204,7 +220,7 @@ public class PipeRawTabletInsertionEvent extends EnrichedEvent implements Tablet
public boolean mayEventPathsOverlappedWithPattern() {
final String deviceId = getDeviceId();
return Objects.isNull(deviceId)
- || pipePattern.mayOverlapWithDevice(IDeviceID.Factory.DEFAULT_FACTORY.create(deviceId));
+ || treePattern.mayOverlapWithDevice(IDeviceID.Factory.DEFAULT_FACTORY.create(deviceId));
}
public void markAsNeedToReport() {
@@ -227,7 +243,7 @@ public class PipeRawTabletInsertionEvent extends EnrichedEvent implements Tablet
final BiConsumer<Row, RowCollector> consumer) {
if (dataContainer == null) {
dataContainer =
- new TabletInsertionDataContainer(pipeTaskMeta, this, tablet, isAligned, pipePattern);
+ new TabletInsertionDataContainer(pipeTaskMeta, this, tablet, isAligned, treePattern);
}
return dataContainer.processRowByRow(consumer);
}
@@ -237,7 +253,7 @@ public class PipeRawTabletInsertionEvent extends EnrichedEvent implements Tablet
final BiConsumer<Tablet, RowCollector> consumer) {
if (dataContainer == null) {
dataContainer =
- new TabletInsertionDataContainer(pipeTaskMeta, this, tablet, isAligned, pipePattern);
+ new TabletInsertionDataContainer(pipeTaskMeta, this, tablet, isAligned, treePattern);
}
return dataContainer.processTablet(consumer);
}
@@ -256,7 +272,7 @@ public class PipeRawTabletInsertionEvent extends EnrichedEvent implements Tablet
// if notNullPattern is not "root", we need to convert the tablet
if (dataContainer == null) {
dataContainer =
- new TabletInsertionDataContainer(pipeTaskMeta, this, tablet, isAligned, pipePattern);
+ new TabletInsertionDataContainer(pipeTaskMeta, this, tablet, isAligned, treePattern);
}
return dataContainer.convertToTablet();
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/TabletInsertionDataContainer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/TabletInsertionDataContainer.java
index 09332c68ecd..ce1e4113b91 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/TabletInsertionDataContainer.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/TabletInsertionDataContainer.java
@@ -20,7 +20,7 @@
package org.apache.iotdb.db.pipe.event.common.tablet;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta;
-import org.apache.iotdb.commons.pipe.datastructure.pattern.PipePattern;
+import org.apache.iotdb.commons.pipe.datastructure.pattern.TreePattern;
import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
import org.apache.iotdb.commons.utils.TestOnly;
import org.apache.iotdb.db.pipe.event.common.row.PipeRow;
@@ -99,7 +99,7 @@ public class TabletInsertionDataContainer {
final PipeTaskMeta pipeTaskMeta,
final EnrichedEvent sourceEvent,
final InsertNode insertNode,
- final PipePattern pattern) {
+ final TreePattern pattern) {
this.pipeTaskMeta = pipeTaskMeta;
this.sourceEvent = sourceEvent;
@@ -118,7 +118,7 @@ public class TabletInsertionDataContainer {
final EnrichedEvent sourceEvent,
final Tablet tablet,
final boolean isAligned,
- final PipePattern pattern) {
+ final TreePattern pattern) {
this.pipeTaskMeta = pipeTaskMeta;
this.sourceEvent = sourceEvent;
@@ -126,7 +126,7 @@ public class TabletInsertionDataContainer {
}
@TestOnly
- public TabletInsertionDataContainer(final InsertNode insertNode, final PipePattern pattern) {
+ public TabletInsertionDataContainer(final InsertNode insertNode, final TreePattern pattern) {
this(null, null, insertNode, pattern);
}
@@ -140,7 +140,7 @@ public class TabletInsertionDataContainer {
//////////////////////////// parse ////////////////////////////
- private void parse(final InsertRowNode insertRowNode, final PipePattern pattern) {
+ private void parse(final InsertRowNode insertRowNode, final TreePattern pattern) {
final int originColumnSize = insertRowNode.getMeasurements().length;
final Integer[] originColumnIndex2FilteredColumnIndexMapperList = new Integer[originColumnSize];
@@ -209,7 +209,7 @@ public class TabletInsertionDataContainer {
}
}
- private void parse(final InsertTabletNode insertTabletNode, final PipePattern pattern) {
+ private void parse(final InsertTabletNode insertTabletNode, final TreePattern pattern) {
final int originColumnSize = insertTabletNode.getMeasurements().length;
final Integer[] originColumnIndex2FilteredColumnIndexMapperList = new Integer[originColumnSize];
@@ -294,7 +294,7 @@ public class TabletInsertionDataContainer {
}
}
- private void parse(final Tablet tablet, final boolean isAligned, final PipePattern pattern) {
+ private void parse(final Tablet tablet, final boolean isAligned, final TreePattern pattern) {
final int originColumnSize = tablet.getSchemas().size();
final Integer[] originColumnIndex2FilteredColumnIndexMapperList = new Integer[originColumnSize];
@@ -389,7 +389,7 @@ public class TabletInsertionDataContainer {
private void generateColumnIndexMapper(
final String[] originMeasurementList,
- final PipePattern pattern,
+ final TreePattern pattern,
final Integer[] originColumnIndex2FilteredColumnIndexMapperList) {
final int originColumnSize = originMeasurementList.length;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/terminate/PipeTerminateEvent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/terminate/PipeTerminateEvent.java
index dbe8aada88d..78c95a37f12 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/terminate/PipeTerminateEvent.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/terminate/PipeTerminateEvent.java
@@ -22,7 +22,8 @@ package org.apache.iotdb.db.pipe.event.common.terminate;
import org.apache.iotdb.commons.consensus.index.ProgressIndex;
import org.apache.iotdb.commons.consensus.index.impl.MinimumProgressIndex;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta;
-import org.apache.iotdb.commons.pipe.datastructure.pattern.PipePattern;
+import org.apache.iotdb.commons.pipe.datastructure.pattern.TablePattern;
+import org.apache.iotdb.commons.pipe.datastructure.pattern.TreePattern;
import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
import org.apache.iotdb.db.pipe.agent.PipeDataNodeAgent;
import org.apache.iotdb.db.pipe.agent.task.PipeDataNodeTask;
@@ -42,7 +43,7 @@ public class PipeTerminateEvent extends EnrichedEvent {
final long creationTime,
final PipeTaskMeta pipeTaskMeta,
final int dataRegionId) {
- super(pipeName, creationTime, pipeTaskMeta, null, Long.MIN_VALUE, Long.MAX_VALUE);
+ super(pipeName, creationTime, pipeTaskMeta, null, null, Long.MIN_VALUE, Long.MAX_VALUE);
this.dataRegionId = dataRegionId;
}
@@ -66,7 +67,8 @@ public class PipeTerminateEvent extends EnrichedEvent {
final String pipeName,
final long creationTime,
final PipeTaskMeta pipeTaskMeta,
- final PipePattern pattern,
+ final TreePattern treePattern,
+ final TablePattern tablePattern,
final long startTime,
final long endTime) {
// Should record PipeTaskMeta, for the terminateEvent shall report progress to
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java
index d026a65e0a8..c78ea9703e4 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java
@@ -22,7 +22,8 @@ package org.apache.iotdb.db.pipe.event.common.tsfile;
import org.apache.iotdb.commons.consensus.index.ProgressIndex;
import org.apache.iotdb.commons.consensus.index.impl.MinimumProgressIndex;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta;
-import org.apache.iotdb.commons.pipe.datastructure.pattern.PipePattern;
+import org.apache.iotdb.commons.pipe.datastructure.pattern.TablePattern;
+import org.apache.iotdb.commons.pipe.datastructure.pattern.TreePattern;
import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
import org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent;
import org.apache.iotdb.db.pipe.event.common.tsfile.container.TsFileInsertionDataContainer;
@@ -90,6 +91,7 @@ public class PipeTsFileInsertionEvent extends EnrichedEvent implements TsFileIns
0,
null,
null,
+ null,
Long.MIN_VALUE,
Long.MAX_VALUE);
}
@@ -103,10 +105,11 @@ public class PipeTsFileInsertionEvent extends EnrichedEvent implements TsFileIns
final String pipeName,
final long creationTime,
final PipeTaskMeta pipeTaskMeta,
- final PipePattern pattern,
+ final TreePattern treePattern,
+ final TablePattern tablePattern,
final long startTime,
final long endTime) {
- super(pipeName, creationTime, pipeTaskMeta, pattern, startTime, endTime);
+ super(pipeName, creationTime, pipeTaskMeta, treePattern, tablePattern, startTime, endTime);
this.resource = resource;
tsFile = resource.getTsFile();
@@ -320,7 +323,8 @@ public class PipeTsFileInsertionEvent extends EnrichedEvent implements TsFileIns
final String pipeName,
final long creationTime,
final PipeTaskMeta pipeTaskMeta,
- final PipePattern pattern,
+ final TreePattern treePattern,
+ final TablePattern tablePattern,
final long startTime,
final long endTime) {
return new PipeTsFileInsertionEvent(
@@ -332,7 +336,8 @@ public class PipeTsFileInsertionEvent extends EnrichedEvent implements TsFileIns
pipeName,
creationTime,
pipeTaskMeta,
- pattern,
+ treePattern,
+ tablePattern,
startTime,
endTime);
}
@@ -365,7 +370,7 @@ public class PipeTsFileInsertionEvent extends EnrichedEvent implements TsFileIns
false);
final Set<IDeviceID> deviceSet =
Objects.nonNull(deviceIsAlignedMap) ? deviceIsAlignedMap.keySet() : resource.getDevices();
- return deviceSet.stream().anyMatch(pipePattern::mayOverlapWithDevice);
+ return deviceSet.stream().anyMatch(treePattern::mayOverlapWithDevice);
} catch (final Exception e) {
LOGGER.warn(
"Pipe {}: failed to get devices from TsFile {}, extract it anyway",
@@ -413,7 +418,7 @@ public class PipeTsFileInsertionEvent extends EnrichedEvent implements TsFileIns
if (dataContainer == null) {
dataContainer =
new TsFileInsertionDataContainerProvider(
- tsFile, pipePattern, startTime, endTime, pipeTaskMeta, this)
+ tsFile, treePattern, startTime, endTime, pipeTaskMeta, this)
.provide();
}
return dataContainer;
@@ -445,7 +450,7 @@ public class PipeTsFileInsertionEvent extends EnrichedEvent implements TsFileIns
}
try (final TsFileInsertionPointCounter counter =
- new TsFileInsertionPointCounter(tsFile, pipePattern)) {
+ new TsFileInsertionPointCounter(tsFile, treePattern)) {
return counter.count();
}
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/TsFileInsertionPointCounter.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/TsFileInsertionPointCounter.java
index 342b7b640ac..e1f2905cae1 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/TsFileInsertionPointCounter.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/TsFileInsertionPointCounter.java
@@ -19,7 +19,7 @@
package org.apache.iotdb.db.pipe.event.common.tsfile;
-import org.apache.iotdb.commons.pipe.datastructure.pattern.PipePattern;
+import org.apache.iotdb.commons.pipe.datastructure.pattern.TreePattern;
import org.apache.tsfile.file.metadata.IDeviceID;
import org.apache.tsfile.file.metadata.TimeseriesMetadata;
@@ -40,7 +40,7 @@ public class TsFileInsertionPointCounter implements AutoCloseable {
private static final Logger LOGGER = LoggerFactory.getLogger(TsFileInsertionPointCounter.class);
- private final PipePattern pattern;
+ private final TreePattern pattern;
private final TsFileSequenceReader tsFileSequenceReader;
@@ -50,7 +50,7 @@ public class TsFileInsertionPointCounter implements AutoCloseable {
private long count = 0;
- public TsFileInsertionPointCounter(final File tsFile, final PipePattern pattern)
+ public TsFileInsertionPointCounter(final File tsFile, final TreePattern pattern)
throws IOException {
this.pattern = pattern;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/container/TsFileInsertionDataContainer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/container/TsFileInsertionDataContainer.java
index 66a6ebb0d25..9c1e435f68f 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/container/TsFileInsertionDataContainer.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/container/TsFileInsertionDataContainer.java
@@ -20,7 +20,7 @@
package org.apache.iotdb.db.pipe.event.common.tsfile.container;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta;
-import org.apache.iotdb.commons.pipe.datastructure.pattern.PipePattern;
+import org.apache.iotdb.commons.pipe.datastructure.pattern.TreePattern;
import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager;
import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryBlock;
@@ -38,7 +38,7 @@ public abstract class TsFileInsertionDataContainer implements AutoCloseable {
private static final Logger LOGGER = LoggerFactory.getLogger(TsFileInsertionDataContainer.class);
- protected final PipePattern pattern; // used to filter data
+ protected final TreePattern pattern; // used to filter data
protected final GlobalTimeExpression timeFilterExpression; // used to filter data
protected final PipeTaskMeta pipeTaskMeta; // used to report progress
@@ -49,7 +49,7 @@ public abstract class TsFileInsertionDataContainer implements AutoCloseable {
protected TsFileSequenceReader tsFileSequenceReader;
protected TsFileInsertionDataContainer(
- final PipePattern pattern,
+ final TreePattern pattern,
final long startTime,
final long endTime,
final PipeTaskMeta pipeTaskMeta,
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/container/TsFileInsertionDataContainerProvider.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/container/TsFileInsertionDataContainerProvider.java
index 64cc9816f36..323593926f9 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/container/TsFileInsertionDataContainerProvider.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/container/TsFileInsertionDataContainerProvider.java
@@ -21,8 +21,8 @@ package org.apache.iotdb.db.pipe.event.common.tsfile.container;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta;
import org.apache.iotdb.commons.pipe.config.PipeConfig;
-import org.apache.iotdb.commons.pipe.datastructure.pattern.IoTDBPipePattern;
-import org.apache.iotdb.commons.pipe.datastructure.pattern.PipePattern;
+import org.apache.iotdb.commons.pipe.datastructure.pattern.IoTDBTreePattern;
+import org.apache.iotdb.commons.pipe.datastructure.pattern.TreePattern;
import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent;
import org.apache.iotdb.db.pipe.event.common.tsfile.container.query.TsFileInsertionQueryDataContainer;
import org.apache.iotdb.db.pipe.event.common.tsfile.container.scan.TsFileInsertionScanDataContainer;
@@ -39,7 +39,7 @@ import java.util.stream.Collectors;
public class TsFileInsertionDataContainerProvider {
private final File tsFile;
- private final PipePattern pattern;
+ private final TreePattern pattern;
private final long startTime;
private final long endTime;
@@ -48,13 +48,13 @@ public class TsFileInsertionDataContainerProvider {
public TsFileInsertionDataContainerProvider(
final File tsFile,
- final PipePattern pipePattern,
+ final TreePattern treePattern,
final long startTime,
final long endTime,
final PipeTaskMeta pipeTaskMeta,
final PipeTsFileInsertionEvent sourceEvent) {
this.tsFile = tsFile;
- this.pattern = pipePattern;
+ this.pattern = treePattern;
this.startTime = startTime;
this.endTime = endTime;
this.pipeTaskMeta = pipeTaskMeta;
@@ -64,8 +64,8 @@ public class TsFileInsertionDataContainerProvider {
public TsFileInsertionDataContainer provide() throws IOException {
if (startTime != Long.MIN_VALUE
|| endTime != Long.MAX_VALUE
- || pattern instanceof IoTDBPipePattern
- && !((IoTDBPipePattern) pattern).mayMatchMultipleTimeSeriesInOneDevice()) {
+ || pattern instanceof IoTDBTreePattern
+ && !((IoTDBTreePattern) pattern).mayMatchMultipleTimeSeriesInOneDevice()) {
// 1. If time filter exists, use query here because the scan container may filter it
// row by row in single page chunk.
// 2. If the pattern matches only one time series in one device, use query container here
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/container/query/TsFileInsertionQueryDataContainer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/container/query/TsFileInsertionQueryDataContainer.java
index fe7eadf8608..d7c81a92e19 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/container/query/TsFileInsertionQueryDataContainer.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/container/query/TsFileInsertionQueryDataContainer.java
@@ -21,7 +21,7 @@ package org.apache.iotdb.db.pipe.event.common.tsfile.container.query;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta;
import org.apache.iotdb.commons.pipe.config.PipeConfig;
-import org.apache.iotdb.commons.pipe.datastructure.pattern.PipePattern;
+import org.apache.iotdb.commons.pipe.datastructure.pattern.TreePattern;
import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
import org.apache.iotdb.commons.utils.TestOnly;
import org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent;
@@ -69,14 +69,14 @@ public class TsFileInsertionQueryDataContainer extends TsFileInsertionDataContai
@TestOnly
public TsFileInsertionQueryDataContainer(
- final File tsFile, final PipePattern pattern, final long startTime, final long endTime)
+ final File tsFile, final TreePattern pattern, final long startTime, final long endTime)
throws IOException {
this(tsFile, pattern, startTime, endTime, null, null);
}
public TsFileInsertionQueryDataContainer(
final File tsFile,
- final PipePattern pattern,
+ final TreePattern pattern,
final long startTime,
final long endTime,
final PipeTaskMeta pipeTaskMeta,
@@ -87,7 +87,7 @@ public class TsFileInsertionQueryDataContainer extends TsFileInsertionDataContai
public TsFileInsertionQueryDataContainer(
final File tsFile,
- final PipePattern pattern,
+ final TreePattern pattern,
final long startTime,
final long endTime,
final PipeTaskMeta pipeTaskMeta,
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/container/scan/TsFileInsertionScanDataContainer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/container/scan/TsFileInsertionScanDataContainer.java
index 8c95d3489c9..5bd0e714413 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/container/scan/TsFileInsertionScanDataContainer.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/container/scan/TsFileInsertionScanDataContainer.java
@@ -20,7 +20,7 @@
package org.apache.iotdb.db.pipe.event.common.tsfile.container.scan;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta;
-import org.apache.iotdb.commons.pipe.datastructure.pattern.PipePattern;
+import org.apache.iotdb.commons.pipe.datastructure.pattern.TreePattern;
import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
import org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent;
import org.apache.iotdb.db.pipe.event.common.tsfile.container.TsFileInsertionDataContainer;
@@ -90,7 +90,7 @@ public class TsFileInsertionScanDataContainer extends TsFileInsertionDataContain
public TsFileInsertionScanDataContainer(
final File tsFile,
- final PipePattern pattern,
+ final TreePattern pattern,
final long startTime,
final long endTime,
final PipeTaskMeta pipeTaskMeta,
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/realtime/PipeRealtimeEvent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/realtime/PipeRealtimeEvent.java
index 150ccaa8c67..b6ce88572a0 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/realtime/PipeRealtimeEvent.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/realtime/PipeRealtimeEvent.java
@@ -21,7 +21,8 @@ package org.apache.iotdb.db.pipe.event.realtime;
import org.apache.iotdb.commons.consensus.index.ProgressIndex;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta;
-import org.apache.iotdb.commons.pipe.datastructure.pattern.PipePattern;
+import org.apache.iotdb.commons.pipe.datastructure.pattern.TablePattern;
+import org.apache.iotdb.commons.pipe.datastructure.pattern.TreePattern;
import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
import org.apache.iotdb.db.pipe.extractor.dataregion.realtime.epoch.TsFileEpoch;
@@ -45,8 +46,17 @@ public class PipeRealtimeEvent extends EnrichedEvent {
final EnrichedEvent event,
final TsFileEpoch tsFileEpoch,
final Map<IDeviceID, String[]> device2Measurements,
- final PipePattern pattern) {
- this(event, tsFileEpoch, device2Measurements, null, pattern, Long.MIN_VALUE, Long.MAX_VALUE);
+ final TreePattern treePattern,
+ final TablePattern tablePattern) {
+ this(
+ event,
+ tsFileEpoch,
+ device2Measurements,
+ null,
+ treePattern,
+ tablePattern,
+ Long.MIN_VALUE,
+ Long.MAX_VALUE);
}
public PipeRealtimeEvent(
@@ -54,7 +64,8 @@ public class PipeRealtimeEvent extends EnrichedEvent {
final TsFileEpoch tsFileEpoch,
final Map<IDeviceID, String[]> device2Measurements,
final PipeTaskMeta pipeTaskMeta,
- final PipePattern pattern,
+ final TreePattern treePattern,
+ final TablePattern tablePattern,
final long startTime,
final long endTime) {
// PipeTaskMeta is used to report the progress of the event, the PipeRealtimeEvent
@@ -64,7 +75,8 @@ public class PipeRealtimeEvent extends EnrichedEvent {
event != null ? event.getPipeName() : null,
event != null ? event.getCreationTime() : 0,
pipeTaskMeta,
- pattern,
+ treePattern,
+ tablePattern,
startTime,
endTime);
@@ -166,16 +178,18 @@ public class PipeRealtimeEvent extends EnrichedEvent {
final String pipeName,
final long creationTime,
final PipeTaskMeta pipeTaskMeta,
- final PipePattern pattern,
+ final TreePattern treePattern,
+ final TablePattern tablePattern,
final long startTime,
final long endTime) {
return new PipeRealtimeEvent(
event.shallowCopySelfAndBindPipeTaskMetaForProgressReport(
- pipeName, creationTime, pipeTaskMeta, pattern, startTime, endTime),
+ pipeName, creationTime, pipeTaskMeta, treePattern, tablePattern, startTime, endTime),
this.tsFileEpoch,
this.device2Measurements,
pipeTaskMeta,
- pattern,
+ treePattern,
+ tablePattern,
startTime,
endTime);
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/realtime/PipeRealtimeEventFactory.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/realtime/PipeRealtimeEventFactory.java
index 3c3023c58ee..3ef134a96cb 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/realtime/PipeRealtimeEventFactory.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/realtime/PipeRealtimeEventFactory.java
@@ -58,16 +58,16 @@ public class PipeRealtimeEventFactory {
public static PipeRealtimeEvent createRealtimeEvent(
final String dataRegionId, final boolean shouldPrintMessage) {
return new PipeRealtimeEvent(
- new PipeHeartbeatEvent(dataRegionId, shouldPrintMessage), null, null, null);
+ new PipeHeartbeatEvent(dataRegionId, shouldPrintMessage), null, null, null, null);
}
public static PipeRealtimeEvent createRealtimeEvent(final DeleteDataNode node) {
return new PipeRealtimeEvent(
- new PipeSchemaRegionWritePlanEvent(node, node.isGeneratedByPipe()), null, null, null);
+ new PipeSchemaRegionWritePlanEvent(node, node.isGeneratedByPipe()), null, null, null, null);
}
public static PipeRealtimeEvent createRealtimeEvent(final ProgressReportEvent event) {
- return new PipeRealtimeEvent(event, null, null, null);
+ return new PipeRealtimeEvent(event, null, null, null, null);
}
private PipeRealtimeEventFactory() {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/IoTDBDataRegionExtractor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/IoTDBDataRegionExtractor.java
index f05bcf2bea5..66d3349d661 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/IoTDBDataRegionExtractor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/IoTDBDataRegionExtractor.java
@@ -21,8 +21,9 @@ package org.apache.iotdb.db.pipe.extractor.dataregion;
import org.apache.iotdb.commons.consensus.DataRegionId;
import org.apache.iotdb.commons.exception.IllegalPathException;
-import org.apache.iotdb.commons.pipe.datastructure.pattern.IoTDBPipePattern;
-import org.apache.iotdb.commons.pipe.datastructure.pattern.PipePattern;
+import org.apache.iotdb.commons.pipe.datastructure.pattern.IoTDBTreePattern;
+import org.apache.iotdb.commons.pipe.datastructure.pattern.TablePattern;
+import org.apache.iotdb.commons.pipe.datastructure.pattern.TreePattern;
import org.apache.iotdb.commons.pipe.extractor.IoTDBExtractor;
import org.apache.iotdb.consensus.ConsensusFactory;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
@@ -139,12 +140,12 @@ public class IoTDBDataRegionExtractor extends IoTDBExtractor {
EXTRACTOR_PATTERN_FORMAT_PREFIX_VALUE,
EXTRACTOR_PATTERN_FORMAT_IOTDB_VALUE);
- // Get the pattern format to check whether the pattern is legal
- final PipePattern pattern =
- PipePattern.parsePipePatternFromSourceParameters(validator.getParameters());
-
- // Check whether the pattern is legal
- validatePattern(pattern);
+ // Validate tree pattern and table pattern
+ final TreePattern treePattern =
+ TreePattern.parsePipePatternFromSourceParameters(validator.getParameters());
+ final TablePattern tablePattern =
+ TablePattern.parsePipePatternFromSourceParameters(validator.getParameters());
+ validatePattern(treePattern, tablePattern);
// Validate extractor.history.enable and extractor.realtime.enable
validator
@@ -218,19 +219,26 @@ public class IoTDBDataRegionExtractor extends IoTDBExtractor {
realtimeExtractor.validate(validator);
}
- private void validatePattern(final PipePattern pattern) {
- if (!pattern.isLegal()) {
- throw new IllegalArgumentException(String.format("Pattern \"%s\" is illegal.", pattern));
+ private void validatePattern(final TreePattern treePattern, final TablePattern tablePattern) {
+ if (!treePattern.isLegal()) {
+ throw new IllegalArgumentException(String.format("Pattern \"%s\" is illegal.", treePattern));
}
if (shouldExtractDeletion
- && !(pattern instanceof IoTDBPipePattern
- && (((IoTDBPipePattern) pattern).isPrefix()
- || ((IoTDBPipePattern) pattern).isFullPath()))) {
+ && !(treePattern instanceof IoTDBTreePattern
+ && (((IoTDBTreePattern) treePattern).isPrefix()
+ || ((IoTDBTreePattern) treePattern).isFullPath()))) {
throw new IllegalArgumentException(
String.format(
"The path pattern %s is not valid for the source. Only prefix or full path is allowed.",
- pattern));
+ treePattern));
+ }
+
+ if (shouldExtractDeletion && tablePattern.hasUserSpecifiedDatabasePatternOrTablePattern()) {
+ throw new IllegalArgumentException(
+ String.format(
+ "The table model pattern %s can not be specified when deletion capture is enabled.",
+ tablePattern));
}
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/historical/PipeHistoricalDataRegionTsFileExtractor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/historical/PipeHistoricalDataRegionTsFileExtractor.java
index 08cccb3aa9f..cfe8b639fa2 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/historical/PipeHistoricalDataRegionTsFileExtractor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/historical/PipeHistoricalDataRegionTsFileExtractor.java
@@ -28,7 +28,8 @@ import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta;
import org.apache.iotdb.commons.pipe.config.constant.PipeExtractorConstant;
import org.apache.iotdb.commons.pipe.config.constant.SystemConstant;
import org.apache.iotdb.commons.pipe.config.plugin.env.PipeTaskExtractorRuntimeEnvironment;
-import org.apache.iotdb.commons.pipe.datastructure.pattern.PipePattern;
+import org.apache.iotdb.commons.pipe.datastructure.pattern.TablePattern;
+import org.apache.iotdb.commons.pipe.datastructure.pattern.TreePattern;
import org.apache.iotdb.db.pipe.event.common.terminate.PipeTerminateEvent;
import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent;
import org.apache.iotdb.db.pipe.extractor.dataregion.DataRegionListeningFilter;
@@ -101,7 +102,8 @@ public class PipeHistoricalDataRegionTsFileExtractor implements PipeHistoricalDa
private int dataRegionId;
- private PipePattern pipePattern;
+ private TreePattern treePattern;
+ private TablePattern tablePattern; // TODO: use like treePattern
private boolean isDbNameCoveredByPattern = false;
private boolean isHistoricalExtractorEnabled = false;
@@ -269,14 +271,18 @@ public class PipeHistoricalDataRegionTsFileExtractor implements PipeHistoricalDa
DATA_REGION_ID_TO_PIPE_FLUSHED_TIME_MAP.putIfAbsent(dataRegionId, 0L);
}
- pipePattern = PipePattern.parsePipePatternFromSourceParameters(parameters);
+ treePattern = TreePattern.parsePipePatternFromSourceParameters(parameters);
+ tablePattern = TablePattern.parsePipePatternFromSourceParameters(parameters);
final DataRegion dataRegion =
StorageEngine.getInstance().getDataRegion(new DataRegionId(environment.getRegionId()));
if (Objects.nonNull(dataRegion)) {
final String databaseName = dataRegion.getDatabaseName();
if (Objects.nonNull(databaseName)) {
- isDbNameCoveredByPattern = pipePattern.coversDb(databaseName);
+ isDbNameCoveredByPattern =
+ treePattern.coversDb(databaseName)
+ && tablePattern.matchesDatabase(
+ databaseName.startsWith("root.") ? databaseName.substring(5) : databaseName);
}
}
@@ -541,7 +547,8 @@ public class PipeHistoricalDataRegionTsFileExtractor implements PipeHistoricalDa
return true;
}
- return deviceSet.stream().anyMatch(deviceID -> pipePattern.mayOverlapWithDevice(deviceID));
+ // TODO: consider tablePattern
+ return deviceSet.stream().anyMatch(deviceID -> treePattern.mayOverlapWithDevice(deviceID));
}
private boolean isTsFileResourceOverlappedWithTimeRange(final TsFileResource resource) {
@@ -609,7 +616,8 @@ public class PipeHistoricalDataRegionTsFileExtractor implements PipeHistoricalDa
pipeName,
creationTime,
pipeTaskMeta,
- pipePattern,
+ treePattern,
+ tablePattern,
historicalDataExtractionStartTime,
historicalDataExtractionEndTime);
if (sloppyPattern || isDbNameCoveredByPattern) {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/PipeRealtimeDataRegionExtractor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/PipeRealtimeDataRegionExtractor.java
index 40c6bdc906f..666f7435701 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/PipeRealtimeDataRegionExtractor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/PipeRealtimeDataRegionExtractor.java
@@ -25,7 +25,8 @@ import org.apache.iotdb.commons.pipe.agent.task.connection.UnboundedBlockingPend
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta;
import org.apache.iotdb.commons.pipe.config.constant.PipeExtractorConstant;
import org.apache.iotdb.commons.pipe.config.plugin.env.PipeTaskExtractorRuntimeEnvironment;
-import org.apache.iotdb.commons.pipe.datastructure.pattern.PipePattern;
+import org.apache.iotdb.commons.pipe.datastructure.pattern.TablePattern;
+import org.apache.iotdb.commons.pipe.datastructure.pattern.TreePattern;
import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
import org.apache.iotdb.commons.pipe.event.ProgressReportEvent;
import org.apache.iotdb.commons.utils.TimePartitionUtils;
@@ -60,6 +61,7 @@ import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;
+import static com.google.common.base.MoreObjects.toStringHelper;
import static org.apache.iotdb.commons.pipe.config.constant.PipeExtractorConstant.EXTRACTOR_END_TIME_KEY;
import static org.apache.iotdb.commons.pipe.config.constant.PipeExtractorConstant.EXTRACTOR_MODS_ENABLE_DEFAULT_VALUE;
import static org.apache.iotdb.commons.pipe.config.constant.PipeExtractorConstant.EXTRACTOR_MODS_ENABLE_KEY;
@@ -87,7 +89,8 @@ public abstract class PipeRealtimeDataRegionExtractor implements PipeExtractor {
protected boolean shouldExtractInsertion;
protected boolean shouldExtractDeletion;
- protected PipePattern pipePattern;
+ protected TreePattern treePattern;
+ protected TablePattern tablePattern;
private boolean isDbNameCoveredByPattern = false;
protected long realtimeDataExtractionStartTime = Long.MIN_VALUE; // Event time
@@ -206,14 +209,18 @@ public abstract class PipeRealtimeDataRegionExtractor implements PipeExtractor {
creationTime = environment.getCreationTime();
taskID = pipeName + "_" + dataRegionId + "_" + creationTime;
- pipePattern = PipePattern.parsePipePatternFromSourceParameters(parameters);
+ treePattern = TreePattern.parsePipePatternFromSourceParameters(parameters);
+ tablePattern = TablePattern.parsePipePatternFromSourceParameters(parameters);
final DataRegion dataRegion =
StorageEngine.getInstance().getDataRegion(new DataRegionId(environment.getRegionId()));
if (dataRegion != null) {
final String databaseName = dataRegion.getDatabaseName();
if (databaseName != null) {
- isDbNameCoveredByPattern = pipePattern.coversDb(databaseName);
+ isDbNameCoveredByPattern =
+ treePattern.coversDb(databaseName)
+ && tablePattern.matchesDatabase(
+ databaseName.startsWith("root.") ? databaseName.substring(5) : databaseName);
}
}
@@ -450,12 +457,12 @@ public abstract class PipeRealtimeDataRegionExtractor implements PipeExtractor {
return shouldExtractDeletion;
}
- public final String getPatternString() {
- return pipePattern != null ? pipePattern.getPattern() : null;
+ public final TreePattern getTreePattern() {
+ return treePattern;
}
- public final PipePattern getPipePattern() {
- return pipePattern;
+ public final TablePattern getTablePattern() {
+ return tablePattern;
}
public final long getRealtimeDataExtractionStartTime() {
@@ -496,14 +503,32 @@ public abstract class PipeRealtimeDataRegionExtractor implements PipeExtractor {
@Override
public String toString() {
- return "PipeRealtimeDataRegionExtractor{"
- + "pipePattern='"
- + pipePattern
- + '\''
- + ", dataRegionId='"
- + dataRegionId
- + '\''
- + '}';
+ return toStringHelper(this)
+ .add("pipeName", pipeName)
+ .add("creationTime", creationTime)
+ .add("dataRegionId", dataRegionId)
+ .add("pipeTaskMeta", pipeTaskMeta)
+ .add("shouldExtractInsertion", shouldExtractInsertion)
+ .add("shouldExtractDeletion", shouldExtractDeletion)
+ .add("treePattern", treePattern)
+ .add("tablePattern", tablePattern)
+ .add("isDbNameCoveredByPattern", isDbNameCoveredByPattern)
+ .add("realtimeDataExtractionStartTime", realtimeDataExtractionStartTime)
+ .add("realtimeDataExtractionEndTime", realtimeDataExtractionEndTime)
+ .add(
+ "disableCheckingDataRegionTimePartitionCovering",
+ disableCheckingDataRegionTimePartitionCovering)
+ .add("startTimePartitionIdLowerBound", startTimePartitionIdLowerBound)
+ .add("endTimePartitionIdUpperBound", endTimePartitionIdUpperBound)
+ .add("dataRegionTimePartitionIdBound", dataRegionTimePartitionIdBound)
+ .add("isForwardingPipeRequests", isForwardingPipeRequests)
+ .add("shouldTransferModFile", shouldTransferModFile)
+ .add("sloppyTimeRange", sloppyTimeRange)
+ .add("sloppyPattern", sloppyPattern)
+ .add("pendingQueue", pendingQueue)
+ .add("isClosed", isClosed)
+ .add("taskID", taskID)
+ .toString();
}
//////////////////////////// APIs provided for metric framework ////////////////////////////
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/assigner/PipeDataRegionAssigner.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/assigner/PipeDataRegionAssigner.java
index 0a5d1230588..2a8eeda9e0a 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/assigner/PipeDataRegionAssigner.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/assigner/PipeDataRegionAssigner.java
@@ -110,7 +110,8 @@ public class PipeDataRegionAssigner implements Closeable {
extractor.getPipeName(),
extractor.getCreationTime(),
extractor.getPipeTaskMeta(),
- extractor.getPipePattern(),
+ extractor.getTreePattern(),
+ extractor.getTablePattern(),
extractor.getRealtimeDataExtractionStartTime(),
extractor.getRealtimeDataExtractionEndTime());
reportEvent.bindProgressIndex(event.getProgressIndex());
@@ -129,7 +130,8 @@ public class PipeDataRegionAssigner implements Closeable {
extractor.getPipeName(),
extractor.getCreationTime(),
extractor.getPipeTaskMeta(),
- extractor.getPipePattern(),
+ extractor.getTreePattern(),
+ extractor.getTablePattern(),
extractor.getRealtimeDataExtractionStartTime(),
extractor.getRealtimeDataExtractionEndTime());
final EnrichedEvent innerEvent = copiedEvent.getEvent();
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/epoch/TsFileEpochManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/epoch/TsFileEpochManager.java
index e4244a1335e..d0ee25838b6 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/epoch/TsFileEpochManager.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/epoch/TsFileEpochManager.java
@@ -71,7 +71,8 @@ public class TsFileEpochManager {
epoch,
resource.getDevices().stream()
.collect(Collectors.toMap(Functions.identity(), device -> EMPTY_MEASUREMENT_ARRAY)),
- event.getPipePattern());
+ event.getTreePattern(),
+ event.getTablePattern());
}
public PipeRealtimeEvent bindPipeInsertNodeTabletInsertionEvent(
@@ -85,7 +86,8 @@ public class TsFileEpochManager {
node instanceof InsertRowsNode
? getDevice2MeasurementsMapFromInsertRowsNode((InsertRowsNode) node)
: Collections.singletonMap(node.getDeviceID(), node.getMeasurements()),
- event.getPipePattern());
+ event.getTreePattern(),
+ event.getTablePattern());
}
private Map<IDeviceID, String[]> getDevice2MeasurementsMapFromInsertRowsNode(
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/matcher/CachedSchemaPatternMatcher.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/matcher/CachedSchemaPatternMatcher.java
index ee554634563..a6cedfe0883 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/matcher/CachedSchemaPatternMatcher.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/matcher/CachedSchemaPatternMatcher.java
@@ -20,7 +20,7 @@
package org.apache.iotdb.db.pipe.extractor.dataregion.realtime.matcher;
import org.apache.iotdb.commons.pipe.config.PipeConfig;
-import org.apache.iotdb.commons.pipe.datastructure.pattern.PipePattern;
+import org.apache.iotdb.commons.pipe.datastructure.pattern.TreePattern;
import org.apache.iotdb.db.pipe.event.common.heartbeat.PipeHeartbeatEvent;
import org.apache.iotdb.db.pipe.event.common.schema.PipeSchemaRegionWritePlanEvent;
import org.apache.iotdb.db.pipe.event.realtime.PipeRealtimeEvent;
@@ -93,6 +93,7 @@ public class CachedSchemaPatternMatcher implements PipeDataRegionMatcher {
}
}
+ // TODO: consider table pattern
@Override
public Set<PipeRealtimeDataRegionExtractor> match(final PipeRealtimeEvent event) {
final Set<PipeRealtimeDataRegionExtractor> matchedExtractors = new HashSet<>();
@@ -144,7 +145,7 @@ public class CachedSchemaPatternMatcher implements PipeDataRegionMatcher {
// Match extractors by measurements.
extractorsFilteredByDevice.forEach(
extractor -> {
- final PipePattern pattern = extractor.getPipePattern();
+ final TreePattern pattern = extractor.getTreePattern();
if (Objects.isNull(pattern) || pattern.isRoot() || pattern.coversDevice(device)) {
// The pattern can match all measurements of the device.
matchedExtractors.add(extractor);
@@ -186,8 +187,8 @@ public class CachedSchemaPatternMatcher implements PipeDataRegionMatcher {
continue;
}
- final PipePattern pipePattern = extractor.getPipePattern();
- if (Objects.isNull(pipePattern) || pipePattern.mayOverlapWithDevice(device)) {
+ final TreePattern treePattern = extractor.getTreePattern();
+ if (Objects.isNull(treePattern) || treePattern.mayOverlapWithDevice(device)) {
filteredExtractors.add(extractor);
}
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/schemaregion/PipePlanPatternParseVisitor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/schemaregion/PipePlanPatternParseVisitor.java
index e7f783158cc..1d75dd66cd1 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/schemaregion/PipePlanPatternParseVisitor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/schemaregion/PipePlanPatternParseVisitor.java
@@ -21,7 +21,7 @@ package org.apache.iotdb.db.pipe.extractor.schemaregion;
import org.apache.iotdb.commons.path.MeasurementPath;
import org.apache.iotdb.commons.path.PartialPath;
-import org.apache.iotdb.commons.pipe.datastructure.pattern.IoTDBPipePattern;
+import org.apache.iotdb.commons.pipe.datastructure.pattern.IoTDBTreePattern;
import org.apache.iotdb.commons.schema.view.viewExpression.ViewExpression;
import org.apache.iotdb.db.pipe.event.common.schema.PipeSchemaRegionWritePlanEvent;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNode;
@@ -54,27 +54,27 @@ import java.util.stream.IntStream;
/**
* The {@link PipePlanPatternParseVisitor} will transform the schema {@link PlanNode}s using {@link
- * IoTDBPipePattern}. Rule:
+ * IoTDBTreePattern}. Rule:
*
* <p>1. All patterns in the output {@link PlanNode} will be the intersection of the original {@link
- * PlanNode}'s patterns and the given {@link IoTDBPipePattern}.
+ * PlanNode}'s patterns and the given {@link IoTDBTreePattern}.
*
- * <p>2. If a pattern does not intersect with the {@link IoTDBPipePattern}, it's dropped.
+ * <p>2. If a pattern does not intersect with the {@link IoTDBTreePattern}, it's dropped.
*
* <p>3. If all the patterns in the {@link PlanNode} is dropped, the {@link PlanNode} is dropped.
*
* <p>4. The output {@link PlanNode} shall be a copied form of the original one because the original
* one is used in the {@link PipeSchemaRegionWritePlanEvent} in {@link SchemaRegionListeningQueue}.
*/
-public class PipePlanPatternParseVisitor extends PlanVisitor<Optional<PlanNode>, IoTDBPipePattern> {
+public class PipePlanPatternParseVisitor extends PlanVisitor<Optional<PlanNode>, IoTDBTreePattern> {
@Override
- public Optional<PlanNode> visitPlan(final PlanNode node, final IoTDBPipePattern pattern) {
+ public Optional<PlanNode> visitPlan(final PlanNode node, final IoTDBTreePattern pattern) {
return Optional.of(node);
}
@Override
public Optional<PlanNode> visitCreateTimeSeries(
- final CreateTimeSeriesNode node, final IoTDBPipePattern pattern) {
+ final CreateTimeSeriesNode node, final IoTDBTreePattern pattern) {
return pattern.matchesMeasurement(
node.getPath().getIDeviceID(), node.getPath().getMeasurement())
? Optional.of(node)
@@ -83,7 +83,7 @@ public class PipePlanPatternParseVisitor extends PlanVisitor<Optional<PlanNode>,
@Override
public Optional<PlanNode> visitCreateAlignedTimeSeries(
- final CreateAlignedTimeSeriesNode node, final IoTDBPipePattern pattern) {
+ final CreateAlignedTimeSeriesNode node, final IoTDBTreePattern pattern) {
final int[] filteredIndexes =
IntStream.range(0, node.getMeasurements().size())
.filter(
@@ -97,19 +97,19 @@ public class PipePlanPatternParseVisitor extends PlanVisitor<Optional<PlanNode>,
new CreateAlignedTimeSeriesNode(
node.getPlanNodeId(),
node.getDevicePath(),
- IoTDBPipePattern.applyIndexesOnList(filteredIndexes, node.getMeasurements()),
- IoTDBPipePattern.applyIndexesOnList(filteredIndexes, node.getDataTypes()),
- IoTDBPipePattern.applyIndexesOnList(filteredIndexes, node.getEncodings()),
- IoTDBPipePattern.applyIndexesOnList(filteredIndexes, node.getCompressors()),
- IoTDBPipePattern.applyIndexesOnList(filteredIndexes, node.getAliasList()),
- IoTDBPipePattern.applyIndexesOnList(filteredIndexes, node.getTagsList()),
- IoTDBPipePattern.applyIndexesOnList(filteredIndexes, node.getAttributesList())))
+ IoTDBTreePattern.applyIndexesOnList(filteredIndexes, node.getMeasurements()),
+ IoTDBTreePattern.applyIndexesOnList(filteredIndexes, node.getDataTypes()),
+ IoTDBTreePattern.applyIndexesOnList(filteredIndexes, node.getEncodings()),
+ IoTDBTreePattern.applyIndexesOnList(filteredIndexes, node.getCompressors()),
+ IoTDBTreePattern.applyIndexesOnList(filteredIndexes, node.getAliasList()),
+ IoTDBTreePattern.applyIndexesOnList(filteredIndexes, node.getTagsList()),
+ IoTDBTreePattern.applyIndexesOnList(filteredIndexes, node.getAttributesList())))
: Optional.empty();
}
@Override
public Optional<PlanNode> visitCreateMultiTimeSeries(
- final CreateMultiTimeSeriesNode node, final IoTDBPipePattern pattern) {
+ final CreateMultiTimeSeriesNode node, final IoTDBTreePattern pattern) {
final Map<PartialPath, MeasurementGroup> filteredMeasurementGroupMap =
node.getMeasurementGroupMap().entrySet().stream()
.filter(entry -> pattern.matchPrefixPath(entry.getKey().getFullPath()))
@@ -128,7 +128,7 @@ public class PipePlanPatternParseVisitor extends PlanVisitor<Optional<PlanNode>,
}
private static MeasurementGroup trimMeasurementGroup(
- final IDeviceID device, final MeasurementGroup group, final IoTDBPipePattern pattern) {
+ final IDeviceID device, final MeasurementGroup group, final IoTDBTreePattern pattern) {
final int[] filteredIndexes =
IntStream.range(0, group.size())
.filter(index -> pattern.matchesMeasurement(device, group.getMeasurements().get(index)))
@@ -163,7 +163,7 @@ public class PipePlanPatternParseVisitor extends PlanVisitor<Optional<PlanNode>,
@Override
public Optional<PlanNode> visitAlterTimeSeries(
- final AlterTimeSeriesNode node, final IoTDBPipePattern pattern) {
+ final AlterTimeSeriesNode node, final IoTDBTreePattern pattern) {
return pattern.matchesMeasurement(
node.getPath().getIDeviceID(), node.getPath().getMeasurement())
? Optional.of(node)
@@ -172,7 +172,7 @@ public class PipePlanPatternParseVisitor extends PlanVisitor<Optional<PlanNode>,
@Override
public Optional<PlanNode> visitInternalCreateTimeSeries(
- final InternalCreateTimeSeriesNode node, final IoTDBPipePattern pattern) {
+ final InternalCreateTimeSeriesNode node, final IoTDBTreePattern pattern) {
final MeasurementGroup group =
pattern.matchPrefixPath(node.getDevicePath().getFullPath())
? trimMeasurementGroup(
@@ -189,7 +189,7 @@ public class PipePlanPatternParseVisitor extends PlanVisitor<Optional<PlanNode>,
@Override
public Optional<PlanNode> visitActivateTemplate(
- final ActivateTemplateNode node, final IoTDBPipePattern pattern) {
+ final ActivateTemplateNode node, final IoTDBTreePattern pattern) {
return pattern.matchDevice(node.getActivatePath().getFullPath())
? Optional.of(node)
: Optional.empty();
@@ -197,7 +197,7 @@ public class PipePlanPatternParseVisitor extends PlanVisitor<Optional<PlanNode>,
@Override
public Optional<PlanNode> visitInternalBatchActivateTemplate(
- final InternalBatchActivateTemplateNode node, final IoTDBPipePattern pattern) {
+ final InternalBatchActivateTemplateNode node, final IoTDBTreePattern pattern) {
final Map<PartialPath, Pair<Integer, Integer>> filteredTemplateActivationMap =
node.getTemplateActivationMap().entrySet().stream()
.filter(entry -> pattern.matchDevice(entry.getKey().getFullPath()))
@@ -211,7 +211,7 @@ public class PipePlanPatternParseVisitor extends PlanVisitor<Optional<PlanNode>,
@Override
public Optional<PlanNode> visitInternalCreateMultiTimeSeries(
- final InternalCreateMultiTimeSeriesNode node, final IoTDBPipePattern pattern) {
+ final InternalCreateMultiTimeSeriesNode node, final IoTDBTreePattern pattern) {
final Map<PartialPath, Pair<Boolean, MeasurementGroup>> filteredDeviceMap =
node.getDeviceMap().entrySet().stream()
.filter(entry -> pattern.matchPrefixPath(entry.getKey().getFullPath()))
@@ -235,7 +235,7 @@ public class PipePlanPatternParseVisitor extends PlanVisitor<Optional<PlanNode>,
@Override
public Optional<PlanNode> visitBatchActivateTemplate(
- final BatchActivateTemplateNode node, final IoTDBPipePattern pattern) {
+ final BatchActivateTemplateNode node, final IoTDBTreePattern pattern) {
final Map<PartialPath, Pair<Integer, Integer>> filteredTemplateActivationMap =
node.getTemplateActivationMap().entrySet().stream()
.filter(entry -> pattern.matchDevice(entry.getKey().getFullPath()))
@@ -248,7 +248,7 @@ public class PipePlanPatternParseVisitor extends PlanVisitor<Optional<PlanNode>,
@Override
public Optional<PlanNode> visitCreateLogicalView(
- final CreateLogicalViewNode node, final IoTDBPipePattern pattern) {
+ final CreateLogicalViewNode node, final IoTDBTreePattern pattern) {
final Map<PartialPath, ViewExpression> filteredViewPathToSourceMap =
node.getViewPathToSourceExpressionMap().entrySet().stream()
.filter(
@@ -263,7 +263,7 @@ public class PipePlanPatternParseVisitor extends PlanVisitor<Optional<PlanNode>,
@Override
public Optional<PlanNode> visitAlterLogicalView(
- final AlterLogicalViewNode node, final IoTDBPipePattern pattern) {
+ final AlterLogicalViewNode node, final IoTDBTreePattern pattern) {
final Map<PartialPath, ViewExpression> filteredViewPathToSourceMap =
node.getViewPathToSourceMap().entrySet().stream()
.filter(
@@ -278,7 +278,7 @@ public class PipePlanPatternParseVisitor extends PlanVisitor<Optional<PlanNode>,
@Override
public Optional<PlanNode> visitDeleteData(
- final DeleteDataNode node, final IoTDBPipePattern pattern) {
+ final DeleteDataNode node, final IoTDBTreePattern pattern) {
final List<MeasurementPath> intersectedPaths =
node.getPathList().stream()
.map(pattern::getIntersection)
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java
index a61c165bf80..1629fca3e3d 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java
@@ -29,7 +29,7 @@ import org.apache.iotdb.commons.pipe.connector.payload.thrift.request.PipeTransf
import org.apache.iotdb.commons.pipe.connector.payload.thrift.request.PipeTransferFileSealReqV1;
import org.apache.iotdb.commons.pipe.connector.payload.thrift.request.PipeTransferFileSealReqV2;
import org.apache.iotdb.commons.pipe.connector.payload.thrift.request.PipeTransferSliceReq;
-import org.apache.iotdb.commons.pipe.datastructure.pattern.IoTDBPipePattern;
+import org.apache.iotdb.commons.pipe.datastructure.pattern.IoTDBTreePattern;
import org.apache.iotdb.commons.pipe.receiver.IoTDBFileReceiver;
import org.apache.iotdb.commons.pipe.receiver.PipeReceiverStatusHandler;
import org.apache.iotdb.commons.utils.FileUtils;
@@ -464,8 +464,8 @@ public class IoTDBDataNodeReceiver extends IoTDBFileReceiver {
final Set<StatementType> executionTypes =
PipeSchemaRegionSnapshotEvent.getStatementTypeSet(
parameters.get(ColumnHeaderConstant.TYPE));
- final IoTDBPipePattern pattern =
- new IoTDBPipePattern(parameters.get(ColumnHeaderConstant.PATH_PATTERN));
+ final IoTDBTreePattern pattern =
+ new IoTDBTreePattern(parameters.get(ColumnHeaderConstant.PATH_PATTERN));
// Clear to avoid previous exceptions
batchVisitor.clear();
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/visitor/PipeStatementDataTypeConvertExecutionVisitor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/visitor/PipeStatementDataTypeConvertExecutionVisitor.java
index d2d12402783..5268ac32e58 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/visitor/PipeStatementDataTypeConvertExecutionVisitor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/visitor/PipeStatementDataTypeConvertExecutionVisitor.java
@@ -20,7 +20,7 @@
package org.apache.iotdb.db.pipe.receiver.visitor;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
-import org.apache.iotdb.commons.pipe.datastructure.pattern.IoTDBPipePattern;
+import org.apache.iotdb.commons.pipe.datastructure.pattern.IoTDBTreePattern;
import org.apache.iotdb.db.pipe.connector.payload.evolvable.request.PipeTransferTabletRawReq;
import org.apache.iotdb.db.pipe.event.common.tsfile.container.scan.TsFileInsertionScanDataContainer;
import org.apache.iotdb.db.pipe.receiver.transform.statement.PipeConvertedInsertRowStatement;
@@ -97,7 +97,7 @@ public class PipeStatementDataTypeConvertExecutionVisitor
for (final File file : loadTsFileStatement.getTsFiles()) {
try (final TsFileInsertionScanDataContainer container =
new TsFileInsertionScanDataContainer(
- file, new IoTDBPipePattern(null), Long.MIN_VALUE, Long.MAX_VALUE, null, null)) {
+ file, new IoTDBTreePattern(null), Long.MIN_VALUE, Long.MAX_VALUE, null, null)) {
for (final Tablet tablet : container.toTablets()) {
final PipeConvertedInsertTabletStatement statement =
new PipeConvertedInsertTabletStatement(
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/visitor/PipeStatementPatternParseVisitor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/visitor/PipeStatementPatternParseVisitor.java
index c85bd0cbcf9..21ec44e0e1b 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/visitor/PipeStatementPatternParseVisitor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/visitor/PipeStatementPatternParseVisitor.java
@@ -19,7 +19,7 @@
package org.apache.iotdb.db.pipe.receiver.visitor;
-import org.apache.iotdb.commons.pipe.datastructure.pattern.IoTDBPipePattern;
+import org.apache.iotdb.commons.pipe.datastructure.pattern.IoTDBTreePattern;
import org.apache.iotdb.db.queryengine.plan.statement.Statement;
import org.apache.iotdb.db.queryengine.plan.statement.StatementNode;
import org.apache.iotdb.db.queryengine.plan.statement.StatementVisitor;
@@ -36,12 +36,12 @@ import java.util.stream.IntStream;
/**
* The {@link PipeStatementPatternParseVisitor} will transform the schema {@link Statement}s using
- * {@link IoTDBPipePattern}. Rule:
+ * {@link IoTDBTreePattern}. Rule:
*
* <p>1. All patterns in the output {@link Statement} will be the intersection of the original
- * {@link Statement}'s patterns and the given {@link IoTDBPipePattern}.
+ * {@link Statement}'s patterns and the given {@link IoTDBTreePattern}.
*
- * <p>2. If a pattern does not intersect with the {@link IoTDBPipePattern}, it's dropped.
+ * <p>2. If a pattern does not intersect with the {@link IoTDBTreePattern}, it's dropped.
*
* <p>3. If all the patterns in the {@link Statement} is dropped, the {@link Statement} is dropped.
*
@@ -49,16 +49,16 @@ import java.util.stream.IntStream;
* from the {@link SRStatementGenerator} and will no longer be used.
*/
public class PipeStatementPatternParseVisitor
- extends StatementVisitor<Optional<Statement>, IoTDBPipePattern> {
+ extends StatementVisitor<Optional<Statement>, IoTDBTreePattern> {
@Override
public Optional<Statement> visitNode(
- final StatementNode statement, final IoTDBPipePattern pattern) {
+ final StatementNode statement, final IoTDBTreePattern pattern) {
return Optional.of((Statement) statement);
}
@Override
public Optional<Statement> visitCreateTimeseries(
- final CreateTimeSeriesStatement statement, final IoTDBPipePattern pattern) {
+ final CreateTimeSeriesStatement statement, final IoTDBTreePattern pattern) {
return pattern.matchesMeasurement(
statement.getPath().getIDeviceID(), statement.getPath().getMeasurement())
? Optional.of(statement)
@@ -67,7 +67,7 @@ public class PipeStatementPatternParseVisitor
@Override
public Optional<Statement> visitCreateAlignedTimeseries(
- final CreateAlignedTimeSeriesStatement statement, final IoTDBPipePattern pattern) {
+ final CreateAlignedTimeSeriesStatement statement, final IoTDBTreePattern pattern) {
final int[] filteredIndexes =
IntStream.range(0, statement.getMeasurements().size())
.filter(
@@ -107,7 +107,7 @@ public class PipeStatementPatternParseVisitor
// For logical view with tags/attributes
@Override
public Optional<Statement> visitAlterTimeSeries(
- final AlterTimeSeriesStatement alterTimeSeriesStatement, final IoTDBPipePattern pattern) {
+ final AlterTimeSeriesStatement alterTimeSeriesStatement, final IoTDBTreePattern pattern) {
return pattern.matchesMeasurement(
alterTimeSeriesStatement.getPath().getIDeviceID(),
alterTimeSeriesStatement.getPath().getMeasurement())
@@ -117,7 +117,7 @@ public class PipeStatementPatternParseVisitor
@Override
public Optional<Statement> visitActivateTemplate(
- final ActivateTemplateStatement activateTemplateStatement, final IoTDBPipePattern pattern) {
+ final ActivateTemplateStatement activateTemplateStatement, final IoTDBTreePattern pattern) {
return pattern.matchDevice(activateTemplateStatement.getPath().getFullPath())
? Optional.of(activateTemplateStatement)
: Optional.empty();
@@ -125,7 +125,7 @@ public class PipeStatementPatternParseVisitor
@Override
public Optional<Statement> visitCreateLogicalView(
- final CreateLogicalViewStatement createLogicalViewStatement, final IoTDBPipePattern pattern) {
+ final CreateLogicalViewStatement createLogicalViewStatement, final IoTDBTreePattern pattern) {
final int[] filteredIndexes =
IntStream.range(0, createLogicalViewStatement.getTargetPathList().size())
.filter(
@@ -141,10 +141,10 @@ public class PipeStatementPatternParseVisitor
return Optional.empty();
}
createLogicalViewStatement.setTargetFullPaths(
- IoTDBPipePattern.applyIndexesOnList(
+ IoTDBTreePattern.applyIndexesOnList(
filteredIndexes, createLogicalViewStatement.getTargetPathList()));
createLogicalViewStatement.setViewExpressions(
- IoTDBPipePattern.applyIndexesOnList(
+ IoTDBTreePattern.applyIndexesOnList(
filteredIndexes, createLogicalViewStatement.getViewExpressions()));
return Optional.of(createLogicalViewStatement);
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/connector/PipeStatementPatternParseVisitorTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/connector/PipeStatementPatternParseVisitorTest.java
index 24c128fd93b..6ef615a5047 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/connector/PipeStatementPatternParseVisitorTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/connector/PipeStatementPatternParseVisitorTest.java
@@ -22,7 +22,7 @@ package org.apache.iotdb.db.pipe.connector;
import org.apache.iotdb.commons.exception.IllegalPathException;
import org.apache.iotdb.commons.path.MeasurementPath;
import org.apache.iotdb.commons.path.PartialPath;
-import org.apache.iotdb.commons.pipe.datastructure.pattern.IoTDBPipePattern;
+import org.apache.iotdb.commons.pipe.datastructure.pattern.IoTDBTreePattern;
import org.apache.iotdb.commons.schema.view.viewExpression.leaf.TimeSeriesViewOperand;
import org.apache.iotdb.db.pipe.receiver.visitor.PipeStatementPatternParseVisitor;
import org.apache.iotdb.db.queryengine.plan.statement.metadata.AlterTimeSeriesStatement;
@@ -43,8 +43,8 @@ import java.util.Map;
public class PipeStatementPatternParseVisitorTest {
- private final IoTDBPipePattern prefixPathPattern = new IoTDBPipePattern("root.db.device.**");
- private final IoTDBPipePattern fullPathPattern = new IoTDBPipePattern("root.db.device.s1");
+ private final IoTDBTreePattern prefixPathPattern = new IoTDBTreePattern("root.db.device.**");
+ private final IoTDBTreePattern fullPathPattern = new IoTDBTreePattern("root.db.device.s1");
@Test
public void testCreateTimeSeries() throws IllegalPathException {
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/PipeTabletInsertionEventTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/PipeTabletInsertionEventTest.java
index 4a7d77eab8d..b1629623196 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/PipeTabletInsertionEventTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/PipeTabletInsertionEventTest.java
@@ -21,7 +21,7 @@ package org.apache.iotdb.db.pipe.event;
import org.apache.iotdb.commons.exception.IllegalPathException;
import org.apache.iotdb.commons.path.PartialPath;
-import org.apache.iotdb.commons.pipe.datastructure.pattern.PrefixPipePattern;
+import org.apache.iotdb.commons.pipe.datastructure.pattern.PrefixTreePattern;
import org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent;
import org.apache.iotdb.db.pipe.event.common.tablet.TabletInsertionDataContainer;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeId;
@@ -259,28 +259,28 @@ public class PipeTabletInsertionEventTest {
@Test
public void convertToTabletForTest() {
TabletInsertionDataContainer container1 =
- new TabletInsertionDataContainer(insertRowNode, new PrefixPipePattern(pattern));
+ new TabletInsertionDataContainer(insertRowNode, new PrefixTreePattern(pattern));
Tablet tablet1 = container1.convertToTablet();
boolean isAligned1 = container1.isAligned();
Assert.assertEquals(tablet1, tabletForInsertRowNode);
Assert.assertFalse(isAligned1);
TabletInsertionDataContainer container2 =
- new TabletInsertionDataContainer(insertTabletNode, new PrefixPipePattern(pattern));
+ new TabletInsertionDataContainer(insertTabletNode, new PrefixTreePattern(pattern));
Tablet tablet2 = container2.convertToTablet();
boolean isAligned2 = container2.isAligned();
Assert.assertEquals(tablet2, tabletForInsertTabletNode);
Assert.assertFalse(isAligned2);
PipeRawTabletInsertionEvent event3 =
- new PipeRawTabletInsertionEvent(tablet1, false, new PrefixPipePattern(pattern));
+ new PipeRawTabletInsertionEvent(tablet1, false, new PrefixTreePattern(pattern));
Tablet tablet3 = event3.convertToTablet();
boolean isAligned3 = event3.isAligned();
Assert.assertEquals(tablet1, tablet3);
Assert.assertFalse(isAligned3);
PipeRawTabletInsertionEvent event4 =
- new PipeRawTabletInsertionEvent(tablet2, false, new PrefixPipePattern(pattern));
+ new PipeRawTabletInsertionEvent(tablet2, false, new PrefixTreePattern(pattern));
Tablet tablet4 = event4.convertToTablet();
boolean isAligned4 = event4.isAligned();
Assert.assertEquals(tablet2, tablet4);
@@ -290,28 +290,28 @@ public class PipeTabletInsertionEventTest {
@Test
public void convertToAlignedTabletForTest() {
TabletInsertionDataContainer container1 =
- new TabletInsertionDataContainer(insertRowNodeAligned, new PrefixPipePattern(pattern));
+ new TabletInsertionDataContainer(insertRowNodeAligned, new PrefixTreePattern(pattern));
Tablet tablet1 = container1.convertToTablet();
boolean isAligned1 = container1.isAligned();
Assert.assertEquals(tablet1, tabletForInsertRowNode);
Assert.assertTrue(isAligned1);
TabletInsertionDataContainer container2 =
- new TabletInsertionDataContainer(insertTabletNodeAligned, new PrefixPipePattern(pattern));
+ new TabletInsertionDataContainer(insertTabletNodeAligned, new PrefixTreePattern(pattern));
Tablet tablet2 = container2.convertToTablet();
boolean isAligned2 = container2.isAligned();
Assert.assertEquals(tablet2, tabletForInsertTabletNode);
Assert.assertTrue(isAligned2);
PipeRawTabletInsertionEvent event3 =
- new PipeRawTabletInsertionEvent(tablet1, true, new PrefixPipePattern(pattern));
+ new PipeRawTabletInsertionEvent(tablet1, true, new PrefixTreePattern(pattern));
Tablet tablet3 = event3.convertToTablet();
boolean isAligned3 = event3.isAligned();
Assert.assertEquals(tablet1, tablet3);
Assert.assertTrue(isAligned3);
PipeRawTabletInsertionEvent event4 =
- new PipeRawTabletInsertionEvent(tablet2, true, new PrefixPipePattern(pattern));
+ new PipeRawTabletInsertionEvent(tablet2, true, new PrefixTreePattern(pattern));
Tablet tablet4 = event4.convertToTablet();
boolean isAligned4 = event4.isAligned();
Assert.assertEquals(tablet2, tablet4);
@@ -325,7 +325,7 @@ public class PipeTabletInsertionEventTest {
null,
new PipeRawTabletInsertionEvent(tabletForInsertRowNode, 111L, 113L),
insertRowNode,
- new PrefixPipePattern(pattern));
+ new PrefixTreePattern(pattern));
Tablet tablet1 = container1.convertToTablet();
Assert.assertEquals(0, tablet1.rowSize);
boolean isAligned1 = container1.isAligned();
@@ -336,7 +336,7 @@ public class PipeTabletInsertionEventTest {
null,
new PipeRawTabletInsertionEvent(tabletForInsertRowNode, 110L, 110L),
insertRowNode,
- new PrefixPipePattern(pattern));
+ new PrefixTreePattern(pattern));
Tablet tablet2 = container2.convertToTablet();
Assert.assertEquals(1, tablet2.rowSize);
boolean isAligned2 = container2.isAligned();
@@ -347,7 +347,7 @@ public class PipeTabletInsertionEventTest {
null,
new PipeRawTabletInsertionEvent(tabletForInsertTabletNode, 111L, 113L),
insertTabletNode,
- new PrefixPipePattern(pattern));
+ new PrefixTreePattern(pattern));
Tablet tablet3 = container3.convertToTablet();
Assert.assertEquals(3, tablet3.rowSize);
boolean isAligned3 = container3.isAligned();
@@ -358,7 +358,7 @@ public class PipeTabletInsertionEventTest {
null,
new PipeRawTabletInsertionEvent(tabletForInsertTabletNode, Long.MIN_VALUE, 109L),
insertTabletNode,
- new PrefixPipePattern(pattern));
+ new PrefixTreePattern(pattern));
Tablet tablet4 = container4.convertToTablet();
Assert.assertEquals(0, tablet4.rowSize);
boolean isAligned4 = container4.isAligned();
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/TsFileInsertionDataContainerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/TsFileInsertionDataContainerTest.java
index 9cfa9328d5a..98b15b39050 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/TsFileInsertionDataContainerTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/TsFileInsertionDataContainerTest.java
@@ -21,9 +21,9 @@ package org.apache.iotdb.db.pipe.event;
import org.apache.iotdb.commons.exception.IllegalPathException;
import org.apache.iotdb.commons.path.PartialPath;
-import org.apache.iotdb.commons.pipe.datastructure.pattern.IoTDBPipePattern;
-import org.apache.iotdb.commons.pipe.datastructure.pattern.PipePattern;
-import org.apache.iotdb.commons.pipe.datastructure.pattern.PrefixPipePattern;
+import org.apache.iotdb.commons.pipe.datastructure.pattern.IoTDBTreePattern;
+import org.apache.iotdb.commons.pipe.datastructure.pattern.PrefixTreePattern;
+import org.apache.iotdb.commons.pipe.datastructure.pattern.TreePattern;
import org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent;
import org.apache.iotdb.db.pipe.event.common.tsfile.container.TsFileInsertionDataContainer;
import org.apache.iotdb.db.pipe.event.common.tsfile.container.query.TsFileInsertionQueryDataContainer;
@@ -350,14 +350,14 @@ public class TsFileInsertionDataContainerTest {
}
}
- final PipePattern rootPattern;
+ final TreePattern rootPattern;
switch (patternFormat) {
case PREFIX_FORMAT:
- rootPattern = new PrefixPipePattern("root");
+ rootPattern = new PrefixTreePattern("root");
break;
case IOTDB_FORMAT:
default:
- rootPattern = new IoTDBPipePattern("root.**");
+ rootPattern = new IoTDBTreePattern("root.**");
break;
}
@@ -411,17 +411,17 @@ public class TsFileInsertionDataContainerTest {
}));
}
- final PipePattern oneAlignedDevicePattern;
- final PipePattern oneNonAlignedDevicePattern;
+ final TreePattern oneAlignedDevicePattern;
+ final TreePattern oneNonAlignedDevicePattern;
switch (patternFormat) {
case PREFIX_FORMAT:
- oneAlignedDevicePattern = new PrefixPipePattern(oneDeviceInAlignedTsFile.get());
- oneNonAlignedDevicePattern = new PrefixPipePattern(oneDeviceInUnalignedTsFile.get());
+ oneAlignedDevicePattern = new PrefixTreePattern(oneDeviceInAlignedTsFile.get());
+ oneNonAlignedDevicePattern = new PrefixTreePattern(oneDeviceInUnalignedTsFile.get());
break;
case IOTDB_FORMAT:
default:
- oneAlignedDevicePattern = new IoTDBPipePattern(oneDeviceInAlignedTsFile.get() + ".**");
- oneNonAlignedDevicePattern = new IoTDBPipePattern(oneDeviceInUnalignedTsFile.get() + ".**");
+ oneAlignedDevicePattern = new IoTDBTreePattern(oneDeviceInAlignedTsFile.get() + ".**");
+ oneNonAlignedDevicePattern = new IoTDBTreePattern(oneDeviceInUnalignedTsFile.get() + ".**");
break;
}
@@ -440,19 +440,19 @@ public class TsFileInsertionDataContainerTest {
isQuery,
expectedRowNumber * measurementNumber);
- final PipePattern oneAlignedMeasurementPattern;
- final PipePattern oneNonAlignedMeasurementPattern;
+ final TreePattern oneAlignedMeasurementPattern;
+ final TreePattern oneNonAlignedMeasurementPattern;
switch (patternFormat) {
case PREFIX_FORMAT:
- oneAlignedMeasurementPattern = new PrefixPipePattern(oneMeasurementInAlignedTsFile.get());
+ oneAlignedMeasurementPattern = new PrefixTreePattern(oneMeasurementInAlignedTsFile.get());
oneNonAlignedMeasurementPattern =
- new PrefixPipePattern(oneMeasurementInUnalignedTsFile.get());
+ new PrefixTreePattern(oneMeasurementInUnalignedTsFile.get());
break;
case IOTDB_FORMAT:
default:
- oneAlignedMeasurementPattern = new IoTDBPipePattern(oneMeasurementInAlignedTsFile.get());
+ oneAlignedMeasurementPattern = new IoTDBTreePattern(oneMeasurementInAlignedTsFile.get());
oneNonAlignedMeasurementPattern =
- new IoTDBPipePattern(oneMeasurementInUnalignedTsFile.get());
+ new IoTDBTreePattern(oneMeasurementInUnalignedTsFile.get());
break;
}
@@ -471,14 +471,14 @@ public class TsFileInsertionDataContainerTest {
isQuery,
expectedRowNumber);
- final PipePattern notExistPattern;
+ final TreePattern notExistPattern;
switch (patternFormat) {
case PREFIX_FORMAT:
- notExistPattern = new PrefixPipePattern("root.`not-exist-pattern`");
+ notExistPattern = new PrefixTreePattern("root.`not-exist-pattern`");
break;
case IOTDB_FORMAT:
default:
- notExistPattern = new IoTDBPipePattern("root.`not-exist-pattern`");
+ notExistPattern = new IoTDBTreePattern("root.`not-exist-pattern`");
break;
}
@@ -514,7 +514,7 @@ public class TsFileInsertionDataContainerTest {
testTsFilePointNum(
resource.getTsFile(),
- new PrefixPipePattern("root"),
+ new PrefixTreePattern("root"),
Long.MIN_VALUE,
Long.MAX_VALUE,
isQuery,
@@ -548,12 +548,12 @@ public class TsFileInsertionDataContainerTest {
writer.writeAligned(t);
}
testTsFilePointNum(
- alignedTsFile, new PrefixPipePattern("root"), Long.MIN_VALUE, Long.MAX_VALUE, isQuery, 4);
+ alignedTsFile, new PrefixTreePattern("root"), Long.MIN_VALUE, Long.MAX_VALUE, isQuery, 4);
}
private void testTsFilePointNum(
final File tsFile,
- final PipePattern pattern,
+ final TreePattern pattern,
final long startTime,
final long endTime,
final boolean isQuery,
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/extractor/PipePlanPatternParseVisitorTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/extractor/PipePlanPatternParseVisitorTest.java
index bbaa45dad06..428faaf112e 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/extractor/PipePlanPatternParseVisitorTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/extractor/PipePlanPatternParseVisitorTest.java
@@ -22,7 +22,7 @@ package org.apache.iotdb.db.pipe.extractor;
import org.apache.iotdb.commons.exception.IllegalPathException;
import org.apache.iotdb.commons.path.MeasurementPath;
import org.apache.iotdb.commons.path.PartialPath;
-import org.apache.iotdb.commons.pipe.datastructure.pattern.IoTDBPipePattern;
+import org.apache.iotdb.commons.pipe.datastructure.pattern.IoTDBTreePattern;
import org.apache.iotdb.commons.schema.view.viewExpression.ViewExpression;
import org.apache.iotdb.commons.schema.view.viewExpression.leaf.TimeSeriesViewOperand;
import org.apache.iotdb.db.pipe.extractor.schemaregion.IoTDBSchemaRegionExtractor;
@@ -56,8 +56,8 @@ import java.util.HashMap;
import java.util.Map;
public class PipePlanPatternParseVisitorTest {
- private final IoTDBPipePattern prefixPathPattern = new IoTDBPipePattern("root.db.device.**");
- private final IoTDBPipePattern fullPathPattern = new IoTDBPipePattern("root.db.device.s1");
+ private final IoTDBTreePattern prefixPathPattern = new IoTDBTreePattern("root.db.device.**");
+ private final IoTDBTreePattern fullPathPattern = new IoTDBTreePattern("root.db.device.s1");
@Test
public void testCreateTimeSeries() throws IllegalPathException {
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/pattern/CachedSchemaPatternMatcherTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/pattern/CachedSchemaPatternMatcherTest.java
index 245b2ac4369..487fabfda38 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/pattern/CachedSchemaPatternMatcherTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/pattern/CachedSchemaPatternMatcherTest.java
@@ -22,8 +22,8 @@ package org.apache.iotdb.db.pipe.pattern;
import org.apache.iotdb.commons.pipe.config.constant.PipeExtractorConstant;
import org.apache.iotdb.commons.pipe.config.plugin.configuraion.PipeTaskRuntimeConfiguration;
import org.apache.iotdb.commons.pipe.config.plugin.env.PipeTaskExtractorRuntimeEnvironment;
-import org.apache.iotdb.commons.pipe.datastructure.pattern.PipePattern;
-import org.apache.iotdb.commons.pipe.datastructure.pattern.PrefixPipePattern;
+import org.apache.iotdb.commons.pipe.datastructure.pattern.PrefixTreePattern;
+import org.apache.iotdb.commons.pipe.datastructure.pattern.TreePattern;
import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
import org.apache.iotdb.db.pipe.event.realtime.PipeRealtimeEvent;
import org.apache.iotdb.db.pipe.extractor.dataregion.realtime.PipeRealtimeDataRegionExtractor;
@@ -59,8 +59,8 @@ public class CachedSchemaPatternMatcherTest {
EnrichedEvent event,
TsFileEpoch tsFileEpoch,
Map<IDeviceID, String[]> device2Measurements,
- PipePattern pattern) {
- super(event, tsFileEpoch, device2Measurements, pattern);
+ TreePattern pattern) {
+ super(event, tsFileEpoch, device2Measurements, pattern, null);
}
@Override
@@ -180,7 +180,7 @@ public class CachedSchemaPatternMatcherTest {
public static class PipeRealtimeDataRegionFakeExtractor extends PipeRealtimeDataRegionExtractor {
public PipeRealtimeDataRegionFakeExtractor() {
- pipePattern = new PrefixPipePattern(null);
+ treePattern = new PrefixTreePattern(null);
}
@Override
@@ -200,13 +200,13 @@ public class CachedSchemaPatternMatcherTest {
match[0] =
match[0]
|| (k + TsFileConstant.PATH_SEPARATOR + s)
- .startsWith(getPatternString());
+ .startsWith(getTreePattern().getPattern());
}
} else {
match[0] =
match[0]
- || (getPatternString().startsWith(k.toString())
- || k.toString().startsWith(getPatternString()));
+ || (getTreePattern().getPattern().startsWith(k.toString())
+ || k.toString().startsWith(getTreePattern().getPattern()));
}
});
Assert.assertTrue(match[0]);
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/pattern/IoTDBPipePatternTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/pattern/IoTDBTreePatternTest.java
similarity index 83%
rename from iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/pattern/IoTDBPipePatternTest.java
rename to iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/pattern/IoTDBTreePatternTest.java
index 471900fbe5e..b246b8a33e5 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/pattern/IoTDBPipePatternTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/pattern/IoTDBTreePatternTest.java
@@ -19,7 +19,7 @@
package org.apache.iotdb.db.pipe.pattern;
-import org.apache.iotdb.commons.pipe.datastructure.pattern.IoTDBPipePattern;
+import org.apache.iotdb.commons.pipe.datastructure.pattern.IoTDBTreePattern;
import org.apache.iotdb.pipe.api.exception.PipeException;
import org.apache.tsfile.file.metadata.IDeviceID;
@@ -27,7 +27,7 @@ import org.apache.tsfile.file.metadata.StringArrayDeviceID;
import org.junit.Assert;
import org.junit.Test;
-public class IoTDBPipePatternTest {
+public class IoTDBTreePatternTest {
@Test
public void testIotdbPipePattern() {
@@ -39,11 +39,11 @@ public class IoTDBPipePatternTest {
"root.", "roo", "", "root..", "root./",
};
for (final String s : legalPatterns) {
- Assert.assertTrue(new IoTDBPipePattern(s).isLegal());
+ Assert.assertTrue(new IoTDBTreePattern(s).isLegal());
}
for (final String t : illegalPatterns) {
try {
- Assert.assertFalse(new IoTDBPipePattern(t).isLegal());
+ Assert.assertFalse(new IoTDBTreePattern(t).isLegal());
} catch (final Exception e) {
Assert.assertTrue(e instanceof PipeException);
}
@@ -58,10 +58,10 @@ public class IoTDBPipePatternTest {
"root.db", "root.*", "root.*.*", "root.db.*.**", "root.db.d1", "root.**.db.**",
};
for (final String s : patternsCoverDb) {
- Assert.assertTrue(new IoTDBPipePattern(s).coversDb(db));
+ Assert.assertTrue(new IoTDBTreePattern(s).coversDb(db));
}
for (final String t : patternsNotCoverDb) {
- Assert.assertFalse(new IoTDBPipePattern(t).coversDb(db));
+ Assert.assertFalse(new IoTDBTreePattern(t).coversDb(db));
}
final IDeviceID device = new StringArrayDeviceID("root.db.d1");
@@ -74,10 +74,10 @@ public class IoTDBPipePatternTest {
"root.*", "root.*.*", "root.db.d1", "root.db.d2.*", "root.**.d2.**",
};
for (final String s : patternsCoverDevice) {
- Assert.assertTrue(new IoTDBPipePattern(s).coversDevice(device));
+ Assert.assertTrue(new IoTDBTreePattern(s).coversDevice(device));
}
for (String t : patternsNotCoverDevice) {
- Assert.assertFalse(new IoTDBPipePattern(t).coversDevice(device));
+ Assert.assertFalse(new IoTDBTreePattern(t).coversDevice(device));
}
// Test pattern may overlap with device
@@ -88,10 +88,10 @@ public class IoTDBPipePatternTest {
"root.db.d2.**", "root.db2.d1.**", "root.db.db.d1.**",
};
for (final String s : patternsOverlapWithDevice) {
- Assert.assertTrue(new IoTDBPipePattern(s).mayOverlapWithDevice(device));
+ Assert.assertTrue(new IoTDBTreePattern(s).mayOverlapWithDevice(device));
}
for (final String t : patternsNotOverlapWithDevice) {
- Assert.assertFalse(new IoTDBPipePattern(t).mayOverlapWithDevice(device));
+ Assert.assertFalse(new IoTDBTreePattern(t).mayOverlapWithDevice(device));
}
// Test pattern match measurement
@@ -103,10 +103,10 @@ public class IoTDBPipePatternTest {
"root.db.d1", "root.db.d1", "root.db.d1.*.*",
};
for (final String s : patternsMatchMeasurement) {
- Assert.assertTrue(new IoTDBPipePattern(s).matchesMeasurement(device, measurement));
+ Assert.assertTrue(new IoTDBTreePattern(s).matchesMeasurement(device, measurement));
}
for (final String t : patternsNotMatchMeasurement) {
- Assert.assertFalse(new IoTDBPipePattern(t).matchesMeasurement(device, measurement));
+ Assert.assertFalse(new IoTDBTreePattern(t).matchesMeasurement(device, measurement));
}
}
}
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/pattern/PrefixPipePatternTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/pattern/PrefixTreePatternTest.java
similarity index 82%
rename from iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/pattern/PrefixPipePatternTest.java
rename to iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/pattern/PrefixTreePatternTest.java
index ab344bfab72..999f266605c 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/pattern/PrefixPipePatternTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/pattern/PrefixTreePatternTest.java
@@ -19,14 +19,14 @@
package org.apache.iotdb.db.pipe.pattern;
-import org.apache.iotdb.commons.pipe.datastructure.pattern.PrefixPipePattern;
+import org.apache.iotdb.commons.pipe.datastructure.pattern.PrefixTreePattern;
import org.apache.tsfile.file.metadata.IDeviceID;
import org.apache.tsfile.file.metadata.StringArrayDeviceID;
import org.junit.Assert;
import org.junit.Test;
-public class PrefixPipePatternTest {
+public class PrefixTreePatternTest {
@Test
public void testPrefixPipePattern() {
@@ -38,10 +38,10 @@ public class PrefixPipePatternTest {
"roo", "", "root..", "root./",
};
for (final String s : legalPatterns) {
- Assert.assertTrue(new PrefixPipePattern(s).isLegal());
+ Assert.assertTrue(new PrefixTreePattern(s).isLegal());
}
for (final String t : illegalPatterns) {
- Assert.assertFalse(new PrefixPipePattern(t).isLegal());
+ Assert.assertFalse(new PrefixTreePattern(t).isLegal());
}
// Test pattern cover db
@@ -53,10 +53,10 @@ public class PrefixPipePatternTest {
"root.**", "root.db.",
};
for (final String s : patternsCoverDb) {
- Assert.assertTrue(new PrefixPipePattern(s).coversDb(db));
+ Assert.assertTrue(new PrefixTreePattern(s).coversDb(db));
}
for (final String t : patternsNotCoverDb) {
- Assert.assertFalse(new PrefixPipePattern(t).coversDb(db));
+ Assert.assertFalse(new PrefixTreePattern(t).coversDb(db));
}
final IDeviceID device = new StringArrayDeviceID("root.db.d1");
@@ -69,10 +69,10 @@ public class PrefixPipePatternTest {
"root.db.d1.", "root.db.d1.s1", "root.**", "root.db.d2",
};
for (final String s : patternsCoverDevice) {
- Assert.assertTrue(new PrefixPipePattern(s).coversDevice(device));
+ Assert.assertTrue(new PrefixTreePattern(s).coversDevice(device));
}
for (final String t : patternsNotCoverDevice) {
- Assert.assertFalse(new PrefixPipePattern(t).coversDevice(device));
+ Assert.assertFalse(new PrefixTreePattern(t).coversDevice(device));
}
// Test pattern may overlap with device
@@ -83,10 +83,10 @@ public class PrefixPipePatternTest {
"root.db.d2", "root.**",
};
for (final String s : patternsOverlapWithDevice) {
- Assert.assertTrue(new PrefixPipePattern(s).mayOverlapWithDevice(device));
+ Assert.assertTrue(new PrefixTreePattern(s).mayOverlapWithDevice(device));
}
for (final String t : patternsNotOverlapWithDevice) {
- Assert.assertFalse(new PrefixPipePattern(t).mayOverlapWithDevice(device));
+ Assert.assertFalse(new PrefixTreePattern(t).mayOverlapWithDevice(device));
}
// Test pattern match measurement
@@ -98,10 +98,10 @@ public class PrefixPipePatternTest {
"root.db.d1.s11", "root.db.d1.s2",
};
for (final String s : patternsMatchMeasurement) {
- Assert.assertTrue(new PrefixPipePattern(s).matchesMeasurement(device, measurement));
+ Assert.assertTrue(new PrefixTreePattern(s).matchesMeasurement(device, measurement));
}
for (final String t : patternsNotMatchMeasurement) {
- Assert.assertFalse(new PrefixPipePattern(t).matchesMeasurement(device, measurement));
+ Assert.assertFalse(new PrefixTreePattern(t).matchesMeasurement(device, measurement));
}
}
}
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/PipeExtractorConstant.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/PipeExtractorConstant.java
index 58b8c06192c..db5cded0688 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/PipeExtractorConstant.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/PipeExtractorConstant.java
@@ -56,6 +56,10 @@ public class PipeExtractorConstant {
public static final String EXTRACTOR_PATTERN_FORMAT_IOTDB_VALUE = "iotdb";
public static final String EXTRACTOR_PATTERN_PREFIX_DEFAULT_VALUE = "root";
public static final String EXTRACTOR_PATTERN_IOTDB_DEFAULT_VALUE = "root.**";
+ public static final String EXTRACTOR_DATABASE_NAME_KEY = "extractor.database-name";
+ public static final String SOURCE_DATABASE_NAME_KEY = "source.database-name";
+ public static final String EXTRACTOR_TABLE_NAME_KEY = "extractor.table-name";
+ public static final String SOURCE_TABLE_NAME_KEY = "source.table-name";
public static final String EXTRACTOR_FORWARDING_PIPE_REQUESTS_KEY =
"extractor.forwarding-pipe-requests";
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/pattern/IoTDBPipePattern.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/pattern/IoTDBTreePattern.java
similarity index 95%
rename from iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/pattern/IoTDBPipePattern.java
rename to iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/pattern/IoTDBTreePattern.java
index a410e37e371..71654f25e69 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/pattern/IoTDBPipePattern.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/pattern/IoTDBTreePattern.java
@@ -37,11 +37,11 @@ import java.util.List;
import java.util.Objects;
import java.util.stream.Collectors;
-public class IoTDBPipePattern extends PipePattern {
+public class IoTDBTreePattern extends TreePattern {
private final PartialPath patternPartialPath;
- public IoTDBPipePattern(final String pattern) {
+ public IoTDBTreePattern(final String pattern) {
super(pattern);
try {
@@ -123,9 +123,9 @@ public class IoTDBPipePattern extends PipePattern {
}
/**
- * Check if the {@link PipePattern} matches the given prefix path. In schema transmission, this
+ * Check if the {@link TreePattern} matches the given prefix path. In schema transmission, this
* can be used to detect whether the given path can act as a parent path of the {@link
- * PipePattern}, and to transmit possibly used schemas like database creation and template
+ * TreePattern}, and to transmit possibly used schemas like database creation and template
* setting.
*/
public boolean matchPrefixPath(final String path) {
@@ -156,7 +156,7 @@ public class IoTDBPipePattern extends PipePattern {
}
/**
- * Get the intersection of the given {@link PartialPath} and the {@link PipePattern}, Only used by
+ * Get the intersection of the given {@link PartialPath} and the {@link TreePattern}, Only used by
* schema transmission. Caller shall ensure that it is a prefix or full path pattern.
*/
public List<PartialPath> getIntersection(final PartialPath partialPath) {
@@ -169,7 +169,7 @@ public class IoTDBPipePattern extends PipePattern {
}
/**
- * Get the intersection of the given {@link PathPatternTree} and the {@link PipePattern}. Only
+ * Get the intersection of the given {@link PathPatternTree} and the {@link TreePattern}. Only
* used by schema transmission. Caller shall ensure that it is a prefix or full path pattern.
*/
public PathPatternTree getIntersection(final PathPatternTree patternTree) {
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/pattern/PrefixPipePattern.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/pattern/PrefixTreePattern.java
similarity index 97%
rename from iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/pattern/PrefixPipePattern.java
rename to iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/pattern/PrefixTreePattern.java
index b7e6bc7b023..0f5ad1b81bd 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/pattern/PrefixPipePattern.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/pattern/PrefixTreePattern.java
@@ -29,9 +29,9 @@ import org.apache.tsfile.file.metadata.IDeviceID;
import java.util.Arrays;
-public class PrefixPipePattern extends PipePattern {
+public class PrefixTreePattern extends TreePattern {
- public PrefixPipePattern(final String pattern) {
+ public PrefixTreePattern(final String pattern) {
super(pattern);
}
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/pattern/TablePattern.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/pattern/TablePattern.java
new file mode 100644
index 00000000000..61a851b7f03
--- /dev/null
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/pattern/TablePattern.java
@@ -0,0 +1,94 @@
+/*
+ * 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.commons.pipe.datastructure.pattern;
+
+import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
+import org.apache.iotdb.pipe.api.exception.PipeException;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.regex.Pattern;
+
+import static org.apache.iotdb.commons.pipe.config.constant.PipeExtractorConstant.EXTRACTOR_DATABASE_NAME_KEY;
+import static org.apache.iotdb.commons.pipe.config.constant.PipeExtractorConstant.EXTRACTOR_TABLE_NAME_KEY;
+import static org.apache.iotdb.commons.pipe.config.constant.PipeExtractorConstant.SOURCE_DATABASE_NAME_KEY;
+import static org.apache.iotdb.commons.pipe.config.constant.PipeExtractorConstant.SOURCE_TABLE_NAME_KEY;
+
+public class TablePattern {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(TablePattern.class);
+
+ private final Pattern databasePattern;
+ private final Pattern tablePattern;
+
+ protected TablePattern(final String databasePatternString, final String tablePatternString) {
+ databasePattern = databasePatternString == null ? null : Pattern.compile(databasePatternString);
+ tablePattern = tablePatternString == null ? null : Pattern.compile(tablePatternString);
+ }
+
+ public boolean hasUserSpecifiedDatabasePatternOrTablePattern() {
+ return databasePattern != null || tablePattern != null;
+ }
+
+ public boolean matchesDatabase(final String database) {
+ return databasePattern == null || databasePattern.matcher(database).matches();
+ }
+
+ public boolean matchesTable(final String table) {
+ return tablePattern == null || tablePattern.matcher(table).matches();
+ }
+
+ public String getDatabasePattern() {
+ return databasePattern == null ? ".*" : databasePattern.pattern();
+ }
+
+ public String getTablePattern() {
+ return tablePattern == null ? ".*" : tablePattern.pattern();
+ }
+
+ /**
+ * Interpret from source parameters and get a pipe pattern.
+ *
+ * @return The interpreted {@link TablePattern} which is not null.
+ */
+ public static TablePattern parsePipePatternFromSourceParameters(
+ final PipeParameters sourceParameters) {
+ final String databaseNamePattern =
+ sourceParameters.getStringByKeys(EXTRACTOR_DATABASE_NAME_KEY, SOURCE_DATABASE_NAME_KEY);
+ final String tableNamePattern =
+ sourceParameters.getStringByKeys(EXTRACTOR_TABLE_NAME_KEY, SOURCE_TABLE_NAME_KEY);
+ try {
+ return new TablePattern(databaseNamePattern, tableNamePattern);
+ } catch (final Exception e) {
+ throw new PipeException("Illegal database or table pattern. Detail: " + e.getMessage(), e);
+ }
+ }
+
+ @Override
+ public String toString() {
+ return "TablePattern{"
+ + "databasePattern="
+ + databasePattern
+ + ", tablePattern="
+ + tablePattern
+ + '}';
+ }
+}
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/pattern/PipePattern.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/pattern/TreePattern.java
similarity index 85%
rename from iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/pattern/PipePattern.java
rename to iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/pattern/TreePattern.java
index ea78bee5e2e..de1d25c2aa9 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/pattern/PipePattern.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/pattern/TreePattern.java
@@ -36,13 +36,13 @@ import static org.apache.iotdb.commons.pipe.config.constant.PipeExtractorConstan
import static org.apache.iotdb.commons.pipe.config.constant.PipeExtractorConstant.SOURCE_PATTERN_FORMAT_KEY;
import static org.apache.iotdb.commons.pipe.config.constant.PipeExtractorConstant.SOURCE_PATTERN_KEY;
-public abstract class PipePattern {
+public abstract class TreePattern {
- private static final Logger LOGGER = LoggerFactory.getLogger(PipePattern.class);
+ private static final Logger LOGGER = LoggerFactory.getLogger(TreePattern.class);
protected final String pattern;
- protected PipePattern(final String pattern) {
+ protected TreePattern(final String pattern) {
this.pattern = pattern != null ? pattern : getDefaultPattern();
}
@@ -57,16 +57,16 @@ public abstract class PipePattern {
/**
* Interpret from source parameters and get a pipe pattern.
*
- * @return The interpreted {@link PipePattern} which is not null.
+ * @return The interpreted {@link TreePattern} which is not null.
*/
- public static PipePattern parsePipePatternFromSourceParameters(
+ public static TreePattern parsePipePatternFromSourceParameters(
final PipeParameters sourceParameters) {
final String path = sourceParameters.getStringByKeys(EXTRACTOR_PATH_KEY, SOURCE_PATH_KEY);
// 1. If "source.path" is specified, it will be interpreted as an IoTDB-style path,
// ignoring the other 2 parameters.
if (path != null) {
- return new IoTDBPipePattern(path);
+ return new IoTDBTreePattern(path);
}
final String pattern =
@@ -80,24 +80,24 @@ public abstract class PipePattern {
// If "source.pattern.format" is not specified, use prefix format by default.
if (patternFormat == null) {
- return new PrefixPipePattern(pattern);
+ return new PrefixTreePattern(pattern);
}
switch (patternFormat.toLowerCase()) {
case EXTRACTOR_PATTERN_FORMAT_IOTDB_VALUE:
- return new IoTDBPipePattern(pattern);
+ return new IoTDBTreePattern(pattern);
case EXTRACTOR_PATTERN_FORMAT_PREFIX_VALUE:
- return new PrefixPipePattern(pattern);
+ return new PrefixTreePattern(pattern);
default:
LOGGER.info(
"Unknown pattern format: {}, use prefix matching format by default.", patternFormat);
- return new PrefixPipePattern(pattern);
+ return new PrefixTreePattern(pattern);
}
}
// 3. If neither "source.path" nor "source.pattern" is specified,
// this pipe source will match all data.
- return new IoTDBPipePattern(null);
+ return new IoTDBTreePattern(null);
}
public abstract String getDefaultPattern();
@@ -114,17 +114,17 @@ public abstract class PipePattern {
/**
* Check if a device may have some measurements matched by the pattern.
*
- * <p>NOTE1: this is only called when {@link PipePattern#coversDevice} is false.
+ * <p>NOTE1: this is only called when {@link TreePattern#coversDevice} is false.
*
* <p>NOTE2: this is just a loose check and may have false positives. To further check if a
- * measurement matches the pattern, please use {@link PipePattern#matchesMeasurement} after this.
+ * measurement matches the pattern, please use {@link TreePattern#matchesMeasurement} after this.
*/
public abstract boolean mayOverlapWithDevice(final IDeviceID device);
/**
* Check if a full path with device and measurement can be matched by pattern.
*
- * <p>NOTE: this is only called when {@link PipePattern#mayOverlapWithDevice} is true.
+ * <p>NOTE: this is only called when {@link TreePattern#mayOverlapWithDevice} is true.
*/
public abstract boolean matchesMeasurement(final IDeviceID device, final String measurement);
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/event/EnrichedEvent.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/event/EnrichedEvent.java
index 39c1174f8ef..afed8d5f3bb 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/event/EnrichedEvent.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/event/EnrichedEvent.java
@@ -24,7 +24,8 @@ import org.apache.iotdb.commons.consensus.index.impl.MinimumProgressIndex;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta;
import org.apache.iotdb.commons.pipe.agent.task.progress.CommitterKey;
import org.apache.iotdb.commons.pipe.agent.task.progress.PipeEventCommitManager;
-import org.apache.iotdb.commons.pipe.datastructure.pattern.PipePattern;
+import org.apache.iotdb.commons.pipe.datastructure.pattern.TablePattern;
+import org.apache.iotdb.commons.pipe.datastructure.pattern.TreePattern;
import org.apache.iotdb.pipe.api.event.Event;
import org.slf4j.Logger;
@@ -60,7 +61,8 @@ public abstract class EnrichedEvent implements Event {
protected long commitId = NO_COMMIT_ID;
protected int rebootTimes = 0;
- protected final PipePattern pipePattern;
+ protected final TreePattern treePattern;
+ protected final TablePattern tablePattern;
protected final long startTime;
protected final long endTime;
@@ -75,19 +77,27 @@ public abstract class EnrichedEvent implements Event {
final String pipeName,
final long creationTime,
final PipeTaskMeta pipeTaskMeta,
- final PipePattern pipePattern,
+ final TreePattern treePattern,
+ final TablePattern tablePattern,
final long startTime,
final long endTime) {
referenceCount = new AtomicInteger(0);
isReleased = new AtomicBoolean(false);
+
this.pipeName = pipeName;
this.creationTime = creationTime;
this.pipeTaskMeta = pipeTaskMeta;
- this.pipePattern = pipePattern;
+ this.treePattern = treePattern;
+ this.tablePattern = tablePattern;
this.startTime = startTime;
this.endTime = endTime;
- isPatternParsed = this.pipePattern == null || this.pipePattern.isRoot();
+
+ isPatternParsed =
+ (treePattern == null || treePattern.isRoot())
+ && (tablePattern == null
+ || !tablePattern.hasUserSpecifiedDatabasePatternOrTablePattern());
isTimeParsed = Long.MIN_VALUE == startTime && Long.MAX_VALUE == endTime;
+
addOnCommittedHook(
() -> {
if (shouldReportOnCommit) {
@@ -306,12 +316,17 @@ public abstract class EnrichedEvent implements Event {
*
* @return the pattern string
*/
- public final String getPatternString() {
- return pipePattern != null ? pipePattern.getPattern() : null;
+ // TODO: consider tablePattern
+ public final String getTreePatternString() {
+ return treePattern != null ? treePattern.getPattern() : null;
+ }
+
+ public final TreePattern getTreePattern() {
+ return treePattern;
}
- public final PipePattern getPipePattern() {
- return pipePattern;
+ public final TablePattern getTablePattern() {
+ return tablePattern;
}
public final long getStartTime() {
@@ -350,7 +365,8 @@ public abstract class EnrichedEvent implements Event {
final String pipeName,
final long creationTime,
final PipeTaskMeta pipeTaskMeta,
- final PipePattern pattern,
+ final TreePattern treePattern,
+ final TablePattern tablePattern,
final long startTime,
final long endTime);
@@ -434,8 +450,10 @@ public abstract class EnrichedEvent implements Event {
+ committerKey
+ "', commitId="
+ commitId
- + ", pattern='"
- + pipePattern
+ + ", treePattern='"
+ + treePattern
+ + "', tablePattern='"
+ + tablePattern
+ "', startTime="
+ startTime
+ ", endTime="
@@ -461,8 +479,10 @@ public abstract class EnrichedEvent implements Event {
+ committerKey
+ "', commitId="
+ commitId
- + ", pattern='"
- + pipePattern
+ + ", treePattern='"
+ + treePattern
+ + "', tablePattern='"
+ + tablePattern
+ "', startTime="
+ startTime
+ ", endTime="
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/event/PipeSnapshotEvent.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/event/PipeSnapshotEvent.java
index 1a80d404bb1..94a3d11c668 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/event/PipeSnapshotEvent.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/event/PipeSnapshotEvent.java
@@ -21,7 +21,8 @@ package org.apache.iotdb.commons.pipe.event;
import org.apache.iotdb.commons.consensus.index.ProgressIndex;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta;
-import org.apache.iotdb.commons.pipe.datastructure.pattern.PipePattern;
+import org.apache.iotdb.commons.pipe.datastructure.pattern.TablePattern;
+import org.apache.iotdb.commons.pipe.datastructure.pattern.TreePattern;
import org.apache.iotdb.commons.pipe.resource.snapshot.PipeSnapshotResourceManager;
import java.util.Set;
@@ -37,9 +38,17 @@ public abstract class PipeSnapshotEvent extends EnrichedEvent implements Seriali
final String pipeName,
final long creationTime,
final PipeTaskMeta pipeTaskMeta,
- final PipePattern pattern,
+ final TreePattern treePattern,
+ final TablePattern tablePattern,
final PipeSnapshotResourceManager resourceManager) {
- super(pipeName, creationTime, pipeTaskMeta, pattern, Long.MIN_VALUE, Long.MAX_VALUE);
+ super(
+ pipeName,
+ creationTime,
+ pipeTaskMeta,
+ treePattern,
+ tablePattern,
+ Long.MIN_VALUE,
+ Long.MAX_VALUE);
this.resourceManager = resourceManager;
}
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/event/PipeWritePlanEvent.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/event/PipeWritePlanEvent.java
index 01f3a881ac2..dc3240793ba 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/event/PipeWritePlanEvent.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/event/PipeWritePlanEvent.java
@@ -21,7 +21,8 @@ package org.apache.iotdb.commons.pipe.event;
import org.apache.iotdb.commons.consensus.index.ProgressIndex;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta;
-import org.apache.iotdb.commons.pipe.datastructure.pattern.PipePattern;
+import org.apache.iotdb.commons.pipe.datastructure.pattern.TablePattern;
+import org.apache.iotdb.commons.pipe.datastructure.pattern.TreePattern;
public abstract class PipeWritePlanEvent extends EnrichedEvent implements SerializableEvent {
@@ -33,9 +34,17 @@ public abstract class PipeWritePlanEvent extends EnrichedEvent implements Serial
final String pipeName,
final long creationTime,
final PipeTaskMeta pipeTaskMeta,
- final PipePattern pattern,
+ final TreePattern treePattern,
+ final TablePattern tablePattern,
final boolean isGeneratedByPipe) {
- super(pipeName, creationTime, pipeTaskMeta, pattern, Long.MIN_VALUE, Long.MAX_VALUE);
+ super(
+ pipeName,
+ creationTime,
+ pipeTaskMeta,
+ treePattern,
+ tablePattern,
+ Long.MIN_VALUE,
+ Long.MAX_VALUE);
this.isGeneratedByPipe = isGeneratedByPipe;
}
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/event/ProgressReportEvent.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/event/ProgressReportEvent.java
index 75620ba0759..5a7dbfe7399 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/event/ProgressReportEvent.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/event/ProgressReportEvent.java
@@ -21,7 +21,8 @@ package org.apache.iotdb.commons.pipe.event;
import org.apache.iotdb.commons.consensus.index.ProgressIndex;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta;
-import org.apache.iotdb.commons.pipe.datastructure.pattern.PipePattern;
+import org.apache.iotdb.commons.pipe.datastructure.pattern.TablePattern;
+import org.apache.iotdb.commons.pipe.datastructure.pattern.TreePattern;
/**
* {@link ProgressReportEvent} is an {@link EnrichedEvent} that is used only for progress report. It
@@ -36,10 +37,11 @@ public class ProgressReportEvent extends EnrichedEvent {
final String pipeName,
final long creationTime,
final PipeTaskMeta pipeTaskMeta,
- final PipePattern pipePattern,
+ final TreePattern treePattern,
+ final TablePattern tablePattern,
final long startTime,
final long endTime) {
- super(pipeName, creationTime, pipeTaskMeta, pipePattern, startTime, endTime);
+ super(pipeName, creationTime, pipeTaskMeta, treePattern, tablePattern, startTime, endTime);
}
@Override
@@ -67,11 +69,12 @@ public class ProgressReportEvent extends EnrichedEvent {
final String pipeName,
final long creationTime,
final PipeTaskMeta pipeTaskMeta,
- final PipePattern pattern,
+ final TreePattern treePattern,
+ final TablePattern tablePattern,
final long startTime,
final long endTime) {
return new ProgressReportEvent(
- pipeName, creationTime, pipeTaskMeta, pattern, startTime, endTime);
+ pipeName, creationTime, pipeTaskMeta, treePattern, tablePattern, startTime, endTime);
}
@Override
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/extractor/IoTDBNonDataRegionExtractor.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/extractor/IoTDBNonDataRegionExtractor.java
index 2815ae5e061..1f9f4250fe5 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/extractor/IoTDBNonDataRegionExtractor.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/extractor/IoTDBNonDataRegionExtractor.java
@@ -22,8 +22,9 @@ package org.apache.iotdb.commons.pipe.extractor;
import org.apache.iotdb.commons.consensus.index.ProgressIndex;
import org.apache.iotdb.commons.consensus.index.impl.MetaProgressIndex;
import org.apache.iotdb.commons.consensus.index.impl.MinimumProgressIndex;
-import org.apache.iotdb.commons.pipe.datastructure.pattern.IoTDBPipePattern;
-import org.apache.iotdb.commons.pipe.datastructure.pattern.PipePattern;
+import org.apache.iotdb.commons.pipe.datastructure.pattern.IoTDBTreePattern;
+import org.apache.iotdb.commons.pipe.datastructure.pattern.TablePattern;
+import org.apache.iotdb.commons.pipe.datastructure.pattern.TreePattern;
import org.apache.iotdb.commons.pipe.datastructure.queue.ConcurrentIterableLinkedQueue;
import org.apache.iotdb.commons.pipe.datastructure.queue.listening.AbstractPipeListeningQueue;
import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
@@ -45,7 +46,7 @@ import java.util.concurrent.atomic.AtomicBoolean;
public abstract class IoTDBNonDataRegionExtractor extends IoTDBExtractor {
- protected IoTDBPipePattern pipePattern;
+ protected IoTDBTreePattern pipePattern;
private List<PipeSnapshotEvent> historicalEvents = new LinkedList<>();
// A fixed size initialized only when the historicalEvents are first
@@ -67,16 +68,22 @@ public abstract class IoTDBNonDataRegionExtractor extends IoTDBExtractor {
throws Exception {
super.customize(parameters, configuration);
- final PipePattern pattern = PipePattern.parsePipePatternFromSourceParameters(parameters);
- if (!(pattern instanceof IoTDBPipePattern
- && (((IoTDBPipePattern) pattern).isPrefix()
- || ((IoTDBPipePattern) pattern).isFullPath()))) {
+ if (TablePattern.parsePipePatternFromSourceParameters(parameters)
+ .hasUserSpecifiedDatabasePatternOrTablePattern()) {
+ throw new IllegalArgumentException(
+ "The table model pattern %s can not be specified for the source.");
+ }
+
+ final TreePattern pattern = TreePattern.parsePipePatternFromSourceParameters(parameters);
+ if (!(pattern instanceof IoTDBTreePattern
+ && (((IoTDBTreePattern) pattern).isPrefix()
+ || ((IoTDBTreePattern) pattern).isFullPath()))) {
throw new IllegalArgumentException(
String.format(
"The path pattern %s is not valid for the source. Only prefix or full path is allowed.",
pattern.getPattern()));
}
- pipePattern = (IoTDBPipePattern) pattern;
+ pipePattern = (IoTDBTreePattern) pattern;
}
@Override
@@ -158,6 +165,7 @@ public abstract class IoTDBNonDataRegionExtractor extends IoTDBExtractor {
creationTime,
pipeTaskMeta,
pipePattern,
+ null,
Long.MIN_VALUE,
Long.MAX_VALUE);
@@ -185,7 +193,13 @@ public abstract class IoTDBNonDataRegionExtractor extends IoTDBExtractor {
|| (!isForwardingPipeRequests && realtimeEvent.isGeneratedByPipe())) {
final ProgressReportEvent event =
new ProgressReportEvent(
- pipeName, creationTime, pipeTaskMeta, pipePattern, Long.MIN_VALUE, Long.MAX_VALUE);
+ pipeName,
+ creationTime,
+ pipeTaskMeta,
+ pipePattern,
+ null,
+ Long.MIN_VALUE,
+ Long.MAX_VALUE);
event.bindProgressIndex(new MetaProgressIndex(iterator.getNextIndex() - 1));
event.increaseReferenceCount(IoTDBNonDataRegionExtractor.class.getName());
return event;
@@ -194,7 +208,13 @@ public abstract class IoTDBNonDataRegionExtractor extends IoTDBExtractor {
realtimeEvent =
(PipeWritePlanEvent)
realtimeEvent.shallowCopySelfAndBindPipeTaskMetaForProgressReport(
- pipeName, creationTime, pipeTaskMeta, pipePattern, Long.MIN_VALUE, Long.MAX_VALUE);
+ pipeName,
+ creationTime,
+ pipeTaskMeta,
+ pipePattern,
+ null,
+ Long.MIN_VALUE,
+ Long.MAX_VALUE);
realtimeEvent.bindProgressIndex(new MetaProgressIndex(iterator.getNextIndex() - 1));
realtimeEvent.increaseReferenceCount(IoTDBNonDataRegionExtractor.class.getName());
return realtimeEvent;