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/08 13:09:52 UTC
(iotdb) branch pipe-new-sql created (now 49eeace70ee)
This is an automated email from the ASF dual-hosted git repository.
rong pushed a change to branch pipe-new-sql
in repository https://gitbox.apache.org/repos/asf/iotdb.git
at 49eeace70ee refactor for extractor table pattern
This branch includes the following new commits:
new 49eeace70ee refactor for extractor table pattern
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(iotdb) 01/01: refactor for extractor table pattern
Posted by ro...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rong pushed a commit to branch pipe-new-sql
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 49eeace70eee1a08189edbc5e3e379db23e0cc3b
Author: Steve Yurong Su <ro...@apache.org>
AuthorDate: Tue Oct 8 21:09:39 2024 +0800
refactor for extractor table pattern
---
...FormatIT.java => IoTDBTreePatternFormatIT.java} | 2 +-
.../pipe/event/PipeConfigRegionSnapshotEvent.java | 6 +-
.../pipe/event/PipeConfigRegionWritePlanEvent.java | 6 +-
.../PipeConfigPhysicalPlanPatternParseVisitor.java | 52 ++++++------
.../receiver/protocol/IoTDBConfigNodeReceiver.java | 6 +-
...eConfigPhysicalPlanPatternParseVisitorTest.java | 6 +-
.../agent/task/connection/PipeEventCollector.java | 4 +-
.../db/pipe/event/UserDefinedEnrichedEvent.java | 4 +-
.../event/common/heartbeat/PipeHeartbeatEvent.java | 4 +-
.../schema/PipeSchemaRegionSnapshotEvent.java | 6 +-
.../schema/PipeSchemaRegionWritePlanEvent.java | 6 +-
.../tablet/PipeInsertNodeTabletInsertionEvent.java | 14 ++--
.../common/tablet/PipeRawTabletInsertionEvent.java | 16 ++--
.../tablet/TabletInsertionDataContainer.java | 16 ++--
.../event/common/terminate/PipeTerminateEvent.java | 4 +-
.../common/tsfile/PipeTsFileInsertionEvent.java | 12 +--
.../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 | 8 +-
.../dataregion/IoTDBDataRegionExtractor.java | 38 +++++----
.../PipeHistoricalDataRegionTsFileExtractor.java | 20 +++--
.../realtime/PipeRealtimeDataRegionExtractor.java | 57 +++++++++----
.../realtime/assigner/PipeDataRegionAssigner.java | 5 +-
.../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 | 14 ++--
...ePatternTest.java => IoTDBTreePatternTest.java} | 24 +++---
...PatternTest.java => PrefixTreePatternTest.java} | 24 +++---
.../config/constant/PipeExtractorConstant.java | 7 ++
...IoTDBPipePattern.java => IoTDBTreePattern.java} | 12 +--
...efixPipePattern.java => PrefixTreePattern.java} | 4 +-
.../pipe/datastructure/pattern/TablePattern.java | 96 ++++++++++++++++++++++
.../pattern/{PipePattern.java => TreePattern.java} | 28 +++----
.../iotdb/commons/pipe/event/EnrichedEvent.java | 22 ++---
.../commons/pipe/event/PipeSnapshotEvent.java | 4 +-
.../commons/pipe/event/PipeWritePlanEvent.java | 4 +-
.../commons/pipe/event/ProgressReportEvent.java | 8 +-
.../extractor/IoTDBNonDataRegionExtractor.java | 23 ++++--
48 files changed, 468 insertions(+), 315 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/event/PipeConfigRegionSnapshotEvent.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/event/PipeConfigRegionSnapshotEvent.java
index 8e17cb9b515..76ff5ed2083 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,7 @@
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.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;
@@ -100,7 +100,7 @@ public class PipeConfigRegionSnapshotEvent extends PipeSnapshotEvent {
final String pipeName,
final long creationTime,
final PipeTaskMeta pipeTaskMeta,
- final PipePattern pattern) {
+ final TreePattern pattern) {
super(pipeName, creationTime, pipeTaskMeta, pattern, PipeConfigNodeResourceManager.snapshot());
this.snapshotPath = snapshotPath;
this.templateFilePath = Objects.nonNull(templateFilePath) ? templateFilePath : "";
@@ -160,7 +160,7 @@ public class PipeConfigRegionSnapshotEvent extends PipeSnapshotEvent {
final String pipeName,
final long creationTime,
final PipeTaskMeta pipeTaskMeta,
- final PipePattern pattern,
+ final TreePattern pattern,
final long startTime,
final long endTime) {
return new PipeConfigRegionSnapshotEvent(
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..a6bddf40c0a 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,7 @@
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.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;
@@ -49,7 +49,7 @@ public class PipeConfigRegionWritePlanEvent extends PipeWritePlanEvent {
final String pipeName,
final long creationTime,
final PipeTaskMeta pipeTaskMeta,
- final PipePattern pattern,
+ final TreePattern pattern,
final boolean isGeneratedByPipe) {
super(pipeName, creationTime, pipeTaskMeta, pattern, isGeneratedByPipe);
this.configPhysicalPlan = configPhysicalPlan;
@@ -64,7 +64,7 @@ public class PipeConfigRegionWritePlanEvent extends PipeWritePlanEvent {
final String pipeName,
final long creationTime,
final PipeTaskMeta pipeTaskMeta,
- final PipePattern pattern,
+ final TreePattern pattern,
final long startTime,
final long endTime) {
return new PipeConfigRegionWritePlanEvent(
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..f9341b95ec4 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.getPipePattern())
.map(
planNode ->
new PipeSchemaRegionWritePlanEvent(
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..23ea2c561e4 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,7 @@ 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.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;
@@ -76,7 +76,7 @@ public class UserDefinedEnrichedEvent extends EnrichedEvent {
final String pipeName,
final long creationTime,
final PipeTaskMeta pipeTaskMeta,
- final PipePattern pattern,
+ final TreePattern pattern,
final long startTime,
final long endTime) {
return enrichedEvent.shallowCopySelfAndBindPipeTaskMetaForProgressReport(
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..de40212b656 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,7 @@ 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.TreePattern;
import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
import org.apache.iotdb.db.pipe.metric.PipeHeartbeatEventMetrics;
import org.apache.iotdb.db.utils.DateTimeUtils;
@@ -102,7 +102,7 @@ public class PipeHeartbeatEvent extends EnrichedEvent {
final String pipeName,
final long creationTime,
final PipeTaskMeta pipeTaskMeta,
- final PipePattern pattern,
+ final TreePattern pattern,
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..63152ea8032 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,7 @@
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.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;
@@ -88,7 +88,7 @@ public class PipeSchemaRegionSnapshotEvent extends PipeSnapshotEvent {
final String pipeName,
final long creationTime,
final PipeTaskMeta pipeTaskMeta,
- final PipePattern pattern) {
+ final TreePattern pattern) {
super(pipeName, creationTime, pipeTaskMeta, pattern, PipeDataNodeResourceManager.snapshot());
this.mTreeSnapshotPath = mTreeSnapshotPath;
this.tagLogSnapshotPath = Objects.nonNull(tagLogSnapshotPath) ? tagLogSnapshotPath : "";
@@ -148,7 +148,7 @@ public class PipeSchemaRegionSnapshotEvent extends PipeSnapshotEvent {
final String pipeName,
final long creationTime,
final PipeTaskMeta pipeTaskMeta,
- final PipePattern pattern,
+ final TreePattern pattern,
final long startTime,
final long endTime) {
return new PipeSchemaRegionSnapshotEvent(
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..0014c2038f6 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,7 @@
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.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;
@@ -48,7 +48,7 @@ public class PipeSchemaRegionWritePlanEvent extends PipeWritePlanEvent {
final String pipeName,
final long creationTime,
final PipeTaskMeta pipeTaskMeta,
- final PipePattern pattern,
+ final TreePattern pattern,
final boolean isGeneratedByPipe) {
super(pipeName, creationTime, pipeTaskMeta, pattern, isGeneratedByPipe);
this.planNode = planNode;
@@ -63,7 +63,7 @@ public class PipeSchemaRegionWritePlanEvent extends PipeWritePlanEvent {
final String pipeName,
final long creationTime,
final PipeTaskMeta pipeTaskMeta,
- final PipePattern pattern,
+ final TreePattern pattern,
final long startTime,
final long endTime) {
return new PipeSchemaRegionWritePlanEvent(
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..97a35caa1b4 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,7 @@ 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.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;
@@ -95,7 +95,7 @@ public class PipeInsertNodeTabletInsertionEvent extends EnrichedEvent
final String pipeName,
final long creationTime,
final PipeTaskMeta pipeTaskMeta,
- final PipePattern pattern,
+ final TreePattern pattern,
final long startTime,
final long endTime) {
super(pipeName, creationTime, pipeTaskMeta, pattern, startTime, endTime);
@@ -179,7 +179,7 @@ public class PipeInsertNodeTabletInsertionEvent extends EnrichedEvent
final String pipeName,
final long creationTime,
final PipeTaskMeta pipeTaskMeta,
- final PipePattern pattern,
+ final TreePattern pattern,
final long startTime,
final long endTime) {
return new PipeInsertNodeTabletInsertionEvent(
@@ -256,7 +256,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 +265,7 @@ public class PipeInsertNodeTabletInsertionEvent extends EnrichedEvent
.anyMatch(
insertRowNode ->
Objects.isNull(insertRowNode.getTargetPath())
- || pipePattern.mayOverlapWithDevice(
+ || treePattern.mayOverlapWithDevice(
insertRowNode.getTargetPath().getIDeviceIDAsFullDevice()));
}
@@ -328,12 +328,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..f08ccb6d450 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,7 @@ 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.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,7 +62,7 @@ public class PipeRawTabletInsertionEvent extends EnrichedEvent implements Tablet
final String pipeName,
final long creationTime,
final PipeTaskMeta pipeTaskMeta,
- final PipePattern pattern,
+ final TreePattern pattern,
final long startTime,
final long endTime) {
super(pipeName, creationTime, pipeTaskMeta, pattern, startTime, endTime);
@@ -100,7 +100,7 @@ public class PipeRawTabletInsertionEvent extends EnrichedEvent implements Tablet
@TestOnly
public PipeRawTabletInsertionEvent(
- final Tablet tablet, final boolean isAligned, final PipePattern pattern) {
+ final Tablet tablet, final boolean isAligned, final TreePattern pattern) {
this(tablet, isAligned, null, false, null, 0, null, pattern, Long.MIN_VALUE, Long.MAX_VALUE);
}
@@ -169,7 +169,7 @@ public class PipeRawTabletInsertionEvent extends EnrichedEvent implements Tablet
final String pipeName,
final long creationTime,
final PipeTaskMeta pipeTaskMeta,
- final PipePattern pattern,
+ final TreePattern pattern,
final long startTime,
final long endTime) {
return new PipeRawTabletInsertionEvent(
@@ -204,7 +204,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 +227,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 +237,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 +256,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..28c4165a7e7 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,7 @@ 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.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;
@@ -66,7 +66,7 @@ public class PipeTerminateEvent extends EnrichedEvent {
final String pipeName,
final long creationTime,
final PipeTaskMeta pipeTaskMeta,
- final PipePattern pattern,
+ final TreePattern pattern,
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..d84a1bbdcae 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,7 @@ 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.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;
@@ -103,7 +103,7 @@ public class PipeTsFileInsertionEvent extends EnrichedEvent implements TsFileIns
final String pipeName,
final long creationTime,
final PipeTaskMeta pipeTaskMeta,
- final PipePattern pattern,
+ final TreePattern pattern,
final long startTime,
final long endTime) {
super(pipeName, creationTime, pipeTaskMeta, pattern, startTime, endTime);
@@ -320,7 +320,7 @@ public class PipeTsFileInsertionEvent extends EnrichedEvent implements TsFileIns
final String pipeName,
final long creationTime,
final PipeTaskMeta pipeTaskMeta,
- final PipePattern pattern,
+ final TreePattern pattern,
final long startTime,
final long endTime) {
return new PipeTsFileInsertionEvent(
@@ -365,7 +365,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 +413,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 +445,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..b03e7caf9fc 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,7 @@ 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.TreePattern;
import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
import org.apache.iotdb.db.pipe.extractor.dataregion.realtime.epoch.TsFileEpoch;
@@ -45,7 +45,7 @@ public class PipeRealtimeEvent extends EnrichedEvent {
final EnrichedEvent event,
final TsFileEpoch tsFileEpoch,
final Map<IDeviceID, String[]> device2Measurements,
- final PipePattern pattern) {
+ final TreePattern pattern) {
this(event, tsFileEpoch, device2Measurements, null, pattern, Long.MIN_VALUE, Long.MAX_VALUE);
}
@@ -54,7 +54,7 @@ public class PipeRealtimeEvent extends EnrichedEvent {
final TsFileEpoch tsFileEpoch,
final Map<IDeviceID, String[]> device2Measurements,
final PipeTaskMeta pipeTaskMeta,
- final PipePattern pattern,
+ final TreePattern pattern,
final long startTime,
final long endTime) {
// PipeTaskMeta is used to report the progress of the event, the PipeRealtimeEvent
@@ -166,7 +166,7 @@ public class PipeRealtimeEvent extends EnrichedEvent {
final String pipeName,
final long creationTime,
final PipeTaskMeta pipeTaskMeta,
- final PipePattern pattern,
+ final TreePattern pattern,
final long startTime,
final long endTime) {
return new PipeRealtimeEvent(
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..82bb5e2f5e2 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,
+ // TODO: consider 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..e80101933d8 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
@@ -105,12 +105,13 @@ public class PipeDataRegionAssigner implements Closeable {
counter = 0;
}
+ // TODO: consider table pattern
final ProgressReportEvent reportEvent =
new ProgressReportEvent(
extractor.getPipeName(),
extractor.getCreationTime(),
extractor.getPipeTaskMeta(),
- extractor.getPipePattern(),
+ extractor.getTreePattern(),
extractor.getRealtimeDataExtractionStartTime(),
extractor.getRealtimeDataExtractionEndTime());
reportEvent.bindProgressIndex(event.getProgressIndex());
@@ -129,7 +130,7 @@ public class PipeDataRegionAssigner implements Closeable {
extractor.getPipeName(),
extractor.getCreationTime(),
extractor.getPipeTaskMeta(),
- extractor.getPipePattern(),
+ extractor.getTreePattern(),
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/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..89c96595253 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,7 +59,7 @@ public class CachedSchemaPatternMatcherTest {
EnrichedEvent event,
TsFileEpoch tsFileEpoch,
Map<IDeviceID, String[]> device2Measurements,
- PipePattern pattern) {
+ TreePattern pattern) {
super(event, tsFileEpoch, device2Measurements, pattern);
}
@@ -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..1615e477e09 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,13 @@ 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_CAPTURE_TABLE_DATABASE_KEY =
+ "extractor.capture.table.database-name";
+ public static final String SOURCE_CAPTURE_TABLE_DATABASE_KEY =
+ "source.capture.table.database-name";
+ public static final String EXTRACTOR_CAPTURE_TABLE_TABLE_KEY =
+ "extractor.capture.table.table-name";
+ public static final String SOURCE_CAPTURE_TABLE_TABLE_KEY = "source.capture.table.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..90220e9be52
--- /dev/null
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/pattern/TablePattern.java
@@ -0,0 +1,96 @@
+/*
+ * 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_CAPTURE_TABLE_DATABASE_KEY;
+import static org.apache.iotdb.commons.pipe.config.constant.PipeExtractorConstant.EXTRACTOR_CAPTURE_TABLE_TABLE_KEY;
+import static org.apache.iotdb.commons.pipe.config.constant.PipeExtractorConstant.SOURCE_CAPTURE_TABLE_DATABASE_KEY;
+import static org.apache.iotdb.commons.pipe.config.constant.PipeExtractorConstant.SOURCE_CAPTURE_TABLE_TABLE_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_CAPTURE_TABLE_DATABASE_KEY, SOURCE_CAPTURE_TABLE_DATABASE_KEY);
+ final String tableNamePattern =
+ sourceParameters.getStringByKeys(
+ EXTRACTOR_CAPTURE_TABLE_TABLE_KEY, SOURCE_CAPTURE_TABLE_TABLE_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..da427705714 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,7 @@ 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.TreePattern;
import org.apache.iotdb.pipe.api.event.Event;
import org.slf4j.Logger;
@@ -60,7 +60,7 @@ 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 long startTime;
protected final long endTime;
@@ -75,7 +75,7 @@ public abstract class EnrichedEvent implements Event {
final String pipeName,
final long creationTime,
final PipeTaskMeta pipeTaskMeta,
- final PipePattern pipePattern,
+ final TreePattern treePattern,
final long startTime,
final long endTime) {
referenceCount = new AtomicInteger(0);
@@ -83,10 +83,10 @@ public abstract class EnrichedEvent implements Event {
this.pipeName = pipeName;
this.creationTime = creationTime;
this.pipeTaskMeta = pipeTaskMeta;
- this.pipePattern = pipePattern;
+ this.treePattern = treePattern;
this.startTime = startTime;
this.endTime = endTime;
- isPatternParsed = this.pipePattern == null || this.pipePattern.isRoot();
+ isPatternParsed = this.treePattern == null || this.treePattern.isRoot();
isTimeParsed = Long.MIN_VALUE == startTime && Long.MAX_VALUE == endTime;
addOnCommittedHook(
() -> {
@@ -307,11 +307,11 @@ public abstract class EnrichedEvent implements Event {
* @return the pattern string
*/
public final String getPatternString() {
- return pipePattern != null ? pipePattern.getPattern() : null;
+ return treePattern != null ? treePattern.getPattern() : null;
}
- public final PipePattern getPipePattern() {
- return pipePattern;
+ public final TreePattern getPipePattern() {
+ return treePattern;
}
public final long getStartTime() {
@@ -350,7 +350,7 @@ public abstract class EnrichedEvent implements Event {
final String pipeName,
final long creationTime,
final PipeTaskMeta pipeTaskMeta,
- final PipePattern pattern,
+ final TreePattern pattern,
final long startTime,
final long endTime);
@@ -435,7 +435,7 @@ public abstract class EnrichedEvent implements Event {
+ "', commitId="
+ commitId
+ ", pattern='"
- + pipePattern
+ + treePattern
+ "', startTime="
+ startTime
+ ", endTime="
@@ -462,7 +462,7 @@ public abstract class EnrichedEvent implements Event {
+ "', commitId="
+ commitId
+ ", pattern='"
- + pipePattern
+ + treePattern
+ "', 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..f94e96588d7 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,7 @@ 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.TreePattern;
import org.apache.iotdb.commons.pipe.resource.snapshot.PipeSnapshotResourceManager;
import java.util.Set;
@@ -37,7 +37,7 @@ public abstract class PipeSnapshotEvent extends EnrichedEvent implements Seriali
final String pipeName,
final long creationTime,
final PipeTaskMeta pipeTaskMeta,
- final PipePattern pattern,
+ final TreePattern pattern,
final PipeSnapshotResourceManager resourceManager) {
super(pipeName, creationTime, pipeTaskMeta, pattern, 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..220974bc71a 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,7 @@ 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.TreePattern;
public abstract class PipeWritePlanEvent extends EnrichedEvent implements SerializableEvent {
@@ -33,7 +33,7 @@ public abstract class PipeWritePlanEvent extends EnrichedEvent implements Serial
final String pipeName,
final long creationTime,
final PipeTaskMeta pipeTaskMeta,
- final PipePattern pattern,
+ final TreePattern pattern,
final boolean isGeneratedByPipe) {
super(pipeName, creationTime, pipeTaskMeta, pattern, 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..97c62cfaaf6 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,7 @@ 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.TreePattern;
/**
* {@link ProgressReportEvent} is an {@link EnrichedEvent} that is used only for progress report. It
@@ -36,10 +36,10 @@ public class ProgressReportEvent extends EnrichedEvent {
final String pipeName,
final long creationTime,
final PipeTaskMeta pipeTaskMeta,
- final PipePattern pipePattern,
+ final TreePattern treePattern,
final long startTime,
final long endTime) {
- super(pipeName, creationTime, pipeTaskMeta, pipePattern, startTime, endTime);
+ super(pipeName, creationTime, pipeTaskMeta, treePattern, startTime, endTime);
}
@Override
@@ -67,7 +67,7 @@ public class ProgressReportEvent extends EnrichedEvent {
final String pipeName,
final long creationTime,
final PipeTaskMeta pipeTaskMeta,
- final PipePattern pattern,
+ final TreePattern pattern,
final long startTime,
final long endTime) {
return new ProgressReportEvent(
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..ca58ceedf83 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