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/05/21 14:07:33 UTC
(iotdb) branch master updated: Pipe: Support parsing pattern when tranferring schema plans (#12464)
This is an automated email from the ASF dual-hosted git repository.
rong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 49dc7650e80 Pipe: Support parsing pattern when tranferring schema plans (#12464)
49dc7650e80 is described below
commit 49dc7650e8021b33baafdbd9ace17bf66dc88da6
Author: Caideyipi <87...@users.noreply.github.com>
AuthorDate: Tue May 21 22:07:26 2024 +0800
Pipe: Support parsing pattern when tranferring schema plans (#12464)
Co-authored-by: Steve Yurong Su <ro...@apache.org>
---
.../pipe/it/autocreate/IoTDBPipeExtractorIT.java | 10 +-
.../iotdb/pipe/it/manual/IoTDBPipeInclusionIT.java | 49 +++
.../pipe/it/manual/IoTDBPipeMetaHistoricalIT.java | 4 +
.../consensus/request/ConfigPhysicalPlan.java | 4 +
.../consensus/request/ConfigPhysicalPlanType.java | 1 +
.../consensus/request/auth/AuthorPlan.java | 21 +-
.../pipe/payload/PipeDeleteLogicalViewPlan.java | 11 +-
.../pipe/payload/PipeDeleteTimeSeriesPlan.java | 11 +-
.../request/write/pipe/payload/PipeSetTTLPlan.java | 103 +++++
.../payload/PipeTransferConfigSnapshotSealReq.java | 33 +-
.../protocol/IoTDBConfigRegionAirGapConnector.java | 2 +
.../protocol/IoTDBConfigRegionConnector.java | 2 +
.../pipe/extractor/IoTDBConfigRegionExtractor.java | 18 +-
.../PipeConfigPhysicalPlanPatternParseVisitor.java | 283 ++++++++++++++
.../receiver/protocol/IoTDBConfigNodeReceiver.java | 42 +-
.../PipeConfigPhysicalPlanTSStatusVisitor.java | 55 +--
.../thrift/ConfigNodeRPCServiceProcessor.java | 65 ++--
.../request/ConfigPhysicalPlanSerDeTest.java | 111 +++---
.../connector/PipeConfigNodeThriftRequestTest.java | 2 +-
.../extractor/IoTDBConfigRegionExtractorTest.java | 7 +-
...eConfigPhysicalPlanPatternParseVisitorTest.java | 408 ++++++++++++++++++++
.../request/PipeTransferSchemaSnapshotSealReq.java | 32 +-
.../airgap/IoTDBSchemaRegionAirGapConnector.java | 2 +
.../thrift/sync/IoTDBSchemaRegionConnector.java | 2 +
.../schema/PipeSchemaRegionWritePlanEvent.java | 4 +-
.../tablet/TabletInsertionDataContainer.java | 32 +-
.../tsfile/TsFileInsertionDataContainer.java | 28 +-
.../common/tsfile/TsFileInsertionPointCounter.java | 10 +-
.../dataregion/IoTDBDataRegionExtractor.java | 13 +
.../schemaregion/IoTDBSchemaRegionExtractor.java | 15 +-
.../schemaregion/PipePlanPatternParseVisitor.java | 282 ++++++++++++++
.../pipe/pattern/CachedSchemaPatternMatcher.java | 10 +-
.../protocol/thrift/IoTDBDataNodeReceiver.java | 13 +-
.../visitor/PipePlanToStatementVisitor.java | 2 +-
.../visitor/PipeStatementPatternParseVisitor.java | 136 +++++++
.../pipe/task/connection/PipeEventCollector.java | 19 +
.../planner/plan/node/write/DeleteDataNode.java | 21 +
.../metadata/CreateAlignedTimeSeriesStatement.java | 55 +++
.../connector/PipeDataNodeThriftRequestTest.java | 2 +-
.../PipeStatementPatternParseVisitorTest.java | 161 ++++++++
.../extractor/IoTDBDataRegionExtractorTest.java | 18 +-
.../extractor/PipePlanPatternParseVisitorTest.java | 428 +++++++++++++++++++++
.../db/pipe/pattern/IoTDBPipePatternTest.java | 46 +--
.../db/pipe/pattern/PrefixPipePatternTest.java | 46 +--
.../extractor/IoTDBNonDataRegionExtractor.java | 48 ++-
.../commons/pipe/pattern/IoTDBPipePattern.java | 99 ++++-
.../iotdb/commons/pipe/pattern/PipePattern.java | 17 +-
.../commons/pipe/pattern/PrefixPipePattern.java | 20 +-
48 files changed, 2487 insertions(+), 316 deletions(-)
diff --git a/integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBPipeExtractorIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBPipeExtractorIT.java
index 541d628aa7e..28852463a77 100644
--- a/integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBPipeExtractorIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBPipeExtractorIT.java
@@ -411,7 +411,7 @@ public class IoTDBPipeExtractorIT extends AbstractPipeDualAutoIT {
final Map<String, String> connectorAttributes = new HashMap<>();
extractorAttributes.put("extractor.pattern", null);
- extractorAttributes.put("extractor.inclusion", "data");
+ extractorAttributes.put("extractor.inclusion", "data.insert");
connectorAttributes.put("connector", "iotdb-thrift-connector");
connectorAttributes.put("connector.batch.enable", "false");
@@ -491,7 +491,7 @@ public class IoTDBPipeExtractorIT extends AbstractPipeDualAutoIT {
final Map<String, String> connectorAttributes = new HashMap<>();
extractorAttributes.put("extractor.pattern", "root.db1");
- extractorAttributes.put("extractor.inclusion", "data");
+ extractorAttributes.put("extractor.inclusion", "data.insert");
connectorAttributes.put("connector", "iotdb-thrift-connector");
connectorAttributes.put("connector.batch.enable", "false");
@@ -589,7 +589,7 @@ public class IoTDBPipeExtractorIT extends AbstractPipeDualAutoIT {
connectorAttributes.put("connector.ip", receiverIp);
connectorAttributes.put("connector.port", Integer.toString(receiverPort));
- extractorAttributes.put("extractor.inclusion", "data");
+ extractorAttributes.put("extractor.inclusion", "data.insert");
extractorAttributes.put("extractor.pattern", "root.db.d2");
extractorAttributes.put("extractor.history.enable", "false");
extractorAttributes.put("extractor.realtime.enable", "true");
@@ -675,7 +675,7 @@ public class IoTDBPipeExtractorIT extends AbstractPipeDualAutoIT {
final Map<String, String> connectorAttributes = new HashMap<>();
extractorAttributes.put("extractor.pattern", "root.db.d1");
- extractorAttributes.put("extractor.inclusion", "data");
+ extractorAttributes.put("extractor.inclusion", "data.insert");
extractorAttributes.put("extractor.history.enable", "true");
// 1970-01-01T08:00:02+08:00
extractorAttributes.put("extractor.history.start-time", "2000");
@@ -825,7 +825,7 @@ public class IoTDBPipeExtractorIT extends AbstractPipeDualAutoIT {
final Map<String, String> connectorAttributes = new HashMap<>();
extractorAttributes.put("source.pattern", "root.db.d1");
- extractorAttributes.put("source.inclusion", "data");
+ extractorAttributes.put("source.inclusion", "data.insert");
extractorAttributes.put("source.start-time", "1970-01-01T08:00:02+08:00");
// 1970-01-01T08:00:04+08:00
extractorAttributes.put("source.end-time", "4000");
diff --git a/integration-test/src/test/java/org/apache/iotdb/pipe/it/manual/IoTDBPipeInclusionIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/manual/IoTDBPipeInclusionIT.java
index c27fa7f46c4..7a8ff6c3f15 100644
--- a/integration-test/src/test/java/org/apache/iotdb/pipe/it/manual/IoTDBPipeInclusionIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/pipe/it/manual/IoTDBPipeInclusionIT.java
@@ -36,6 +36,7 @@ import org.junit.runner.RunWith;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.Map;
@RunWith(IoTDBTestRunner.class)
@@ -144,6 +145,54 @@ public class IoTDBPipeInclusionIT extends AbstractPipeDualManualIT {
}
}
+ @Test
+ public void testAuthInclusionWithPattern() throws Exception {
+ final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+
+ final String receiverIp = receiverDataNode.getIp();
+ final int receiverPort = receiverDataNode.getPort();
+
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+ final Map<String, String> extractorAttributes = new HashMap<>();
+ final Map<String, String> processorAttributes = new HashMap<>();
+ final Map<String, String> connectorAttributes = new HashMap<>();
+
+ extractorAttributes.put("extractor.inclusion", "auth");
+ extractorAttributes.put("path", "root.ln.**");
+
+ connectorAttributes.put("connector", "iotdb-thrift-connector");
+ connectorAttributes.put("connector.ip", receiverIp);
+ connectorAttributes.put("connector.port", Integer.toString(receiverPort));
+
+ final TSStatus status =
+ client.createPipe(
+ new TCreatePipeReq("testPipe", connectorAttributes)
+ .setExtractorAttributes(extractorAttributes)
+ .setProcessorAttributes(processorAttributes));
+
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
+
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("testPipe").getCode());
+
+ if (!TestUtils.tryExecuteNonQueriesWithRetry(
+ senderEnv,
+ Arrays.asList(
+ "create user `ln_write_user` 'write_pwd'",
+ "GRANT READ_DATA, WRITE_DATA ON root.** TO USER ln_write_user;"))) {
+ return;
+ }
+
+ TestUtils.assertDataAlwaysOnEnv(
+ receiverEnv,
+ "LIST PRIVILEGES OF USER ln_write_user",
+ "ROLE,PATH,PRIVILEGES,GRANT OPTION,",
+ new HashSet<>(
+ Arrays.asList(",root.ln.**,READ_DATA,false,", ",root.ln.**,WRITE_DATA,false,")));
+ }
+ }
+
@Test
public void testPureDeleteInclusion() throws Exception {
final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
diff --git a/integration-test/src/test/java/org/apache/iotdb/pipe/it/manual/IoTDBPipeMetaHistoricalIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/manual/IoTDBPipeMetaHistoricalIT.java
index 89662c19f63..70316d3b600 100644
--- a/integration-test/src/test/java/org/apache/iotdb/pipe/it/manual/IoTDBPipeMetaHistoricalIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/pipe/it/manual/IoTDBPipeMetaHistoricalIT.java
@@ -97,6 +97,7 @@ public class IoTDBPipeMetaHistoricalIT extends AbstractPipeDualManualIT {
senderEnv,
Arrays.asList(
"create database root.ln",
+ "create database root.db",
"set ttl to root.ln 3600000",
"create user `thulab` 'passwd'",
"create role `admin`",
@@ -104,7 +105,9 @@ public class IoTDBPipeMetaHistoricalIT extends AbstractPipeDualManualIT {
"grant read on root.** to role `admin`",
"create schema template t1 (temperature FLOAT encoding=RLE, status BOOLEAN encoding=PLAIN compression=SNAPPY)",
"set schema template t1 to root.ln.wf01",
+ "set schema template t1 to root.db.wf01",
"create timeseries using schema template on root.ln.wf01.wt01",
+ "create timeseries using schema template on root.db.wf01.wt01",
"create timeseries root.ln.wf02.wt01.status with datatype=BOOLEAN,encoding=PLAIN",
// Insert large timestamp to avoid deletion by ttl
"insert into root.ln.wf01.wt01(time, temperature, status) values (1800000000000, 23, true)"))) {
@@ -118,6 +121,7 @@ public class IoTDBPipeMetaHistoricalIT extends AbstractPipeDualManualIT {
extractorAttributes.put("extractor.inclusion", "data, schema");
extractorAttributes.put(
"extractor.inclusion.exclusion", "schema.timeseries.ordinary, schema.ttl");
+ extractorAttributes.put("extractor.path", "root.ln.**");
connectorAttributes.put("connector", "iotdb-thrift-connector");
connectorAttributes.put("connector.ip", receiverIp);
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlan.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlan.java
index 8ad8c9b2fd2..3c93330ae11 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlan.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlan.java
@@ -82,6 +82,7 @@ import org.apache.iotdb.confignode.consensus.request.write.pipe.payload.PipeDeac
import org.apache.iotdb.confignode.consensus.request.write.pipe.payload.PipeDeleteLogicalViewPlan;
import org.apache.iotdb.confignode.consensus.request.write.pipe.payload.PipeDeleteTimeSeriesPlan;
import org.apache.iotdb.confignode.consensus.request.write.pipe.payload.PipeEnrichedPlan;
+import org.apache.iotdb.confignode.consensus.request.write.pipe.payload.PipeSetTTLPlan;
import org.apache.iotdb.confignode.consensus.request.write.pipe.payload.PipeUnsetSchemaTemplatePlan;
import org.apache.iotdb.confignode.consensus.request.write.pipe.plugin.CreatePipePluginPlan;
import org.apache.iotdb.confignode.consensus.request.write.pipe.plugin.DropPipePluginPlan;
@@ -492,6 +493,9 @@ public abstract class ConfigPhysicalPlan implements IConsensusRequest {
case PipeDeactivateTemplate:
plan = new PipeDeactivateTemplatePlan();
break;
+ case PipeSetTTL:
+ plan = new PipeSetTTLPlan();
+ break;
case GetRegionId:
plan = new GetRegionIdPlan();
break;
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlanType.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlanType.java
index 031518be944..1318c2165b4 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlanType.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlanType.java
@@ -213,6 +213,7 @@ public enum ConfigPhysicalPlanType {
PipeDeleteTimeSeries((short) 1702),
PipeDeleteLogicalView((short) 1703),
PipeDeactivateTemplate((short) 1704),
+ PipeSetTTL((short) 1705),
/** Subscription */
CreateTopic((short) 1800),
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/auth/AuthorPlan.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/auth/AuthorPlan.java
index 0204d013e06..be162647f8f 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/auth/AuthorPlan.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/auth/AuthorPlan.java
@@ -19,7 +19,6 @@
package org.apache.iotdb.confignode.consensus.request.auth;
-import org.apache.iotdb.commons.auth.AuthException;
import org.apache.iotdb.commons.auth.entity.PrivilegeType;
import org.apache.iotdb.commons.exception.MetadataException;
import org.apache.iotdb.commons.path.PartialPath;
@@ -51,7 +50,7 @@ public class AuthorPlan extends ConfigPhysicalPlan {
private String userName;
private boolean grantOpt;
- public AuthorPlan(ConfigPhysicalPlanType type) {
+ public AuthorPlan(final ConfigPhysicalPlanType type) {
super(type);
authorType = type;
}
@@ -67,18 +66,16 @@ public class AuthorPlan extends ConfigPhysicalPlan {
* @param permissions permissions
* @param grantOpt with grant option, only grant statement can set grantOpt = true
* @param nodeNameList node name in Path structure
- * @throws AuthException Authentication Exception
*/
public AuthorPlan(
- ConfigPhysicalPlanType authorType,
- String userName,
- String roleName,
- String password,
- String newPassword,
- Set<Integer> permissions,
- boolean grantOpt,
- List<PartialPath> nodeNameList)
- throws AuthException {
+ final ConfigPhysicalPlanType authorType,
+ final String userName,
+ final String roleName,
+ final String password,
+ final String newPassword,
+ final Set<Integer> permissions,
+ final boolean grantOpt,
+ final List<PartialPath> nodeNameList) {
this(authorType);
this.authorType = authorType;
this.userName = userName;
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/pipe/payload/PipeDeleteLogicalViewPlan.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/pipe/payload/PipeDeleteLogicalViewPlan.java
index 0db5952175a..6865b135c7b 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/pipe/payload/PipeDeleteLogicalViewPlan.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/pipe/payload/PipeDeleteLogicalViewPlan.java
@@ -38,9 +38,8 @@ public class PipeDeleteLogicalViewPlan extends ConfigPhysicalPlan {
super(ConfigPhysicalPlanType.PipeDeleteLogicalView);
}
- public PipeDeleteLogicalViewPlan(ByteBuffer patternTreeBytes) {
+ public PipeDeleteLogicalViewPlan(final ByteBuffer patternTreeBytes) {
super(ConfigPhysicalPlanType.PipeDeleteLogicalView);
- patternTreeBytes.flip();
this.patternTreeBytes = patternTreeBytes;
}
@@ -49,25 +48,25 @@ public class PipeDeleteLogicalViewPlan extends ConfigPhysicalPlan {
}
@Override
- protected void serializeImpl(DataOutputStream stream) throws IOException {
+ protected void serializeImpl(final DataOutputStream stream) throws IOException {
stream.writeShort(getType().getPlanType());
ReadWriteIOUtils.write(patternTreeBytes, stream);
}
@Override
- protected void deserializeImpl(ByteBuffer buffer) throws IOException {
+ protected void deserializeImpl(final ByteBuffer buffer) throws IOException {
patternTreeBytes = ByteBuffer.wrap(ReadWriteIOUtils.readBinary(buffer).getValues());
}
@Override
- public boolean equals(Object obj) {
+ public boolean equals(final Object obj) {
if (this == obj) {
return true;
}
if (obj == null || getClass() != obj.getClass()) {
return false;
}
- PipeDeleteLogicalViewPlan that = (PipeDeleteLogicalViewPlan) obj;
+ final PipeDeleteLogicalViewPlan that = (PipeDeleteLogicalViewPlan) obj;
return Arrays.equals(patternTreeBytes.array(), that.patternTreeBytes.array());
}
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/pipe/payload/PipeDeleteTimeSeriesPlan.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/pipe/payload/PipeDeleteTimeSeriesPlan.java
index d97346f0155..2b651cb034d 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/pipe/payload/PipeDeleteTimeSeriesPlan.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/pipe/payload/PipeDeleteTimeSeriesPlan.java
@@ -38,9 +38,8 @@ public class PipeDeleteTimeSeriesPlan extends ConfigPhysicalPlan {
super(ConfigPhysicalPlanType.PipeDeleteTimeSeries);
}
- public PipeDeleteTimeSeriesPlan(ByteBuffer patternTreeBytes) {
+ public PipeDeleteTimeSeriesPlan(final ByteBuffer patternTreeBytes) {
super(ConfigPhysicalPlanType.PipeDeleteTimeSeries);
- patternTreeBytes.flip();
this.patternTreeBytes = patternTreeBytes;
}
@@ -49,25 +48,25 @@ public class PipeDeleteTimeSeriesPlan extends ConfigPhysicalPlan {
}
@Override
- protected void serializeImpl(DataOutputStream stream) throws IOException {
+ protected void serializeImpl(final DataOutputStream stream) throws IOException {
stream.writeShort(getType().getPlanType());
ReadWriteIOUtils.write(patternTreeBytes, stream);
}
@Override
- protected void deserializeImpl(ByteBuffer buffer) throws IOException {
+ protected void deserializeImpl(final ByteBuffer buffer) throws IOException {
patternTreeBytes = ByteBuffer.wrap(ReadWriteIOUtils.readBinary(buffer).getValues());
}
@Override
- public boolean equals(Object obj) {
+ public boolean equals(final Object obj) {
if (this == obj) {
return true;
}
if (obj == null || getClass() != obj.getClass()) {
return false;
}
- PipeDeleteTimeSeriesPlan that = (PipeDeleteTimeSeriesPlan) obj;
+ final PipeDeleteTimeSeriesPlan that = (PipeDeleteTimeSeriesPlan) obj;
return Arrays.equals(patternTreeBytes.array(), that.patternTreeBytes.array());
}
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/pipe/payload/PipeSetTTLPlan.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/pipe/payload/PipeSetTTLPlan.java
new file mode 100644
index 00000000000..67ab85d898c
--- /dev/null
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/pipe/payload/PipeSetTTLPlan.java
@@ -0,0 +1,103 @@
+/*
+ * 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.confignode.consensus.request.write.pipe.payload;
+
+import org.apache.iotdb.commons.pipe.extractor.IoTDBNonDataRegionExtractor;
+import org.apache.iotdb.commons.pipe.pattern.IoTDBPipePattern;
+import org.apache.iotdb.confignode.consensus.request.ConfigPhysicalPlan;
+import org.apache.iotdb.confignode.consensus.request.ConfigPhysicalPlanType;
+import org.apache.iotdb.confignode.consensus.request.write.database.SetTTLPlan;
+import org.apache.iotdb.confignode.manager.pipe.event.PipeConfigRegionWritePlanEvent;
+import org.apache.iotdb.confignode.manager.pipe.extractor.ConfigRegionListeningQueue;
+import org.apache.iotdb.confignode.manager.pipe.receiver.protocol.IoTDBConfigNodeReceiver;
+
+import org.apache.tsfile.utils.ReadWriteIOUtils;
+
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Objects;
+
+/**
+ * This is a set of {@link SetTTLPlan}s generated by an {@link IoTDBPipePattern} and a normal {@link
+ * SetTTLPlan}. This is used to persist the "at most one" match between {@link
+ * PipeConfigRegionWritePlanEvent} in the {@link ConfigRegionListeningQueue} and the {@link
+ * IoTDBNonDataRegionExtractor#supply()} and will be executed serially in {@link
+ * IoTDBConfigNodeReceiver}
+ */
+public class PipeSetTTLPlan extends ConfigPhysicalPlan {
+ private List<SetTTLPlan> planList;
+
+ public PipeSetTTLPlan() {
+ super(ConfigPhysicalPlanType.PipeSetTTL);
+ }
+
+ public PipeSetTTLPlan(final List<SetTTLPlan> planList) {
+ super(ConfigPhysicalPlanType.PipeSetTTL);
+ this.planList = planList;
+ }
+
+ public List<SetTTLPlan> getSetTTLPlans() {
+ return planList;
+ }
+
+ @Override
+ protected void serializeImpl(final DataOutputStream stream) throws IOException {
+ stream.writeShort(getType().getPlanType());
+ ReadWriteIOUtils.write(planList.size(), stream);
+ for (final SetTTLPlan setTTLPlan : planList) {
+ final ByteBuffer buffer = setTTLPlan.serializeToByteBuffer();
+ stream.write(buffer.array(), 0, buffer.limit());
+ }
+ }
+
+ @Override
+ protected void deserializeImpl(final ByteBuffer buffer) throws IOException {
+ final int size = ReadWriteIOUtils.readInt(buffer);
+ planList = new ArrayList<>();
+ for (int i = 0; i < size; ++i) {
+ planList.add((SetTTLPlan) ConfigPhysicalPlan.Factory.create(buffer));
+ }
+ }
+
+ @Override
+ public boolean equals(final Object obj) {
+ if (this == obj) {
+ return true;
+ }
+ if (obj == null || getClass() != obj.getClass()) {
+ return false;
+ }
+ final PipeSetTTLPlan that = (PipeSetTTLPlan) obj;
+ return Objects.equals(planList, that.planList);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(planList);
+ }
+
+ @Override
+ public String toString() {
+ return "PipeSetTTLPlan{" + "planList='" + planList + "'}";
+ }
+}
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/connector/payload/PipeTransferConfigSnapshotSealReq.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/connector/payload/PipeTransferConfigSnapshotSealReq.java
index 0397cff3a94..07664f6523c 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/connector/payload/PipeTransferConfigSnapshotSealReq.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/connector/payload/PipeTransferConfigSnapshotSealReq.java
@@ -48,16 +48,19 @@ public class PipeTransferConfigSnapshotSealReq extends PipeTransferFileSealReqV2
/////////////////////////////// Thrift ///////////////////////////////
public static PipeTransferConfigSnapshotSealReq toTPipeTransferReq(
- String snapshotName,
- long snapshotLength,
- String templateFileName,
- long templateFileLength,
- CNSnapshotFileType fileType,
- String typeString)
+ final String pattern,
+ final String snapshotName,
+ final long snapshotLength,
+ final String templateFileName,
+ final long templateFileLength,
+ final CNSnapshotFileType fileType,
+ final String typeString)
throws IOException {
final Map<String, String> parameters = new HashMap<>();
+ parameters.put(ColumnHeaderConstant.PATH_PATTERN, pattern);
parameters.put(FILE_TYPE, Byte.toString(fileType.getType()));
parameters.put(ColumnHeaderConstant.TYPE, typeString);
+
return (PipeTransferConfigSnapshotSealReq)
new PipeTransferConfigSnapshotSealReq()
.convertToTPipeTransferReq(
@@ -70,7 +73,7 @@ public class PipeTransferConfigSnapshotSealReq extends PipeTransferFileSealReqV2
parameters);
}
- public static PipeTransferConfigSnapshotSealReq fromTPipeTransferReq(TPipeTransferReq req) {
+ public static PipeTransferConfigSnapshotSealReq fromTPipeTransferReq(final TPipeTransferReq req) {
return (PipeTransferConfigSnapshotSealReq)
new PipeTransferConfigSnapshotSealReq().translateFromTPipeTransferReq(req);
}
@@ -78,14 +81,16 @@ public class PipeTransferConfigSnapshotSealReq extends PipeTransferFileSealReqV2
/////////////////////////////// Air Gap ///////////////////////////////
public static byte[] toTPipeTransferBytes(
- String snapshotName,
- long snapshotLength,
- String templateFileName,
- long templateFileLength,
- CNSnapshotFileType fileType,
- String typeString)
+ final String pattern,
+ final String snapshotName,
+ final long snapshotLength,
+ final String templateFileName,
+ final long templateFileLength,
+ final CNSnapshotFileType fileType,
+ final String typeString)
throws IOException {
final Map<String, String> parameters = new HashMap<>();
+ parameters.put(ColumnHeaderConstant.PATH_PATTERN, pattern);
parameters.put(FILE_TYPE, Byte.toString(fileType.getType()));
parameters.put(ColumnHeaderConstant.TYPE, typeString);
return new PipeTransferConfigSnapshotSealReq()
@@ -102,7 +107,7 @@ public class PipeTransferConfigSnapshotSealReq extends PipeTransferFileSealReqV2
/////////////////////////////// Object ///////////////////////////////
@Override
- public boolean equals(Object obj) {
+ public boolean equals(final Object obj) {
return obj instanceof PipeTransferConfigSnapshotSealReq && super.equals(obj);
}
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/connector/protocol/IoTDBConfigRegionAirGapConnector.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/connector/protocol/IoTDBConfigRegionAirGapConnector.java
index 3ca8031d268..ec09b546648 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/connector/protocol/IoTDBConfigRegionAirGapConnector.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/connector/protocol/IoTDBConfigRegionAirGapConnector.java
@@ -201,6 +201,8 @@ public class IoTDBConfigRegionAirGapConnector extends IoTDBAirGapConnector {
socket,
compressIfNeeded(
PipeTransferConfigSnapshotSealReq.toTPipeTransferBytes(
+ // The pattern is surely Non-null
+ pipeConfigRegionSnapshotEvent.getPatternString(),
snapshot.getName(),
snapshot.length(),
Objects.nonNull(templateFile) ? templateFile.getName() : null,
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/connector/protocol/IoTDBConfigRegionConnector.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/connector/protocol/IoTDBConfigRegionConnector.java
index a2962c194f3..b199df4c390 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/connector/protocol/IoTDBConfigRegionConnector.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/connector/protocol/IoTDBConfigRegionConnector.java
@@ -195,6 +195,8 @@ public class IoTDBConfigRegionConnector extends IoTDBSslSyncConnector {
.pipeTransfer(
compressIfNeeded(
PipeTransferConfigSnapshotSealReq.toTPipeTransferReq(
+ // The pattern is surely Non-null
+ snapshotEvent.getPatternString(),
snapshotFile.getName(),
snapshotFile.length(),
Objects.nonNull(templateFile) ? templateFile.getName() : null,
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/extractor/IoTDBConfigRegionExtractor.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/extractor/IoTDBConfigRegionExtractor.java
index a5eb544c95a..ea897465b8b 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/extractor/IoTDBConfigRegionExtractor.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/extractor/IoTDBConfigRegionExtractor.java
@@ -24,6 +24,7 @@ import org.apache.iotdb.commons.pipe.config.PipeConfig;
import org.apache.iotdb.commons.pipe.datastructure.queue.listening.AbstractPipeListeningQueue;
import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
import org.apache.iotdb.commons.pipe.event.PipeSnapshotEvent;
+import org.apache.iotdb.commons.pipe.event.PipeWritePlanEvent;
import org.apache.iotdb.commons.pipe.extractor.IoTDBNonDataRegionExtractor;
import org.apache.iotdb.commons.pipe.progress.PipeEventCommitManager;
import org.apache.iotdb.confignode.conf.ConfigNodeDescriptor;
@@ -38,15 +39,18 @@ import org.apache.iotdb.consensus.ConsensusFactory;
import org.apache.iotdb.consensus.exception.ConsensusException;
import org.apache.iotdb.pipe.api.customizer.configuration.PipeExtractorRuntimeConfiguration;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
-import org.apache.iotdb.pipe.api.event.Event;
import org.apache.iotdb.pipe.api.exception.PipeException;
import java.util.HashSet;
import java.util.Objects;
+import java.util.Optional;
import java.util.Set;
public class IoTDBConfigRegionExtractor extends IoTDBNonDataRegionExtractor {
+ public static final PipeConfigPhysicalPlanPatternParseVisitor PATTERN_PARSE_VISITOR =
+ new PipeConfigPhysicalPlanPatternParseVisitor();
+
private Set<ConfigPhysicalPlanType> listenedTypeSet = new HashSet<>();
@Override
@@ -110,7 +114,17 @@ public class IoTDBConfigRegionExtractor extends IoTDBNonDataRegionExtractor {
}
@Override
- protected boolean isTypeListened(final Event event) {
+ protected Optional<PipeWritePlanEvent> trimRealtimeEventByPipePattern(
+ final PipeWritePlanEvent event) {
+ return PATTERN_PARSE_VISITOR
+ .process(((PipeConfigRegionWritePlanEvent) event).getConfigPhysicalPlan(), pipePattern)
+ .map(
+ configPhysicalPlan ->
+ new PipeConfigRegionWritePlanEvent(configPhysicalPlan, event.isGeneratedByPipe()));
+ }
+
+ @Override
+ protected boolean isTypeListened(final PipeWritePlanEvent event) {
return listenedTypeSet.contains(
((PipeConfigRegionWritePlanEvent) event).getConfigPhysicalPlan().getType());
}
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
new file mode 100644
index 00000000000..e5ea0cccd3b
--- /dev/null
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/extractor/PipeConfigPhysicalPlanPatternParseVisitor.java
@@ -0,0 +1,283 @@
+/*
+ * 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.confignode.manager.pipe.extractor;
+
+import org.apache.iotdb.commons.path.PartialPath;
+import org.apache.iotdb.commons.path.PathPatternTree;
+import org.apache.iotdb.commons.pipe.pattern.IoTDBPipePattern;
+import org.apache.iotdb.confignode.consensus.request.ConfigPhysicalPlan;
+import org.apache.iotdb.confignode.consensus.request.ConfigPhysicalPlanVisitor;
+import org.apache.iotdb.confignode.consensus.request.auth.AuthorPlan;
+import org.apache.iotdb.confignode.consensus.request.write.database.DatabaseSchemaPlan;
+import org.apache.iotdb.confignode.consensus.request.write.database.DeleteDatabasePlan;
+import org.apache.iotdb.confignode.consensus.request.write.database.SetTTLPlan;
+import org.apache.iotdb.confignode.consensus.request.write.pipe.payload.PipeDeactivateTemplatePlan;
+import org.apache.iotdb.confignode.consensus.request.write.pipe.payload.PipeDeleteLogicalViewPlan;
+import org.apache.iotdb.confignode.consensus.request.write.pipe.payload.PipeDeleteTimeSeriesPlan;
+import org.apache.iotdb.confignode.consensus.request.write.pipe.payload.PipeSetTTLPlan;
+import org.apache.iotdb.confignode.consensus.request.write.pipe.payload.PipeUnsetSchemaTemplatePlan;
+import org.apache.iotdb.confignode.consensus.request.write.template.CommitSetSchemaTemplatePlan;
+import org.apache.iotdb.confignode.consensus.request.write.template.CreateSchemaTemplatePlan;
+import org.apache.iotdb.confignode.consensus.request.write.template.ExtendSchemaTemplatePlan;
+import org.apache.iotdb.confignode.manager.pipe.event.PipeConfigRegionWritePlanEvent;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNode;
+import org.apache.iotdb.db.schemaengine.template.Template;
+import org.apache.iotdb.db.schemaengine.template.alter.TemplateExtendInfo;
+
+import org.apache.tsfile.utils.Pair;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import java.util.stream.Stream;
+
+/**
+ * The {@link PipeConfigPhysicalPlanPatternParseVisitor} will transform the schema {@link
+ * ConfigPhysicalPlan}s using {@link IoTDBPipePattern}. 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}.
+ *
+ * <p>2. If a pattern does not intersect with the {@link IoTDBPipePattern}, it's dropped.
+ *
+ * <p>3. If all the patterns in the {@link ConfigPhysicalPlan} is dropped, the {@link
+ * ConfigPhysicalPlan} 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 PipeConfigRegionWritePlanEvent} in {@link ConfigRegionListeningQueue}.
+ */
+public class PipeConfigPhysicalPlanPatternParseVisitor
+ extends ConfigPhysicalPlanVisitor<Optional<ConfigPhysicalPlan>, IoTDBPipePattern> {
+ private static final Logger LOGGER =
+ LoggerFactory.getLogger(PipeConfigPhysicalPlanPatternParseVisitor.class);
+
+ @Override
+ public Optional<ConfigPhysicalPlan> visitPlan(
+ final ConfigPhysicalPlan plan, final IoTDBPipePattern pattern) {
+ return Optional.of(plan);
+ }
+
+ // Note: This will return true even if the pattern matches the database precisely,
+ // like database is "root.db.a" and pattern is "root.db.a". In this case, none of
+ // the data and time series under this database will be transferred, however we
+ // interpret user's pattern as "send precisely the database" and transfer it
+ // nonetheless.
+ // Other matches using "matchPrefixPath" are with the same principle.
+ @Override
+ public Optional<ConfigPhysicalPlan> visitCreateDatabase(
+ final DatabaseSchemaPlan createDatabasePlan, final IoTDBPipePattern pattern) {
+ return pattern.matchPrefixPath(createDatabasePlan.getSchema().getName())
+ ? Optional.of(createDatabasePlan)
+ : Optional.empty();
+ }
+
+ @Override
+ public Optional<ConfigPhysicalPlan> visitAlterDatabase(
+ final DatabaseSchemaPlan alterDatabasePlan, final IoTDBPipePattern pattern) {
+ return pattern.matchPrefixPath(alterDatabasePlan.getSchema().getName())
+ ? Optional.of(alterDatabasePlan)
+ : Optional.empty();
+ }
+
+ @Override
+ public Optional<ConfigPhysicalPlan> visitDeleteDatabase(
+ final DeleteDatabasePlan deleteDatabasePlan, final IoTDBPipePattern pattern) {
+ return pattern.matchPrefixPath(deleteDatabasePlan.getName())
+ ? Optional.of(deleteDatabasePlan)
+ : Optional.empty();
+ }
+
+ @Override
+ public Optional<ConfigPhysicalPlan> visitCreateSchemaTemplate(
+ final CreateSchemaTemplatePlan createSchemaTemplatePlan, final IoTDBPipePattern pattern) {
+ // This is a deserialized template and can be arbitrarily altered
+ final Template template = createSchemaTemplatePlan.getTemplate();
+ template.getSchemaMap().keySet().removeIf(measurement -> !pattern.matchTailNode(measurement));
+ return !template.getSchemaMap().isEmpty()
+ ? Optional.of(new CreateSchemaTemplatePlan(template.serialize().array()))
+ : Optional.empty();
+ }
+
+ @Override
+ public Optional<ConfigPhysicalPlan> visitCommitSetSchemaTemplate(
+ final CommitSetSchemaTemplatePlan commitSetSchemaTemplatePlan,
+ final IoTDBPipePattern pattern) {
+ return pattern.matchPrefixPath(commitSetSchemaTemplatePlan.getPath())
+ ? Optional.of(commitSetSchemaTemplatePlan)
+ : Optional.empty();
+ }
+
+ @Override
+ public Optional<ConfigPhysicalPlan> visitPipeUnsetSchemaTemplate(
+ final PipeUnsetSchemaTemplatePlan pipeUnsetSchemaTemplatePlan,
+ final IoTDBPipePattern pattern) {
+ return pattern.matchPrefixPath(pipeUnsetSchemaTemplatePlan.getPath())
+ ? Optional.of(pipeUnsetSchemaTemplatePlan)
+ : Optional.empty();
+ }
+
+ @Override
+ public Optional<ConfigPhysicalPlan> visitExtendSchemaTemplate(
+ final ExtendSchemaTemplatePlan extendSchemaTemplatePlan, final IoTDBPipePattern pattern) {
+ final TemplateExtendInfo extendInfo = extendSchemaTemplatePlan.getTemplateExtendInfo();
+ final int[] filteredIndexes =
+ IntStream.range(0, extendInfo.getMeasurements().size())
+ .filter(index -> pattern.matchTailNode(extendInfo.getMeasurements().get(index)))
+ .toArray();
+ return filteredIndexes.length > 0
+ ? Optional.of(
+ new ExtendSchemaTemplatePlan(
+ new TemplateExtendInfo(
+ extendInfo.getTemplateName(),
+ IoTDBPipePattern.applyIndexesOnList(
+ filteredIndexes, extendInfo.getMeasurements()),
+ IoTDBPipePattern.applyIndexesOnList(filteredIndexes, extendInfo.getDataTypes()),
+ IoTDBPipePattern.applyIndexesOnList(filteredIndexes, extendInfo.getEncodings()),
+ IoTDBPipePattern.applyIndexesOnList(
+ filteredIndexes, extendInfo.getCompressors()))))
+ : Optional.empty();
+ }
+
+ @Override
+ public Optional<ConfigPhysicalPlan> visitGrantUser(
+ final AuthorPlan grantUserPlan, final IoTDBPipePattern pattern) {
+ return visitPathRelatedAuthorPlan(grantUserPlan, pattern);
+ }
+
+ @Override
+ public Optional<ConfigPhysicalPlan> visitRevokeUser(
+ final AuthorPlan revokeUserPlan, final IoTDBPipePattern pattern) {
+ return visitPathRelatedAuthorPlan(revokeUserPlan, pattern);
+ }
+
+ @Override
+ public Optional<ConfigPhysicalPlan> visitGrantRole(
+ final AuthorPlan revokeUserPlan, final IoTDBPipePattern pattern) {
+ return visitPathRelatedAuthorPlan(revokeUserPlan, pattern);
+ }
+
+ @Override
+ public Optional<ConfigPhysicalPlan> visitRevokeRole(
+ final AuthorPlan revokeUserPlan, final IoTDBPipePattern pattern) {
+ return visitPathRelatedAuthorPlan(revokeUserPlan, pattern);
+ }
+
+ private Optional<ConfigPhysicalPlan> visitPathRelatedAuthorPlan(
+ final AuthorPlan pathRelatedAuthorPlan, final IoTDBPipePattern pattern) {
+ final List<PartialPath> intersectedPaths =
+ pathRelatedAuthorPlan.getNodeNameList().stream()
+ .map(pattern::getIntersection)
+ .flatMap(Collection::stream)
+ .collect(Collectors.toList());
+ return !intersectedPaths.isEmpty()
+ ? Optional.of(
+ new AuthorPlan(
+ pathRelatedAuthorPlan.getAuthorType(),
+ pathRelatedAuthorPlan.getUserName(),
+ pathRelatedAuthorPlan.getRoleName(),
+ pathRelatedAuthorPlan.getPassword(),
+ pathRelatedAuthorPlan.getNewPassword(),
+ pathRelatedAuthorPlan.getPermissions(),
+ pathRelatedAuthorPlan.getGrantOpt(),
+ intersectedPaths))
+ : Optional.empty();
+ }
+
+ @Override
+ public Optional<ConfigPhysicalPlan> visitPipeDeleteTimeSeries(
+ final PipeDeleteTimeSeriesPlan pipeDeleteTimeSeriesPlan, final IoTDBPipePattern pattern) {
+ try {
+ final PathPatternTree intersectedTree =
+ pattern.getIntersection(
+ PathPatternTree.deserialize(pipeDeleteTimeSeriesPlan.getPatternTreeBytes()));
+ return !intersectedTree.isEmpty()
+ ? Optional.of(new PipeDeleteTimeSeriesPlan(intersectedTree.serialize()))
+ : Optional.empty();
+ } catch (final IOException e) {
+ LOGGER.warn(
+ "Serialization failed for the delete time series plan in pipe transmission, skip transfer",
+ e);
+ return Optional.empty();
+ }
+ }
+
+ @Override
+ public Optional<ConfigPhysicalPlan> visitPipeDeleteLogicalView(
+ final PipeDeleteLogicalViewPlan pipeDeleteLogicalViewPlan, final IoTDBPipePattern pattern) {
+ try {
+ final PathPatternTree intersectedTree =
+ pattern.getIntersection(
+ PathPatternTree.deserialize(pipeDeleteLogicalViewPlan.getPatternTreeBytes()));
+ return !intersectedTree.isEmpty()
+ ? Optional.of(new PipeDeleteTimeSeriesPlan(intersectedTree.serialize()))
+ : Optional.empty();
+ } catch (final IOException e) {
+ LOGGER.warn(
+ "Serialization failed for the delete logical view plan in pipe transmission, skip transfer",
+ e);
+ return Optional.empty();
+ }
+ }
+
+ @Override
+ public Optional<ConfigPhysicalPlan> visitPipeDeactivateTemplate(
+ final PipeDeactivateTemplatePlan pipeDeactivateTemplatePlan, final IoTDBPipePattern pattern) {
+ final Map<PartialPath, List<Template>> newTemplateSetInfo =
+ pipeDeactivateTemplatePlan.getTemplateSetInfo().entrySet().stream()
+ .flatMap(
+ entry ->
+ pattern.getIntersection(entry.getKey()).stream()
+ .map(partialPath -> new Pair<>(partialPath, entry.getValue())))
+ .collect(
+ Collectors.toMap(
+ Pair::getLeft,
+ Pair::getRight,
+ (oldTemplates, newTemplates) ->
+ Stream.of(oldTemplates, newTemplates)
+ .flatMap(Collection::stream)
+ .distinct()
+ .collect(Collectors.toList())));
+ return !newTemplateSetInfo.isEmpty()
+ ? Optional.of(new PipeDeactivateTemplatePlan(newTemplateSetInfo))
+ : Optional.empty();
+ }
+
+ @Override
+ public Optional<ConfigPhysicalPlan> visitTTL(
+ final SetTTLPlan setTTLPlan, final IoTDBPipePattern pattern) {
+ final List<PartialPath> intersectionList =
+ pattern.getIntersection(new PartialPath(setTTLPlan.getDatabasePathPattern()));
+ return !intersectionList.isEmpty()
+ ? Optional.of(
+ new PipeSetTTLPlan(
+ intersectionList.stream()
+ .map(
+ path -> new SetTTLPlan(Arrays.asList(path.getNodes()), setTTLPlan.getTTL()))
+ .collect(Collectors.toList())))
+ : Optional.empty();
+ }
+}
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 0a78553516a..513c910ce11 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
@@ -27,6 +27,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.pattern.IoTDBPipePattern;
import org.apache.iotdb.commons.pipe.receiver.IoTDBFileReceiver;
import org.apache.iotdb.confignode.conf.ConfigNodeDescriptor;
import org.apache.iotdb.confignode.consensus.request.ConfigPhysicalPlan;
@@ -39,6 +40,7 @@ import org.apache.iotdb.confignode.consensus.request.write.pipe.payload.PipeDeac
import org.apache.iotdb.confignode.consensus.request.write.pipe.payload.PipeDeleteLogicalViewPlan;
import org.apache.iotdb.confignode.consensus.request.write.pipe.payload.PipeDeleteTimeSeriesPlan;
import org.apache.iotdb.confignode.consensus.request.write.pipe.payload.PipeEnrichedPlan;
+import org.apache.iotdb.confignode.consensus.request.write.pipe.payload.PipeSetTTLPlan;
import org.apache.iotdb.confignode.consensus.request.write.pipe.payload.PipeUnsetSchemaTemplatePlan;
import org.apache.iotdb.confignode.consensus.request.write.template.CommitSetSchemaTemplatePlan;
import org.apache.iotdb.confignode.consensus.request.write.template.ExtendSchemaTemplatePlan;
@@ -50,6 +52,7 @@ import org.apache.iotdb.confignode.manager.pipe.connector.payload.PipeTransferCo
import org.apache.iotdb.confignode.manager.pipe.connector.payload.PipeTransferConfigSnapshotPieceReq;
import org.apache.iotdb.confignode.manager.pipe.connector.payload.PipeTransferConfigSnapshotSealReq;
import org.apache.iotdb.confignode.manager.pipe.event.PipeConfigRegionSnapshotEvent;
+import org.apache.iotdb.confignode.manager.pipe.extractor.IoTDBConfigRegionExtractor;
import org.apache.iotdb.confignode.manager.pipe.receiver.visitor.PipeConfigPhysicalPlanExceptionVisitor;
import org.apache.iotdb.confignode.manager.pipe.receiver.visitor.PipeConfigPhysicalPlanTSStatusVisitor;
import org.apache.iotdb.confignode.persistence.schema.CNPhysicalPlanGenerator;
@@ -82,6 +85,7 @@ import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
public class IoTDBConfigNodeReceiver extends IoTDBFileReceiver {
@@ -143,7 +147,7 @@ public class IoTDBConfigNodeReceiver extends IoTDBFileReceiver {
receiverId.get(),
status);
return new TPipeTransferResp(status);
- } catch (Exception e) {
+ } catch (final Exception e) {
final String error =
"Exception encountered while handling pipe transfer request. Root cause: "
+ e.getMessage();
@@ -157,19 +161,19 @@ public class IoTDBConfigNodeReceiver extends IoTDBFileReceiver {
// to notify its configurations.
// Note that the sender needs not to reconstruct its client because the client
// is directly linked to the preceding DataNode and has not broken.
- private boolean needHandshake(PipeRequestType type) {
+ private boolean needHandshake(final PipeRequestType type) {
return Objects.isNull(receiverFileDirWithIdSuffix.get())
&& type != PipeRequestType.HANDSHAKE_CONFIGNODE_V1
&& type != PipeRequestType.HANDSHAKE_CONFIGNODE_V2;
}
- private TPipeTransferResp handleTransferConfigPlan(PipeTransferConfigPlanReq req)
+ private TPipeTransferResp handleTransferConfigPlan(final PipeTransferConfigPlanReq req)
throws IOException {
return new TPipeTransferResp(
executePlanAndClassifyExceptions(ConfigPhysicalPlan.Factory.create(req.body)));
}
- private TSStatus executePlanAndClassifyExceptions(ConfigPhysicalPlan plan) {
+ private TSStatus executePlanAndClassifyExceptions(final ConfigPhysicalPlan plan) {
TSStatus result;
try {
result = executePlan(plan);
@@ -181,7 +185,7 @@ public class IoTDBConfigNodeReceiver extends IoTDBFileReceiver {
result);
result = STATUS_VISITOR.process(plan, result);
}
- } catch (Exception e) {
+ } catch (final Exception e) {
LOGGER.warn(
"Receiver id = {}: Exception encountered while executing plan {}: ",
receiverId.get(),
@@ -192,7 +196,7 @@ public class IoTDBConfigNodeReceiver extends IoTDBFileReceiver {
return result;
}
- private TSStatus executePlan(ConfigPhysicalPlan plan) throws ConsensusException {
+ private TSStatus executePlan(final ConfigPhysicalPlan plan) throws ConsensusException {
switch (plan.getType()) {
case CreateDatabase:
// Here we only reserve database name and substitute the sender's local information
@@ -269,6 +273,13 @@ public class IoTDBConfigNodeReceiver extends IoTDBFileReceiver {
.setIsGeneratedByPipe(true));
case SetTTL:
return configManager.getClusterSchemaManager().setTTL((SetTTLPlan) plan, true);
+ case PipeSetTTL:
+ // The prior status won't be altered by the status visitor
+ return PipeReceiverStatusHandler.getPriorStatus(
+ ((PipeSetTTLPlan) plan)
+ .getSetTTLPlans().stream()
+ .map(this::executePlanAndClassifyExceptions)
+ .collect(Collectors.toList()));
case DropUser:
case DropRole:
case GrantRole:
@@ -304,13 +315,15 @@ public class IoTDBConfigNodeReceiver extends IoTDBFileReceiver {
}
@Override
- protected TSStatus loadFileV1(PipeTransferFileSealReqV1 req, String fileAbsolutePath) {
+ protected TSStatus loadFileV1(
+ final PipeTransferFileSealReqV1 req, final String fileAbsolutePath) {
throw new UnsupportedOperationException(
"IoTDBConfigNodeReceiver does not support load file V1.");
}
@Override
- protected TSStatus loadFileV2(PipeTransferFileSealReqV2 req, List<String> fileAbsolutePaths)
+ protected TSStatus loadFileV2(
+ final PipeTransferFileSealReqV2 req, final List<String> fileAbsolutePaths)
throws IOException {
final Map<String, String> parameters = req.getParameters();
final CNPhysicalPlanGenerator generator =
@@ -326,13 +339,16 @@ 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 List<TSStatus> results = new ArrayList<>();
while (generator.hasNext()) {
- final ConfigPhysicalPlan plan = generator.next();
- if (executionTypes.contains(plan.getType())) {
- // Here we apply the statements as many as possible
- results.add(executePlanAndClassifyExceptions(plan));
- }
+ IoTDBConfigRegionExtractor.PATTERN_PARSE_VISITOR
+ .process(generator.next(), pattern)
+ .filter(configPhysicalPlan -> executionTypes.contains(configPhysicalPlan.getType()))
+ .ifPresent(
+ configPhysicalPlan ->
+ results.add(executePlanAndClassifyExceptions(configPhysicalPlan)));
}
return PipeReceiverStatusHandler.getPriorStatus(results);
}
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/receiver/visitor/PipeConfigPhysicalPlanTSStatusVisitor.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/receiver/visitor/PipeConfigPhysicalPlanTSStatusVisitor.java
index 89a9fd4d471..f2db9d8c3ee 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/receiver/visitor/PipeConfigPhysicalPlanTSStatusVisitor.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/receiver/visitor/PipeConfigPhysicalPlanTSStatusVisitor.java
@@ -45,12 +45,12 @@ public class PipeConfigPhysicalPlanTSStatusVisitor
extends ConfigPhysicalPlanVisitor<TSStatus, TSStatus> {
@Override
- public TSStatus visitPlan(ConfigPhysicalPlan plan, TSStatus context) {
+ public TSStatus visitPlan(final ConfigPhysicalPlan plan, final TSStatus context) {
return context;
}
@Override
- public TSStatus visitCreateDatabase(DatabaseSchemaPlan plan, TSStatus context) {
+ public TSStatus visitCreateDatabase(final DatabaseSchemaPlan plan, final TSStatus context) {
if (context.getCode() == TSStatusCode.DATABASE_ALREADY_EXISTS.getStatusCode()) {
if (context
.getMessage()
@@ -77,7 +77,7 @@ public class PipeConfigPhysicalPlanTSStatusVisitor
}
@Override
- public TSStatus visitAlterDatabase(DatabaseSchemaPlan plan, TSStatus context) {
+ public TSStatus visitAlterDatabase(final DatabaseSchemaPlan plan, final TSStatus context) {
if (context.getCode() == TSStatusCode.DATABASE_CONFIG_ERROR.getStatusCode()) {
return new TSStatus(TSStatusCode.PIPE_RECEIVER_IDEMPOTENT_CONFLICT_EXCEPTION.getStatusCode())
.setMessage(context.getMessage());
@@ -89,7 +89,7 @@ public class PipeConfigPhysicalPlanTSStatusVisitor
}
@Override
- public TSStatus visitDeleteDatabase(DeleteDatabasePlan plan, TSStatus context) {
+ public TSStatus visitDeleteDatabase(final DeleteDatabasePlan plan, final TSStatus context) {
if (context.getCode() == TSStatusCode.PATH_NOT_EXIST.getStatusCode()) {
return new TSStatus(TSStatusCode.PIPE_RECEIVER_IDEMPOTENT_CONFLICT_EXCEPTION.getStatusCode())
.setMessage(context.getMessage());
@@ -98,7 +98,8 @@ public class PipeConfigPhysicalPlanTSStatusVisitor
}
@Override
- public TSStatus visitCreateSchemaTemplate(CreateSchemaTemplatePlan plan, TSStatus context) {
+ public TSStatus visitCreateSchemaTemplate(
+ final CreateSchemaTemplatePlan plan, final TSStatus context) {
if (context.getCode() == TSStatusCode.METADATA_ERROR.getStatusCode()) {
return new TSStatus(TSStatusCode.PIPE_RECEIVER_IDEMPOTENT_CONFLICT_EXCEPTION.getStatusCode())
.setMessage(context.getMessage());
@@ -107,7 +108,8 @@ public class PipeConfigPhysicalPlanTSStatusVisitor
}
@Override
- public TSStatus visitExtendSchemaTemplate(ExtendSchemaTemplatePlan plan, TSStatus context) {
+ public TSStatus visitExtendSchemaTemplate(
+ final ExtendSchemaTemplatePlan plan, final TSStatus context) {
if (context.getCode() == TSStatusCode.MEASUREMENT_ALREADY_EXISTS_IN_TEMPLATE.getStatusCode()) {
return new TSStatus(TSStatusCode.PIPE_RECEIVER_IDEMPOTENT_CONFLICT_EXCEPTION.getStatusCode())
.setMessage(context.getMessage());
@@ -119,7 +121,8 @@ public class PipeConfigPhysicalPlanTSStatusVisitor
}
@Override
- public TSStatus visitCommitSetSchemaTemplate(CommitSetSchemaTemplatePlan plan, TSStatus context) {
+ public TSStatus visitCommitSetSchemaTemplate(
+ final CommitSetSchemaTemplatePlan plan, final TSStatus context) {
if (context.getCode() == TSStatusCode.METADATA_ERROR.getStatusCode()) {
if (context.getMessage().contains("Template already exists")) {
return new TSStatus(
@@ -138,7 +141,7 @@ public class PipeConfigPhysicalPlanTSStatusVisitor
@Override
public TSStatus visitPipeUnsetSchemaTemplate(
- PipeUnsetSchemaTemplatePlan pipeUnsetSchemaTemplatePlan, TSStatus context) {
+ final PipeUnsetSchemaTemplatePlan pipeUnsetSchemaTemplatePlan, final TSStatus context) {
if (context.getCode() == TSStatusCode.TEMPLATE_NOT_SET.getStatusCode()) {
return new TSStatus(TSStatusCode.PIPE_RECEIVER_IDEMPOTENT_CONFLICT_EXCEPTION.getStatusCode())
.setMessage(context.getMessage());
@@ -152,7 +155,7 @@ public class PipeConfigPhysicalPlanTSStatusVisitor
@Override
public TSStatus visitDropSchemaTemplate(
- DropSchemaTemplatePlan dropSchemaTemplatePlan, TSStatus context) {
+ final DropSchemaTemplatePlan dropSchemaTemplatePlan, final TSStatus context) {
if (context.getCode() == TSStatusCode.UNDEFINED_TEMPLATE.getStatusCode()) {
return new TSStatus(TSStatusCode.PIPE_RECEIVER_IDEMPOTENT_CONFLICT_EXCEPTION.getStatusCode())
.setMessage(context.getMessage());
@@ -162,7 +165,7 @@ public class PipeConfigPhysicalPlanTSStatusVisitor
@Override
public TSStatus visitPipeDeleteTimeSeries(
- PipeDeleteTimeSeriesPlan pipeDeleteTimeSeriesPlan, TSStatus context) {
+ final PipeDeleteTimeSeriesPlan pipeDeleteTimeSeriesPlan, final TSStatus context) {
if (context.getCode() == TSStatusCode.PATH_NOT_EXIST.getStatusCode()) {
return new TSStatus(TSStatusCode.PIPE_RECEIVER_IDEMPOTENT_CONFLICT_EXCEPTION.getStatusCode())
.setMessage(context.getMessage());
@@ -172,7 +175,7 @@ public class PipeConfigPhysicalPlanTSStatusVisitor
@Override
public TSStatus visitPipeDeleteLogicalView(
- PipeDeleteLogicalViewPlan pipeDeleteLogicalViewPlan, TSStatus context) {
+ final PipeDeleteLogicalViewPlan pipeDeleteLogicalViewPlan, final TSStatus context) {
if (context.getCode() == TSStatusCode.PATH_NOT_EXIST.getStatusCode()) {
return new TSStatus(TSStatusCode.PIPE_RECEIVER_IDEMPOTENT_CONFLICT_EXCEPTION.getStatusCode())
.setMessage(context.getMessage());
@@ -182,7 +185,7 @@ public class PipeConfigPhysicalPlanTSStatusVisitor
@Override
public TSStatus visitPipeDeactivateTemplate(
- PipeDeactivateTemplatePlan pipeDeactivateTemplatePlan, TSStatus context) {
+ final PipeDeactivateTemplatePlan pipeDeactivateTemplatePlan, final TSStatus context) {
if (context.getCode() == TSStatusCode.TEMPLATE_NOT_ACTIVATED.getStatusCode()) {
return new TSStatus(TSStatusCode.PIPE_RECEIVER_IDEMPOTENT_CONFLICT_EXCEPTION.getStatusCode())
.setMessage(context.getMessage());
@@ -191,7 +194,7 @@ public class PipeConfigPhysicalPlanTSStatusVisitor
}
@Override
- public TSStatus visitCreateUser(AuthorPlan plan, TSStatus context) {
+ public TSStatus visitCreateUser(final AuthorPlan plan, final TSStatus context) {
if (context.getCode() == TSStatusCode.USER_ALREADY_EXIST.getStatusCode()) {
return new TSStatus(TSStatusCode.PIPE_RECEIVER_IDEMPOTENT_CONFLICT_EXCEPTION.getStatusCode())
.setMessage(context.getMessage());
@@ -200,7 +203,7 @@ public class PipeConfigPhysicalPlanTSStatusVisitor
}
@Override
- public TSStatus visitCreateRawUser(AuthorPlan plan, TSStatus context) {
+ public TSStatus visitCreateRawUser(final AuthorPlan plan, final TSStatus context) {
if (context.getCode() == TSStatusCode.USER_ALREADY_EXIST.getStatusCode()) {
return new TSStatus(TSStatusCode.PIPE_RECEIVER_IDEMPOTENT_CONFLICT_EXCEPTION.getStatusCode())
.setMessage(context.getMessage());
@@ -209,7 +212,7 @@ public class PipeConfigPhysicalPlanTSStatusVisitor
}
@Override
- public TSStatus visitUpdateUser(AuthorPlan plan, TSStatus context) {
+ public TSStatus visitUpdateUser(final AuthorPlan plan, final TSStatus context) {
if (context.getCode() == TSStatusCode.USER_NOT_EXIST.getStatusCode()) {
return new TSStatus(TSStatusCode.PIPE_RECEIVER_USER_CONFLICT_EXCEPTION.getStatusCode())
.setMessage(context.getMessage());
@@ -218,7 +221,7 @@ public class PipeConfigPhysicalPlanTSStatusVisitor
}
@Override
- public TSStatus visitDropUser(AuthorPlan plan, TSStatus context) {
+ public TSStatus visitDropUser(final AuthorPlan plan, final TSStatus context) {
if (context.getCode() == TSStatusCode.USER_NOT_EXIST.getStatusCode()) {
return new TSStatus(TSStatusCode.PIPE_RECEIVER_IDEMPOTENT_CONFLICT_EXCEPTION.getStatusCode())
.setMessage(context.getMessage());
@@ -227,7 +230,7 @@ public class PipeConfigPhysicalPlanTSStatusVisitor
}
@Override
- public TSStatus visitGrantUser(AuthorPlan plan, TSStatus context) {
+ public TSStatus visitGrantUser(final AuthorPlan plan, final TSStatus context) {
if (context.getCode() == TSStatusCode.NO_PERMISSION.getStatusCode()) {
// Admin user
return new TSStatus(TSStatusCode.PIPE_RECEIVER_IDEMPOTENT_CONFLICT_EXCEPTION.getStatusCode())
@@ -241,7 +244,7 @@ public class PipeConfigPhysicalPlanTSStatusVisitor
}
@Override
- public TSStatus visitRevokeUser(AuthorPlan plan, TSStatus context) {
+ public TSStatus visitRevokeUser(final AuthorPlan plan, final TSStatus context) {
if (context.getCode() == TSStatusCode.NOT_HAS_PRIVILEGE.getStatusCode()) {
return new TSStatus(TSStatusCode.PIPE_RECEIVER_IDEMPOTENT_CONFLICT_EXCEPTION.getStatusCode())
.setMessage(context.getMessage());
@@ -255,7 +258,7 @@ public class PipeConfigPhysicalPlanTSStatusVisitor
}
@Override
- public TSStatus visitCreateRole(AuthorPlan plan, TSStatus context) {
+ public TSStatus visitCreateRole(final AuthorPlan plan, final TSStatus context) {
if (context.getCode() == TSStatusCode.ROLE_ALREADY_EXIST.getStatusCode()) {
return new TSStatus(TSStatusCode.PIPE_RECEIVER_IDEMPOTENT_CONFLICT_EXCEPTION.getStatusCode())
.setMessage(context.getMessage());
@@ -264,7 +267,7 @@ public class PipeConfigPhysicalPlanTSStatusVisitor
}
@Override
- public TSStatus visitDropRole(AuthorPlan plan, TSStatus context) {
+ public TSStatus visitDropRole(final AuthorPlan plan, final TSStatus context) {
if (context.getCode() == TSStatusCode.ROLE_NOT_EXIST.getStatusCode()) {
return new TSStatus(TSStatusCode.PIPE_RECEIVER_IDEMPOTENT_CONFLICT_EXCEPTION.getStatusCode())
.setMessage(context.getMessage());
@@ -273,7 +276,7 @@ public class PipeConfigPhysicalPlanTSStatusVisitor
}
@Override
- public TSStatus visitGrantRole(AuthorPlan plan, TSStatus context) {
+ public TSStatus visitGrantRole(final AuthorPlan plan, final TSStatus context) {
if (context.getCode() == TSStatusCode.ROLE_NOT_EXIST.getStatusCode()) {
return new TSStatus(TSStatusCode.PIPE_RECEIVER_USER_CONFLICT_EXCEPTION.getStatusCode())
.setMessage(context.getMessage());
@@ -282,7 +285,7 @@ public class PipeConfigPhysicalPlanTSStatusVisitor
}
@Override
- public TSStatus visitRevokeRole(AuthorPlan plan, TSStatus context) {
+ public TSStatus visitRevokeRole(final AuthorPlan plan, final TSStatus context) {
if (context.getCode() == TSStatusCode.NOT_HAS_PRIVILEGE.getStatusCode()) {
return new TSStatus(TSStatusCode.PIPE_RECEIVER_IDEMPOTENT_CONFLICT_EXCEPTION.getStatusCode())
.setMessage(context.getMessage());
@@ -294,7 +297,7 @@ public class PipeConfigPhysicalPlanTSStatusVisitor
}
@Override
- public TSStatus visitGrantRoleToUser(AuthorPlan plan, TSStatus context) {
+ public TSStatus visitGrantRoleToUser(final AuthorPlan plan, final TSStatus context) {
if (context.getCode() == TSStatusCode.USER_ALREADY_HAS_ROLE.getStatusCode()) {
return new TSStatus(TSStatusCode.PIPE_RECEIVER_IDEMPOTENT_CONFLICT_EXCEPTION.getStatusCode())
.setMessage(context.getMessage());
@@ -303,7 +306,7 @@ public class PipeConfigPhysicalPlanTSStatusVisitor
}
@Override
- public TSStatus visitRevokeRoleFromUser(AuthorPlan plan, TSStatus context) {
+ public TSStatus visitRevokeRoleFromUser(final AuthorPlan plan, final TSStatus context) {
if (context.getCode() == TSStatusCode.USER_NOT_HAS_ROLE.getStatusCode()) {
return new TSStatus(TSStatusCode.PIPE_RECEIVER_IDEMPOTENT_CONFLICT_EXCEPTION.getStatusCode())
.setMessage(context.getMessage());
@@ -315,9 +318,9 @@ public class PipeConfigPhysicalPlanTSStatusVisitor
}
@Override
- public TSStatus visitTTL(SetTTLPlan plan, TSStatus context) {
+ public TSStatus visitTTL(final SetTTLPlan plan, final TSStatus context) {
if (context.getCode() == TSStatusCode.DATABASE_NOT_EXIST.getStatusCode()) {
- return new TSStatus(TSStatusCode.PIPE_RECEIVER_USER_CONFLICT_EXCEPTION.getStatusCode())
+ return new TSStatus(TSStatusCode.PIPE_RECEIVER_IDEMPOTENT_CONFLICT_EXCEPTION.getStatusCode())
.setMessage(context.getMessage());
}
return super.visitTTL(plan, context);
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java
index 6cb52ef7f27..afb2f8415de 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java
@@ -26,7 +26,6 @@ import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.common.rpc.thrift.TSetSpaceQuotaReq;
import org.apache.iotdb.common.rpc.thrift.TSetTTLReq;
import org.apache.iotdb.common.rpc.thrift.TSetThrottleQuotaReq;
-import org.apache.iotdb.commons.auth.AuthException;
import org.apache.iotdb.commons.conf.CommonDescriptor;
import org.apache.iotdb.commons.consensus.ConsensusGroupId;
import org.apache.iotdb.commons.path.PartialPath;
@@ -545,52 +544,42 @@ public class ConfigNodeRPCServiceProcessor implements IConfigNodeRPCService.Ifac
}
@Override
- public TSStatus operatePermission(TAuthorizerReq req) {
+ public TSStatus operatePermission(final TAuthorizerReq req) {
if (req.getAuthorType() < 0 || req.getAuthorType() >= AuthorType.values().length) {
throw new IndexOutOfBoundsException("Invalid Author Type ordinal");
}
- AuthorPlan plan = null;
- try {
- plan =
- new AuthorPlan(
- ConfigPhysicalPlanType.values()[
- req.getAuthorType() + ConfigPhysicalPlanType.CreateUser.ordinal()],
- req.getUserName(),
- req.getRoleName(),
- req.getPassword(),
- req.getNewPassword(),
- req.getPermissions(),
- req.isGrantOpt(),
- AuthUtils.deserializePartialPathList(ByteBuffer.wrap(req.getNodeNameList())));
- } catch (AuthException e) {
- LOGGER.error(e.getMessage());
- }
- return configManager.operatePermission(plan);
+ return configManager.operatePermission(
+ new AuthorPlan(
+ ConfigPhysicalPlanType.values()[
+ req.getAuthorType() + ConfigPhysicalPlanType.CreateUser.ordinal()],
+ req.getUserName(),
+ req.getRoleName(),
+ req.getPassword(),
+ req.getNewPassword(),
+ req.getPermissions(),
+ req.isGrantOpt(),
+ AuthUtils.deserializePartialPathList(ByteBuffer.wrap(req.getNodeNameList()))));
}
@Override
- public TAuthorizerResp queryPermission(TAuthorizerReq req) {
+ public TAuthorizerResp queryPermission(final TAuthorizerReq req) {
if (req.getAuthorType() < 0 || req.getAuthorType() >= AuthorType.values().length) {
throw new IndexOutOfBoundsException("Invalid Author Type ordinal");
}
- AuthorPlan plan = null;
- try {
- plan =
- new AuthorPlan(
- ConfigPhysicalPlanType.values()[
- req.getAuthorType() + ConfigPhysicalPlanType.CreateUser.ordinal()],
- req.getUserName(),
- req.getRoleName(),
- req.getPassword(),
- req.getNewPassword(),
- req.getPermissions(),
- req.isGrantOpt(),
- AuthUtils.deserializePartialPathList(ByteBuffer.wrap(req.getNodeNameList())));
- } catch (AuthException e) {
- LOGGER.error(e.getMessage());
- }
- PermissionInfoResp dataSet = (PermissionInfoResp) configManager.queryPermission(plan);
- TAuthorizerResp resp = new TAuthorizerResp(dataSet.getStatus());
+ final PermissionInfoResp dataSet =
+ (PermissionInfoResp)
+ configManager.queryPermission(
+ new AuthorPlan(
+ ConfigPhysicalPlanType.values()[
+ req.getAuthorType() + ConfigPhysicalPlanType.CreateUser.ordinal()],
+ req.getUserName(),
+ req.getRoleName(),
+ req.getPassword(),
+ req.getNewPassword(),
+ req.getPermissions(),
+ req.isGrantOpt(),
+ AuthUtils.deserializePartialPathList(ByteBuffer.wrap(req.getNodeNameList()))));
+ final TAuthorizerResp resp = new TAuthorizerResp(dataSet.getStatus());
resp.setMemberInfo(dataSet.getMemberList());
resp.setPermissionInfo(dataSet.getPermissionInfoResp());
resp.setTag(dataSet.getTag());
diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlanSerDeTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlanSerDeTest.java
index b6c57369f67..5c60f2dc3d3 100644
--- a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlanSerDeTest.java
+++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlanSerDeTest.java
@@ -33,7 +33,6 @@ import org.apache.iotdb.common.rpc.thrift.TThrottleQuota;
import org.apache.iotdb.common.rpc.thrift.TTimePartitionSlot;
import org.apache.iotdb.common.rpc.thrift.TTimedQuota;
import org.apache.iotdb.common.rpc.thrift.ThrottleType;
-import org.apache.iotdb.commons.auth.AuthException;
import org.apache.iotdb.commons.consensus.index.impl.IoTProgressIndex;
import org.apache.iotdb.commons.consensus.index.impl.MinimumProgressIndex;
import org.apache.iotdb.commons.exception.IllegalPathException;
@@ -111,6 +110,7 @@ import org.apache.iotdb.confignode.consensus.request.write.pipe.payload.PipeDeac
import org.apache.iotdb.confignode.consensus.request.write.pipe.payload.PipeDeleteLogicalViewPlan;
import org.apache.iotdb.confignode.consensus.request.write.pipe.payload.PipeDeleteTimeSeriesPlan;
import org.apache.iotdb.confignode.consensus.request.write.pipe.payload.PipeEnrichedPlan;
+import org.apache.iotdb.confignode.consensus.request.write.pipe.payload.PipeSetTTLPlan;
import org.apache.iotdb.confignode.consensus.request.write.pipe.payload.PipeUnsetSchemaTemplatePlan;
import org.apache.iotdb.confignode.consensus.request.write.pipe.plugin.CreatePipePluginPlan;
import org.apache.iotdb.confignode.consensus.request.write.pipe.plugin.DropPipePluginPlan;
@@ -145,6 +145,7 @@ import org.apache.iotdb.confignode.consensus.request.write.sync.SetPipeStatusPla
import org.apache.iotdb.confignode.consensus.request.write.sync.ShowPipePlanV1;
import org.apache.iotdb.confignode.consensus.request.write.template.CreateSchemaTemplatePlan;
import org.apache.iotdb.confignode.consensus.request.write.template.DropSchemaTemplatePlan;
+import org.apache.iotdb.confignode.consensus.request.write.template.ExtendSchemaTemplatePlan;
import org.apache.iotdb.confignode.consensus.request.write.template.PreUnsetSchemaTemplatePlan;
import org.apache.iotdb.confignode.consensus.request.write.template.RollbackPreUnsetSchemaTemplatePlan;
import org.apache.iotdb.confignode.consensus.request.write.template.SetSchemaTemplatePlan;
@@ -166,6 +167,7 @@ import org.apache.iotdb.confignode.rpc.thrift.TShowRegionReq;
import org.apache.iotdb.confignode.rpc.thrift.TTimeSlotList;
import org.apache.iotdb.confignode.rpc.thrift.TTriggerState;
import org.apache.iotdb.db.schemaengine.template.Template;
+import org.apache.iotdb.db.schemaengine.template.alter.TemplateExtendInfo;
import org.apache.iotdb.trigger.api.enums.FailureStrategy;
import org.apache.iotdb.trigger.api.enums.TriggerEvent;
@@ -560,11 +562,12 @@ public class ConfigPhysicalPlanSerDeTest {
}
@Test
- public void AuthorPlanTest() throws IOException, AuthException, IllegalPathException {
-
+ public void AuthorPlanTest() throws IOException, IllegalPathException {
AuthorPlan req0;
AuthorPlan req1;
Set<Integer> permissions = new HashSet<>();
+ permissions.add(1);
+ permissions.add(2);
// create user
req0 =
@@ -966,6 +969,20 @@ public class ConfigPhysicalPlanSerDeTest {
return new Template(name, measurements, dataTypes, encodings, compressors);
}
+ @Test
+ public void ExtendSchemaTemplatePlanTest() throws IOException {
+ final ExtendSchemaTemplatePlan plan =
+ new ExtendSchemaTemplatePlan(
+ new TemplateExtendInfo(
+ "template_name",
+ Arrays.asList(
+ "template_name" + "_" + "temperature", "template_name" + "_" + "status"),
+ Arrays.asList(TSDataType.FLOAT, TSDataType.BOOLEAN),
+ Arrays.asList(TSEncoding.RLE, TSEncoding.PLAIN),
+ Arrays.asList(CompressionType.SNAPPY, CompressionType.SNAPPY)));
+ Assert.assertEquals(plan, ConfigPhysicalPlan.Factory.create(plan.serializeToByteBuffer()));
+ }
+
@Test
public void GetSchemaTemplatePlanTest() throws IOException {
GetSchemaTemplatePlan getSchemaTemplatePlan = new GetSchemaTemplatePlan("template1");
@@ -1869,72 +1886,78 @@ public class ConfigPhysicalPlanSerDeTest {
@Test
public void pipeEnrichedPlanTest() throws IOException {
- DatabaseSchemaPlan req0 =
- new DatabaseSchemaPlan(
- ConfigPhysicalPlanType.CreateDatabase,
- new TDatabaseSchema()
- .setName("sg")
- .setTTL(Long.MAX_VALUE)
- .setSchemaReplicationFactor(3)
- .setDataReplicationFactor(3)
- .setTimePartitionInterval(604800));
- PipeEnrichedPlan plan = new PipeEnrichedPlan(req0);
- PipeEnrichedPlan plan1 =
- (PipeEnrichedPlan) ConfigPhysicalPlan.Factory.create(plan.serializeToByteBuffer());
- Assert.assertEquals(plan, plan1);
+ final PipeEnrichedPlan plan =
+ new PipeEnrichedPlan(
+ new DatabaseSchemaPlan(
+ ConfigPhysicalPlanType.CreateDatabase,
+ new TDatabaseSchema()
+ .setName("sg")
+ .setTTL(Long.MAX_VALUE)
+ .setSchemaReplicationFactor(3)
+ .setDataReplicationFactor(3)
+ .setTimePartitionInterval(604800)));
+ Assert.assertEquals(plan, ConfigPhysicalPlan.Factory.create(plan.serializeToByteBuffer()));
}
@Test
public void pipeUnsetSchemaTemplatePlanTest() throws IOException {
- PipeUnsetSchemaTemplatePlan pipeUnsetSchemaTemplatePlan =
+ final PipeUnsetSchemaTemplatePlan pipeUnsetSchemaTemplatePlan =
new PipeUnsetSchemaTemplatePlan("template0", "root.sg");
- PipeUnsetSchemaTemplatePlan deserializedPlan =
- (PipeUnsetSchemaTemplatePlan)
- ConfigPhysicalPlan.Factory.create(pipeUnsetSchemaTemplatePlan.serializeToByteBuffer());
- Assert.assertEquals(pipeUnsetSchemaTemplatePlan, deserializedPlan);
+ Assert.assertEquals(
+ pipeUnsetSchemaTemplatePlan,
+ ConfigPhysicalPlan.Factory.create(pipeUnsetSchemaTemplatePlan.serializeToByteBuffer()));
}
@Test
public void pipeDeleteTimeSeriesPlanTest() throws IOException, IllegalPathException {
- PathPatternTree patternTree = new PathPatternTree();
+ final PathPatternTree patternTree = new PathPatternTree();
patternTree.appendPathPattern(new PartialPath("root.**.s1"));
patternTree.constructTree();
- PipeDeleteTimeSeriesPlan pipeDeleteTimeSeriesPlan =
+ final PipeDeleteTimeSeriesPlan pipeDeleteTimeSeriesPlan =
new PipeDeleteTimeSeriesPlan(patternTree.serialize());
- PipeDeleteTimeSeriesPlan deserializedPlan =
- (PipeDeleteTimeSeriesPlan)
- ConfigPhysicalPlan.Factory.create(pipeDeleteTimeSeriesPlan.serializeToByteBuffer());
- Assert.assertEquals(pipeDeleteTimeSeriesPlan, deserializedPlan);
+ Assert.assertEquals(
+ pipeDeleteTimeSeriesPlan,
+ ConfigPhysicalPlan.Factory.create(pipeDeleteTimeSeriesPlan.serializeToByteBuffer()));
}
@Test
public void pipeDeleteLogicalViewPlanTest() throws IOException, IllegalPathException {
- PathPatternTree patternTree = new PathPatternTree();
+ final PathPatternTree patternTree = new PathPatternTree();
patternTree.appendPathPattern(new PartialPath("root.**.s1"));
patternTree.constructTree();
- PipeDeleteLogicalViewPlan pipeDeleteLogicalViewPlan =
+ final PipeDeleteLogicalViewPlan pipeDeleteLogicalViewPlan =
new PipeDeleteLogicalViewPlan(patternTree.serialize());
- PipeDeleteLogicalViewPlan deserializedPlan =
- (PipeDeleteLogicalViewPlan)
- ConfigPhysicalPlan.Factory.create(pipeDeleteLogicalViewPlan.serializeToByteBuffer());
- Assert.assertEquals(pipeDeleteLogicalViewPlan, deserializedPlan);
+ Assert.assertEquals(
+ pipeDeleteLogicalViewPlan,
+ ConfigPhysicalPlan.Factory.create(pipeDeleteLogicalViewPlan.serializeToByteBuffer()));
}
@Test
public void pipeDeactivateTemplatePlanTest() throws IllegalPathException, IOException {
- Map<PartialPath, List<Template>> templateSetInfo = new HashMap<>();
- templateSetInfo.put(
- new PartialPath("root.**.s1"),
- Collections.singletonList(newSchemaTemplate("template_name")));
-
- PipeDeactivateTemplatePlan pipeDeactivateTemplatePlan =
- new PipeDeactivateTemplatePlan(templateSetInfo);
- PipeDeactivateTemplatePlan deserializedPlan =
- (PipeDeactivateTemplatePlan)
- ConfigPhysicalPlan.Factory.create(pipeDeactivateTemplatePlan.serializeToByteBuffer());
- Assert.assertEquals(pipeDeactivateTemplatePlan, deserializedPlan);
+ final PipeDeactivateTemplatePlan pipeDeactivateTemplatePlan =
+ new PipeDeactivateTemplatePlan(
+ new HashMap<PartialPath, List<Template>>() {
+ {
+ put(
+ new PartialPath("root.**.s1"),
+ Collections.singletonList(newSchemaTemplate("template_name")));
+ }
+ });
+ Assert.assertEquals(
+ pipeDeactivateTemplatePlan,
+ ConfigPhysicalPlan.Factory.create(pipeDeactivateTemplatePlan.serializeToByteBuffer()));
+ }
+
+ @Test
+ public void pipeSetTTLPlanTest() throws IOException {
+ final PipeSetTTLPlan plan =
+ new PipeSetTTLPlan(
+ Arrays.asList(
+ new SetTTLPlan(Arrays.asList("root", "db", "**", "a", "**"), Long.MAX_VALUE),
+ new SetTTLPlan(Arrays.asList("root", "db", "a", "**"), Long.MAX_VALUE)));
+ Assert.assertEquals(plan, ConfigPhysicalPlan.Factory.create(plan.serializeToByteBuffer()));
}
@Test
diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/pipe/connector/PipeConfigNodeThriftRequestTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/pipe/connector/PipeConfigNodeThriftRequestTest.java
index 934fa214e06..c2c04f94529 100644
--- a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/pipe/connector/PipeConfigNodeThriftRequestTest.java
+++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/pipe/connector/PipeConfigNodeThriftRequestTest.java
@@ -89,7 +89,7 @@ public class PipeConfigNodeThriftRequestTest {
PipeTransferConfigSnapshotSealReq req =
PipeTransferConfigSnapshotSealReq.toTPipeTransferReq(
- snapshotName, 100, templateInfoName, 10, fileType, typeString);
+ "root.**", snapshotName, 100, templateInfoName, 10, fileType, typeString);
PipeTransferConfigSnapshotSealReq deserializeReq =
PipeTransferConfigSnapshotSealReq.fromTPipeTransferReq(req);
diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/pipe/extractor/IoTDBConfigRegionExtractorTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/pipe/extractor/IoTDBConfigRegionExtractorTest.java
index 46f514dbaf1..f3ccf42a0c2 100644
--- a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/pipe/extractor/IoTDBConfigRegionExtractorTest.java
+++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/pipe/extractor/IoTDBConfigRegionExtractorTest.java
@@ -30,9 +30,8 @@ import java.util.HashMap;
public class IoTDBConfigRegionExtractorTest {
@Test
- public void testIoTDBSchemaExtractor() {
- IoTDBConfigRegionExtractor extractor = new IoTDBConfigRegionExtractor();
- try {
+ public void testIoTDBConfigExtractor() {
+ try (final IoTDBConfigRegionExtractor extractor = new IoTDBConfigRegionExtractor()) {
extractor.validate(
new PipeParameterValidator(
new PipeParameters(
@@ -52,7 +51,7 @@ public class IoTDBConfigRegionExtractorTest {
PipeExtractorConstant.EXTRACTOR_INCLUSION_DEFAULT_VALUE);
}
})));
- } catch (Exception e) {
+ } catch (final Exception e) {
Assert.fail();
}
}
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
new file mode 100644
index 00000000000..a2ff81a9da4
--- /dev/null
+++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/pipe/extractor/PipeConfigPhysicalPlanPatternParseVisitorTest.java
@@ -0,0 +1,408 @@
+/*
+ * 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.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.pattern.IoTDBPipePattern;
+import org.apache.iotdb.confignode.consensus.request.ConfigPhysicalPlanType;
+import org.apache.iotdb.confignode.consensus.request.auth.AuthorPlan;
+import org.apache.iotdb.confignode.consensus.request.write.database.DatabaseSchemaPlan;
+import org.apache.iotdb.confignode.consensus.request.write.database.DeleteDatabasePlan;
+import org.apache.iotdb.confignode.consensus.request.write.database.SetTTLPlan;
+import org.apache.iotdb.confignode.consensus.request.write.pipe.payload.PipeDeactivateTemplatePlan;
+import org.apache.iotdb.confignode.consensus.request.write.pipe.payload.PipeDeleteLogicalViewPlan;
+import org.apache.iotdb.confignode.consensus.request.write.pipe.payload.PipeDeleteTimeSeriesPlan;
+import org.apache.iotdb.confignode.consensus.request.write.pipe.payload.PipeSetTTLPlan;
+import org.apache.iotdb.confignode.consensus.request.write.pipe.payload.PipeUnsetSchemaTemplatePlan;
+import org.apache.iotdb.confignode.consensus.request.write.template.CommitSetSchemaTemplatePlan;
+import org.apache.iotdb.confignode.consensus.request.write.template.CreateSchemaTemplatePlan;
+import org.apache.iotdb.confignode.consensus.request.write.template.ExtendSchemaTemplatePlan;
+import org.apache.iotdb.confignode.rpc.thrift.TDatabaseSchema;
+import org.apache.iotdb.db.schemaengine.template.Template;
+import org.apache.iotdb.db.schemaengine.template.alter.TemplateExtendInfo;
+
+import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.file.metadata.enums.CompressionType;
+import org.apache.tsfile.file.metadata.enums.TSEncoding;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.stream.Collectors;
+
+public class PipeConfigPhysicalPlanPatternParseVisitorTest {
+
+ private final IoTDBPipePattern prefixPathPattern = new IoTDBPipePattern("root.db.device.**");
+ private final IoTDBPipePattern fullPathPattern = new IoTDBPipePattern("root.db.device.s1");
+
+ @Test
+ public void testCreateDatabase() {
+ final DatabaseSchemaPlan createDatabasePlan =
+ new DatabaseSchemaPlan(
+ ConfigPhysicalPlanType.CreateDatabase, new TDatabaseSchema("root.db"));
+ final DatabaseSchemaPlan createDatabasePlanToFilter =
+ new DatabaseSchemaPlan(
+ ConfigPhysicalPlanType.CreateDatabase, new TDatabaseSchema("root.db1"));
+
+ Assert.assertEquals(
+ createDatabasePlan,
+ IoTDBConfigRegionExtractor.PATTERN_PARSE_VISITOR
+ .visitCreateDatabase(createDatabasePlan, prefixPathPattern)
+ .orElseThrow(AssertionError::new));
+ Assert.assertFalse(
+ IoTDBConfigRegionExtractor.PATTERN_PARSE_VISITOR
+ .visitCreateDatabase(createDatabasePlanToFilter, prefixPathPattern)
+ .isPresent());
+ }
+
+ @Test
+ public void testAlterDatabase() {
+ final DatabaseSchemaPlan alterDatabasePlan =
+ new DatabaseSchemaPlan(
+ ConfigPhysicalPlanType.AlterDatabase, new TDatabaseSchema("root.db"));
+ final DatabaseSchemaPlan alterDatabasePlanToFilter =
+ new DatabaseSchemaPlan(
+ ConfigPhysicalPlanType.AlterDatabase, new TDatabaseSchema("root.db1"));
+
+ Assert.assertEquals(
+ alterDatabasePlan,
+ IoTDBConfigRegionExtractor.PATTERN_PARSE_VISITOR
+ .visitAlterDatabase(alterDatabasePlan, prefixPathPattern)
+ .orElseThrow(AssertionError::new));
+ Assert.assertFalse(
+ IoTDBConfigRegionExtractor.PATTERN_PARSE_VISITOR
+ .visitAlterDatabase(alterDatabasePlanToFilter, prefixPathPattern)
+ .isPresent());
+ }
+
+ @Test
+ public void testDeleteDatabase() {
+ final DeleteDatabasePlan deleteDatabasePlan = new DeleteDatabasePlan("root.db");
+ final DeleteDatabasePlan deleteDatabasePlanToFilter = new DeleteDatabasePlan("root.db1");
+
+ Assert.assertEquals(
+ deleteDatabasePlan,
+ IoTDBConfigRegionExtractor.PATTERN_PARSE_VISITOR
+ .visitDeleteDatabase(deleteDatabasePlan, prefixPathPattern)
+ .orElseThrow(AssertionError::new));
+ Assert.assertFalse(
+ IoTDBConfigRegionExtractor.PATTERN_PARSE_VISITOR
+ .visitDeleteDatabase(deleteDatabasePlanToFilter, prefixPathPattern)
+ .isPresent());
+ }
+
+ @Test
+ public void testCreateSchemaTemplate() throws IllegalPathException {
+ final CreateSchemaTemplatePlan createSchemaTemplatePlan =
+ new CreateSchemaTemplatePlan(
+ new Template(
+ "template_name",
+ Arrays.asList("s1", "s2"),
+ Arrays.asList(TSDataType.FLOAT, TSDataType.BOOLEAN),
+ Arrays.asList(TSEncoding.RLE, TSEncoding.PLAIN),
+ Arrays.asList(CompressionType.SNAPPY, CompressionType.SNAPPY))
+ .serialize()
+ .array());
+
+ Assert.assertEquals(
+ createSchemaTemplatePlan,
+ IoTDBConfigRegionExtractor.PATTERN_PARSE_VISITOR
+ .visitCreateSchemaTemplate(createSchemaTemplatePlan, prefixPathPattern)
+ .orElseThrow(AssertionError::new));
+
+ final CreateSchemaTemplatePlan parsedTemplatePlan =
+ (CreateSchemaTemplatePlan)
+ IoTDBConfigRegionExtractor.PATTERN_PARSE_VISITOR
+ .visitCreateSchemaTemplate(createSchemaTemplatePlan, fullPathPattern)
+ .orElseThrow(AssertionError::new);
+ Assert.assertEquals(
+ Collections.singleton("s1"), parsedTemplatePlan.getTemplate().getSchemaMap().keySet());
+ Assert.assertEquals(
+ createSchemaTemplatePlan.getTemplate().getSchemaMap().get("s1"),
+ parsedTemplatePlan.getTemplate().getSchemaMap().get("s1"));
+ }
+
+ @Test
+ public void testCommitSetSchemaTemplate() {
+ final CommitSetSchemaTemplatePlan setSchemaTemplatePlanOnPrefix =
+ new CommitSetSchemaTemplatePlan("t1", "root.db");
+ final CommitSetSchemaTemplatePlan setSchemaTemplatePlanOnFullPath =
+ new CommitSetSchemaTemplatePlan("t1", "root.db.device.s1");
+
+ Assert.assertEquals(
+ setSchemaTemplatePlanOnPrefix,
+ IoTDBConfigRegionExtractor.PATTERN_PARSE_VISITOR
+ .visitCommitSetSchemaTemplate(setSchemaTemplatePlanOnPrefix, fullPathPattern)
+ .orElseThrow(AssertionError::new));
+ Assert.assertEquals(
+ setSchemaTemplatePlanOnFullPath,
+ IoTDBConfigRegionExtractor.PATTERN_PARSE_VISITOR
+ .visitCommitSetSchemaTemplate(setSchemaTemplatePlanOnFullPath, fullPathPattern)
+ .orElseThrow(AssertionError::new));
+ }
+
+ @Test
+ public void testPipeUnsetSchemaTemplate() {
+ final PipeUnsetSchemaTemplatePlan pipeUnsetSchemaTemplatePlanOnPrefix =
+ new PipeUnsetSchemaTemplatePlan("t1", "root.db");
+ final PipeUnsetSchemaTemplatePlan pipeUnsetSchemaTemplatePlanOrFullPath =
+ new PipeUnsetSchemaTemplatePlan("t1", "root.db.device.s1");
+
+ Assert.assertEquals(
+ pipeUnsetSchemaTemplatePlanOnPrefix,
+ IoTDBConfigRegionExtractor.PATTERN_PARSE_VISITOR
+ .visitPipeUnsetSchemaTemplate(pipeUnsetSchemaTemplatePlanOnPrefix, fullPathPattern)
+ .orElseThrow(AssertionError::new));
+ Assert.assertEquals(
+ pipeUnsetSchemaTemplatePlanOrFullPath,
+ IoTDBConfigRegionExtractor.PATTERN_PARSE_VISITOR
+ .visitPipeUnsetSchemaTemplate(pipeUnsetSchemaTemplatePlanOrFullPath, fullPathPattern)
+ .orElseThrow(AssertionError::new));
+ }
+
+ @Test
+ public void testExtendSchemaTemplate() {
+ final ExtendSchemaTemplatePlan extendSchemaTemplatePlan =
+ new ExtendSchemaTemplatePlan(
+ new TemplateExtendInfo(
+ "template_name",
+ Arrays.asList("s1", "s2"),
+ Arrays.asList(TSDataType.FLOAT, TSDataType.BOOLEAN),
+ Arrays.asList(TSEncoding.RLE, TSEncoding.PLAIN),
+ Arrays.asList(CompressionType.SNAPPY, CompressionType.SNAPPY)));
+
+ final ExtendSchemaTemplatePlan parsedTemplatePlan =
+ (ExtendSchemaTemplatePlan)
+ IoTDBConfigRegionExtractor.PATTERN_PARSE_VISITOR
+ .visitExtendSchemaTemplate(extendSchemaTemplatePlan, fullPathPattern)
+ .orElseThrow(AssertionError::new);
+ Assert.assertEquals(
+ Collections.singletonList("s1"),
+ parsedTemplatePlan.getTemplateExtendInfo().getMeasurements());
+ Assert.assertEquals(
+ extendSchemaTemplatePlan.getTemplateExtendInfo().getTemplateName(),
+ parsedTemplatePlan.getTemplateExtendInfo().getTemplateName());
+ Assert.assertEquals(
+ extendSchemaTemplatePlan.getTemplateExtendInfo().getDataTypes().get(0),
+ parsedTemplatePlan.getTemplateExtendInfo().getDataTypes().get(0));
+ Assert.assertEquals(
+ extendSchemaTemplatePlan.getTemplateExtendInfo().getEncodings().get(0),
+ parsedTemplatePlan.getTemplateExtendInfo().getEncodings().get(0));
+ Assert.assertEquals(
+ extendSchemaTemplatePlan.getTemplateExtendInfo().getCompressors().get(0),
+ parsedTemplatePlan.getTemplateExtendInfo().getCompressors().get(0));
+ }
+
+ @Test
+ public void testGrantUser() throws IllegalPathException {
+ Assert.assertEquals(
+ Collections.singletonList(new PartialPath("root.db.device.**")),
+ ((AuthorPlan)
+ IoTDBConfigRegionExtractor.PATTERN_PARSE_VISITOR
+ .visitGrantUser(
+ new AuthorPlan(
+ ConfigPhysicalPlanType.GrantUser,
+ "tempUser",
+ "",
+ "",
+ "",
+ new HashSet<>(Arrays.asList(1, 2)),
+ true,
+ Arrays.asList(
+ new PartialPath("root.db.**"), new PartialPath("root.abc.**"))),
+ prefixPathPattern)
+ .orElseThrow(AssertionError::new))
+ .getNodeNameList());
+ }
+
+ @Test
+ public void testRevokeUser() throws IllegalPathException {
+ Assert.assertEquals(
+ Collections.singletonList(new PartialPath("root.db.device.**")),
+ ((AuthorPlan)
+ IoTDBConfigRegionExtractor.PATTERN_PARSE_VISITOR
+ .visitRevokeUser(
+ new AuthorPlan(
+ ConfigPhysicalPlanType.RevokeUser,
+ "tempUser",
+ "",
+ "",
+ "",
+ new HashSet<>(Arrays.asList(1, 2)),
+ false,
+ Arrays.asList(
+ new PartialPath("root.db.**"), new PartialPath("root.abc.**"))),
+ prefixPathPattern)
+ .orElseThrow(AssertionError::new))
+ .getNodeNameList());
+ }
+
+ @Test
+ public void testGrantRole() throws IllegalPathException {
+ Assert.assertEquals(
+ Collections.singletonList(new PartialPath("root.db.device.**")),
+ ((AuthorPlan)
+ IoTDBConfigRegionExtractor.PATTERN_PARSE_VISITOR
+ .visitGrantRole(
+ new AuthorPlan(
+ ConfigPhysicalPlanType.GrantRole,
+ "",
+ "tempRole",
+ "",
+ "",
+ new HashSet<>(Arrays.asList(1, 2)),
+ true,
+ Arrays.asList(
+ new PartialPath("root.db.**"), new PartialPath("root.abc.**"))),
+ prefixPathPattern)
+ .orElseThrow(AssertionError::new))
+ .getNodeNameList());
+ }
+
+ @Test
+ public void testRevokeRole() throws IllegalPathException {
+ Assert.assertEquals(
+ Collections.singletonList(new PartialPath("root.db.device.**")),
+ ((AuthorPlan)
+ IoTDBConfigRegionExtractor.PATTERN_PARSE_VISITOR
+ .visitRevokeRole(
+ new AuthorPlan(
+ ConfigPhysicalPlanType.RevokeRole,
+ "",
+ "tempRole",
+ "",
+ "",
+ new HashSet<>(Arrays.asList(1, 2)),
+ false,
+ Arrays.asList(
+ new PartialPath("root.db.**"), new PartialPath("root.abc.**"))),
+ prefixPathPattern)
+ .orElseThrow(AssertionError::new))
+ .getNodeNameList());
+ }
+
+ @Test
+ public void testPipeDeleteTimeSeries() throws IllegalPathException, IOException {
+ final PathPatternTree patternTree = new PathPatternTree();
+ patternTree.appendPathPattern(new PartialPath("root.*.device.s1"));
+ patternTree.constructTree();
+
+ Assert.assertEquals(
+ Collections.singletonList(new PartialPath("root.db.device.s1")),
+ PathPatternTree.deserialize(
+ ((PipeDeleteTimeSeriesPlan)
+ IoTDBConfigRegionExtractor.PATTERN_PARSE_VISITOR
+ .visitPipeDeleteTimeSeries(
+ new PipeDeleteTimeSeriesPlan(patternTree.serialize()),
+ prefixPathPattern)
+ .orElseThrow(AssertionError::new))
+ .getPatternTreeBytes())
+ .getAllPathPatterns());
+ }
+
+ @Test
+ public void testPipeDeleteLogicalView() throws IllegalPathException, IOException {
+ final PathPatternTree patternTree = new PathPatternTree();
+ patternTree.appendPathPattern(new PartialPath("root.*.device.s1"));
+ patternTree.constructTree();
+
+ Assert.assertEquals(
+ Collections.singletonList(new PartialPath("root.db.device.s1")),
+ PathPatternTree.deserialize(
+ ((PipeDeleteTimeSeriesPlan)
+ IoTDBConfigRegionExtractor.PATTERN_PARSE_VISITOR
+ .visitPipeDeleteLogicalView(
+ new PipeDeleteLogicalViewPlan(patternTree.serialize()),
+ prefixPathPattern)
+ .orElseThrow(AssertionError::new))
+ .getPatternTreeBytes())
+ .getAllPathPatterns());
+ }
+
+ @Test
+ public void testPipeDeactivateTemplate() throws IllegalPathException {
+ final Template template1 = newSchemaTemplate("template1");
+ final Template template2 = newSchemaTemplate("template2");
+ final Template template3 = newSchemaTemplate("template3");
+
+ Assert.assertEquals(
+ new HashMap<PartialPath, List<Template>>() {
+ {
+ put(
+ new PartialPath("root.db.device.s1"),
+ Arrays.asList(template1, template2, template3));
+ }
+ },
+ ((PipeDeactivateTemplatePlan)
+ IoTDBConfigRegionExtractor.PATTERN_PARSE_VISITOR
+ .visitPipeDeactivateTemplate(
+ new PipeDeactivateTemplatePlan(
+ new HashMap<PartialPath, List<Template>>() {
+ {
+ put(
+ new PartialPath("root.*.device.s1"),
+ Collections.singletonList(template1));
+ put(
+ new PartialPath("root.db.*.s1"),
+ Arrays.asList(template2, template3));
+ }
+ }),
+ prefixPathPattern)
+ .orElseThrow(AssertionError::new))
+ .getTemplateSetInfo());
+ }
+
+ private Template newSchemaTemplate(final String name) throws IllegalPathException {
+ return new Template(
+ name,
+ Arrays.asList(name + "_" + "temperature", name + "_" + "status"),
+ Arrays.asList(TSDataType.FLOAT, TSDataType.BOOLEAN),
+ Arrays.asList(TSEncoding.RLE, TSEncoding.PLAIN),
+ Arrays.asList(CompressionType.SNAPPY, CompressionType.SNAPPY));
+ }
+
+ @Test
+ public void testSetTTL() throws IllegalPathException {
+ final List<SetTTLPlan> plans =
+ ((PipeSetTTLPlan)
+ IoTDBConfigRegionExtractor.PATTERN_PARSE_VISITOR
+ .visitTTL(
+ new SetTTLPlan(Arrays.asList("root", "*", "device", "s1"), Long.MAX_VALUE),
+ prefixPathPattern)
+ .orElseThrow(AssertionError::new))
+ .getSetTTLPlans();
+
+ Assert.assertEquals(
+ Collections.singletonList(new PartialPath("root.db.device.s1")),
+ plans.stream()
+ .map(setTTLPlan -> new PartialPath(setTTLPlan.getDatabasePathPattern()))
+ .collect(Collectors.toList()));
+ Assert.assertEquals(
+ Collections.singletonList(Long.MAX_VALUE),
+ plans.stream().map(SetTTLPlan::getTTL).collect(Collectors.toList()));
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/payload/evolvable/request/PipeTransferSchemaSnapshotSealReq.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/payload/evolvable/request/PipeTransferSchemaSnapshotSealReq.java
index 9b896a00203..70835e0ea4f 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/payload/evolvable/request/PipeTransferSchemaSnapshotSealReq.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/payload/evolvable/request/PipeTransferSchemaSnapshotSealReq.java
@@ -45,14 +45,16 @@ public class PipeTransferSchemaSnapshotSealReq extends PipeTransferFileSealReqV2
/////////////////////////////// Thrift ///////////////////////////////
public static PipeTransferSchemaSnapshotSealReq toTPipeTransferReq(
- String mTreeSnapshotName,
- long mTreeSnapshotLength,
- String tLogName,
- long tLogLength,
- String databaseName,
- String typeString)
+ final String pattern,
+ final String mTreeSnapshotName,
+ final long mTreeSnapshotLength,
+ final String tLogName,
+ final long tLogLength,
+ final String databaseName,
+ final String typeString)
throws IOException {
final Map<String, String> parameters = new HashMap<>();
+ parameters.put(ColumnHeaderConstant.PATH_PATTERN, pattern);
parameters.put(ColumnHeaderConstant.DATABASE, databaseName);
parameters.put(ColumnHeaderConstant.TYPE, typeString);
return (PipeTransferSchemaSnapshotSealReq)
@@ -67,7 +69,7 @@ public class PipeTransferSchemaSnapshotSealReq extends PipeTransferFileSealReqV2
parameters);
}
- public static PipeTransferSchemaSnapshotSealReq fromTPipeTransferReq(TPipeTransferReq req) {
+ public static PipeTransferSchemaSnapshotSealReq fromTPipeTransferReq(final TPipeTransferReq req) {
return (PipeTransferSchemaSnapshotSealReq)
new PipeTransferSchemaSnapshotSealReq().translateFromTPipeTransferReq(req);
}
@@ -75,14 +77,16 @@ public class PipeTransferSchemaSnapshotSealReq extends PipeTransferFileSealReqV2
/////////////////////////////// Air Gap ///////////////////////////////
public static byte[] toTPipeTransferBytes(
- String mTreeSnapshotName,
- long mTreeSnapshotLength,
- String tLogName,
- long tLogLength,
- String databaseName,
- String typeString)
+ final String pattern,
+ final String mTreeSnapshotName,
+ final long mTreeSnapshotLength,
+ final String tLogName,
+ final long tLogLength,
+ final String databaseName,
+ final String typeString)
throws IOException {
final Map<String, String> parameters = new HashMap<>();
+ parameters.put(ColumnHeaderConstant.PATH_PATTERN, pattern);
parameters.put(ColumnHeaderConstant.DATABASE, databaseName);
parameters.put(ColumnHeaderConstant.TYPE, typeString);
return new PipeTransferSchemaSnapshotSealReq()
@@ -99,7 +103,7 @@ public class PipeTransferSchemaSnapshotSealReq extends PipeTransferFileSealReqV2
/////////////////////////////// Object ///////////////////////////////
@Override
- public boolean equals(Object obj) {
+ public boolean equals(final Object obj) {
return obj instanceof PipeTransferSchemaSnapshotSealReq && super.equals(obj);
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/airgap/IoTDBSchemaRegionAirGapConnector.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/airgap/IoTDBSchemaRegionAirGapConnector.java
index d4e6706a5a4..f390994e4f0 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/airgap/IoTDBSchemaRegionAirGapConnector.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/airgap/IoTDBSchemaRegionAirGapConnector.java
@@ -116,6 +116,8 @@ public class IoTDBSchemaRegionAirGapConnector extends IoTDBDataNodeAirGapConnect
socket,
compressIfNeeded(
PipeTransferSchemaSnapshotSealReq.toTPipeTransferBytes(
+ // The pattern is surely Non-null
+ pipeSchemaRegionSnapshotEvent.getPatternString(),
mtreeSnapshotFile.getName(),
mtreeSnapshotFile.length(),
Objects.nonNull(tagLogSnapshotFile) ? tagLogSnapshotFile.getName() : null,
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/sync/IoTDBSchemaRegionConnector.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/sync/IoTDBSchemaRegionConnector.java
index 667050320d0..2b2cba03f20 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/sync/IoTDBSchemaRegionConnector.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/sync/IoTDBSchemaRegionConnector.java
@@ -105,6 +105,8 @@ public class IoTDBSchemaRegionConnector extends IoTDBDataNodeSyncConnector {
.pipeTransfer(
compressIfNeeded(
PipeTransferSchemaSnapshotSealReq.toTPipeTransferReq(
+ // The pattern is surely Non-null
+ snapshotEvent.getPatternString(),
mTreeSnapshotFile.getName(),
mTreeSnapshotFile.length(),
Objects.nonNull(tagLogSnapshotFile) ? tagLogSnapshotFile.getName() : null,
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/schema/PipeSchemaRegionWritePlanEvent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/schema/PipeSchemaRegionWritePlanEvent.java
index 6e62fe80610..6e0b5cccf86 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
@@ -88,14 +88,14 @@ public class PipeSchemaRegionWritePlanEvent extends PipeWritePlanEvent {
@Override
public String toString() {
- return String.format("PipeConfigRegionWritePlanEvent{planNode=%s}", planNode)
+ return String.format("PipeSchemaRegionWritePlanEvent{planNode=%s}", planNode)
+ " - "
+ super.toString();
}
@Override
public String coreReportMessage() {
- return String.format("PipeConfigRegionWritePlanEvent{planNode=%s}", planNode)
+ return String.format("PipeSchemaRegionWritePlanEvent{planNode=%s}", planNode)
+ " - "
+ super.coreReportMessage();
}
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 e95bc5373b1..83fa0497f18 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
@@ -89,10 +89,10 @@ public class TabletInsertionDataContainer {
}
public TabletInsertionDataContainer(
- PipeTaskMeta pipeTaskMeta,
- EnrichedEvent sourceEvent,
- InsertNode insertNode,
- PipePattern pattern) {
+ final PipeTaskMeta pipeTaskMeta,
+ final EnrichedEvent sourceEvent,
+ final InsertNode insertNode,
+ final PipePattern pattern) {
this.pipeTaskMeta = pipeTaskMeta;
this.sourceEvent = sourceEvent;
@@ -107,11 +107,11 @@ public class TabletInsertionDataContainer {
}
public TabletInsertionDataContainer(
- PipeTaskMeta pipeTaskMeta,
- EnrichedEvent sourceEvent,
- Tablet tablet,
- boolean isAligned,
- PipePattern pattern) {
+ final PipeTaskMeta pipeTaskMeta,
+ final EnrichedEvent sourceEvent,
+ final Tablet tablet,
+ final boolean isAligned,
+ final PipePattern pattern) {
this.pipeTaskMeta = pipeTaskMeta;
this.sourceEvent = sourceEvent;
@@ -119,7 +119,7 @@ public class TabletInsertionDataContainer {
}
@TestOnly
- public TabletInsertionDataContainer(InsertNode insertNode, PipePattern pattern) {
+ public TabletInsertionDataContainer(final InsertNode insertNode, final PipePattern pattern) {
this(null, null, insertNode, pattern);
}
@@ -133,7 +133,7 @@ public class TabletInsertionDataContainer {
//////////////////////////// parse ////////////////////////////
- private void parse(InsertRowNode insertRowNode, PipePattern pattern) {
+ private void parse(final InsertRowNode insertRowNode, final PipePattern pattern) {
final int originColumnSize = insertRowNode.getMeasurements().length;
final Integer[] originColumnIndex2FilteredColumnIndexMapperList = new Integer[originColumnSize];
@@ -200,7 +200,7 @@ public class TabletInsertionDataContainer {
}
}
- private void parse(InsertTabletNode insertTabletNode, PipePattern pattern) {
+ private void parse(final InsertTabletNode insertTabletNode, final PipePattern pattern) {
final int originColumnSize = insertTabletNode.getMeasurements().length;
final Integer[] originColumnIndex2FilteredColumnIndexMapperList = new Integer[originColumnSize];
@@ -283,7 +283,7 @@ public class TabletInsertionDataContainer {
}
}
- private void parse(Tablet tablet, boolean isAligned, PipePattern pattern) {
+ private void parse(final Tablet tablet, final boolean isAligned, final PipePattern pattern) {
final int originColumnSize = tablet.getSchemas().size();
final Integer[] originColumnIndex2FilteredColumnIndexMapperList = new Integer[originColumnSize];
@@ -375,9 +375,9 @@ public class TabletInsertionDataContainer {
}
private void generateColumnIndexMapper(
- String[] originMeasurementList,
- PipePattern pattern,
- Integer[] originColumnIndex2FilteredColumnIndexMapperList) {
+ final String[] originMeasurementList,
+ final PipePattern pattern,
+ final Integer[] originColumnIndex2FilteredColumnIndexMapperList) {
final int originColumnSize = originMeasurementList.length;
// case 1: for example, pattern is root.a.b or pattern is null and device is root.a.b.c
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/TsFileInsertionDataContainer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/TsFileInsertionDataContainer.java
index 2081a540fea..36dfc3df458 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/TsFileInsertionDataContainer.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/TsFileInsertionDataContainer.java
@@ -78,17 +78,18 @@ public class TsFileInsertionDataContainer implements AutoCloseable {
private boolean shouldParsePattern = false;
public TsFileInsertionDataContainer(
- File tsFile, PipePattern pattern, long startTime, long endTime) throws IOException {
+ final File tsFile, final PipePattern pattern, final long startTime, final long endTime)
+ throws IOException {
this(tsFile, pattern, startTime, endTime, null, null);
}
public TsFileInsertionDataContainer(
- File tsFile,
- PipePattern pattern,
- long startTime,
- long endTime,
- PipeTaskMeta pipeTaskMeta,
- EnrichedEvent sourceEvent)
+ final File tsFile,
+ final PipePattern pattern,
+ final long startTime,
+ final long endTime,
+ final PipeTaskMeta pipeTaskMeta,
+ final EnrichedEvent sourceEvent)
throws IOException {
this.pattern = pattern;
timeFilterExpression =
@@ -135,16 +136,17 @@ public class TsFileInsertionDataContainer implements AutoCloseable {
// No longer need this. Help GC.
tsFileSequenceReader.clearCachedDeviceMetadata();
- } catch (Exception e) {
+ } catch (final Exception e) {
close();
throw e;
}
}
private Map<IDeviceID, List<String>> filterDeviceMeasurementsMapByPattern(
- Map<IDeviceID, List<String>> originalDeviceMeasurementsMap) {
+ final Map<IDeviceID, List<String>> originalDeviceMeasurementsMap) {
final Map<IDeviceID, List<String>> filteredDeviceMeasurementsMap = new HashMap<>();
- for (Map.Entry<IDeviceID, List<String>> entry : originalDeviceMeasurementsMap.entrySet()) {
+ for (final Map.Entry<IDeviceID, List<String>> entry :
+ originalDeviceMeasurementsMap.entrySet()) {
final String deviceId = ((PlainDeviceID) entry.getKey()).toStringID();
// case 1: for example, pattern is root.a.b or pattern is null and device is root.a.b.c
@@ -223,7 +225,7 @@ public class TsFileInsertionDataContainer implements AutoCloseable {
((PlainDeviceID) entry.getKey()).toStringID(),
entry.getValue(),
timeFilterExpression);
- } catch (IOException e) {
+ } catch (final IOException e) {
close();
throw new PipeException("failed to create TsFileInsertionDataTabletIterator", e);
}
@@ -279,7 +281,7 @@ public class TsFileInsertionDataContainer implements AutoCloseable {
if (tsFileReader != null) {
tsFileReader.close();
}
- } catch (IOException e) {
+ } catch (final IOException e) {
LOGGER.warn("Failed to close TsFileReader", e);
}
@@ -287,7 +289,7 @@ public class TsFileInsertionDataContainer implements AutoCloseable {
if (tsFileSequenceReader != null) {
tsFileSequenceReader.close();
}
- } catch (IOException e) {
+ } catch (final IOException e) {
LOGGER.warn("Failed to close TsFileSequenceReader", e);
}
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 086b03c4558..f76c220dd2b 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
@@ -52,7 +52,8 @@ public class TsFileInsertionPointCounter implements AutoCloseable {
private long count = 0;
- public TsFileInsertionPointCounter(File tsFile, PipePattern pattern) throws IOException {
+ public TsFileInsertionPointCounter(final File tsFile, final PipePattern pattern)
+ throws IOException {
this.pattern = pattern;
try {
@@ -69,7 +70,7 @@ public class TsFileInsertionPointCounter implements AutoCloseable {
// No longer need this. Help GC.
tsFileSequenceReader.clearCachedDeviceMetadata();
- } catch (Exception e) {
+ } catch (final Exception e) {
close();
throw e;
}
@@ -80,7 +81,8 @@ public class TsFileInsertionPointCounter implements AutoCloseable {
tsFileSequenceReader.getDeviceMeasurementsMap();
final Map<IDeviceID, Set<String>> filteredDeviceMeasurementsMap = new HashMap<>();
- for (Map.Entry<IDeviceID, List<String>> entry : originalDeviceMeasurementsMap.entrySet()) {
+ for (final Map.Entry<IDeviceID, List<String>> entry :
+ originalDeviceMeasurementsMap.entrySet()) {
final String deviceId = ((PlainDeviceID) entry.getKey()).toStringID();
// case 1: for example, pattern is root.a.b or pattern is null and device is root.a.b.c
@@ -162,7 +164,7 @@ public class TsFileInsertionPointCounter implements AutoCloseable {
if (tsFileSequenceReader != null) {
tsFileSequenceReader.close();
}
- } catch (IOException e) {
+ } catch (final IOException e) {
LOGGER.warn("Failed to close TsFileSequenceReader", e);
}
}
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 b8ee7eb0620..d89b223df5b 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
@@ -22,6 +22,7 @@ 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.extractor.IoTDBExtractor;
+import org.apache.iotdb.commons.pipe.pattern.IoTDBPipePattern;
import org.apache.iotdb.commons.pipe.pattern.PipePattern;
import org.apache.iotdb.consensus.ConsensusFactory;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
@@ -94,6 +95,7 @@ public class IoTDBDataRegionExtractor extends IoTDBExtractor {
private boolean hasNoExtractionNeed = true;
private boolean shouldExtractInsertion = false;
+ private boolean shouldExtractDeletion = false;
@Override
public void validate(final PipeParameterValidator validator) throws Exception {
@@ -108,6 +110,7 @@ public class IoTDBDataRegionExtractor extends IoTDBExtractor {
}
hasNoExtractionNeed = false;
shouldExtractInsertion = insertionDeletionListeningOptionPair.getLeft();
+ shouldExtractDeletion = insertionDeletionListeningOptionPair.getRight();
if (insertionDeletionListeningOptionPair.getLeft().equals(true)
&& IoTDBDescriptor.getInstance()
@@ -214,6 +217,16 @@ public class IoTDBDataRegionExtractor extends IoTDBExtractor {
if (!pattern.isLegal()) {
throw new IllegalArgumentException(String.format("Pattern \"%s\" is illegal.", pattern));
}
+
+ if (shouldExtractDeletion
+ && !(pattern instanceof IoTDBPipePattern
+ && (((IoTDBPipePattern) pattern).isPrefix()
+ || ((IoTDBPipePattern) 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));
+ }
}
private void constructHistoricalExtractor() {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/schemaregion/IoTDBSchemaRegionExtractor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/schemaregion/IoTDBSchemaRegionExtractor.java
index a0798e3818f..3682095999b 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/schemaregion/IoTDBSchemaRegionExtractor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/schemaregion/IoTDBSchemaRegionExtractor.java
@@ -23,6 +23,7 @@ import org.apache.iotdb.commons.consensus.SchemaRegionId;
import org.apache.iotdb.commons.pipe.datastructure.queue.listening.AbstractPipeListeningQueue;
import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
import org.apache.iotdb.commons.pipe.event.PipeSnapshotEvent;
+import org.apache.iotdb.commons.pipe.event.PipeWritePlanEvent;
import org.apache.iotdb.commons.pipe.extractor.IoTDBNonDataRegionExtractor;
import org.apache.iotdb.consensus.ConsensusFactory;
import org.apache.iotdb.consensus.exception.ConsensusException;
@@ -40,14 +41,16 @@ import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metedata.write.Alt
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.pipe.PipeOperateSchemaQueueNode;
import org.apache.iotdb.pipe.api.customizer.configuration.PipeExtractorRuntimeConfiguration;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
-import org.apache.iotdb.pipe.api.event.Event;
import org.apache.iotdb.pipe.api.exception.PipeException;
import java.util.HashSet;
import java.util.Objects;
+import java.util.Optional;
import java.util.Set;
public class IoTDBSchemaRegionExtractor extends IoTDBNonDataRegionExtractor {
+ public static final PipePlanPatternParseVisitor PATTERN_PARSE_VISITOR =
+ new PipePlanPatternParseVisitor();
private SchemaRegionId schemaRegionId;
@@ -120,13 +123,21 @@ public class IoTDBSchemaRegionExtractor extends IoTDBNonDataRegionExtractor {
return 0;
}
+ @Override
+ protected Optional<PipeWritePlanEvent> trimRealtimeEventByPipePattern(
+ final PipeWritePlanEvent event) {
+ return PATTERN_PARSE_VISITOR
+ .process(((PipeSchemaRegionWritePlanEvent) event).getPlanNode(), pipePattern)
+ .map(planNode -> new PipeSchemaRegionWritePlanEvent(planNode, event.isGeneratedByPipe()));
+ }
+
@Override
protected AbstractPipeListeningQueue getListeningQueue() {
return PipeAgent.runtime().schemaListener(schemaRegionId);
}
@Override
- protected boolean isTypeListened(final Event event) {
+ protected boolean isTypeListened(final PipeWritePlanEvent event) {
final PlanNode planNode = ((PipeSchemaRegionWritePlanEvent) event).getPlanNode();
return listenedTypeSet.contains(
(planNode.getType() == PlanNodeType.ALTER_TIME_SERIES
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
new file mode 100644
index 00000000000..150775884aa
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/schemaregion/PipePlanPatternParseVisitor.java
@@ -0,0 +1,282 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.pipe.extractor.schemaregion;
+
+import org.apache.iotdb.commons.path.PartialPath;
+import org.apache.iotdb.commons.pipe.pattern.IoTDBPipePattern;
+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;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanVisitor;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metedata.write.ActivateTemplateNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metedata.write.AlterTimeSeriesNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metedata.write.BatchActivateTemplateNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metedata.write.CreateAlignedTimeSeriesNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metedata.write.CreateMultiTimeSeriesNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metedata.write.CreateTimeSeriesNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metedata.write.InternalBatchActivateTemplateNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metedata.write.InternalCreateMultiTimeSeriesNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metedata.write.InternalCreateTimeSeriesNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metedata.write.MeasurementGroup;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metedata.write.view.AlterLogicalViewNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metedata.write.view.CreateLogicalViewNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.DeleteDataNode;
+
+import org.apache.tsfile.utils.Pair;
+
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+/**
+ * The {@link PipePlanPatternParseVisitor} will transform the schema {@link PlanNode}s using {@link
+ * IoTDBPipePattern}. 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}.
+ *
+ * <p>2. If a pattern does not intersect with the {@link IoTDBPipePattern}, 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> {
+ @Override
+ public Optional<PlanNode> visitPlan(final PlanNode node, final IoTDBPipePattern pattern) {
+ return Optional.of(node);
+ }
+
+ @Override
+ public Optional<PlanNode> visitCreateTimeSeries(
+ final CreateTimeSeriesNode node, final IoTDBPipePattern pattern) {
+ return pattern.matchesMeasurement(node.getPath().getDevice(), node.getPath().getMeasurement())
+ ? Optional.of(node)
+ : Optional.empty();
+ }
+
+ @Override
+ public Optional<PlanNode> visitCreateAlignedTimeSeries(
+ final CreateAlignedTimeSeriesNode node, final IoTDBPipePattern pattern) {
+ final int[] filteredIndexes =
+ IntStream.range(0, node.getMeasurements().size())
+ .filter(
+ index ->
+ pattern.matchesMeasurement(
+ node.getDevicePath().getFullPath(), node.getMeasurements().get(index)))
+ .toArray();
+ return filteredIndexes.length > 0
+ ? Optional.of(
+ 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())))
+ : Optional.empty();
+ }
+
+ @Override
+ public Optional<PlanNode> visitCreateMultiTimeSeries(
+ final CreateMultiTimeSeriesNode node, final IoTDBPipePattern pattern) {
+ final Map<PartialPath, MeasurementGroup> filteredMeasurementGroupMap =
+ node.getMeasurementGroupMap().entrySet().stream()
+ .filter(entry -> pattern.matchPrefixPath(entry.getKey().getFullPath()))
+ .map(
+ entry ->
+ new Pair<>(
+ entry.getKey(),
+ trimMeasurementGroup(
+ entry.getKey().getFullPath(), entry.getValue(), pattern)))
+ .filter(pair -> Objects.nonNull(pair.getRight()))
+ .collect(Collectors.toMap(Pair::getLeft, Pair::getRight));
+ return !filteredMeasurementGroupMap.isEmpty()
+ ? Optional.of(
+ new CreateMultiTimeSeriesNode(node.getPlanNodeId(), filteredMeasurementGroupMap))
+ : Optional.empty();
+ }
+
+ private static MeasurementGroup trimMeasurementGroup(
+ final String device, final MeasurementGroup group, final IoTDBPipePattern pattern) {
+ final int[] filteredIndexes =
+ IntStream.range(0, group.size())
+ .filter(index -> pattern.matchesMeasurement(device, group.getMeasurements().get(index)))
+ .toArray();
+ if (filteredIndexes.length == 0) {
+ return null;
+ }
+ final MeasurementGroup targetMeasurementGroup = new MeasurementGroup();
+ Arrays.stream(filteredIndexes)
+ .forEach(
+ index -> {
+ targetMeasurementGroup.addMeasurement(
+ group.getMeasurements().get(index),
+ group.getDataTypes().get(index),
+ group.getEncodings().get(index),
+ group.getCompressors().get(index));
+ targetMeasurementGroup.addTags(group.getTagsList().get(index));
+ targetMeasurementGroup.addAttributes(group.getAttributesList().get(index));
+ targetMeasurementGroup.addAlias(group.getAliasList().get(index));
+ targetMeasurementGroup.addProps(group.getPropsList().get(index));
+ });
+ return targetMeasurementGroup;
+ }
+
+ @Override
+ public Optional<PlanNode> visitAlterTimeSeries(
+ final AlterTimeSeriesNode node, final IoTDBPipePattern pattern) {
+ return pattern.matchesMeasurement(node.getPath().getDevice(), node.getPath().getMeasurement())
+ ? Optional.of(node)
+ : Optional.empty();
+ }
+
+ @Override
+ public Optional<PlanNode> visitInternalCreateTimeSeries(
+ final InternalCreateTimeSeriesNode node, final IoTDBPipePattern pattern) {
+ final MeasurementGroup group =
+ pattern.matchPrefixPath(node.getDevicePath().getFullPath())
+ ? trimMeasurementGroup(
+ node.getDevicePath().getFullPath(), node.getMeasurementGroup(), pattern)
+ : null;
+ return Objects.nonNull(group)
+ ? Optional.of(
+ new InternalCreateTimeSeriesNode(
+ node.getPlanNodeId(), node.getDevicePath(), group, node.isAligned()))
+ : Optional.empty();
+ }
+
+ @Override
+ public Optional<PlanNode> visitActivateTemplate(
+ final ActivateTemplateNode node, final IoTDBPipePattern pattern) {
+ return pattern.matchDevice(node.getActivatePath().getFullPath())
+ ? Optional.of(node)
+ : Optional.empty();
+ }
+
+ @Override
+ public Optional<PlanNode> visitInternalBatchActivateTemplate(
+ final InternalBatchActivateTemplateNode node, final IoTDBPipePattern pattern) {
+ final Map<PartialPath, Pair<Integer, Integer>> filteredTemplateActivationMap =
+ node.getTemplateActivationMap().entrySet().stream()
+ .filter(entry -> pattern.matchDevice(entry.getKey().getFullPath()))
+ .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+ return !filteredTemplateActivationMap.isEmpty()
+ ? Optional.of(
+ new InternalBatchActivateTemplateNode(
+ node.getPlanNodeId(), filteredTemplateActivationMap))
+ : Optional.empty();
+ }
+
+ @Override
+ public Optional<PlanNode> visitInternalCreateMultiTimeSeries(
+ final InternalCreateMultiTimeSeriesNode node, final IoTDBPipePattern pattern) {
+ final Map<PartialPath, Pair<Boolean, MeasurementGroup>> filteredDeviceMap =
+ node.getDeviceMap().entrySet().stream()
+ .filter(entry -> pattern.matchPrefixPath(entry.getKey().getFullPath()))
+ .map(
+ entry ->
+ new Pair<>(
+ entry.getKey(),
+ new Pair<>(
+ entry.getValue().getLeft(),
+ trimMeasurementGroup(
+ entry.getKey().getFullPath(),
+ entry.getValue().getRight(),
+ pattern))))
+ .filter(pair -> Objects.nonNull(pair.getRight().getRight()))
+ .collect(Collectors.toMap(Pair::getLeft, Pair::getRight));
+ return !filteredDeviceMap.isEmpty()
+ ? Optional.of(
+ new InternalCreateMultiTimeSeriesNode(node.getPlanNodeId(), filteredDeviceMap))
+ : Optional.empty();
+ }
+
+ @Override
+ public Optional<PlanNode> visitBatchActivateTemplate(
+ final BatchActivateTemplateNode node, final IoTDBPipePattern pattern) {
+ final Map<PartialPath, Pair<Integer, Integer>> filteredTemplateActivationMap =
+ node.getTemplateActivationMap().entrySet().stream()
+ .filter(entry -> pattern.matchDevice(entry.getKey().getFullPath()))
+ .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+ return !filteredTemplateActivationMap.isEmpty()
+ ? Optional.of(
+ new BatchActivateTemplateNode(node.getPlanNodeId(), filteredTemplateActivationMap))
+ : Optional.empty();
+ }
+
+ @Override
+ public Optional<PlanNode> visitCreateLogicalView(
+ final CreateLogicalViewNode node, final IoTDBPipePattern pattern) {
+ final Map<PartialPath, ViewExpression> filteredViewPathToSourceMap =
+ node.getViewPathToSourceExpressionMap().entrySet().stream()
+ .filter(
+ entry ->
+ pattern.matchesMeasurement(
+ entry.getKey().getDevice(), entry.getKey().getMeasurement()))
+ .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+ return !filteredViewPathToSourceMap.isEmpty()
+ ? Optional.of(new CreateLogicalViewNode(node.getPlanNodeId(), filteredViewPathToSourceMap))
+ : Optional.empty();
+ }
+
+ @Override
+ public Optional<PlanNode> visitAlterLogicalView(
+ final AlterLogicalViewNode node, final IoTDBPipePattern pattern) {
+ final Map<PartialPath, ViewExpression> filteredViewPathToSourceMap =
+ node.getViewPathToSourceMap().entrySet().stream()
+ .filter(
+ entry ->
+ pattern.matchesMeasurement(
+ entry.getKey().getDevice(), entry.getKey().getMeasurement()))
+ .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+ return !filteredViewPathToSourceMap.isEmpty()
+ ? Optional.of(new AlterLogicalViewNode(node.getPlanNodeId(), filteredViewPathToSourceMap))
+ : Optional.empty();
+ }
+
+ @Override
+ public Optional<PlanNode> visitDeleteData(
+ final DeleteDataNode node, final IoTDBPipePattern pattern) {
+ final List<PartialPath> intersectedPaths =
+ node.getPathList().stream()
+ .map(pattern::getIntersection)
+ .flatMap(Collection::stream)
+ .distinct()
+ .collect(Collectors.toList());
+ return !intersectedPaths.isEmpty()
+ ? Optional.of(
+ new DeleteDataNode(
+ node.getPlanNodeId(),
+ intersectedPaths,
+ node.getDeleteStartTime(),
+ node.getDeleteEndTime()))
+ : Optional.empty();
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/pattern/CachedSchemaPatternMatcher.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/pattern/CachedSchemaPatternMatcher.java
index 1de43b7042e..4be44a0fecc 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/pattern/CachedSchemaPatternMatcher.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/pattern/CachedSchemaPatternMatcher.java
@@ -61,7 +61,7 @@ public class CachedSchemaPatternMatcher implements PipeDataRegionMatcher {
}
@Override
- public void register(PipeRealtimeDataRegionExtractor extractor) {
+ public void register(final PipeRealtimeDataRegionExtractor extractor) {
lock.writeLock().lock();
try {
extractors.add(extractor);
@@ -72,7 +72,7 @@ public class CachedSchemaPatternMatcher implements PipeDataRegionMatcher {
}
@Override
- public void deregister(PipeRealtimeDataRegionExtractor extractor) {
+ public void deregister(final PipeRealtimeDataRegionExtractor extractor) {
lock.writeLock().lock();
try {
extractors.remove(extractor);
@@ -93,7 +93,7 @@ public class CachedSchemaPatternMatcher implements PipeDataRegionMatcher {
}
@Override
- public Set<PipeRealtimeDataRegionExtractor> match(PipeRealtimeEvent event) {
+ public Set<PipeRealtimeDataRegionExtractor> match(final PipeRealtimeEvent event) {
final Set<PipeRealtimeDataRegionExtractor> matchedExtractors = new HashSet<>();
lock.readLock().lock();
@@ -176,10 +176,10 @@ public class CachedSchemaPatternMatcher implements PipeDataRegionMatcher {
return matchedExtractors;
}
- protected Set<PipeRealtimeDataRegionExtractor> filterExtractorsByDevice(String device) {
+ protected Set<PipeRealtimeDataRegionExtractor> filterExtractorsByDevice(final String device) {
final Set<PipeRealtimeDataRegionExtractor> filteredExtractors = new HashSet<>();
- for (PipeRealtimeDataRegionExtractor extractor : extractors) {
+ for (final PipeRealtimeDataRegionExtractor extractor : extractors) {
// Return if the extractor only extract deletion
if (!extractor.shouldExtractInsertion()) {
continue;
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 dfd8be444b6..55fe372ea8f 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
@@ -28,6 +28,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.pattern.IoTDBPipePattern;
import org.apache.iotdb.commons.pipe.receiver.IoTDBFileReceiver;
import org.apache.iotdb.db.auth.AuthorityChecker;
import org.apache.iotdb.db.conf.IoTDBConfig;
@@ -50,6 +51,7 @@ import org.apache.iotdb.db.pipe.connector.payload.evolvable.request.PipeTransfer
import org.apache.iotdb.db.pipe.event.common.schema.PipeSchemaRegionSnapshotEvent;
import org.apache.iotdb.db.pipe.receiver.visitor.PipePlanToStatementVisitor;
import org.apache.iotdb.db.pipe.receiver.visitor.PipeStatementExceptionVisitor;
+import org.apache.iotdb.db.pipe.receiver.visitor.PipeStatementPatternParseVisitor;
import org.apache.iotdb.db.pipe.receiver.visitor.PipeStatementTSStatusVisitor;
import org.apache.iotdb.db.pipe.receiver.visitor.PipeStatementToBatchVisitor;
import org.apache.iotdb.db.protocol.session.SessionManager;
@@ -112,6 +114,8 @@ public class IoTDBDataNodeReceiver extends IoTDBFileReceiver {
new PipeStatementTSStatusVisitor();
private static final PipeStatementExceptionVisitor STATEMENT_EXCEPTION_VISITOR =
new PipeStatementExceptionVisitor();
+ private static final PipeStatementPatternParseVisitor STATEMENT_PATTERN_PARSE_VISITOR =
+ new PipeStatementPatternParseVisitor();
private final PipeStatementToBatchVisitor batchVisitor = new PipeStatementToBatchVisitor();
// Used for data transfer: confignode (cluster A) -> datanode (cluster B) -> confignode (cluster
@@ -207,7 +211,7 @@ public class IoTDBDataNodeReceiver extends IoTDBFileReceiver {
receiverId.get(),
status);
return new TPipeTransferResp(status);
- } catch (Exception e) {
+ } catch (final Exception e) {
final String error =
String.format("Exception %s encountered while handling request %s.", e.getMessage(), req);
LOGGER.warn("Receiver id = {}: {}", receiverId.get(), error, e);
@@ -303,6 +307,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));
// Clear to avoid previous exceptions
batchVisitor.clear();
@@ -316,8 +322,9 @@ public class IoTDBDataNodeReceiver extends IoTDBFileReceiver {
// The statements do not contain AlterLogicalViewStatements
// Here we apply the statements as many as possible
// Even if there are failed statements
- batchVisitor
- .process(originalStatement, null)
+ STATEMENT_PATTERN_PARSE_VISITOR
+ .process(originalStatement, pattern)
+ .flatMap(parsedStatement -> batchVisitor.process(parsedStatement, null))
.ifPresent(statement -> results.add(executeStatementAndClassifyExceptions(statement)));
}
batchVisitor.getRemainBatches().stream()
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/visitor/PipePlanToStatementVisitor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/visitor/PipePlanToStatementVisitor.java
index 32e3e6947ed..593e3715c88 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/visitor/PipePlanToStatementVisitor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/visitor/PipePlanToStatementVisitor.java
@@ -257,7 +257,7 @@ public class PipePlanToStatementVisitor extends PlanVisitor<Statement, Void> {
// We do not support AlterLogicalViewNode parsing and use direct rpc instead
@Override
- public DeleteDataStatement visitDeleteData(DeleteDataNode node, final Void context) {
+ public DeleteDataStatement visitDeleteData(final DeleteDataNode node, final Void context) {
final DeleteDataStatement statement = new DeleteDataStatement();
statement.setDeleteEndTime(node.getDeleteEndTime());
statement.setDeleteStartTime(node.getDeleteStartTime());
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
new file mode 100644
index 00000000000..1fd404375f4
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/visitor/PipeStatementPatternParseVisitor.java
@@ -0,0 +1,136 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.pipe.receiver.visitor;
+
+import org.apache.iotdb.commons.pipe.pattern.IoTDBPipePattern;
+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;
+import org.apache.iotdb.db.queryengine.plan.statement.metadata.CreateAlignedTimeSeriesStatement;
+import org.apache.iotdb.db.queryengine.plan.statement.metadata.CreateTimeSeriesStatement;
+import org.apache.iotdb.db.queryengine.plan.statement.metadata.template.ActivateTemplateStatement;
+import org.apache.iotdb.db.queryengine.plan.statement.metadata.view.CreateLogicalViewStatement;
+import org.apache.iotdb.db.tools.schema.SRStatementGenerator;
+
+import java.util.Arrays;
+import java.util.Optional;
+import java.util.stream.IntStream;
+
+/**
+ * The {@link PipeStatementPatternParseVisitor} will transform the schema {@link Statement}s using
+ * {@link IoTDBPipePattern}. 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}.
+ *
+ * <p>2. If a pattern does not intersect with the {@link IoTDBPipePattern}, it's dropped.
+ *
+ * <p>3. If all the patterns in the {@link Statement} is dropped, the {@link Statement} is dropped.
+ *
+ * <p>4. The output {@link Statement} can be the altered form of the original one because it's read
+ * from the {@link SRStatementGenerator} and will no longer be used.
+ */
+public class PipeStatementPatternParseVisitor
+ extends StatementVisitor<Optional<Statement>, IoTDBPipePattern> {
+ @Override
+ public Optional<Statement> visitNode(
+ final StatementNode statement, final IoTDBPipePattern pattern) {
+ return Optional.of((Statement) statement);
+ }
+
+ @Override
+ public Optional<Statement> visitCreateTimeseries(
+ final CreateTimeSeriesStatement statement, final IoTDBPipePattern pattern) {
+ return pattern.matchesMeasurement(
+ statement.getPath().getDevice(), statement.getPath().getMeasurement())
+ ? Optional.of(statement)
+ : Optional.empty();
+ }
+
+ @Override
+ public Optional<Statement> visitCreateAlignedTimeseries(
+ final CreateAlignedTimeSeriesStatement statement, final IoTDBPipePattern pattern) {
+ final int[] filteredIndexes =
+ IntStream.range(0, statement.getMeasurements().size())
+ .filter(
+ index ->
+ pattern.matchesMeasurement(
+ statement.getDevicePath().getFullPath(),
+ statement.getMeasurements().get(index)))
+ .toArray();
+ if (filteredIndexes.length == 0) {
+ return Optional.empty();
+ }
+ final CreateAlignedTimeSeriesStatement targetCreateAlignedTimeSeriesStatement =
+ new CreateAlignedTimeSeriesStatement();
+ targetCreateAlignedTimeSeriesStatement.setDevicePath(statement.getDevicePath());
+ Arrays.stream(filteredIndexes)
+ .forEach(
+ index -> {
+ targetCreateAlignedTimeSeriesStatement.addMeasurement(
+ statement.getMeasurements().get(index));
+ targetCreateAlignedTimeSeriesStatement.addDataType(
+ statement.getDataTypes().get(index));
+ targetCreateAlignedTimeSeriesStatement.addEncoding(
+ statement.getEncodings().get(index));
+ targetCreateAlignedTimeSeriesStatement.addCompressor(
+ statement.getCompressors().get(index));
+ targetCreateAlignedTimeSeriesStatement.addTagsList(
+ statement.getTagsList().get(index));
+ targetCreateAlignedTimeSeriesStatement.addAttributesList(
+ statement.getAttributesList().get(index));
+ targetCreateAlignedTimeSeriesStatement.addAliasList(
+ statement.getAliasList().get(index));
+ });
+ return Optional.of(targetCreateAlignedTimeSeriesStatement);
+ }
+
+ @Override
+ public Optional<Statement> visitActivateTemplate(
+ final ActivateTemplateStatement activateTemplateStatement, final IoTDBPipePattern pattern) {
+ return pattern.matchDevice(activateTemplateStatement.getPath().getFullPath())
+ ? Optional.of(activateTemplateStatement)
+ : Optional.empty();
+ }
+
+ @Override
+ public Optional<Statement> visitCreateLogicalView(
+ final CreateLogicalViewStatement createLogicalViewStatement, final IoTDBPipePattern pattern) {
+ final int[] filteredIndexes =
+ IntStream.range(0, createLogicalViewStatement.getTargetPathList().size())
+ .filter(
+ index ->
+ pattern.matchesMeasurement(
+ createLogicalViewStatement.getTargetPathList().get(index).getDevice(),
+ createLogicalViewStatement.getTargetPathList().get(index).getMeasurement()))
+ .toArray();
+ if (filteredIndexes.length == 0) {
+ return Optional.empty();
+ }
+ createLogicalViewStatement.setTargetFullPaths(
+ IoTDBPipePattern.applyIndexesOnList(
+ filteredIndexes, createLogicalViewStatement.getTargetPathList()));
+ createLogicalViewStatement.setViewExpressions(
+ IoTDBPipePattern.applyIndexesOnList(
+ filteredIndexes, createLogicalViewStatement.getViewExpressions()));
+
+ return Optional.of(createLogicalViewStatement);
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/connection/PipeEventCollector.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/connection/PipeEventCollector.java
index cdd73598571..a3f2227e126 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/connection/PipeEventCollector.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/connection/PipeEventCollector.java
@@ -21,12 +21,15 @@ package org.apache.iotdb.db.pipe.task.connection;
import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
import org.apache.iotdb.commons.pipe.event.ProgressReportEvent;
+import org.apache.iotdb.commons.pipe.pattern.IoTDBPipePattern;
import org.apache.iotdb.commons.pipe.progress.PipeEventCommitManager;
import org.apache.iotdb.commons.pipe.task.connection.UnboundedBlockingPendingQueue;
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.common.tablet.PipeInsertNodeTabletInsertionEvent;
import org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent;
import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent;
+import org.apache.iotdb.db.pipe.extractor.schemaregion.IoTDBSchemaRegionExtractor;
import org.apache.iotdb.pipe.api.collector.EventCollector;
import org.apache.iotdb.pipe.api.event.Event;
import org.apache.iotdb.pipe.api.event.dml.insertion.TabletInsertionEvent;
@@ -67,6 +70,8 @@ public class PipeEventCollector implements EventCollector {
parseAndCollectEvent((PipeRawTabletInsertionEvent) event);
} else if (event instanceof PipeTsFileInsertionEvent) {
parseAndCollectEvent((PipeTsFileInsertionEvent) event);
+ } else if (event instanceof PipeSchemaRegionWritePlanEvent) {
+ parseAndCollectEvent((PipeSchemaRegionWritePlanEvent) event);
} else if (!(event instanceof ProgressReportEvent)) {
collectEvent(event);
}
@@ -121,6 +126,20 @@ public class PipeEventCollector implements EventCollector {
}
}
+ private void parseAndCollectEvent(final PipeSchemaRegionWritePlanEvent deleteDataEvent) {
+ IoTDBSchemaRegionExtractor.PATTERN_PARSE_VISITOR
+ .process(deleteDataEvent.getPlanNode(), (IoTDBPipePattern) deleteDataEvent.getPipePattern())
+ .map(
+ planNode ->
+ new PipeSchemaRegionWritePlanEvent(
+ planNode,
+ deleteDataEvent.getPipeName(),
+ deleteDataEvent.getPipeTaskMeta(),
+ deleteDataEvent.getPipePattern(),
+ deleteDataEvent.isGeneratedByPipe()))
+ .ifPresent(this::collectEvent);
+ }
+
private void collectEvent(final Event event) {
collectInvocationCount.incrementAndGet();
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/DeleteDataNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/DeleteDataNode.java
index 08a80133926..877187be5a6 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/DeleteDataNode.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/DeleteDataNode.java
@@ -48,6 +48,7 @@ import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.Objects;
import java.util.stream.Collectors;
import static org.apache.iotdb.commons.conf.IoTDBConstant.MULTI_LEVEL_PATH_WILDCARD;
@@ -251,6 +252,26 @@ public class DeleteDataNode extends WritePlanNode implements WALEntryValue {
this.regionReplicaSet = regionReplicaSet;
}
+ @Override
+ public boolean equals(final Object obj) {
+ if (this == obj) {
+ return true;
+ }
+ if (obj == null || getClass() != obj.getClass()) {
+ return false;
+ }
+ final DeleteDataNode that = (DeleteDataNode) obj;
+ return this.getPlanNodeId().equals(that.getPlanNodeId())
+ && Objects.equals(this.pathList, that.pathList)
+ && Objects.equals(this.deleteStartTime, that.deleteStartTime)
+ && Objects.equals(this.deleteEndTime, that.deleteEndTime);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(getPlanNodeId(), pathList, deleteStartTime, deleteEndTime);
+ }
+
public String toString() {
return String.format(
"DeleteDataNode-%s[ Paths: %s, Region: %s ]",
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/metadata/CreateAlignedTimeSeriesStatement.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/metadata/CreateAlignedTimeSeriesStatement.java
index 5c441253697..81ee69f0bd2 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/metadata/CreateAlignedTimeSeriesStatement.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/metadata/CreateAlignedTimeSeriesStatement.java
@@ -35,6 +35,7 @@ import org.apache.tsfile.file.metadata.enums.TSEncoding;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
+import java.util.Objects;
/**
* CREATE ALIGNED TIMESERIES statement.
@@ -178,4 +179,58 @@ public class CreateAlignedTimeSeriesStatement extends Statement {
public <R, C> R accept(StatementVisitor<R, C> visitor, C context) {
return visitor.visitCreateAlignedTimeseries(this, context);
}
+
+ @Override
+ public boolean equals(final Object obj) {
+ if (this == obj) {
+ return true;
+ }
+ if (obj == null || getClass() != obj.getClass()) {
+ return false;
+ }
+ final CreateAlignedTimeSeriesStatement that = (CreateAlignedTimeSeriesStatement) obj;
+ return Objects.equals(this.devicePath, that.devicePath)
+ && Objects.equals(this.measurements, that.measurements)
+ && Objects.equals(this.dataTypes, that.dataTypes)
+ && Objects.equals(this.encodings, that.encodings)
+ && Objects.equals(this.compressors, that.compressors)
+ && Objects.equals(this.aliasList, that.aliasList)
+ && Objects.equals(this.tagsList, that.tagsList)
+ && Objects.equals(this.attributesList, that.attributesList);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(
+ devicePath,
+ measurements,
+ dataTypes,
+ encodings,
+ compressors,
+ aliasList,
+ tagsList,
+ attributesList);
+ }
+
+ @Override
+ public String toString() {
+ return "CreateAlignedTimeSeriesStatement{"
+ + "devicePath='"
+ + devicePath
+ + "', measurements="
+ + measurements
+ + ", dataTypes='"
+ + dataTypes
+ + "', encodings="
+ + encodings
+ + "', compressors="
+ + compressors
+ + "', aliasList="
+ + aliasList
+ + "', tagsList="
+ + tagsList
+ + "', attributesList="
+ + attributesList
+ + "'}";
+ }
}
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/connector/PipeDataNodeThriftRequestTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/connector/PipeDataNodeThriftRequestTest.java
index fd3e06f4588..0d8e5718c9e 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/connector/PipeDataNodeThriftRequestTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/connector/PipeDataNodeThriftRequestTest.java
@@ -323,7 +323,7 @@ public class PipeDataNodeThriftRequestTest {
PipeTransferSchemaSnapshotSealReq req =
PipeTransferSchemaSnapshotSealReq.toTPipeTransferReq(
- mTreeSnapshotName, 100, tLogName, 10, databaseName, typeString);
+ "root.**", mTreeSnapshotName, 100, tLogName, 10, databaseName, typeString);
PipeTransferSchemaSnapshotSealReq deserializeReq =
PipeTransferSchemaSnapshotSealReq.fromTPipeTransferReq(req);
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
new file mode 100644
index 00000000000..b11e9a0e3e6
--- /dev/null
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/connector/PipeStatementPatternParseVisitorTest.java
@@ -0,0 +1,161 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.pipe.connector;
+
+import org.apache.iotdb.commons.exception.IllegalPathException;
+import org.apache.iotdb.commons.path.PartialPath;
+import org.apache.iotdb.commons.pipe.pattern.IoTDBPipePattern;
+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.CreateAlignedTimeSeriesStatement;
+import org.apache.iotdb.db.queryengine.plan.statement.metadata.CreateTimeSeriesStatement;
+import org.apache.iotdb.db.queryengine.plan.statement.metadata.template.ActivateTemplateStatement;
+import org.apache.iotdb.db.queryengine.plan.statement.metadata.view.CreateLogicalViewStatement;
+
+import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.file.metadata.enums.CompressionType;
+import org.apache.tsfile.file.metadata.enums.TSEncoding;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+
+public class PipeStatementPatternParseVisitorTest {
+
+ private final IoTDBPipePattern prefixPathPattern = new IoTDBPipePattern("root.db.device.**");
+ private final IoTDBPipePattern fullPathPattern = new IoTDBPipePattern("root.db.device.s1");
+
+ @Test
+ public void testCreateTimeSeries() throws IllegalPathException {
+ final CreateTimeSeriesStatement createTimeSeriesStatement = new CreateTimeSeriesStatement();
+ createTimeSeriesStatement.setPath(new PartialPath("root.db.device.s1"));
+ createTimeSeriesStatement.setDataType(TSDataType.FLOAT);
+ createTimeSeriesStatement.setEncoding(TSEncoding.RLE);
+ createTimeSeriesStatement.setCompressor(CompressionType.SNAPPY);
+ createTimeSeriesStatement.setProps(new HashMap<>());
+ createTimeSeriesStatement.setTags(new HashMap<>());
+ createTimeSeriesStatement.setAttributes(new HashMap<>());
+ createTimeSeriesStatement.setAlias("a1");
+
+ final CreateTimeSeriesStatement createTimeSeriesStatementToFilter =
+ new CreateTimeSeriesStatement();
+ createTimeSeriesStatementToFilter.setPath(new PartialPath("root.db1.device.s1"));
+ createTimeSeriesStatementToFilter.setDataType(TSDataType.FLOAT);
+ createTimeSeriesStatementToFilter.setEncoding(TSEncoding.RLE);
+ createTimeSeriesStatementToFilter.setCompressor(CompressionType.SNAPPY);
+ createTimeSeriesStatementToFilter.setProps(new HashMap<>());
+ createTimeSeriesStatementToFilter.setTags(new HashMap<>());
+ createTimeSeriesStatementToFilter.setAttributes(new HashMap<>());
+ createTimeSeriesStatementToFilter.setAlias("a2");
+
+ Assert.assertEquals(
+ createTimeSeriesStatement,
+ new PipeStatementPatternParseVisitor()
+ .visitCreateTimeseries(createTimeSeriesStatement, prefixPathPattern)
+ .orElseThrow(AssertionError::new));
+ Assert.assertFalse(
+ new PipeStatementPatternParseVisitor()
+ .visitCreateTimeseries(createTimeSeriesStatementToFilter, prefixPathPattern)
+ .isPresent());
+ }
+
+ @Test
+ public void testCreateAlignedTimeSeries() throws IllegalPathException {
+ final CreateAlignedTimeSeriesStatement expectedCreateAlignedTimeSeriesStatement =
+ new CreateAlignedTimeSeriesStatement();
+ expectedCreateAlignedTimeSeriesStatement.setDevicePath(new PartialPath("root.db.device"));
+ expectedCreateAlignedTimeSeriesStatement.setMeasurements(Collections.singletonList("s1"));
+ expectedCreateAlignedTimeSeriesStatement.setDataTypes(
+ Collections.singletonList(TSDataType.FLOAT));
+ expectedCreateAlignedTimeSeriesStatement.setEncodings(
+ Collections.singletonList(TSEncoding.RLE));
+ expectedCreateAlignedTimeSeriesStatement.setCompressors(
+ Collections.singletonList(CompressionType.SNAPPY));
+ expectedCreateAlignedTimeSeriesStatement.setTagsList(
+ Collections.singletonList(new HashMap<>()));
+ expectedCreateAlignedTimeSeriesStatement.setAttributesList(
+ Collections.singletonList(new HashMap<>()));
+ expectedCreateAlignedTimeSeriesStatement.setAliasList(Collections.singletonList("a1"));
+
+ final CreateAlignedTimeSeriesStatement originalCreateAlignedTimeSeriesStatement =
+ new CreateAlignedTimeSeriesStatement();
+ originalCreateAlignedTimeSeriesStatement.setDevicePath(new PartialPath("root.db.device"));
+ originalCreateAlignedTimeSeriesStatement.setMeasurements(Arrays.asList("s1", "s2"));
+ originalCreateAlignedTimeSeriesStatement.setDataTypes(
+ Arrays.asList(TSDataType.FLOAT, TSDataType.BOOLEAN));
+ originalCreateAlignedTimeSeriesStatement.setEncodings(
+ Arrays.asList(TSEncoding.RLE, TSEncoding.PLAIN));
+ originalCreateAlignedTimeSeriesStatement.setCompressors(
+ Arrays.asList(CompressionType.SNAPPY, CompressionType.SNAPPY));
+ originalCreateAlignedTimeSeriesStatement.setTagsList(
+ Arrays.asList(new HashMap<>(), new HashMap<>()));
+ originalCreateAlignedTimeSeriesStatement.setAttributesList(
+ Arrays.asList(new HashMap<>(), new HashMap<>()));
+ originalCreateAlignedTimeSeriesStatement.setAliasList(Arrays.asList("a1", "a2"));
+
+ Assert.assertEquals(
+ expectedCreateAlignedTimeSeriesStatement,
+ new PipeStatementPatternParseVisitor()
+ .visitCreateAlignedTimeseries(originalCreateAlignedTimeSeriesStatement, fullPathPattern)
+ .orElseThrow(AssertionError::new));
+ }
+
+ @Test
+ public void testActivateTemplate() throws IllegalPathException {
+ final ActivateTemplateStatement activateTemplateStatement =
+ new ActivateTemplateStatement(new PartialPath("root.db.device"));
+ final ActivateTemplateStatement activateTemplateStatementToFilter =
+ new ActivateTemplateStatement(new PartialPath("root.db"));
+
+ Assert.assertEquals(
+ activateTemplateStatement,
+ new PipeStatementPatternParseVisitor()
+ .visitActivateTemplate(activateTemplateStatement, prefixPathPattern)
+ .orElseThrow(AssertionError::new));
+ Assert.assertFalse(
+ new PipeStatementPatternParseVisitor()
+ .visitActivateTemplate(activateTemplateStatementToFilter, prefixPathPattern)
+ .isPresent());
+ }
+
+ @Test
+ public void testCreateLogicalView() throws IllegalPathException {
+ final CreateLogicalViewStatement createLogicalViewStatement = new CreateLogicalViewStatement();
+ createLogicalViewStatement.setTargetFullPaths(
+ Arrays.asList(new PartialPath("root.db.device.a1"), new PartialPath("root.db1.device.a1")));
+ createLogicalViewStatement.setViewExpressions(
+ Arrays.asList(
+ new TimeSeriesViewOperand("root.sg1.d1"), new TimeSeriesViewOperand("root.sg1.d2")));
+
+ final CreateLogicalViewStatement targetLogicalViewStatement =
+ (CreateLogicalViewStatement)
+ new PipeStatementPatternParseVisitor()
+ .visitCreateLogicalView(createLogicalViewStatement, prefixPathPattern)
+ .orElseThrow(AssertionError::new);
+ Assert.assertEquals(
+ Collections.singletonList(new PartialPath("root.db.device.a1")),
+ targetLogicalViewStatement.getPaths());
+ Assert.assertEquals(
+ Collections.singletonList(new TimeSeriesViewOperand("root.sg1.d1")),
+ targetLogicalViewStatement.getViewExpressions());
+ }
+}
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/extractor/IoTDBDataRegionExtractorTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/extractor/IoTDBDataRegionExtractorTest.java
index bdf63161b96..8b133b67391 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/extractor/IoTDBDataRegionExtractorTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/extractor/IoTDBDataRegionExtractorTest.java
@@ -32,8 +32,7 @@ import java.util.HashMap;
public class IoTDBDataRegionExtractorTest {
@Test
public void testIoTDBDataRegionExtractor() {
- final IoTDBDataRegionExtractor extractor = new IoTDBDataRegionExtractor();
- try {
+ try (final IoTDBDataRegionExtractor extractor = new IoTDBDataRegionExtractor()) {
extractor.validate(
new PipeParameterValidator(
new PipeParameters(
@@ -102,4 +101,19 @@ public class IoTDBDataRegionExtractorTest {
}
return null;
}
+
+ @Test(expected = IllegalArgumentException.class)
+ public void testIoTDBDataRegionExtractorWithDeletionAndPattern() throws Exception {
+ try (final IoTDBDataRegionExtractor extractor = new IoTDBDataRegionExtractor()) {
+ extractor.validate(
+ new PipeParameterValidator(
+ new PipeParameters(
+ new HashMap<String, String>() {
+ {
+ put(PipeExtractorConstant.EXTRACTOR_PATTERN_KEY, "root.db");
+ put(PipeExtractorConstant.EXTRACTOR_INCLUSION_KEY, "data.delete");
+ }
+ })));
+ }
+ }
}
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
new file mode 100644
index 00000000000..2cd3e02822d
--- /dev/null
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/extractor/PipePlanPatternParseVisitorTest.java
@@ -0,0 +1,428 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.pipe.extractor;
+
+import org.apache.iotdb.commons.exception.IllegalPathException;
+import org.apache.iotdb.commons.path.PartialPath;
+import org.apache.iotdb.commons.pipe.pattern.IoTDBPipePattern;
+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;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeId;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metedata.write.ActivateTemplateNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metedata.write.AlterTimeSeriesNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metedata.write.BatchActivateTemplateNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metedata.write.CreateAlignedTimeSeriesNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metedata.write.CreateMultiTimeSeriesNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metedata.write.CreateTimeSeriesNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metedata.write.InternalBatchActivateTemplateNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metedata.write.InternalCreateMultiTimeSeriesNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metedata.write.InternalCreateTimeSeriesNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metedata.write.MeasurementGroup;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metedata.write.view.AlterLogicalViewNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metedata.write.view.CreateLogicalViewNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.DeleteDataNode;
+import org.apache.iotdb.db.queryengine.plan.statement.metadata.AlterTimeSeriesStatement;
+
+import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.file.metadata.enums.CompressionType;
+import org.apache.tsfile.file.metadata.enums.TSEncoding;
+import org.apache.tsfile.utils.Pair;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+import java.util.Collections;
+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");
+
+ @Test
+ public void testCreateTimeSeries() throws IllegalPathException {
+ final CreateTimeSeriesNode createTimeSeriesNode =
+ new CreateTimeSeriesNode(
+ new PlanNodeId("2024-04-30-1"),
+ new PartialPath("root.db.device.s1"),
+ TSDataType.FLOAT,
+ TSEncoding.RLE,
+ CompressionType.SNAPPY,
+ new HashMap<>(),
+ new HashMap<>(),
+ new HashMap<>(),
+ "a1");
+ final CreateTimeSeriesNode createTimeSeriesNodeToFilter =
+ new CreateTimeSeriesNode(
+ new PlanNodeId("2024-04-30-2"),
+ new PartialPath("root.db1.device.s1"),
+ TSDataType.FLOAT,
+ TSEncoding.RLE,
+ CompressionType.SNAPPY,
+ new HashMap<>(),
+ new HashMap<>(),
+ new HashMap<>(),
+ "a1");
+
+ Assert.assertEquals(
+ createTimeSeriesNode,
+ IoTDBSchemaRegionExtractor.PATTERN_PARSE_VISITOR
+ .visitCreateTimeSeries(createTimeSeriesNode, prefixPathPattern)
+ .orElseThrow(AssertionError::new));
+ Assert.assertFalse(
+ IoTDBSchemaRegionExtractor.PATTERN_PARSE_VISITOR
+ .visitCreateTimeSeries(createTimeSeriesNodeToFilter, prefixPathPattern)
+ .isPresent());
+ }
+
+ @Test
+ public void testCreateAlignedTimeSeries() throws IllegalPathException {
+ Assert.assertEquals(
+ new CreateAlignedTimeSeriesNode(
+ new PlanNodeId("2024-04-30-1"),
+ new PartialPath("root.db.device"),
+ Collections.singletonList("s1"),
+ Collections.singletonList(TSDataType.FLOAT),
+ Collections.singletonList(TSEncoding.RLE),
+ Collections.singletonList(CompressionType.SNAPPY),
+ Collections.singletonList("a1"),
+ Collections.singletonList(new HashMap<>()),
+ Collections.singletonList(new HashMap<>())),
+ IoTDBSchemaRegionExtractor.PATTERN_PARSE_VISITOR
+ .visitCreateAlignedTimeSeries(
+ new CreateAlignedTimeSeriesNode(
+ new PlanNodeId("2024-04-30-1"),
+ new PartialPath("root.db.device"),
+ Arrays.asList("s1", "s2"),
+ Arrays.asList(TSDataType.FLOAT, TSDataType.BOOLEAN),
+ Arrays.asList(TSEncoding.RLE, TSEncoding.PLAIN),
+ Arrays.asList(CompressionType.SNAPPY, CompressionType.SNAPPY),
+ Arrays.asList("a1", "a2"),
+ Arrays.asList(new HashMap<>(), new HashMap<>()),
+ Arrays.asList(new HashMap<>(), new HashMap<>())),
+ fullPathPattern)
+ .orElseThrow(AssertionError::new));
+ }
+
+ @Test
+ public void testCreateMultiTimeSeries() throws IllegalPathException {
+ Assert.assertEquals(
+ new CreateMultiTimeSeriesNode(
+ new PlanNodeId("2024-04-30-1"),
+ Collections.singletonList(new PartialPath("root.db.device.s1")),
+ Collections.singletonList(TSDataType.FLOAT),
+ Collections.singletonList(TSEncoding.RLE),
+ Collections.singletonList(CompressionType.SNAPPY),
+ Collections.singletonList(new HashMap<>()),
+ Collections.singletonList("a1"),
+ Collections.singletonList(new HashMap<>()),
+ Collections.singletonList(new HashMap<>())),
+ IoTDBSchemaRegionExtractor.PATTERN_PARSE_VISITOR
+ .visitCreateMultiTimeSeries(
+ new CreateMultiTimeSeriesNode(
+ new PlanNodeId("2024-04-30-1"),
+ Arrays.asList(
+ new PartialPath("root.db.device.s1"),
+ new PartialPath("root.db1.device.s1")),
+ Arrays.asList(TSDataType.FLOAT, TSDataType.BOOLEAN),
+ Arrays.asList(TSEncoding.RLE, TSEncoding.PLAIN),
+ Arrays.asList(CompressionType.SNAPPY, CompressionType.SNAPPY),
+ Arrays.asList(new HashMap<>(), new HashMap<>()),
+ Arrays.asList("a1", "a2"),
+ Arrays.asList(new HashMap<>(), new HashMap<>()),
+ Arrays.asList(new HashMap<>(), new HashMap<>())),
+ fullPathPattern)
+ .orElseThrow(AssertionError::new));
+ }
+
+ @Test
+ public void testAlterTimeSeries() throws IllegalPathException {
+ final Map<String, String> attributesMap =
+ new HashMap<String, String>() {
+ {
+ put("k1", "v1");
+ }
+ };
+ final AlterTimeSeriesNode alterTimeSeriesNode =
+ new AlterTimeSeriesNode(
+ new PlanNodeId("2024-04-30-1"),
+ new PartialPath("root.db.device.s1"),
+ AlterTimeSeriesStatement.AlterType.ADD_ATTRIBUTES,
+ attributesMap,
+ "",
+ new HashMap<>(),
+ attributesMap,
+ false);
+ final AlterTimeSeriesNode alterTimeSeriesNodeToFilter =
+ new AlterTimeSeriesNode(
+ new PlanNodeId("2024-04-30-2"),
+ new PartialPath("root.db1.device.s1"),
+ AlterTimeSeriesStatement.AlterType.ADD_ATTRIBUTES,
+ attributesMap,
+ "",
+ new HashMap<>(),
+ attributesMap,
+ false);
+
+ Assert.assertEquals(
+ alterTimeSeriesNode,
+ IoTDBSchemaRegionExtractor.PATTERN_PARSE_VISITOR
+ .visitAlterTimeSeries(alterTimeSeriesNode, prefixPathPattern)
+ .orElseThrow(AssertionError::new));
+ Assert.assertFalse(
+ IoTDBSchemaRegionExtractor.PATTERN_PARSE_VISITOR
+ .visitAlterTimeSeries(alterTimeSeriesNodeToFilter, prefixPathPattern)
+ .isPresent());
+ }
+
+ @Test
+ public void testInternalCreateTimeSeries() throws IllegalPathException {
+ final MeasurementGroup expectedMeasurementGroup = new MeasurementGroup();
+ expectedMeasurementGroup.addMeasurement(
+ "s1", TSDataType.FLOAT, TSEncoding.RLE, CompressionType.SNAPPY);
+ expectedMeasurementGroup.addProps(new HashMap<>());
+ expectedMeasurementGroup.addAlias("a1");
+ expectedMeasurementGroup.addTags(new HashMap<>());
+ expectedMeasurementGroup.addAttributes(new HashMap<>());
+
+ final ByteBuffer byteBuffer = ByteBuffer.allocate(1000);
+ expectedMeasurementGroup.serialize(byteBuffer);
+ byteBuffer.flip();
+ final MeasurementGroup originalMeasurementGroup = new MeasurementGroup();
+ originalMeasurementGroup.deserialize(byteBuffer);
+
+ originalMeasurementGroup.addMeasurement(
+ "s2", TSDataType.BOOLEAN, TSEncoding.PLAIN, CompressionType.SNAPPY);
+ originalMeasurementGroup.addProps(new HashMap<>());
+ originalMeasurementGroup.addAlias("a2");
+ originalMeasurementGroup.addTags(new HashMap<>());
+ originalMeasurementGroup.addAttributes(new HashMap<>());
+
+ Assert.assertEquals(
+ new InternalCreateTimeSeriesNode(
+ new PlanNodeId("2024-04-30-1"),
+ new PartialPath("root.db.device"),
+ expectedMeasurementGroup,
+ true),
+ IoTDBSchemaRegionExtractor.PATTERN_PARSE_VISITOR
+ .visitInternalCreateTimeSeries(
+ new InternalCreateTimeSeriesNode(
+ new PlanNodeId("2024-04-30-1"),
+ new PartialPath("root.db.device"),
+ originalMeasurementGroup,
+ true),
+ fullPathPattern)
+ .orElseThrow(AssertionError::new));
+ }
+
+ @Test
+ public void testActivateTemplate() throws IllegalPathException {
+ final ActivateTemplateNode activateTemplateNode =
+ new ActivateTemplateNode(
+ new PlanNodeId("2024-04-30-1"), new PartialPath("root.db.device"), 3, 1);
+ final ActivateTemplateNode activateTemplateNodeToFilter =
+ new ActivateTemplateNode(new PlanNodeId("2024-04-30-2"), new PartialPath("root.db"), 2, 1);
+
+ Assert.assertEquals(
+ activateTemplateNode,
+ IoTDBSchemaRegionExtractor.PATTERN_PARSE_VISITOR
+ .visitActivateTemplate(activateTemplateNode, prefixPathPattern)
+ .orElseThrow(AssertionError::new));
+ Assert.assertFalse(
+ IoTDBSchemaRegionExtractor.PATTERN_PARSE_VISITOR
+ .visitActivateTemplate(activateTemplateNodeToFilter, prefixPathPattern)
+ .isPresent());
+ }
+
+ @Test
+ public void testInternalBatchActivateTemplate() throws IllegalPathException {
+ Assert.assertEquals(
+ new InternalBatchActivateTemplateNode(
+ new PlanNodeId("2024-04-30-1"),
+ new HashMap<PartialPath, Pair<Integer, Integer>>() {
+ {
+ put(new PartialPath("root.db.device"), new Pair<>(1, 1));
+ }
+ }),
+ IoTDBSchemaRegionExtractor.PATTERN_PARSE_VISITOR
+ .visitInternalBatchActivateTemplate(
+ new InternalBatchActivateTemplateNode(
+ new PlanNodeId("2024-04-30-1"),
+ new HashMap<PartialPath, Pair<Integer, Integer>>() {
+ {
+ put(new PartialPath("root.db.device"), new Pair<>(1, 1));
+ put(new PartialPath("root.db"), new Pair<>(2, 2));
+ }
+ }),
+ fullPathPattern)
+ .orElseThrow(AssertionError::new));
+ }
+
+ @Test
+ public void testInternalCreateMultiTimeSeries() throws IllegalPathException {
+ final MeasurementGroup expectedMeasurementGroup = new MeasurementGroup();
+ expectedMeasurementGroup.addMeasurement(
+ "s1", TSDataType.FLOAT, TSEncoding.RLE, CompressionType.SNAPPY);
+ expectedMeasurementGroup.addProps(new HashMap<>());
+ expectedMeasurementGroup.addAlias("a1");
+ expectedMeasurementGroup.addTags(new HashMap<>());
+ expectedMeasurementGroup.addAttributes(new HashMap<>());
+
+ final ByteBuffer byteBuffer = ByteBuffer.allocate(1000);
+ expectedMeasurementGroup.serialize(byteBuffer);
+ byteBuffer.flip();
+ final MeasurementGroup originalMeasurementGroup = new MeasurementGroup();
+ originalMeasurementGroup.deserialize(byteBuffer);
+
+ originalMeasurementGroup.addMeasurement(
+ "s2", TSDataType.BOOLEAN, TSEncoding.PLAIN, CompressionType.SNAPPY);
+ originalMeasurementGroup.addProps(new HashMap<>());
+ originalMeasurementGroup.addAlias("a2");
+ originalMeasurementGroup.addTags(new HashMap<>());
+ originalMeasurementGroup.addAttributes(new HashMap<>());
+
+ Assert.assertEquals(
+ new InternalCreateMultiTimeSeriesNode(
+ new PlanNodeId("2024-04-30-1"),
+ new HashMap<PartialPath, Pair<Boolean, MeasurementGroup>>() {
+ {
+ put(new PartialPath("root.db.device"), new Pair<>(false, expectedMeasurementGroup));
+ }
+ }),
+ IoTDBSchemaRegionExtractor.PATTERN_PARSE_VISITOR
+ .visitInternalCreateMultiTimeSeries(
+ new InternalCreateMultiTimeSeriesNode(
+ new PlanNodeId("2024-04-30-1"),
+ new HashMap<PartialPath, Pair<Boolean, MeasurementGroup>>() {
+ {
+ put(
+ new PartialPath("root.db.device"),
+ new Pair<>(false, originalMeasurementGroup));
+ put(
+ new PartialPath("root.db1.device"),
+ new Pair<>(false, originalMeasurementGroup));
+ }
+ }),
+ fullPathPattern)
+ .orElseThrow(AssertionError::new));
+ }
+
+ @Test
+ public void testBatchActivateTemplate() throws IllegalPathException {
+ Assert.assertEquals(
+ new BatchActivateTemplateNode(
+ new PlanNodeId("2024-04-30-1"),
+ new HashMap<PartialPath, Pair<Integer, Integer>>() {
+ {
+ put(new PartialPath("root.db.device"), new Pair<>(1, 1));
+ }
+ }),
+ IoTDBSchemaRegionExtractor.PATTERN_PARSE_VISITOR
+ .visitBatchActivateTemplate(
+ new BatchActivateTemplateNode(
+ new PlanNodeId("2024-04-30-1"),
+ new HashMap<PartialPath, Pair<Integer, Integer>>() {
+ {
+ put(new PartialPath("root.db.device"), new Pair<>(1, 1));
+ put(new PartialPath("root.db"), new Pair<>(2, 2));
+ }
+ }),
+ fullPathPattern)
+ .orElseThrow(AssertionError::new));
+ }
+
+ @Test
+ public void testCreateLogicalView() throws IllegalPathException {
+ Assert.assertEquals(
+ new CreateLogicalViewNode(
+ new PlanNodeId("2024-04-30-1"),
+ new HashMap<PartialPath, ViewExpression>() {
+ {
+ put(new PartialPath("root.db.device.a1"), new TimeSeriesViewOperand("root.sg1.d1"));
+ }
+ }),
+ IoTDBSchemaRegionExtractor.PATTERN_PARSE_VISITOR
+ .visitCreateLogicalView(
+ new CreateLogicalViewNode(
+ new PlanNodeId("2024-04-30-1"),
+ new HashMap<PartialPath, ViewExpression>() {
+ {
+ put(
+ new PartialPath("root.db.device.a1"),
+ new TimeSeriesViewOperand("root.sg1.d1"));
+ put(
+ new PartialPath("root.db1.device.a1"),
+ new TimeSeriesViewOperand("root.sg1.d2"));
+ }
+ }),
+ prefixPathPattern)
+ .orElseThrow(AssertionError::new));
+ }
+
+ @Test
+ public void testAlterLogicalView() throws IllegalPathException {
+ Assert.assertEquals(
+ new AlterLogicalViewNode(
+ new PlanNodeId("2024-04-30-1"),
+ new HashMap<PartialPath, ViewExpression>() {
+ {
+ put(new PartialPath("root.db.device.a1"), new TimeSeriesViewOperand("root.sg1.d1"));
+ }
+ }),
+ IoTDBSchemaRegionExtractor.PATTERN_PARSE_VISITOR
+ .visitAlterLogicalView(
+ new AlterLogicalViewNode(
+ new PlanNodeId("2024-04-30-1"),
+ new HashMap<PartialPath, ViewExpression>() {
+ {
+ put(
+ new PartialPath("root.db.device.a1"),
+ new TimeSeriesViewOperand("root.sg1.d1"));
+ put(
+ new PartialPath("root.db1.device.a1"),
+ new TimeSeriesViewOperand("root.sg1.d2"));
+ }
+ }),
+ prefixPathPattern)
+ .orElseThrow(AssertionError::new));
+ }
+
+ @Test
+ public void testDeleteData() throws IllegalPathException {
+ Assert.assertEquals(
+ new DeleteDataNode(
+ new PlanNodeId("2024-04-30-1"),
+ Collections.singletonList(new PartialPath("root.db.device.s1")),
+ Long.MIN_VALUE,
+ Long.MAX_VALUE),
+ IoTDBSchemaRegionExtractor.PATTERN_PARSE_VISITOR
+ .visitDeleteData(
+ new DeleteDataNode(
+ new PlanNodeId("2024-04-30-1"),
+ Arrays.asList(
+ new PartialPath("root.*.device.s1"), new PartialPath("root.db.*.s1")),
+ Long.MIN_VALUE,
+ Long.MAX_VALUE),
+ prefixPathPattern)
+ .orElseThrow(AssertionError::new));
+ }
+}
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/IoTDBPipePatternTest.java
index 5c74f3ea21c..db42d713012 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/IoTDBPipePatternTest.java
@@ -30,48 +30,48 @@ public class IoTDBPipePatternTest {
@Test
public void testIotdbPipePattern() {
// Test legal and illegal pattern
- String[] legalPatterns = {
+ final String[] legalPatterns = {
"root", "root.db", "root.db.d1.s", "root.db.`1`", "root.*.d.*s.s",
};
- String[] illegalPatterns = {
+ final String[] illegalPatterns = {
"root.", "roo", "", "root..", "root./",
};
- for (String s : legalPatterns) {
+ for (final String s : legalPatterns) {
Assert.assertTrue(new IoTDBPipePattern(s).isLegal());
}
- for (String t : illegalPatterns) {
+ for (final String t : illegalPatterns) {
try {
Assert.assertFalse(new IoTDBPipePattern(t).isLegal());
- } catch (Exception e) {
+ } catch (final Exception e) {
Assert.assertTrue(e instanceof PipeException);
}
}
// Test pattern cover db
- String db = "root.db";
- String[] patternsCoverDb = {
+ final String db = "root.db";
+ final String[] patternsCoverDb = {
"root.**", "root.db.**", "root.*db*.**",
};
- String[] patternsNotCoverDb = {
+ final String[] patternsNotCoverDb = {
"root.db", "root.*", "root.*.*", "root.db.*.**", "root.db.d1", "root.**.db.**",
};
- for (String s : patternsCoverDb) {
+ for (final String s : patternsCoverDb) {
Assert.assertTrue(new IoTDBPipePattern(s).coversDb(db));
}
- for (String t : patternsNotCoverDb) {
+ for (final String t : patternsNotCoverDb) {
Assert.assertFalse(new IoTDBPipePattern(t).coversDb(db));
}
- String device = "root.db.d1";
+ final String device = "root.db.d1";
// Test pattern cover device
- String[] patternsCoverDevice = {
+ final String[] patternsCoverDevice = {
"root.**", "root.db.**", "root.*.*.*", "root.db.d1.*", "root.*db*.*d*.*", "root.**.*1.*",
};
- String[] patternsNotCoverDevice = {
+ final String[] patternsNotCoverDevice = {
"root.*", "root.*.*", "root.db.d1", "root.db.d2.*", "root.**.d2.**",
};
- for (String s : patternsCoverDevice) {
+ for (final String s : patternsCoverDevice) {
Assert.assertTrue(new IoTDBPipePattern(s).coversDevice(device));
}
for (String t : patternsNotCoverDevice) {
@@ -79,31 +79,31 @@ public class IoTDBPipePatternTest {
}
// Test pattern may overlap with device
- String[] patternsOverlapWithDevice = {
+ final String[] patternsOverlapWithDevice = {
"root.db.**", "root.db.d1", "root.db.d1.*", "root.db.d1.s1", "root.**.d2.**", "root.*.d*.**",
};
- String[] patternsNotOverlapWithDevice = {
+ final String[] patternsNotOverlapWithDevice = {
"root.db.d2.**", "root.db2.d1.**", "root.db.db.d1.**",
};
- for (String s : patternsOverlapWithDevice) {
+ for (final String s : patternsOverlapWithDevice) {
Assert.assertTrue(new IoTDBPipePattern(s).mayOverlapWithDevice(device));
}
- for (String t : patternsNotOverlapWithDevice) {
+ for (final String t : patternsNotOverlapWithDevice) {
Assert.assertFalse(new IoTDBPipePattern(t).mayOverlapWithDevice(device));
}
// Test pattern match measurement
- String measurement = "s1";
- String[] patternsMatchMeasurement = {
+ final String measurement = "s1";
+ final String[] patternsMatchMeasurement = {
"root.db.d1.s1", "root.db.d1.*",
};
- String[] patternsNotMatchMeasurement = {
+ final String[] patternsNotMatchMeasurement = {
"root.db.d1", "root.db.d1", "root.db.d1.*.*",
};
- for (String s : patternsMatchMeasurement) {
+ for (final String s : patternsMatchMeasurement) {
Assert.assertTrue(new IoTDBPipePattern(s).matchesMeasurement(device, measurement));
}
- for (String t : patternsNotMatchMeasurement) {
+ for (final String t : patternsNotMatchMeasurement) {
Assert.assertFalse(new IoTDBPipePattern(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/PrefixPipePatternTest.java
index cb327d2dac0..d81c50a57ce 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/PrefixPipePatternTest.java
@@ -29,76 +29,76 @@ public class PrefixPipePatternTest {
@Test
public void testPrefixPipePattern() {
// Test legal and illegal pattern
- String[] legalPatterns = {
+ final String[] legalPatterns = {
"root", "root.", "root.db", "root.db.d1.s", "root.db.`1`",
};
- String[] illegalPatterns = {
+ final String[] illegalPatterns = {
"roo", "", "root..", "root./",
};
- for (String s : legalPatterns) {
+ for (final String s : legalPatterns) {
Assert.assertTrue(new PrefixPipePattern(s).isLegal());
}
- for (String t : illegalPatterns) {
+ for (final String t : illegalPatterns) {
Assert.assertFalse(new PrefixPipePattern(t).isLegal());
}
// Test pattern cover db
- String db = "root.db";
- String[] patternsCoverDb = {
+ final String db = "root.db";
+ final String[] patternsCoverDb = {
"root", "root.", "root.d", "root.db",
};
- String[] patternsNotCoverDb = {
+ final String[] patternsNotCoverDb = {
"root.**", "root.db.",
};
- for (String s : patternsCoverDb) {
+ for (final String s : patternsCoverDb) {
Assert.assertTrue(new PrefixPipePattern(s).coversDb(db));
}
- for (String t : patternsNotCoverDb) {
+ for (final String t : patternsNotCoverDb) {
Assert.assertFalse(new PrefixPipePattern(t).coversDb(db));
}
- String device = "root.db.d1";
+ final String device = "root.db.d1";
// Test pattern cover device
- String[] patternsCoverDevice = {
+ final String[] patternsCoverDevice = {
"root", "root.", "root.d", "root.db", "root.db.", "root.db.d", "root.db.d1",
};
- String[] patternsNotCoverDevice = {
+ final String[] patternsNotCoverDevice = {
"root.db.d1.", "root.db.d1.s1", "root.**", "root.db.d2",
};
- for (String s : patternsCoverDevice) {
+ for (final String s : patternsCoverDevice) {
Assert.assertTrue(new PrefixPipePattern(s).coversDevice(device));
}
- for (String t : patternsNotCoverDevice) {
+ for (final String t : patternsNotCoverDevice) {
Assert.assertFalse(new PrefixPipePattern(t).coversDevice(device));
}
// Test pattern may overlap with device
- String[] patternsOverlapWithDevice = {
+ final String[] patternsOverlapWithDevice = {
"root", "root.db.d1", "root.db.d1.", "root.db.d1.s1",
};
- String[] patternsNotOverlapWithDevice = {
+ final String[] patternsNotOverlapWithDevice = {
"root.db.d2", "root.**",
};
- for (String s : patternsOverlapWithDevice) {
+ for (final String s : patternsOverlapWithDevice) {
Assert.assertTrue(new PrefixPipePattern(s).mayOverlapWithDevice(device));
}
- for (String t : patternsNotOverlapWithDevice) {
+ for (final String t : patternsNotOverlapWithDevice) {
Assert.assertFalse(new PrefixPipePattern(t).mayOverlapWithDevice(device));
}
// Test pattern match measurement
- String measurement = "s1";
- String[] patternsMatchMeasurement = {
+ final String measurement = "s1";
+ final String[] patternsMatchMeasurement = {
"root.db.d1", "root.db.d1.", "root.db.d1.s", "root.db.d1.s1",
};
- String[] patternsNotMatchMeasurement = {
+ final String[] patternsNotMatchMeasurement = {
"root.db.d1.s11", "root.db.d1.s2",
};
- for (String s : patternsMatchMeasurement) {
+ for (final String s : patternsMatchMeasurement) {
Assert.assertTrue(new PrefixPipePattern(s).matchesMeasurement(device, measurement));
}
- for (String t : patternsNotMatchMeasurement) {
+ for (final String t : patternsNotMatchMeasurement) {
Assert.assertFalse(new PrefixPipePattern(t).matchesMeasurement(device, measurement));
}
}
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 e4195745895..9d6ce0ffde8 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
@@ -26,7 +26,12 @@ import org.apache.iotdb.commons.pipe.datastructure.queue.ConcurrentIterableLinke
import org.apache.iotdb.commons.pipe.datastructure.queue.listening.AbstractPipeListeningQueue;
import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
import org.apache.iotdb.commons.pipe.event.PipeSnapshotEvent;
+import org.apache.iotdb.commons.pipe.event.PipeWritePlanEvent;
import org.apache.iotdb.commons.pipe.event.ProgressReportEvent;
+import org.apache.iotdb.commons.pipe.pattern.IoTDBPipePattern;
+import org.apache.iotdb.commons.pipe.pattern.PipePattern;
+import org.apache.iotdb.pipe.api.customizer.configuration.PipeExtractorRuntimeConfiguration;
+import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
import org.apache.iotdb.pipe.api.event.Event;
import org.apache.iotdb.pipe.api.exception.PipeException;
@@ -35,10 +40,13 @@ import org.apache.tsfile.utils.Pair;
import java.util.LinkedList;
import java.util.List;
import java.util.Objects;
+import java.util.Optional;
import java.util.concurrent.atomic.AtomicBoolean;
public abstract class IoTDBNonDataRegionExtractor extends IoTDBExtractor {
+ protected IoTDBPipePattern pipePattern;
+
private List<PipeSnapshotEvent> historicalEvents = new LinkedList<>();
// A fixed size initialized only when the historicalEvents are first
// filled. Used only for metric framework.
@@ -53,6 +61,24 @@ public abstract class IoTDBNonDataRegionExtractor extends IoTDBExtractor {
protected abstract AbstractPipeListeningQueue getListeningQueue();
+ @Override
+ public void customize(
+ final PipeParameters parameters, final PipeExtractorRuntimeConfiguration configuration)
+ throws Exception {
+ super.customize(parameters, configuration);
+
+ final PipePattern pattern = PipePattern.parsePipePatternFromSourceParameters(parameters);
+ if (!(pattern instanceof IoTDBPipePattern
+ && (((IoTDBPipePattern) pattern).isPrefix()
+ || ((IoTDBPipePattern) 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;
+ }
+
@Override
public void start() throws Exception {
if (hasBeenStarted.get() || hasBeenClosed.get()) {
@@ -128,7 +154,7 @@ public abstract class IoTDBNonDataRegionExtractor extends IoTDBExtractor {
historicalEvents
.remove(0)
.shallowCopySelfAndBindPipeTaskMetaForProgressReport(
- pipeName, pipeTaskMeta, null, Long.MIN_VALUE, Long.MAX_VALUE);
+ pipeName, pipeTaskMeta, pipePattern, Long.MIN_VALUE, Long.MAX_VALUE);
if (historicalEvents.isEmpty()) {
// We only report progress for the last snapshot event.
@@ -143,23 +169,27 @@ public abstract class IoTDBNonDataRegionExtractor extends IoTDBExtractor {
}
// Realtime
- EnrichedEvent realtimeEvent = (EnrichedEvent) iterator.next(getMaxBlockingTimeMs());
+ PipeWritePlanEvent realtimeEvent = (PipeWritePlanEvent) iterator.next(getMaxBlockingTimeMs());
if (Objects.isNull(realtimeEvent)) {
return null;
}
- if (!isTypeListened(realtimeEvent)
+ realtimeEvent = trimRealtimeEventByPipePattern(realtimeEvent).orElse(null);
+ if (Objects.isNull(realtimeEvent)
+ || !isTypeListened(realtimeEvent)
|| (!isForwardingPipeRequests && realtimeEvent.isGeneratedByPipe())) {
final ProgressReportEvent event =
- new ProgressReportEvent(pipeName, pipeTaskMeta, null, Long.MIN_VALUE, Long.MAX_VALUE);
+ new ProgressReportEvent(
+ pipeName, pipeTaskMeta, pipePattern, Long.MIN_VALUE, Long.MAX_VALUE);
event.bindProgressIndex(new MetaProgressIndex(iterator.getNextIndex() - 1));
event.increaseReferenceCount(IoTDBNonDataRegionExtractor.class.getName());
return event;
}
realtimeEvent =
- realtimeEvent.shallowCopySelfAndBindPipeTaskMetaForProgressReport(
- pipeName, pipeTaskMeta, null, Long.MIN_VALUE, Long.MAX_VALUE);
+ (PipeWritePlanEvent)
+ realtimeEvent.shallowCopySelfAndBindPipeTaskMetaForProgressReport(
+ pipeName, pipeTaskMeta, pipePattern, Long.MIN_VALUE, Long.MAX_VALUE);
realtimeEvent.bindProgressIndex(new MetaProgressIndex(iterator.getNextIndex() - 1));
realtimeEvent.increaseReferenceCount(IoTDBNonDataRegionExtractor.class.getName());
return realtimeEvent;
@@ -167,7 +197,11 @@ public abstract class IoTDBNonDataRegionExtractor extends IoTDBExtractor {
protected abstract long getMaxBlockingTimeMs();
- protected abstract boolean isTypeListened(final Event event);
+ // The trimmed event shall be non-null.
+ protected abstract Optional<PipeWritePlanEvent> trimRealtimeEventByPipePattern(
+ final PipeWritePlanEvent event);
+
+ protected abstract boolean isTypeListened(final PipeWritePlanEvent event);
protected abstract void confineHistoricalEventTransferTypes(final PipeSnapshotEvent event);
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/pattern/IoTDBPipePattern.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/pattern/IoTDBPipePattern.java
index ec4192adff3..6c82c7e3d9c 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/pattern/IoTDBPipePattern.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/pattern/IoTDBPipePattern.java
@@ -22,26 +22,37 @@ package org.apache.iotdb.commons.pipe.pattern;
import org.apache.iotdb.commons.conf.IoTDBConstant;
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.path.PathPatternUtil;
import org.apache.iotdb.commons.pipe.config.constant.PipeExtractorConstant;
import org.apache.iotdb.commons.utils.PathUtils;
import org.apache.iotdb.pipe.api.exception.PipeException;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
import java.util.Objects;
+import java.util.stream.Collectors;
public class IoTDBPipePattern extends PipePattern {
private final PartialPath patternPartialPath;
- public IoTDBPipePattern(String pattern) {
+ public IoTDBPipePattern(final String pattern) {
super(pattern);
try {
patternPartialPath = new PartialPath(getPattern());
- } catch (IllegalPathException e) {
+ } catch (final IllegalPathException e) {
throw new PipeException("Illegal IoTDBPipePattern: " + getPattern(), e);
}
}
+ public static <T> List<T> applyIndexesOnList(
+ final int[] filteredIndexes, final List<T> originalList) {
+ return Arrays.stream(filteredIndexes).mapToObj(originalList::get).collect(Collectors.toList());
+ }
+
@Override
public String getDefaultPattern() {
return PipeExtractorConstant.EXTRACTOR_PATTERN_IOTDB_DEFAULT_VALUE;
@@ -56,44 +67,44 @@ public class IoTDBPipePattern extends PipePattern {
try {
PathUtils.isLegalPath(pattern);
return true;
- } catch (IllegalPathException e) {
+ } catch (final IllegalPathException e) {
return false;
}
}
@Override
- public boolean coversDb(String db) {
+ public boolean coversDb(final String db) {
try {
return patternPartialPath.include(
new PartialPath(db, IoTDBConstant.MULTI_LEVEL_PATH_WILDCARD));
- } catch (IllegalPathException e) {
+ } catch (final IllegalPathException e) {
return false;
}
}
@Override
- public boolean coversDevice(String device) {
+ public boolean coversDevice(final String device) {
try {
return patternPartialPath.include(
new PartialPath(device, IoTDBConstant.ONE_LEVEL_PATH_WILDCARD));
- } catch (IllegalPathException e) {
+ } catch (final IllegalPathException e) {
return false;
}
}
@Override
- public boolean mayOverlapWithDevice(String device) {
+ public boolean mayOverlapWithDevice(final String device) {
try {
// Another way is to use patternPath.overlapWith("device.*"),
// there will be no false positives but time cost may be higher.
return patternPartialPath.matchPrefixPath(new PartialPath(device));
- } catch (IllegalPathException e) {
+ } catch (final IllegalPathException e) {
return false;
}
}
@Override
- public boolean matchesMeasurement(String device, String measurement) {
+ public boolean matchesMeasurement(final String device, final String measurement) {
// For aligned timeseries, empty measurement is an alias of the time column.
if (Objects.isNull(measurement) || measurement.isEmpty()) {
return false;
@@ -101,11 +112,77 @@ public class IoTDBPipePattern extends PipePattern {
try {
return patternPartialPath.matchFullPath(new PartialPath(device, measurement));
- } catch (IllegalPathException e) {
+ } catch (final IllegalPathException e) {
+ return false;
+ }
+ }
+
+ /**
+ * Check if the {@link PipePattern} matches the given prefix path.
+ *
+ * <p>NOTE: In schema transmission, {@link #mayOverlapWithDevice(String)} 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 setting.
+ */
+ public boolean matchPrefixPath(final String path) {
+ return mayOverlapWithDevice(path);
+ }
+
+ /**
+ * This is the precise form of the device overlap and is used only be device template transfer.
+ */
+ public boolean matchDevice(final String devicePath) {
+ try {
+ return patternPartialPath.overlapWith(new PartialPath(devicePath, "*"));
+ } catch (final IllegalPathException e) {
return false;
}
}
+ /**
+ * Return if the given tail node matches the pattern's tail node. Caller shall ensure that it is a
+ * prefix or full path pattern.
+ */
+ public boolean matchTailNode(final String tailNode) {
+ return !isFullPath() || patternPartialPath.getTailNode().equals(tailNode);
+ }
+
+ /**
+ * Get the intersection of the given {@link PartialPath} and the {@link PipePattern}, Only used by
+ * schema transmission. Caller shall ensure that it is a prefix or full path pattern.
+ */
+ public List<PartialPath> getIntersection(final PartialPath partialPath) {
+ if (isFullPath()) {
+ return partialPath.matchFullPath(patternPartialPath)
+ ? Collections.singletonList(patternPartialPath)
+ : Collections.emptyList();
+ }
+ return partialPath.intersectWithPrefixPattern(patternPartialPath);
+ }
+
+ /**
+ * Get the intersection of the given {@link PathPatternTree} and the {@link PipePattern}. Only
+ * used by schema transmission. Caller shall ensure that it is a prefix or full path pattern.
+ */
+ public PathPatternTree getIntersection(final PathPatternTree patternTree) {
+ final PathPatternTree thisPatternTree = new PathPatternTree();
+ thisPatternTree.appendPathPattern(patternPartialPath);
+ thisPatternTree.constructTree();
+ return patternTree.intersectWithFullPathPrefixTree(thisPatternTree);
+ }
+
+ public boolean isPrefix() {
+ return PathPatternUtil.isMultiLevelMatchWildcard(patternPartialPath.getTailNode())
+ && !new PartialPath(
+ Arrays.copyOfRange(
+ patternPartialPath.getNodes(), 0, patternPartialPath.getNodeLength() - 1))
+ .hasWildcard();
+ }
+
+ public boolean isFullPath() {
+ return !patternPartialPath.hasWildcard();
+ }
+
@Override
public String toString() {
return "IoTDBPipePattern" + super.toString();
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/pattern/PipePattern.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/pattern/PipePattern.java
index 04be541f625..803c24621cf 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/pattern/PipePattern.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/pattern/PipePattern.java
@@ -41,7 +41,7 @@ public abstract class PipePattern {
protected final String pattern;
- protected PipePattern(String pattern) {
+ protected PipePattern(final String pattern) {
this.pattern = pattern != null ? pattern : getDefaultPattern();
}
@@ -58,7 +58,8 @@ public abstract class PipePattern {
*
* @return The interpreted {@link PipePattern} which is not null.
*/
- public static PipePattern parsePipePatternFromSourceParameters(PipeParameters sourceParameters) {
+ public static PipePattern 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,
@@ -95,7 +96,7 @@ public abstract class PipePattern {
// 3. If neither "source.path" nor "source.pattern" is specified,
// this pipe source will match all data.
- return new PrefixPipePattern(null);
+ return new IoTDBPipePattern(null);
}
public abstract String getDefaultPattern();
@@ -104,10 +105,10 @@ public abstract class PipePattern {
public abstract boolean isLegal();
/** Check if this pattern matches all time-series under a database. */
- public abstract boolean coversDb(String db);
+ public abstract boolean coversDb(final String db);
/** Check if a device's all measurements are covered by this pattern. */
- public abstract boolean coversDevice(String device);
+ public abstract boolean coversDevice(final String device);
/**
* Check if a device may have some measurements matched by the pattern.
@@ -117,14 +118,14 @@ public abstract class PipePattern {
* <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.
*/
- public abstract boolean mayOverlapWithDevice(String device);
+ public abstract boolean mayOverlapWithDevice(final String 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 PipePattern#mayOverlapWithDevice(String)} is true.
*/
- public abstract boolean matchesMeasurement(String device, String measurement);
+ public abstract boolean matchesMeasurement(final String device, final String measurement);
@Override
public String toString() {
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/pattern/PrefixPipePattern.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/pattern/PrefixPipePattern.java
index 5e9c3a5aa44..52f70057052 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/pattern/PrefixPipePattern.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/pattern/PrefixPipePattern.java
@@ -30,7 +30,7 @@ import java.util.Arrays;
public class PrefixPipePattern extends PipePattern {
- public PrefixPipePattern(String pattern) {
+ public PrefixPipePattern(final String pattern) {
super(pattern);
}
@@ -47,25 +47,25 @@ public class PrefixPipePattern extends PipePattern {
try {
PathUtils.isLegalPath(pattern);
- } catch (IllegalPathException e) {
+ } catch (final IllegalPathException e) {
try {
if ("root".equals(pattern) || "root.".equals(pattern)) {
return true;
}
// Split the pattern to nodes.
- String[] pathNodes = StringUtils.splitPreserveAllTokens(pattern, "\\.");
+ final String[] pathNodes = StringUtils.splitPreserveAllTokens(pattern, "\\.");
// Check whether the pattern without last node is legal.
PathUtils.splitPathToDetachedNodes(
String.join(".", Arrays.copyOfRange(pathNodes, 0, pathNodes.length - 1)));
- String lastNode = pathNodes[pathNodes.length - 1];
+ final String lastNode = pathNodes[pathNodes.length - 1];
// Check whether the last node is legal.
if (!"".equals(lastNode)) {
Double.parseDouble(lastNode);
}
- } catch (Exception ignored) {
+ } catch (final Exception ignored) {
return false;
}
}
@@ -73,19 +73,19 @@ public class PrefixPipePattern extends PipePattern {
}
@Override
- public boolean coversDb(String db) {
+ public boolean coversDb(final String db) {
return pattern.length() <= db.length() && db.startsWith(pattern);
}
@Override
- public boolean coversDevice(String device) {
+ public boolean coversDevice(final String device) {
// for example, pattern is root.a.b and device is root.a.b.c
// in this case, the extractor can be matched without checking the measurements
return pattern.length() <= device.length() && device.startsWith(pattern);
}
@Override
- public boolean mayOverlapWithDevice(String device) {
+ public boolean mayOverlapWithDevice(final String device) {
return (
// for example, pattern is root.a.b and device is root.a.b.c
// in this case, the extractor can be matched without checking the measurements
@@ -97,7 +97,7 @@ public class PrefixPipePattern extends PipePattern {
}
@Override
- public boolean matchesMeasurement(String device, String measurement) {
+ public boolean matchesMeasurement(final String device, final String measurement) {
// We assume that the device is already matched.
if (pattern.length() <= device.length()) {
return true;
@@ -106,7 +106,7 @@ public class PrefixPipePattern extends PipePattern {
// For example, pattern is "root.a.b.c", device is "root.a.b",
// then measurements "c" and "cc" can be matched,
// measurements "d" or "dc" can't be matched.
- String dotAndMeasurement = TsFileConstant.PATH_SEPARATOR + measurement;
+ final String dotAndMeasurement = TsFileConstant.PATH_SEPARATOR + measurement;
return
// low cost check comes first
pattern.length() <= device.length() + dotAndMeasurement.length()