From 2db21c618973029162e36e443e77d51f8e3436e3 Mon Sep 17 00:00:00 2001 From: Steve Yurong Su Date: Mon, 17 Jun 2024 12:52:12 +0800 Subject: [PATCH 01/19] parameters --- ...peHistoricalDataRegionTsFileExtractor.java | 2 +- .../PipeRealtimeDataRegionExtractor.java | 47 ++++++++++++++----- .../constant/PipeExtractorConstant.java | 3 ++ 3 files changed, 39 insertions(+), 13 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/historical/PipeHistoricalDataRegionTsFileExtractor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/historical/PipeHistoricalDataRegionTsFileExtractor.java index 237dace7c02f7..a0dadcfc9fa6c 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/historical/PipeHistoricalDataRegionTsFileExtractor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/historical/PipeHistoricalDataRegionTsFileExtractor.java @@ -109,8 +109,8 @@ public class PipeHistoricalDataRegionTsFileExtractor implements PipeHistoricalDa private long historicalDataExtractionEndTime = Long.MAX_VALUE; // Event time private long historicalDataExtractionTimeLowerBound; // Arrival time - private boolean sloppyPattern; private boolean sloppyTimeRange; // true to disable time range filter after extraction + private boolean sloppyPattern; // true to disable pattern filter after extraction private Pair listeningOptionPair; private boolean shouldExtractInsertion; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/PipeRealtimeDataRegionExtractor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/PipeRealtimeDataRegionExtractor.java index c26db376da096..262f2d824aa45 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/PipeRealtimeDataRegionExtractor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/PipeRealtimeDataRegionExtractor.java @@ -54,6 +54,7 @@ import java.util.Arrays; import java.util.List; import java.util.Objects; +import java.util.Set; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicReference; import java.util.stream.Collectors; @@ -61,7 +62,10 @@ import static org.apache.iotdb.commons.pipe.config.constant.PipeExtractorConstant.EXTRACTOR_END_TIME_KEY; import static org.apache.iotdb.commons.pipe.config.constant.PipeExtractorConstant.EXTRACTOR_MODS_ENABLE_DEFAULT_VALUE; import static org.apache.iotdb.commons.pipe.config.constant.PipeExtractorConstant.EXTRACTOR_MODS_ENABLE_KEY; +import static org.apache.iotdb.commons.pipe.config.constant.PipeExtractorConstant.EXTRACTOR_REALTIME_LOOSE_RANGE_DEFAULT_VALUE; import static org.apache.iotdb.commons.pipe.config.constant.PipeExtractorConstant.EXTRACTOR_REALTIME_LOOSE_RANGE_KEY; +import static org.apache.iotdb.commons.pipe.config.constant.PipeExtractorConstant.EXTRACTOR_REALTIME_LOOSE_RANGE_PATH_VALUE; +import static org.apache.iotdb.commons.pipe.config.constant.PipeExtractorConstant.EXTRACTOR_REALTIME_LOOSE_RANGE_TIME_VALUE; import static org.apache.iotdb.commons.pipe.config.constant.PipeExtractorConstant.EXTRACTOR_START_TIME_KEY; import static org.apache.iotdb.commons.pipe.config.constant.PipeExtractorConstant.SOURCE_END_TIME_KEY; import static org.apache.iotdb.commons.pipe.config.constant.PipeExtractorConstant.SOURCE_MODS_ENABLE_KEY; @@ -102,6 +106,7 @@ public abstract class PipeRealtimeDataRegionExtractor implements PipeExtractor { private boolean shouldTransferModFile; // Whether to transfer mods private boolean sloppyTimeRange; // true to disable time range filter after extraction + private boolean sloppyPattern; // true to disable pattern filter after extraction // This queue is used to store pending events extracted by the method extract(). The method // supply() will poll events from this queue and send them to the next pipe plugin. @@ -142,10 +147,33 @@ public void validate(final PipeParameterValidator validator) throws Exception { EXTRACTOR_END_TIME_KEY, realtimeDataExtractionEndTime)); } + } catch (final PipeParameterNotValidException e) { + throw e; } catch (final Exception e) { // compatible with the current validation framework throw new PipeParameterNotValidException(e.getMessage()); } + + final Set sloppyOptionSet = + Arrays.stream( + parameters + .getStringOrDefault( + Arrays.asList( + EXTRACTOR_REALTIME_LOOSE_RANGE_KEY, SOURCE_REALTIME_LOOSE_RANGE_KEY), + EXTRACTOR_REALTIME_LOOSE_RANGE_DEFAULT_VALUE) + .split(",")) + .map(String::trim) + .map(String::toLowerCase) + .collect(Collectors.toSet()); + // Avoid empty string + sloppyOptionSet.remove(""); + sloppyTimeRange = sloppyOptionSet.contains(EXTRACTOR_REALTIME_LOOSE_RANGE_TIME_VALUE); + sloppyPattern = sloppyOptionSet.contains(EXTRACTOR_REALTIME_LOOSE_RANGE_PATH_VALUE); + if (!sloppyOptionSet.isEmpty()) { + throw new PipeParameterNotValidException( + String.format( + "Parameters in set %s are not allowed in 'realtime.loose-range'", sloppyOptionSet)); + } } @Override @@ -203,18 +231,13 @@ public void customize( Arrays.asList(SOURCE_MODS_ENABLE_KEY, EXTRACTOR_MODS_ENABLE_KEY), EXTRACTOR_MODS_ENABLE_DEFAULT_VALUE || shouldExtractDeletion); - sloppyTimeRange = - Arrays.stream( - parameters - .getStringOrDefault( - Arrays.asList( - EXTRACTOR_REALTIME_LOOSE_RANGE_KEY, SOURCE_REALTIME_LOOSE_RANGE_KEY), - "") - .split(",")) - .map(String::trim) - .map(String::toLowerCase) - .collect(Collectors.toSet()) - .contains("time"); + if (LOGGER.isInfoEnabled()) { + LOGGER.info( + "Pipe {}@{}: realtime data region extractor is initialized with parameters: {}.", + pipeName, + dataRegionId, + parameters); + } } @Override diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/PipeExtractorConstant.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/PipeExtractorConstant.java index 99efcd48d613d..1bf147e998b15 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/PipeExtractorConstant.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/PipeExtractorConstant.java @@ -84,6 +84,9 @@ public class PipeExtractorConstant { public static final String EXTRACTOR_REALTIME_MODE_BATCH_MODE_VALUE = "batch"; public static final String EXTRACTOR_REALTIME_LOOSE_RANGE_KEY = "extractor.realtime.loose-range"; public static final String SOURCE_REALTIME_LOOSE_RANGE_KEY = "source.realtime.loose-range"; + public static final String EXTRACTOR_REALTIME_LOOSE_RANGE_TIME_VALUE = "time"; + public static final String EXTRACTOR_REALTIME_LOOSE_RANGE_PATH_VALUE = "path"; + public static final String EXTRACTOR_REALTIME_LOOSE_RANGE_DEFAULT_VALUE = ""; public static final String EXTRACTOR_START_TIME_KEY = "extractor.start-time"; public static final String SOURCE_START_TIME_KEY = "source.start-time"; From 161583cbea96888f65dd8487aea03b20a4ba91be Mon Sep 17 00:00:00 2001 From: Steve Yurong Su Date: Mon, 17 Jun 2024 18:35:13 +0800 Subject: [PATCH 02/19] Pipe: Support "source.realtime.loose-range" = "path" in iotdb-source --- .../tsfile/PipeTsFileInsertionEvent.java | 4 ++ .../PipeRealtimeDataRegionExtractor.java | 57 +++++++++++++++++++ 2 files changed, 61 insertions(+) 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 a7c0cc5c255d8..8744eb6cb6999 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 @@ -159,6 +159,10 @@ public boolean waitForTsFileClose() throws InterruptedException { return !resource.isEmpty(); } + public TsFileResource getResource() { + return resource; + } + public File getTsFile() { return tsFile; } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/PipeRealtimeDataRegionExtractor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/PipeRealtimeDataRegionExtractor.java index 262f2d824aa45..62b665441d8d7 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/PipeRealtimeDataRegionExtractor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/PipeRealtimeDataRegionExtractor.java @@ -30,13 +30,18 @@ import org.apache.iotdb.commons.utils.TimePartitionUtils; import org.apache.iotdb.db.pipe.agent.PipeAgent; import org.apache.iotdb.db.pipe.event.common.heartbeat.PipeHeartbeatEvent; +import org.apache.iotdb.db.pipe.event.common.tablet.PipeInsertNodeTabletInsertionEvent; +import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent; import org.apache.iotdb.db.pipe.event.realtime.PipeRealtimeEvent; import org.apache.iotdb.db.pipe.extractor.dataregion.DataRegionListeningFilter; import org.apache.iotdb.db.pipe.extractor.dataregion.realtime.listener.PipeInsertionDataNodeListener; import org.apache.iotdb.db.pipe.extractor.dataregion.realtime.listener.PipeTimePartitionListener; import org.apache.iotdb.db.pipe.metric.PipeDataRegionEventCounter; +import org.apache.iotdb.db.pipe.resource.PipeResourceManager; +import org.apache.iotdb.db.pipe.resource.tsfile.PipeTsFileResourceManager; import org.apache.iotdb.db.storageengine.StorageEngine; import org.apache.iotdb.db.storageengine.dataregion.DataRegion; +import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource; import org.apache.iotdb.db.utils.DateTimeUtils; import org.apache.iotdb.pipe.api.PipeExtractor; import org.apache.iotdb.pipe.api.customizer.configuration.PipeExtractorRuntimeConfiguration; @@ -45,14 +50,18 @@ import org.apache.iotdb.pipe.api.event.Event; import org.apache.iotdb.pipe.api.exception.PipeParameterNotValidException; +import org.apache.tsfile.file.metadata.IDeviceID; +import org.apache.tsfile.file.metadata.PlainDeviceID; import org.apache.tsfile.utils.Pair; import org.checkerframework.checker.nullness.qual.NonNull; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import java.io.IOException; import java.util.ArrayList; import java.util.Arrays; import java.util.List; +import java.util.Map; import java.util.Objects; import java.util.Set; import java.util.concurrent.atomic.AtomicBoolean; @@ -307,6 +316,11 @@ public final void extract(final PipeRealtimeEvent event) { // only skip parsing time for events whose data timestamps may intersect with the time range event.skipParsingTime(); } + if (sloppyPattern && mayEventPathsOverlappedWithPattern(event)) { + // only skip parsing pattern for events whose data paths may intersect with the pattern + event.skipParsingPattern(); + } + doExtract(event); } else { event.decreaseReferenceCount(PipeRealtimeDataRegionExtractor.class.getName(), false); @@ -319,6 +333,49 @@ public final void extract(final PipeRealtimeEvent event) { } } + private boolean mayEventPathsOverlappedWithPattern(final PipeRealtimeEvent realtimeEvent) { + final EnrichedEvent event = realtimeEvent.getEvent(); + + if (event instanceof PipeInsertNodeTabletInsertionEvent) { + final String deviceId = ((PipeInsertNodeTabletInsertionEvent) event).getDeviceId(); + return Objects.isNull(deviceId) || pipePattern.mayOverlapWithDevice(deviceId); + } + + if (event instanceof PipeTsFileInsertionEvent) { + final TsFileResource resource = ((PipeTsFileInsertionEvent) event).getResource(); + if (!resource.isClosed()) { + return true; + } + + try { + final Map deviceIsAlignedMap = + PipeResourceManager.tsfile() + .getDeviceIsAlignedMapFromCache( + PipeTsFileResourceManager.getHardlinkOrCopiedFileInPipeDir( + resource.getTsFile())); + final Set deviceSet = + Objects.nonNull(deviceIsAlignedMap) + ? deviceIsAlignedMap.keySet() + : resource.getDevices(); + return deviceSet.stream() + .anyMatch( + // TODO: use IDeviceID + deviceID -> + pipePattern.mayOverlapWithDevice(((PlainDeviceID) deviceID).toStringID())); + } catch (final IOException e) { + LOGGER.warn( + "Pipe {}@{}: failed to get devices from TsFile {}, extract it anyway", + pipeName, + dataRegionId, + resource.getTsFilePath(), + e); + return true; + } + } + + return false; + } + protected abstract void doExtract(final PipeRealtimeEvent event); protected void extractHeartbeat(final PipeRealtimeEvent event) { From a051e099fc03fdec65144d849fc72eafa4add930 Mon Sep 17 00:00:00 2001 From: Steve Yurong Su Date: Mon, 17 Jun 2024 18:52:44 +0800 Subject: [PATCH 03/19] Update IoTDBPipeExtractorIT.java --- .../it/autocreate/IoTDBPipeExtractorIT.java | 59 ++++++++++++++++++- 1 file changed, 58 insertions(+), 1 deletion(-) 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 f83699f30b7a4..a1244f25b01d6 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 @@ -904,7 +904,7 @@ public void testSourceStartTimeAndEndTimeWorkingWithOrWithoutPattern() throws Ex } @Test - public void testLooseRange() throws Exception { + public void testHistoryLooseRange() throws Exception { final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0); final String receiverIp = receiverDataNode.getIp(); @@ -964,6 +964,63 @@ public void testLooseRange() throws Exception { } } + @Test + public void testRealtimeLooseRange() throws Exception { + final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0); + final String receiverIp = receiverDataNode.getIp(); + final int receiverPort = receiverDataNode.getPort(); + + final Map extractorAttributes = new HashMap<>(); + final Map processorAttributes = new HashMap<>(); + final Map connectorAttributes = new HashMap<>(); + + extractorAttributes.put("source.path", "root.db.d1.at1"); + extractorAttributes.put("source.inclusion", "data.insert"); + extractorAttributes.put("source.realtime.loose-range", "time, path"); + extractorAttributes.put("source.realtime.start-time", "2000"); + extractorAttributes.put("source.realtime.end-time", "10000"); + + connectorAttributes.put("connector", "iotdb-thrift-connector"); + connectorAttributes.put("connector.batch.enable", "false"); + connectorAttributes.put("connector.ip", receiverIp); + connectorAttributes.put("connector.port", Integer.toString(receiverPort)); + + try (final SyncConfigNodeIServiceClient client = + (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) { + TSStatus status = + client.createPipe( + new TCreatePipeReq("p1", connectorAttributes) + .setExtractorAttributes(extractorAttributes) + .setProcessorAttributes(processorAttributes)); + Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode()); + Assert.assertEquals( + TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("p1").getCode()); + + if (!TestUtils.tryExecuteNonQueriesWithRetry( + senderEnv, + Arrays.asList( + "insert into root.db.d1 (time, at1, at2)" + " values (1000, 1, 2), (3000, 3, 4)", + "insert into root.db1.d1 (time, at1, at2)" + " values (1000, 1, 2), (3000, 3, 4)", + "flush"))) { + return; + } + + if (!TestUtils.tryExecuteNonQueriesWithRetry( + senderEnv, + Arrays.asList( + "insert into root.db.d1 (time, at1)" + " values (5000, 1), (16000, 3)", + "insert into root.db.d1 (time, at1, at2)" + " values (5001, 1, 2), (6001, 3, 4)"))) { + return; + } + + TestUtils.assertDataEventuallyOnEnv( + receiverEnv, + "select count(*) from root.** group by level=0", + "count(root.*.*.*),", + Collections.singleton("10,")); + } + } + private void assertTimeseriesCountOnReceiver(BaseEnv receiverEnv, int count) { TestUtils.assertDataEventuallyOnEnv( receiverEnv, "count timeseries", "count(timeseries),", Collections.singleton(count + ",")); From b687d55f574158466002e596ea19924c94e7a818 Mon Sep 17 00:00:00 2001 From: Steve Yurong Su Date: Mon, 17 Jun 2024 18:53:36 +0800 Subject: [PATCH 04/19] Update IoTDBPipeExtractorIT.java --- .../apache/iotdb/pipe/it/autocreate/IoTDBPipeExtractorIT.java | 4 ++-- 1 file changed, 2 insertions(+), 2 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 a1244f25b01d6..874f499f81b63 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 @@ -977,8 +977,8 @@ public void testRealtimeLooseRange() throws Exception { extractorAttributes.put("source.path", "root.db.d1.at1"); extractorAttributes.put("source.inclusion", "data.insert"); extractorAttributes.put("source.realtime.loose-range", "time, path"); - extractorAttributes.put("source.realtime.start-time", "2000"); - extractorAttributes.put("source.realtime.end-time", "10000"); + extractorAttributes.put("source.start-time", "2000"); + extractorAttributes.put("source.end-time", "10000"); connectorAttributes.put("connector", "iotdb-thrift-connector"); connectorAttributes.put("connector.batch.enable", "false"); From b6318f8911890ff4b8b18eca8aae4f062c5a5a6a Mon Sep 17 00:00:00 2001 From: Steve Yurong Su Date: Mon, 17 Jun 2024 19:18:37 +0800 Subject: [PATCH 05/19] refactor --- .../it/autocreate/IoTDBPipeExtractorIT.java | 7 ++ .../pipe/event/UserDefinedEnrichedEvent.java | 5 ++ .../common/heartbeat/PipeHeartbeatEvent.java | 5 ++ .../PipeInsertNodeTabletInsertionEvent.java | 6 ++ .../tablet/PipeRawTabletInsertionEvent.java | 6 ++ .../common/terminate/PipeTerminateEvent.java | 5 ++ .../tsfile/PipeTsFileInsertionEvent.java | 38 ++++++++-- .../event/realtime/PipeRealtimeEvent.java | 5 ++ .../PipeRealtimeDataRegionExtractor.java | 69 +++---------------- .../commons/pipe/event/EnrichedEvent.java | 2 + .../commons/pipe/event/PipeSnapshotEvent.java | 5 ++ .../pipe/event/PipeWritePlanEvent.java | 5 ++ .../pipe/event/ProgressReportEvent.java | 5 ++ 13 files changed, 101 insertions(+), 62 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 874f499f81b63..19abf0620a4c1 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 @@ -1000,6 +1000,13 @@ public void testRealtimeLooseRange() throws Exception { senderEnv, Arrays.asList( "insert into root.db.d1 (time, at1, at2)" + " values (1000, 1, 2), (3000, 3, 4)", + "flush"))) { + return; + } + + if (!TestUtils.tryExecuteNonQueriesWithRetry( + senderEnv, + Arrays.asList( "insert into root.db1.d1 (time, at1, at2)" + " values (1000, 1, 2), (3000, 3, 4)", "flush"))) { return; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/UserDefinedEnrichedEvent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/UserDefinedEnrichedEvent.java index 4803158db2e06..82829a4542bec 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/UserDefinedEnrichedEvent.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/UserDefinedEnrichedEvent.java @@ -92,4 +92,9 @@ public boolean isGeneratedByPipe() { public boolean mayEventTimeOverlappedWithTimeRange() { return enrichedEvent.mayEventTimeOverlappedWithTimeRange(); } + + @Override + public boolean mayEventPathsOverlappedWithPattern() { + return enrichedEvent.mayEventPathsOverlappedWithPattern(); + } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/heartbeat/PipeHeartbeatEvent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/heartbeat/PipeHeartbeatEvent.java index f82d799b7d3d9..d3830b6dcf020 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/heartbeat/PipeHeartbeatEvent.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/heartbeat/PipeHeartbeatEvent.java @@ -121,6 +121,11 @@ public boolean mayEventTimeOverlappedWithTimeRange() { return true; } + @Override + public boolean mayEventPathsOverlappedWithPattern() { + return true; + } + /////////////////////////////// Whether to print /////////////////////////////// public boolean isShouldPrintMessage() { 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 b9ddfd0d55d90..cdb16f374ce8b 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 @@ -239,6 +239,12 @@ public boolean mayEventTimeOverlappedWithTimeRange() { } } + @Override + public boolean mayEventPathsOverlappedWithPattern() { + final String deviceId = getDeviceId(); + return Objects.isNull(deviceId) || pipePattern.mayOverlapWithDevice(deviceId); + } + /////////////////////////// TabletInsertionEvent /////////////////////////// @Override 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 239f73b3c328d..9769299ae68cd 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 @@ -191,6 +191,12 @@ public boolean mayEventTimeOverlappedWithTimeRange() { return startTime <= timestamps[timestamps.length - 1] && timestamps[0] <= endTime; } + @Override + public boolean mayEventPathsOverlappedWithPattern() { + final String deviceId = getDeviceId(); + return Objects.isNull(deviceId) || pipePattern.mayOverlapWithDevice(deviceId); + } + public void markAsNeedToReport() { this.needToReport = true; } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/terminate/PipeTerminateEvent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/terminate/PipeTerminateEvent.java index cc423f1bf3d65..146873ed3d20b 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/terminate/PipeTerminateEvent.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/terminate/PipeTerminateEvent.java @@ -84,6 +84,11 @@ public boolean mayEventTimeOverlappedWithTimeRange() { return true; } + @Override + public boolean mayEventPathsOverlappedWithPattern() { + return true; + } + @Override public void reportProgress() { PipeAgent.task().markCompleted(pipeName, dataRegionId); 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 8744eb6cb6999..84abacc00945a 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 @@ -26,6 +26,7 @@ import org.apache.iotdb.commons.pipe.task.meta.PipeTaskMeta; import org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent; import org.apache.iotdb.db.pipe.resource.PipeResourceManager; +import org.apache.iotdb.db.pipe.resource.tsfile.PipeTsFileResourceManager; import org.apache.iotdb.db.storageengine.dataregion.memtable.TsFileProcessor; import org.apache.iotdb.db.storageengine.dataregion.modification.ModificationFile; import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource; @@ -33,12 +34,17 @@ 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; import java.io.File; import java.io.IOException; import java.util.Collections; +import java.util.Map; +import java.util.Objects; +import java.util.Set; import java.util.concurrent.atomic.AtomicBoolean; public class PipeTsFileInsertionEvent extends EnrichedEvent implements TsFileInsertionEvent { @@ -159,10 +165,6 @@ public boolean waitForTsFileClose() throws InterruptedException { return !resource.isEmpty(); } - public TsFileResource getResource() { - return resource; - } - public File getTsFile() { return tsFile; } @@ -281,6 +283,34 @@ public boolean mayEventTimeOverlappedWithTimeRange() { : resource.getFileStartTime() <= endTime; } + @Override + public boolean mayEventPathsOverlappedWithPattern() { + if (!resource.isClosed()) { + return true; + } + + try { + final Map deviceIsAlignedMap = + PipeResourceManager.tsfile() + .getDeviceIsAlignedMapFromCache( + PipeTsFileResourceManager.getHardlinkOrCopiedFileInPipeDir(resource.getTsFile())); + final Set deviceSet = + Objects.nonNull(deviceIsAlignedMap) ? deviceIsAlignedMap.keySet() : resource.getDevices(); + return deviceSet.stream() + .anyMatch( + // TODO: use IDeviceID + deviceID -> + pipePattern.mayOverlapWithDevice(((PlainDeviceID) deviceID).toStringID())); + } catch (final IOException e) { + LOGGER.warn( + "Pipe {}: failed to get devices from TsFile {}, extract it anyway", + pipeName, + resource.getTsFilePath(), + e); + return true; + } + } + /////////////////////////// TsFileInsertionEvent /////////////////////////// @Override diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/realtime/PipeRealtimeEvent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/realtime/PipeRealtimeEvent.java index ce8e7d9f0fa94..22f816becabe4 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/realtime/PipeRealtimeEvent.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/realtime/PipeRealtimeEvent.java @@ -172,6 +172,11 @@ public boolean mayEventTimeOverlappedWithTimeRange() { return event.mayEventTimeOverlappedWithTimeRange(); } + @Override + public boolean mayEventPathsOverlappedWithPattern() { + return event.mayEventPathsOverlappedWithPattern(); + } + @Override public String toString() { return "PipeRealtimeEvent{" diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/PipeRealtimeDataRegionExtractor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/PipeRealtimeDataRegionExtractor.java index 62b665441d8d7..4f1d8ae5a148d 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/PipeRealtimeDataRegionExtractor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/PipeRealtimeDataRegionExtractor.java @@ -30,18 +30,13 @@ import org.apache.iotdb.commons.utils.TimePartitionUtils; import org.apache.iotdb.db.pipe.agent.PipeAgent; import org.apache.iotdb.db.pipe.event.common.heartbeat.PipeHeartbeatEvent; -import org.apache.iotdb.db.pipe.event.common.tablet.PipeInsertNodeTabletInsertionEvent; -import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent; import org.apache.iotdb.db.pipe.event.realtime.PipeRealtimeEvent; import org.apache.iotdb.db.pipe.extractor.dataregion.DataRegionListeningFilter; import org.apache.iotdb.db.pipe.extractor.dataregion.realtime.listener.PipeInsertionDataNodeListener; import org.apache.iotdb.db.pipe.extractor.dataregion.realtime.listener.PipeTimePartitionListener; import org.apache.iotdb.db.pipe.metric.PipeDataRegionEventCounter; -import org.apache.iotdb.db.pipe.resource.PipeResourceManager; -import org.apache.iotdb.db.pipe.resource.tsfile.PipeTsFileResourceManager; import org.apache.iotdb.db.storageengine.StorageEngine; import org.apache.iotdb.db.storageengine.dataregion.DataRegion; -import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource; import org.apache.iotdb.db.utils.DateTimeUtils; import org.apache.iotdb.pipe.api.PipeExtractor; import org.apache.iotdb.pipe.api.customizer.configuration.PipeExtractorRuntimeConfiguration; @@ -50,18 +45,14 @@ import org.apache.iotdb.pipe.api.event.Event; import org.apache.iotdb.pipe.api.exception.PipeParameterNotValidException; -import org.apache.tsfile.file.metadata.IDeviceID; -import org.apache.tsfile.file.metadata.PlainDeviceID; import org.apache.tsfile.utils.Pair; import org.checkerframework.checker.nullness.qual.NonNull; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.io.IOException; import java.util.ArrayList; import java.util.Arrays; import java.util.List; -import java.util.Map; import java.util.Objects; import java.util.Set; import java.util.concurrent.atomic.AtomicBoolean; @@ -307,16 +298,21 @@ public final void extract(final PipeRealtimeEvent event) { // 1. Check if time parsing is necessary. If not, it means that the timestamps of the data // contained in this event are definitely within the time range [start time, end time]. - // Otherwise, - // 2. Check if the timestamps of the data contained in this event intersect with the time range. - // If there is no intersection, it indicates that this data will be filtered out by the - // extractor, and the extract process is skipped. - if (!event.shouldParseTime() || event.getEvent().mayEventTimeOverlappedWithTimeRange()) { + // 2. Check if pattern parsing is necessary. If not, it means that the paths of the data + // contained in this event are definitely covered by the pattern. + // 3. Check if the event's data timestamps may intersect with the time range. If not, it means + // that the data timestamps of this event are definitely not within the time range. + // 4. Check if the event's data paths may intersect with the pattern. If not, it means that the + // data of this event is definitely not overlapped with the pattern. + if (!event.shouldParseTime() + || !event.shouldParsePattern() + || event.getEvent().mayEventTimeOverlappedWithTimeRange() + || event.getEvent().mayEventPathsOverlappedWithPattern()) { if (sloppyTimeRange) { // only skip parsing time for events whose data timestamps may intersect with the time range event.skipParsingTime(); } - if (sloppyPattern && mayEventPathsOverlappedWithPattern(event)) { + if (sloppyPattern) { // only skip parsing pattern for events whose data paths may intersect with the pattern event.skipParsingPattern(); } @@ -333,49 +329,6 @@ public final void extract(final PipeRealtimeEvent event) { } } - private boolean mayEventPathsOverlappedWithPattern(final PipeRealtimeEvent realtimeEvent) { - final EnrichedEvent event = realtimeEvent.getEvent(); - - if (event instanceof PipeInsertNodeTabletInsertionEvent) { - final String deviceId = ((PipeInsertNodeTabletInsertionEvent) event).getDeviceId(); - return Objects.isNull(deviceId) || pipePattern.mayOverlapWithDevice(deviceId); - } - - if (event instanceof PipeTsFileInsertionEvent) { - final TsFileResource resource = ((PipeTsFileInsertionEvent) event).getResource(); - if (!resource.isClosed()) { - return true; - } - - try { - final Map deviceIsAlignedMap = - PipeResourceManager.tsfile() - .getDeviceIsAlignedMapFromCache( - PipeTsFileResourceManager.getHardlinkOrCopiedFileInPipeDir( - resource.getTsFile())); - final Set deviceSet = - Objects.nonNull(deviceIsAlignedMap) - ? deviceIsAlignedMap.keySet() - : resource.getDevices(); - return deviceSet.stream() - .anyMatch( - // TODO: use IDeviceID - deviceID -> - pipePattern.mayOverlapWithDevice(((PlainDeviceID) deviceID).toStringID())); - } catch (final IOException e) { - LOGGER.warn( - "Pipe {}@{}: failed to get devices from TsFile {}, extract it anyway", - pipeName, - dataRegionId, - resource.getTsFilePath(), - e); - return true; - } - } - - return false; - } - protected abstract void doExtract(final PipeRealtimeEvent event); protected void extractHeartbeat(final PipeRealtimeEvent event) { diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/event/EnrichedEvent.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/event/EnrichedEvent.java index 8de596a5447e1..6d68b6072941c 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/event/EnrichedEvent.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/event/EnrichedEvent.java @@ -322,6 +322,8 @@ public boolean needToCommit() { public abstract boolean mayEventTimeOverlappedWithTimeRange(); + public abstract boolean mayEventPathsOverlappedWithPattern(); + public void setCommitterKeyAndCommitId(final String committerKey, final long commitId) { this.committerKey = committerKey; this.commitId = commitId; diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/event/PipeSnapshotEvent.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/event/PipeSnapshotEvent.java index f130a83c0ee16..1c6dd689e648c 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/event/PipeSnapshotEvent.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/event/PipeSnapshotEvent.java @@ -63,6 +63,11 @@ public boolean mayEventTimeOverlappedWithTimeRange() { return true; } + @Override + public boolean mayEventPathsOverlappedWithPattern() { + return true; + } + /////////////////////////////// Type parsing /////////////////////////////// public String toSealTypeString() { diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/event/PipeWritePlanEvent.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/event/PipeWritePlanEvent.java index c553e6ce99625..4a7a25dbd23ac 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/event/PipeWritePlanEvent.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/event/PipeWritePlanEvent.java @@ -71,6 +71,11 @@ public boolean mayEventTimeOverlappedWithTimeRange() { return true; } + @Override + public boolean mayEventPathsOverlappedWithPattern() { + return true; + } + /////////////////////////// Object /////////////////////////// @Override diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/event/ProgressReportEvent.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/event/ProgressReportEvent.java index 4f00c89c5e61d..700f8f1638748 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/event/ProgressReportEvent.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/event/ProgressReportEvent.java @@ -83,4 +83,9 @@ public boolean isGeneratedByPipe() { public boolean mayEventTimeOverlappedWithTimeRange() { return true; } + + @Override + public boolean mayEventPathsOverlappedWithPattern() { + return true; + } } From 69cda4a618f8d9c77dd8e01162ef219c668b8a98 Mon Sep 17 00:00:00 2001 From: Steve Yurong Su Date: Mon, 17 Jun 2024 19:47:32 +0800 Subject: [PATCH 06/19] Update PipeRealtimeDataRegionExtractor.java --- .../realtime/PipeRealtimeDataRegionExtractor.java | 12 +++++------- 1 file changed, 5 insertions(+), 7 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/PipeRealtimeDataRegionExtractor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/PipeRealtimeDataRegionExtractor.java index 4f1d8ae5a148d..a215236ccbdcb 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/PipeRealtimeDataRegionExtractor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/PipeRealtimeDataRegionExtractor.java @@ -298,16 +298,14 @@ public final void extract(final PipeRealtimeEvent event) { // 1. Check if time parsing is necessary. If not, it means that the timestamps of the data // contained in this event are definitely within the time range [start time, end time]. - // 2. Check if pattern parsing is necessary. If not, it means that the paths of the data - // contained in this event are definitely covered by the pattern. - // 3. Check if the event's data timestamps may intersect with the time range. If not, it means + // 2. Check if the event's data timestamps may intersect with the time range. If not, it means // that the data timestamps of this event are definitely not within the time range. + // 3. Check if pattern parsing is necessary. If not, it means that the paths of the data + // contained in this event are definitely covered by the pattern. // 4. Check if the event's data paths may intersect with the pattern. If not, it means that the // data of this event is definitely not overlapped with the pattern. - if (!event.shouldParseTime() - || !event.shouldParsePattern() - || event.getEvent().mayEventTimeOverlappedWithTimeRange() - || event.getEvent().mayEventPathsOverlappedWithPattern()) { + if ((!event.shouldParseTime() || event.getEvent().mayEventTimeOverlappedWithTimeRange()) + && (!event.shouldParsePattern() || event.getEvent().mayEventPathsOverlappedWithPattern())) { if (sloppyTimeRange) { // only skip parsing time for events whose data timestamps may intersect with the time range event.skipParsingTime(); From ca587c638aab9d753cf98fa11c7d5cfa47304948 Mon Sep 17 00:00:00 2001 From: Steve Yurong Su Date: Mon, 17 Jun 2024 19:53:08 +0800 Subject: [PATCH 07/19] Update PipeRealtimeDataRegionExtractor.java --- .../dataregion/realtime/PipeRealtimeDataRegionExtractor.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/PipeRealtimeDataRegionExtractor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/PipeRealtimeDataRegionExtractor.java index a215236ccbdcb..3c2d9de916564 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/PipeRealtimeDataRegionExtractor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/PipeRealtimeDataRegionExtractor.java @@ -167,8 +167,8 @@ public void validate(final PipeParameterValidator validator) throws Exception { .collect(Collectors.toSet()); // Avoid empty string sloppyOptionSet.remove(""); - sloppyTimeRange = sloppyOptionSet.contains(EXTRACTOR_REALTIME_LOOSE_RANGE_TIME_VALUE); - sloppyPattern = sloppyOptionSet.contains(EXTRACTOR_REALTIME_LOOSE_RANGE_PATH_VALUE); + sloppyTimeRange = sloppyOptionSet.remove(EXTRACTOR_REALTIME_LOOSE_RANGE_TIME_VALUE); + sloppyPattern = sloppyOptionSet.remove(EXTRACTOR_REALTIME_LOOSE_RANGE_PATH_VALUE); if (!sloppyOptionSet.isEmpty()) { throw new PipeParameterNotValidException( String.format( From 3902b8a60eddd17da7a40d8c3d0c23ac4af96b67 Mon Sep 17 00:00:00 2001 From: Steve Yurong Su Date: Mon, 17 Jun 2024 19:54:13 +0800 Subject: [PATCH 08/19] Update PipeHistoricalDataRegionTsFileExtractor.java --- .../historical/PipeHistoricalDataRegionTsFileExtractor.java | 2 ++ 1 file changed, 2 insertions(+) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/historical/PipeHistoricalDataRegionTsFileExtractor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/historical/PipeHistoricalDataRegionTsFileExtractor.java index a0dadcfc9fa6c..8927a4e48f0a8 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/historical/PipeHistoricalDataRegionTsFileExtractor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/historical/PipeHistoricalDataRegionTsFileExtractor.java @@ -183,6 +183,8 @@ public void validate(final PipeParameterValidator validator) { EXTRACTOR_END_TIME_KEY, historicalDataExtractionEndTime)); } + } catch (final PipeParameterNotValidException e) { + throw e; } catch (final Exception e) { // compatible with the current validation framework throw new PipeParameterNotValidException(e.getMessage()); From 5f79e07b1ea0c2b5f8c713d229a36b5c2fbcc403 Mon Sep 17 00:00:00 2001 From: Steve Yurong Su Date: Mon, 17 Jun 2024 20:25:03 +0800 Subject: [PATCH 09/19] refactor --- .../it/autocreate/IoTDBPipeExtractorIT.java | 4 +- .../PipeInsertNodeTabletInsertionEvent.java | 53 ++++++++++++++++--- ...peHistoricalDataRegionTsFileExtractor.java | 3 +- .../PipeRealtimeDataRegionExtractor.java | 3 +- 4 files changed, 50 insertions(+), 13 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 19abf0620a4c1..9dd2b31e28192 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 @@ -979,6 +979,7 @@ public void testRealtimeLooseRange() throws Exception { extractorAttributes.put("source.realtime.loose-range", "time, path"); extractorAttributes.put("source.start-time", "2000"); extractorAttributes.put("source.end-time", "10000"); + extractorAttributes.put("source.realtime.mode", "batch"); connectorAttributes.put("connector", "iotdb-thrift-connector"); connectorAttributes.put("connector.batch.enable", "false"); @@ -1016,7 +1017,8 @@ public void testRealtimeLooseRange() throws Exception { senderEnv, Arrays.asList( "insert into root.db.d1 (time, at1)" + " values (5000, 1), (16000, 3)", - "insert into root.db.d1 (time, at1, at2)" + " values (5001, 1, 2), (6001, 3, 4)"))) { + "insert into root.db.d1 (time, at1, at2)" + " values (5001, 1, 2), (6001, 3, 4)", + "flush"))) { return; } 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 cdb16f374ce8b..68d0c84eaff7b 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 @@ -205,18 +205,26 @@ public boolean isGeneratedByPipe() { @Override public boolean mayEventTimeOverlappedWithTimeRange() { try { - final InsertNode insertNode = getInsertNode(); + final InsertNode insertNode = getInsertNodeViaCacheIfPossible(); + if (Objects.isNull(insertNode)) { + return true; + } + if (insertNode instanceof InsertRowNode) { final long timestamp = ((InsertRowNode) insertNode).getTime(); return startTime <= timestamp && timestamp <= endTime; - } else if (insertNode instanceof InsertTabletNode) { + } + + if (insertNode instanceof InsertTabletNode) { final long[] timestamps = ((InsertTabletNode) insertNode).getTimes(); if (Objects.isNull(timestamps) || timestamps.length == 0) { return false; } // We assume that `timestamps` is ordered. return startTime <= timestamps[timestamps.length - 1] && timestamps[0] <= endTime; - } else if (insertNode instanceof InsertRowsNode) { + } + + if (insertNode instanceof InsertRowsNode) { for (final InsertRowNode node : ((InsertRowsNode) insertNode).getInsertRowNodeList()) { final long timestamp = node.getTime(); if (startTime <= timestamp && timestamp <= endTime) { @@ -224,10 +232,9 @@ public boolean mayEventTimeOverlappedWithTimeRange() { } } return false; - } else { - throw new UnSupportedDataTypeException( - String.format("InsertNode type %s is not supported.", insertNode.getClass().getName())); } + + return true; } catch (final Exception e) { LOGGER.warn( "Exception occurred when determining the event time of PipeInsertNodeTabletInsertionEvent({}) overlaps with the time range: [{}, {}]. Returning true to ensure data integrity.", @@ -241,8 +248,38 @@ public boolean mayEventTimeOverlappedWithTimeRange() { @Override public boolean mayEventPathsOverlappedWithPattern() { - final String deviceId = getDeviceId(); - return Objects.isNull(deviceId) || pipePattern.mayOverlapWithDevice(deviceId); + try { + final InsertNode insertNode = getInsertNodeViaCacheIfPossible(); + if (Objects.isNull(insertNode)) { + return true; + } + + if (insertNode instanceof InsertRowNode || insertNode instanceof InsertTabletNode) { + final PartialPath devicePartialPath = insertNode.getDevicePath(); + return Objects.isNull(devicePartialPath) + || pipePattern.mayOverlapWithDevice(devicePartialPath.getFullPath()); + } + + if (insertNode instanceof InsertRowsNode) { + return ((InsertRowsNode) insertNode) + .getInsertRowNodeList().stream() + .anyMatch( + insertRowNode -> + Objects.isNull(insertRowNode.getDevicePath()) + || pipePattern.mayOverlapWithDevice( + insertRowNode.getDevicePath().getFullPath())); + } + + return true; + } catch (final Exception e) { + LOGGER.warn( + "Exception occurred when determining the event time of PipeInsertNodeTabletInsertionEvent({}) overlaps with the time range: [{}, {}]. Returning true to ensure data integrity.", + this, + startTime, + endTime, + e); + return true; + } } /////////////////////////// TabletInsertionEvent /////////////////////////// diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/historical/PipeHistoricalDataRegionTsFileExtractor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/historical/PipeHistoricalDataRegionTsFileExtractor.java index 8927a4e48f0a8..1f36e98bec09f 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/historical/PipeHistoricalDataRegionTsFileExtractor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/historical/PipeHistoricalDataRegionTsFileExtractor.java @@ -142,10 +142,9 @@ public void validate(final PipeParameterValidator validator) { EXTRACTOR_HISTORY_LOOSE_RANGE_DEFAULT_VALUE) .split(",")) .map(String::trim) + .filter(s -> !s.isEmpty()) .map(String::toLowerCase) .collect(Collectors.toSet()); - // Avoid empty string - sloppyOptionSet.remove(""); sloppyTimeRange = sloppyOptionSet.remove(EXTRACTOR_HISTORY_LOOSE_RANGE_TIME_VALUE); sloppyPattern = sloppyOptionSet.remove(EXTRACTOR_HISTORY_LOOSE_RANGE_PATH_VALUE); if (!sloppyOptionSet.isEmpty()) { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/PipeRealtimeDataRegionExtractor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/PipeRealtimeDataRegionExtractor.java index 3c2d9de916564..3142df68cb4c5 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/PipeRealtimeDataRegionExtractor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/PipeRealtimeDataRegionExtractor.java @@ -163,10 +163,9 @@ public void validate(final PipeParameterValidator validator) throws Exception { EXTRACTOR_REALTIME_LOOSE_RANGE_DEFAULT_VALUE) .split(",")) .map(String::trim) + .filter(s -> !s.isEmpty()) .map(String::toLowerCase) .collect(Collectors.toSet()); - // Avoid empty string - sloppyOptionSet.remove(""); sloppyTimeRange = sloppyOptionSet.remove(EXTRACTOR_REALTIME_LOOSE_RANGE_TIME_VALUE); sloppyPattern = sloppyOptionSet.remove(EXTRACTOR_REALTIME_LOOSE_RANGE_PATH_VALUE); if (!sloppyOptionSet.isEmpty()) { From 0b6a4faec5d992fab4bc8363f045c5f92b74dcf2 Mon Sep 17 00:00:00 2001 From: Steve Yurong Su Date: Mon, 17 Jun 2024 21:36:39 +0800 Subject: [PATCH 10/19] refactor --- .../PipeInsertNodeTabletInsertionEvent.java | 14 +-- .../event/realtime/PipeRealtimeEvent.java | 85 +++++++++++++++++++ .../PipeRealtimeDataRegionExtractor.java | 7 +- 3 files changed, 96 insertions(+), 10 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 68d0c84eaff7b..32e176da3a2bc 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 @@ -225,13 +225,13 @@ public boolean mayEventTimeOverlappedWithTimeRange() { } if (insertNode instanceof InsertRowsNode) { - for (final InsertRowNode node : ((InsertRowsNode) insertNode).getInsertRowNodeList()) { - final long timestamp = node.getTime(); - if (startTime <= timestamp && timestamp <= endTime) { - return true; - } - } - return false; + return ((InsertRowsNode) insertNode) + .getInsertRowNodeList().stream() + .anyMatch( + insertRowNode -> { + final long timestamp = insertRowNode.getTime(); + return startTime <= timestamp && timestamp <= endTime; + }); } return true; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/realtime/PipeRealtimeEvent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/realtime/PipeRealtimeEvent.java index 22f816becabe4..c8a44f24ce8db 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/realtime/PipeRealtimeEvent.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/realtime/PipeRealtimeEvent.java @@ -177,6 +177,91 @@ public boolean mayEventPathsOverlappedWithPattern() { return event.mayEventPathsOverlappedWithPattern(); } + @Override + public void skipReportOnCommit() { + event.skipReportOnCommit(); + } + + @Override + public void bindProgressIndex(ProgressIndex progressIndex) { + event.bindProgressIndex(progressIndex); + } + + @Override + public int getReferenceCount() { + return event.getReferenceCount(); + } + + @Override + public boolean shouldParseTimeOrPattern() { + return event.shouldParseTimeOrPattern(); + } + + @Override + public boolean shouldParsePattern() { + return event.shouldParsePattern(); + } + + @Override + public boolean shouldParseTime() { + return event.shouldParseTime(); + } + + @Override + public PipeTaskMeta getPipeTaskMeta() { + return event.getPipeTaskMeta(); + } + + @Override + public boolean needToCommit() { + return event.needToCommit(); + } + + @Override + public void setCommitterKeyAndCommitId(String committerKey, long commitId) { + event.setCommitterKeyAndCommitId(committerKey, commitId); + } + + @Override + public void setRebootTimes(int rebootTimes) { + event.setRebootTimes(rebootTimes); + } + + @Override + public int getRebootTimes() { + return event.getRebootTimes(); + } + + @Override + public String getCommitterKey() { + return event.getCommitterKey(); + } + + @Override + public long getCommitId() { + return event.getCommitId(); + } + + @Override + public void onCommitted() { + event.onCommitted(); + } + + @Override + public boolean isReleased() { + return event.isReleased(); + } + + @Override + public boolean equalsInPipeConsensus(Object o) { + return event.equalsInPipeConsensus(o); + } + + @Override + public String coreReportMessage() { + return event.coreReportMessage(); + } + @Override public String toString() { return "PipeRealtimeEvent{" diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/PipeRealtimeDataRegionExtractor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/PipeRealtimeDataRegionExtractor.java index 3142df68cb4c5..89e42ea0974c6 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/PipeRealtimeDataRegionExtractor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/PipeRealtimeDataRegionExtractor.java @@ -91,7 +91,7 @@ public abstract class PipeRealtimeDataRegionExtractor implements PipeExtractor { protected long realtimeDataExtractionStartTime = Long.MIN_VALUE; // Event time protected long realtimeDataExtractionEndTime = Long.MAX_VALUE; // Event time - private boolean disableSkippingTimeParse = false; + private boolean disableCheckingDataRegionTimePartitionCovering = false; private long startTimePartitionIdLowerBound; // calculated by realtimeDataExtractionStartTime private long endTimePartitionIdUpperBound; // calculated by realtimeDataExtractionEndTime @@ -284,14 +284,15 @@ public final void extract(final PipeRealtimeEvent event) { event.skipParsingPattern(); } - if (!disableSkippingTimeParse && Objects.nonNull(dataRegionTimePartitionIdBound.get())) { + if (!disableCheckingDataRegionTimePartitionCovering + && Objects.nonNull(dataRegionTimePartitionIdBound.get())) { if (isDataRegionTimePartitionCoveredByTimeRange()) { event.skipParsingTime(); } else { // Since we only record the upper and lower bounds that time partition has ever reached, if // the time partition cannot be covered by the time range during query, it will not be // possible later. - disableSkippingTimeParse = true; + disableCheckingDataRegionTimePartitionCovering = true; } } From 0434368bd8b54735015cb5ff7764a850689950c7 Mon Sep 17 00:00:00 2001 From: Steve Yurong Su Date: Tue, 18 Jun 2024 10:13:21 +0800 Subject: [PATCH 11/19] Update IoTDBPipeExtractorIT.java --- .../pipe/it/autocreate/IoTDBPipeExtractorIT.java | 12 ++---------- 1 file changed, 2 insertions(+), 10 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 9dd2b31e28192..8ce885935b8fd 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 @@ -1013,20 +1013,12 @@ public void testRealtimeLooseRange() throws Exception { return; } - if (!TestUtils.tryExecuteNonQueriesWithRetry( + TestUtils.tryExecuteNonQueriesWithRetry( senderEnv, Arrays.asList( "insert into root.db.d1 (time, at1)" + " values (5000, 1), (16000, 3)", "insert into root.db.d1 (time, at1, at2)" + " values (5001, 1, 2), (6001, 3, 4)", - "flush"))) { - return; - } - - TestUtils.assertDataEventuallyOnEnv( - receiverEnv, - "select count(*) from root.** group by level=0", - "count(root.*.*.*),", - Collections.singleton("10,")); + "flush")); } } From 0690fa47787a81ed1da51a2598d7604971a0bb0e Mon Sep 17 00:00:00 2001 From: Steve Yurong Su Date: Tue, 18 Jun 2024 10:18:20 +0800 Subject: [PATCH 12/19] Update IoTDBSubscriptionTopicIT.java --- .../subscription/it/dual/IoTDBSubscriptionTopicIT.java | 9 --------- 1 file changed, 9 deletions(-) diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/dual/IoTDBSubscriptionTopicIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/dual/IoTDBSubscriptionTopicIT.java index 19c033af151ba..cca6dbd3bf1ee 100644 --- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/dual/IoTDBSubscriptionTopicIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/dual/IoTDBSubscriptionTopicIT.java @@ -530,15 +530,6 @@ public void testTopicInvalidTimeRangeConfig() throws Exception { assertTopicCount(0); } - @Test - public void testTopicInvalidPathConfig() throws Exception { - // Test invalid path when using tsfile format - final Properties config = new Properties(); - config.put(TopicConstant.FORMAT_KEY, TopicConstant.FORMAT_TS_FILE_HANDLER_VALUE); - config.put(TopicConstant.PATH_KEY, "root.db.*.s"); - testTopicInvalidRuntimeConfigTemplate("topic9", config); - } - @Test public void testTopicInvalidProcessorConfig() throws Exception { // Test invalid processor when using tsfile format From bcd5b885ce192a8186064b630147a0ca0008a73a Mon Sep 17 00:00:00 2001 From: Steve Yurong Su Date: Tue, 18 Jun 2024 10:20:01 +0800 Subject: [PATCH 13/19] Update IoTDBSubscriptionTopicIT.java --- .../iotdb/subscription/it/dual/IoTDBSubscriptionTopicIT.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/dual/IoTDBSubscriptionTopicIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/dual/IoTDBSubscriptionTopicIT.java index cca6dbd3bf1ee..06fde3b12d270 100644 --- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/dual/IoTDBSubscriptionTopicIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/dual/IoTDBSubscriptionTopicIT.java @@ -538,7 +538,7 @@ public void testTopicInvalidProcessorConfig() throws Exception { config.put("processor", "tumbling-time-sampling-processor"); config.put("processor.tumbling-time.interval-seconds", "1"); config.put("processor.down-sampling.split-file", "true"); - testTopicInvalidRuntimeConfigTemplate("topic10", config); + testTopicInvalidRuntimeConfigTemplate("topic9", config); } @Test @@ -556,7 +556,7 @@ public void testTopicWithQueryMode() throws Exception { } // Create topic - final String topicName = "topic11"; + final String topicName = "topic10"; final String host = senderEnv.getIP(); final int port = Integer.parseInt(senderEnv.getPort()); try (final SubscriptionSession session = new SubscriptionSession(host, port)) { From fb3119cd8cf527aa7a7f7ebd2669add82efa37c2 Mon Sep 17 00:00:00 2001 From: Steve Yurong Su Date: Tue, 18 Jun 2024 12:09:14 +0800 Subject: [PATCH 14/19] fix --- .../it/autocreate/IoTDBPipeExtractorIT.java | 12 ++- .../it/dual/IoTDBSubscriptionTopicIT.java | 13 +++- .../PipeInsertNodeTabletInsertionEvent.java | 17 ----- .../tsfile/PipeTsFileInsertionEvent.java | 5 -- .../event/realtime/PipeRealtimeEvent.java | 75 ------------------- .../CachedSchemaPatternMatcherTest.java | 30 +++++++- 6 files changed, 48 insertions(+), 104 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 8ce885935b8fd..9dd2b31e28192 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 @@ -1013,12 +1013,20 @@ public void testRealtimeLooseRange() throws Exception { return; } - TestUtils.tryExecuteNonQueriesWithRetry( + if (!TestUtils.tryExecuteNonQueriesWithRetry( senderEnv, Arrays.asList( "insert into root.db.d1 (time, at1)" + " values (5000, 1), (16000, 3)", "insert into root.db.d1 (time, at1, at2)" + " values (5001, 1, 2), (6001, 3, 4)", - "flush")); + "flush"))) { + return; + } + + TestUtils.assertDataEventuallyOnEnv( + receiverEnv, + "select count(*) from root.** group by level=0", + "count(root.*.*.*),", + Collections.singleton("10,")); } } diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/dual/IoTDBSubscriptionTopicIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/dual/IoTDBSubscriptionTopicIT.java index 06fde3b12d270..19c033af151ba 100644 --- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/dual/IoTDBSubscriptionTopicIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/dual/IoTDBSubscriptionTopicIT.java @@ -530,6 +530,15 @@ public void testTopicInvalidTimeRangeConfig() throws Exception { assertTopicCount(0); } + @Test + public void testTopicInvalidPathConfig() throws Exception { + // Test invalid path when using tsfile format + final Properties config = new Properties(); + config.put(TopicConstant.FORMAT_KEY, TopicConstant.FORMAT_TS_FILE_HANDLER_VALUE); + config.put(TopicConstant.PATH_KEY, "root.db.*.s"); + testTopicInvalidRuntimeConfigTemplate("topic9", config); + } + @Test public void testTopicInvalidProcessorConfig() throws Exception { // Test invalid processor when using tsfile format @@ -538,7 +547,7 @@ public void testTopicInvalidProcessorConfig() throws Exception { config.put("processor", "tumbling-time-sampling-processor"); config.put("processor.tumbling-time.interval-seconds", "1"); config.put("processor.down-sampling.split-file", "true"); - testTopicInvalidRuntimeConfigTemplate("topic9", config); + testTopicInvalidRuntimeConfigTemplate("topic10", config); } @Test @@ -556,7 +565,7 @@ public void testTopicWithQueryMode() throws Exception { } // Create topic - final String topicName = "topic10"; + final String topicName = "topic11"; final String host = senderEnv.getIP(); final int port = Integer.parseInt(senderEnv.getPort()); try (final SubscriptionSession session = new SubscriptionSession(host, port)) { 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 32e176da3a2bc..abb0fcb572c39 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 @@ -26,7 +26,6 @@ import org.apache.iotdb.commons.pipe.pattern.PipePattern; import org.apache.iotdb.commons.pipe.task.meta.PipeTaskMeta; import org.apache.iotdb.db.pipe.resource.PipeResourceManager; -import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeType; 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; @@ -361,22 +360,6 @@ public long count() { /////////////////////////// parsePatternOrTime /////////////////////////// - @Override - public boolean shouldParsePattern() { - final InsertNode node = getInsertNodeViaCacheIfPossible(); - return super.shouldParsePattern() - && Objects.nonNull(pipePattern) - && (Objects.isNull(node) - || (node.getType() == PlanNodeType.INSERT_ROWS - ? ((InsertRowsNode) node) - .getInsertRowNodeList().stream() - .anyMatch( - insertRowNode -> - !pipePattern.coversDevice( - insertRowNode.getDevicePath().getFullPath())) - : !pipePattern.coversDevice(node.getDevicePath().getFullPath()))); - } - public List toRawTabletInsertionEvents() { final List events = convertToTablets().stream() 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 84abacc00945a..9a1897864f0cd 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 @@ -329,11 +329,6 @@ public boolean shouldParseTimeOrPattern() { } } - @Override - public boolean shouldParsePattern() { - return super.shouldParsePattern() && initDataContainer().shouldParsePattern(); - } - @Override public Iterable toTabletInsertionEvents() { try { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/realtime/PipeRealtimeEvent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/realtime/PipeRealtimeEvent.java index c8a44f24ce8db..150050fab427a 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/realtime/PipeRealtimeEvent.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/realtime/PipeRealtimeEvent.java @@ -177,26 +177,6 @@ public boolean mayEventPathsOverlappedWithPattern() { return event.mayEventPathsOverlappedWithPattern(); } - @Override - public void skipReportOnCommit() { - event.skipReportOnCommit(); - } - - @Override - public void bindProgressIndex(ProgressIndex progressIndex) { - event.bindProgressIndex(progressIndex); - } - - @Override - public int getReferenceCount() { - return event.getReferenceCount(); - } - - @Override - public boolean shouldParseTimeOrPattern() { - return event.shouldParseTimeOrPattern(); - } - @Override public boolean shouldParsePattern() { return event.shouldParsePattern(); @@ -207,61 +187,6 @@ public boolean shouldParseTime() { return event.shouldParseTime(); } - @Override - public PipeTaskMeta getPipeTaskMeta() { - return event.getPipeTaskMeta(); - } - - @Override - public boolean needToCommit() { - return event.needToCommit(); - } - - @Override - public void setCommitterKeyAndCommitId(String committerKey, long commitId) { - event.setCommitterKeyAndCommitId(committerKey, commitId); - } - - @Override - public void setRebootTimes(int rebootTimes) { - event.setRebootTimes(rebootTimes); - } - - @Override - public int getRebootTimes() { - return event.getRebootTimes(); - } - - @Override - public String getCommitterKey() { - return event.getCommitterKey(); - } - - @Override - public long getCommitId() { - return event.getCommitId(); - } - - @Override - public void onCommitted() { - event.onCommitted(); - } - - @Override - public boolean isReleased() { - return event.isReleased(); - } - - @Override - public boolean equalsInPipeConsensus(Object o) { - return event.equalsInPipeConsensus(o); - } - - @Override - public String coreReportMessage() { - return event.coreReportMessage(); - } - @Override public String toString() { return "PipeRealtimeEvent{" 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 1e7587153fb88..8649183c1ec74 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/pattern/CachedSchemaPatternMatcherTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/pattern/CachedSchemaPatternMatcherTest.java @@ -22,9 +22,12 @@ import org.apache.iotdb.commons.pipe.config.constant.PipeExtractorConstant; import org.apache.iotdb.commons.pipe.config.plugin.configuraion.PipeTaskRuntimeConfiguration; import org.apache.iotdb.commons.pipe.config.plugin.env.PipeTaskExtractorRuntimeEnvironment; +import org.apache.iotdb.commons.pipe.event.EnrichedEvent; +import org.apache.iotdb.commons.pipe.pattern.PipePattern; import org.apache.iotdb.commons.pipe.pattern.PrefixPipePattern; import org.apache.iotdb.db.pipe.event.realtime.PipeRealtimeEvent; import org.apache.iotdb.db.pipe.extractor.dataregion.realtime.PipeRealtimeDataRegionExtractor; +import org.apache.iotdb.db.pipe.extractor.dataregion.realtime.epoch.TsFileEpoch; import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters; import org.apache.iotdb.pipe.api.event.Event; @@ -47,6 +50,27 @@ public class CachedSchemaPatternMatcherTest { + private static class MockedPipeRealtimeEvent extends PipeRealtimeEvent { + + public MockedPipeRealtimeEvent( + EnrichedEvent event, + TsFileEpoch tsFileEpoch, + Map device2Measurements, + PipePattern pattern) { + super(event, tsFileEpoch, device2Measurements, pattern); + } + + @Override + public boolean shouldParseTime() { + return false; + } + + @Override + public boolean shouldParsePattern() { + return false; + } + } + private CachedSchemaPatternMatcher matcher; private ExecutorService executorService; private List extractors; @@ -125,14 +149,14 @@ public void testCachedMatcher() throws Exception { long totalTime = 0; for (int i = 0; i < epochNum; i++) { for (int j = 0; j < deviceNum; j++) { - PipeRealtimeEvent event = - new PipeRealtimeEvent( + MockedPipeRealtimeEvent event = + new MockedPipeRealtimeEvent( null, null, Collections.singletonMap("root." + i, measurements), null); long startTime = System.currentTimeMillis(); matcher.match(event).forEach(extractor -> extractor.extract(event)); totalTime += (System.currentTimeMillis() - startTime); } - PipeRealtimeEvent event = new PipeRealtimeEvent(null, null, deviceMap, null); + MockedPipeRealtimeEvent event = new MockedPipeRealtimeEvent(null, null, deviceMap, null); long startTime = System.currentTimeMillis(); matcher.match(event).forEach(extractor -> extractor.extract(event)); totalTime += (System.currentTimeMillis() - startTime); From 7c3451baf1a6eddb7f5e100487083a74fda4c3b9 Mon Sep 17 00:00:00 2001 From: Steve Yurong Su Date: Tue, 18 Jun 2024 12:31:09 +0800 Subject: [PATCH 15/19] Update PipeRealtimeEvent.java --- .../event/realtime/PipeRealtimeEvent.java | 44 +++++++++---------- 1 file changed, 20 insertions(+), 24 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/realtime/PipeRealtimeEvent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/realtime/PipeRealtimeEvent.java index 150050fab427a..7e12a909b1f19 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/realtime/PipeRealtimeEvent.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/realtime/PipeRealtimeEvent.java @@ -129,10 +129,6 @@ public ProgressIndex getProgressIndex() { return event.getProgressIndex(); } - /** - * If pipe's pattern is database-level, then no need to parse event by pattern cause pipes are - * data-region-level. - */ @Override public void skipParsingPattern() { event.skipParsingPattern(); @@ -143,6 +139,26 @@ public void skipParsingTime() { event.skipParsingTime(); } + @Override + public boolean shouldParseTime() { + return event.shouldParseTime(); + } + + @Override + public boolean shouldParsePattern() { + return event.shouldParsePattern(); + } + + @Override + public boolean mayEventTimeOverlappedWithTimeRange() { + return event.mayEventTimeOverlappedWithTimeRange(); + } + + @Override + public boolean mayEventPathsOverlappedWithPattern() { + return event.mayEventPathsOverlappedWithPattern(); + } + @Override public PipeRealtimeEvent shallowCopySelfAndBindPipeTaskMetaForProgressReport( final String pipeName, @@ -167,26 +183,6 @@ public boolean isGeneratedByPipe() { return event.isGeneratedByPipe(); } - @Override - public boolean mayEventTimeOverlappedWithTimeRange() { - return event.mayEventTimeOverlappedWithTimeRange(); - } - - @Override - public boolean mayEventPathsOverlappedWithPattern() { - return event.mayEventPathsOverlappedWithPattern(); - } - - @Override - public boolean shouldParsePattern() { - return event.shouldParsePattern(); - } - - @Override - public boolean shouldParseTime() { - return event.shouldParseTime(); - } - @Override public String toString() { return "PipeRealtimeEvent{" From d1918c0f699b5fbdfae137407ee0fbf5e0a4819d Mon Sep 17 00:00:00 2001 From: Steve Yurong Su Date: Tue, 18 Jun 2024 12:38:45 +0800 Subject: [PATCH 16/19] Update PipeTsFileInsertionEvent.java --- .../common/tsfile/PipeTsFileInsertionEvent.java | 16 ---------------- 1 file changed, 16 deletions(-) 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 9a1897864f0cd..b7a883712dfd5 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 @@ -313,22 +313,6 @@ public boolean mayEventPathsOverlappedWithPattern() { /////////////////////////// TsFileInsertionEvent /////////////////////////// - @Override - public boolean shouldParseTimeOrPattern() { - boolean shouldParseTimeOrPattern = false; - try { - shouldParseTimeOrPattern = super.shouldParseTimeOrPattern(); - return shouldParseTimeOrPattern; - } finally { - // Super method will call shouldParsePattern() and then init dataContainer at - // shouldParsePattern(). If shouldParsePattern() returns false, dataContainer will - // not be used, so we need to close the resource here. - if (!shouldParseTimeOrPattern) { - close(); - } - } - } - @Override public Iterable toTabletInsertionEvents() { try { From fab40aaeaab9086eac885031b22d2579df13963f Mon Sep 17 00:00:00 2001 From: Steve Yurong Su Date: Tue, 18 Jun 2024 14:40:26 +0800 Subject: [PATCH 17/19] Update IoTDBPipeExtractorIT.java --- .../iotdb/pipe/it/autocreate/IoTDBPipeExtractorIT.java | 6 ------ 1 file changed, 6 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 9dd2b31e28192..4da7c39fb22c0 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 @@ -30,7 +30,6 @@ import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper; import org.apache.iotdb.it.framework.IoTDBTestRunner; import org.apache.iotdb.itbase.category.MultiClusterIT2AutoCreateSchema; -import org.apache.iotdb.itbase.env.BaseEnv; import org.apache.iotdb.rpc.TSStatusCode; import org.junit.Assert; @@ -1030,11 +1029,6 @@ public void testRealtimeLooseRange() throws Exception { } } - private void assertTimeseriesCountOnReceiver(BaseEnv receiverEnv, int count) { - TestUtils.assertDataEventuallyOnEnv( - receiverEnv, "count timeseries", "count(timeseries),", Collections.singleton(count + ",")); - } - private void assertPipeCount(int count) throws Exception { try (final SyncConfigNodeIServiceClient client = (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) { From d3facd2e97ce6d06724c183cacad6345609b9cee Mon Sep 17 00:00:00 2001 From: Steve Yurong Su Date: Tue, 18 Jun 2024 15:29:48 +0800 Subject: [PATCH 18/19] Revert "Update IoTDBPipeExtractorIT.java" This reverts commit fab40aaeaab9086eac885031b22d2579df13963f. --- .../iotdb/pipe/it/autocreate/IoTDBPipeExtractorIT.java | 6 ++++++ 1 file changed, 6 insertions(+) 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 4da7c39fb22c0..9dd2b31e28192 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 @@ -30,6 +30,7 @@ import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper; import org.apache.iotdb.it.framework.IoTDBTestRunner; import org.apache.iotdb.itbase.category.MultiClusterIT2AutoCreateSchema; +import org.apache.iotdb.itbase.env.BaseEnv; import org.apache.iotdb.rpc.TSStatusCode; import org.junit.Assert; @@ -1029,6 +1030,11 @@ public void testRealtimeLooseRange() throws Exception { } } + private void assertTimeseriesCountOnReceiver(BaseEnv receiverEnv, int count) { + TestUtils.assertDataEventuallyOnEnv( + receiverEnv, "count timeseries", "count(timeseries),", Collections.singleton(count + ",")); + } + private void assertPipeCount(int count) throws Exception { try (final SyncConfigNodeIServiceClient client = (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) { From 6688cd5897539db0c23372bbd3f5a0f7169c075a Mon Sep 17 00:00:00 2001 From: Steve Yurong Su Date: Tue, 18 Jun 2024 15:30:42 +0800 Subject: [PATCH 19/19] Update IoTDBPipeExtractorIT.java --- .../pipe/it/autocreate/IoTDBPipeExtractorIT.java | 12 ++---------- 1 file changed, 2 insertions(+), 10 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 9dd2b31e28192..8ce885935b8fd 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 @@ -1013,20 +1013,12 @@ public void testRealtimeLooseRange() throws Exception { return; } - if (!TestUtils.tryExecuteNonQueriesWithRetry( + TestUtils.tryExecuteNonQueriesWithRetry( senderEnv, Arrays.asList( "insert into root.db.d1 (time, at1)" + " values (5000, 1), (16000, 3)", "insert into root.db.d1 (time, at1, at2)" + " values (5001, 1, 2), (6001, 3, 4)", - "flush"))) { - return; - } - - TestUtils.assertDataEventuallyOnEnv( - receiverEnv, - "select count(*) from root.** group by level=0", - "count(root.*.*.*),", - Collections.singleton("10,")); + "flush")); } }