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 2025/07/18 04:33:50 UTC
(iotdb) 02/02: fix
This is an automated email from the ASF dual-hosted git repository.
rong pushed a commit to branch pipe-report-realtime-filtered-evemts
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 687e88d3992f9070cbce7ff9ee28b2f50227e9ad
Author: Steve Yurong Su <ro...@apache.org>
AuthorDate: Fri Jul 18 12:33:33 2025 +0800
fix
---
.../realtime/assigner/PipeDataRegionAssigner.java | 43 +++++++++++++++++++++-
.../matcher/CachedSchemaPatternMatcher.java | 27 ++++++++++----
.../realtime/matcher/PipeDataRegionMatcher.java | 4 +-
3 files changed, 63 insertions(+), 11 deletions(-)
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/assigner/PipeDataRegionAssigner.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/assigner/PipeDataRegionAssigner.java
index 6b92f4bccdc..7c5cfc0e37c 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/assigner/PipeDataRegionAssigner.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/assigner/PipeDataRegionAssigner.java
@@ -45,12 +45,16 @@ import org.apache.iotdb.db.pipe.metric.source.PipeDataRegionEventCounter;
import org.apache.iotdb.db.pipe.processor.pipeconsensus.PipeConsensusProcessor;
import org.apache.iotdb.db.storageengine.StorageEngine;
import org.apache.iotdb.db.storageengine.dataregion.DataRegion;
+import org.apache.iotdb.pipe.api.event.dml.insertion.TabletInsertionEvent;
+import org.apache.iotdb.pipe.api.event.dml.insertion.TsFileInsertionEvent;
+import org.apache.tsfile.utils.Pair;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.Closeable;
import java.util.Objects;
+import java.util.Set;
import java.util.concurrent.atomic.AtomicReference;
public class PipeDataRegionAssigner implements Closeable {
@@ -139,8 +143,10 @@ public class PipeDataRegionAssigner implements Closeable {
return;
}
- matcher
- .match(event)
+ final Pair<Set<PipeRealtimeDataRegionExtractor>, Set<PipeRealtimeDataRegionExtractor>>
+ matchedAndUnmatched = matcher.match(event);
+
+ matchedAndUnmatched
.getLeft()
.forEach(
extractor -> {
@@ -226,6 +232,39 @@ public class PipeDataRegionAssigner implements Closeable {
}
extractor.extract(copiedEvent);
});
+
+ matchedAndUnmatched
+ .getRight()
+ .forEach(
+ extractor -> {
+ if (disruptor.isClosed()) {
+ return;
+ }
+
+ final EnrichedEvent innerEvent = event.getEvent();
+ if (innerEvent instanceof TabletInsertionEvent
+ || innerEvent instanceof TsFileInsertionEvent) {
+ final ProgressReportEvent reportEvent =
+ new ProgressReportEvent(
+ extractor.getPipeName(),
+ extractor.getCreationTime(),
+ extractor.getPipeTaskMeta(),
+ extractor.getTreePattern(),
+ extractor.getTablePattern(),
+ extractor.getUserName(),
+ extractor.isSkipIfNoPrivileges(),
+ extractor.getRealtimeDataExtractionStartTime(),
+ extractor.getRealtimeDataExtractionEndTime());
+ reportEvent.bindProgressIndex(event.getProgressIndex());
+ if (!reportEvent.increaseReferenceCount(PipeDataRegionAssigner.class.getName())) {
+ LOGGER.warn(
+ "The reference count of the event {} cannot be increased, skipping it.",
+ reportEvent);
+ return;
+ }
+ extractor.extract(PipeRealtimeEventFactory.createRealtimeEvent(reportEvent));
+ }
+ });
}
public void startAssignTo(final PipeRealtimeDataRegionExtractor extractor) {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/matcher/CachedSchemaPatternMatcher.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/matcher/CachedSchemaPatternMatcher.java
index 5699850c5bf..e22cf7045b7 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/matcher/CachedSchemaPatternMatcher.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/matcher/CachedSchemaPatternMatcher.java
@@ -39,13 +39,13 @@ import org.apache.tsfile.utils.Pair;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.Collections;
import java.util.HashSet;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.locks.ReentrantReadWriteLock;
-import java.util.stream.Collectors;
import static org.apache.tsfile.common.constant.TsFileConstant.PATH_ROOT;
import static org.apache.tsfile.common.constant.TsFileConstant.PATH_SEPARATOR;
@@ -126,26 +126,28 @@ public class CachedSchemaPatternMatcher implements PipeDataRegionMatcher {
}
@Override
- public Set<PipeRealtimeDataRegionExtractor> match(final PipeRealtimeEvent event) {
+ public Pair<Set<PipeRealtimeDataRegionExtractor>, Set<PipeRealtimeDataRegionExtractor>> match(
+ final PipeRealtimeEvent event) {
final Set<PipeRealtimeDataRegionExtractor> matchedExtractors = new HashSet<>();
lock.readLock().lock();
try {
if (extractors.isEmpty()) {
- return matchedExtractors;
+ return new Pair<>(matchedExtractors, extractors);
}
// HeartbeatEvent will be assigned to all extractors
if (event.getEvent() instanceof PipeHeartbeatEvent) {
- return extractors;
+ return new Pair<>(extractors, Collections.EMPTY_SET);
}
// TODO: consider table pattern?
// Deletion event will be assigned to extractors listened to it
if (event.getEvent() instanceof PipeDeleteDataNodeEvent) {
- return extractors.stream()
+ extractors.stream()
.filter(PipeRealtimeDataRegionExtractor::shouldExtractDeletion)
- .collect(Collectors.toSet());
+ .forEach(matchedExtractors::add);
+ return new Pair<>(matchedExtractors, findUnmatchedExtractors(matchedExtractors));
}
for (final Map.Entry<IDeviceID, String[]> entry : event.getSchemaInfo().entrySet()) {
@@ -171,11 +173,22 @@ public class CachedSchemaPatternMatcher implements PipeDataRegionMatcher {
break;
}
}
+
+ return new Pair<>(matchedExtractors, findUnmatchedExtractors(matchedExtractors));
} finally {
lock.readLock().unlock();
}
+ }
- return matchedExtractors;
+ private Set<PipeRealtimeDataRegionExtractor> findUnmatchedExtractors(
+ final Set<PipeRealtimeDataRegionExtractor> matchedExtractors) {
+ final Set<PipeRealtimeDataRegionExtractor> unmatchedExtractors = new HashSet<>();
+ for (final PipeRealtimeDataRegionExtractor extractor : extractors) {
+ if (!matchedExtractors.contains(extractor)) {
+ unmatchedExtractors.add(extractor);
+ }
+ }
+ return unmatchedExtractors;
}
protected void matchTreeModelEvent(
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/matcher/PipeDataRegionMatcher.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/matcher/PipeDataRegionMatcher.java
index 6efb3681cc9..b6c9545c729 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/matcher/PipeDataRegionMatcher.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/matcher/PipeDataRegionMatcher.java
@@ -45,10 +45,10 @@ public interface PipeDataRegionMatcher {
/**
* Match the event's schema info with the registered extractors' patterns. If the event's schema
- * info matches the pattern of a extractor, the extractor will be returned.
+ * info matches the pattern of an extractor, the extractor will be returned.
*
* @param event the event to be matched
- * @return the matched extractors
+ * @return pair of matched extractors and unmatched extractors.
*/
Pair<Set<PipeRealtimeDataRegionExtractor>, Set<PipeRealtimeDataRegionExtractor>> match(
PipeRealtimeEvent event);