You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ro...@apache.org on 2024/10/11 05:05:59 UTC
(iotdb) branch in-pipe-table-pattern-matching created (now da8441c23f4)
This is an automated email from the ASF dual-hosted git repository.
rong pushed a change to branch in-pipe-table-pattern-matching
in repository https://gitbox.apache.org/repos/asf/iotdb.git
at da8441c23f4 Pipe: Consider table related events in EnrichedEvent#mayEventPathsOverlappedWithPattern
This branch includes the following new commits:
new da8441c23f4 Pipe: Consider table related events in EnrichedEvent#mayEventPathsOverlappedWithPattern
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(iotdb) 01/01: Pipe: Consider table related events in EnrichedEvent#mayEventPathsOverlappedWithPattern
Posted by ro...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rong pushed a commit to branch in-pipe-table-pattern-matching
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit da8441c23f488368ce3f29d0af0a07397f9604f2
Author: Steve Yurong Su <ro...@apache.org>
AuthorDate: Fri Oct 11 13:05:47 2024 +0800
Pipe: Consider table related events in EnrichedEvent#mayEventPathsOverlappedWithPattern
---
.../common/tablet/PipeInsertNodeTabletInsertionEvent.java | 9 +++++++++
.../event/common/tablet/PipeRawTabletInsertionEvent.java | 5 +----
.../pipe/event/common/tsfile/PipeTsFileInsertionEvent.java | 14 +++++++++++++-
.../db/pipe/pattern/CachedSchemaPatternMatcherTest.java | 7 ++++---
4 files changed, 27 insertions(+), 8 deletions(-)
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeInsertNodeTabletInsertionEvent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeInsertNodeTabletInsertionEvent.java
index 4f6157928fa..d58ebffc89a 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeInsertNodeTabletInsertionEvent.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeInsertNodeTabletInsertionEvent.java
@@ -31,6 +31,9 @@ import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowsNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertTabletNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertRowNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertRowsNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertTabletNode;
import org.apache.iotdb.db.storageengine.dataregion.wal.exception.WALPipeException;
import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALEntryHandler;
import org.apache.iotdb.pipe.api.access.Row;
@@ -270,6 +273,12 @@ public class PipeInsertNodeTabletInsertionEvent extends PipeInsertionEvent
return true;
}
+ if (insertNode instanceof RelationalInsertRowNode
+ || insertNode instanceof RelationalInsertTabletNode
+ || insertNode instanceof RelationalInsertRowsNode) {
+ return true;
+ }
+
if (insertNode instanceof InsertRowNode || insertNode instanceof InsertTabletNode) {
final PartialPath devicePartialPath = insertNode.getTargetPath();
return Objects.isNull(devicePartialPath)
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeRawTabletInsertionEvent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeRawTabletInsertionEvent.java
index d09d9579c7e..364ebca2807 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeRawTabletInsertionEvent.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeRawTabletInsertionEvent.java
@@ -35,7 +35,6 @@ import org.apache.iotdb.pipe.api.access.Row;
import org.apache.iotdb.pipe.api.collector.RowCollector;
import org.apache.iotdb.pipe.api.event.dml.insertion.TabletInsertionEvent;
-import org.apache.tsfile.file.metadata.IDeviceID;
import org.apache.tsfile.write.record.Tablet;
import java.util.Objects;
@@ -245,9 +244,7 @@ public class PipeRawTabletInsertionEvent extends PipeInsertionEvent
@Override
public boolean mayEventPathsOverlappedWithPattern() {
- final String deviceId = getDeviceId();
- return Objects.isNull(deviceId)
- || treePattern.mayOverlapWithDevice(IDeviceID.Factory.DEFAULT_FACTORY.create(deviceId));
+ return sourceEvent == null || sourceEvent.mayEventPathsOverlappedWithPattern();
}
public void markAsNeedToReport() {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java
index 55ce6343185..dcee18d0311 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java
@@ -39,6 +39,7 @@ import org.apache.iotdb.pipe.api.event.dml.insertion.TsFileInsertionEvent;
import org.apache.iotdb.pipe.api.exception.PipeException;
import org.apache.tsfile.file.metadata.IDeviceID;
+import org.apache.tsfile.file.metadata.PlainDeviceID;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -50,6 +51,9 @@ import java.util.Objects;
import java.util.Set;
import java.util.concurrent.atomic.AtomicBoolean;
+import static org.apache.tsfile.common.constant.TsFileConstant.PATH_ROOT;
+import static org.apache.tsfile.common.constant.TsFileConstant.PATH_SEPARATOR;
+
public class PipeTsFileInsertionEvent extends PipeInsertionEvent implements TsFileInsertionEvent {
private static final Logger LOGGER = LoggerFactory.getLogger(PipeTsFileInsertionEvent.class);
@@ -382,7 +386,15 @@ public class PipeTsFileInsertionEvent extends PipeInsertionEvent implements TsFi
false);
final Set<IDeviceID> deviceSet =
Objects.nonNull(deviceIsAlignedMap) ? deviceIsAlignedMap.keySet() : resource.getDevices();
- return deviceSet.stream().anyMatch(treePattern::mayOverlapWithDevice);
+ return deviceSet.stream()
+ .anyMatch(
+ deviceID ->
+ // Table model
+ !(deviceID instanceof PlainDeviceID
+ || deviceID.getTableName().startsWith(PATH_ROOT + PATH_SEPARATOR)
+ || deviceID.getTableName().equals(PATH_ROOT))
+ // Tree model
+ || treePattern.mayOverlapWithDevice(deviceID));
} catch (final Exception e) {
LOGGER.warn(
"Pipe {}: failed to get devices from TsFile {}, extract it anyway",
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/pattern/CachedSchemaPatternMatcherTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/pattern/CachedSchemaPatternMatcherTest.java
index e51c648a036..487fabfda38 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/pattern/CachedSchemaPatternMatcherTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/pattern/CachedSchemaPatternMatcherTest.java
@@ -34,7 +34,7 @@ import org.apache.iotdb.pipe.api.event.Event;
import org.apache.tsfile.common.constant.TsFileConstant;
import org.apache.tsfile.file.metadata.IDeviceID;
-import org.apache.tsfile.file.metadata.PlainDeviceID;
+import org.apache.tsfile.file.metadata.StringArrayDeviceID;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
@@ -146,7 +146,8 @@ public class CachedSchemaPatternMatcherTest {
Map<IDeviceID, String[]> deviceMap =
IntStream.range(0, deviceNum)
.mapToObj(String::valueOf)
- .collect(Collectors.toMap(s -> new PlainDeviceID("root.db" + s), s -> new String[0]));
+ .collect(
+ Collectors.toMap(s -> new StringArrayDeviceID("root.db" + s), s -> new String[0]));
String[] measurements =
IntStream.range(0, seriesNum).mapToObj(num -> "s" + num).toArray(String[]::new);
long totalTime = 0;
@@ -156,7 +157,7 @@ public class CachedSchemaPatternMatcherTest {
new MockedPipeRealtimeEvent(
null,
null,
- Collections.singletonMap(new PlainDeviceID("root.db" + i), measurements),
+ Collections.singletonMap(new StringArrayDeviceID("root.db" + i), measurements),
null);
long startTime = System.currentTimeMillis();
matcher.match(event).forEach(extractor -> extractor.extract(event));