This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new e3c4a4e5178 [Pipe] Support region-level realtime downgrading (#18441)
e3c4a4e5178 is described below
commit e3c4a4e5178d4bb34141a10e7f545e3cb47f8028
Author: Caideyipi <[email protected]>
AuthorDate: Mon Oct 12 10:05:22 2026 +0800
[Pipe] Support region-level realtime downgrading (#18441)
* Pipe: support region-level realtime downgrading
* Pipe: make region-level downgrading pipe-specific
* Pipe: address region-level downgrading review feedback
* [Pipe] Preserve non-deferred TsFile transfer completion
* [Pipe] Remove redundant progress index local
* [Pipe] Preserve query priority during region-level downgrading
* [Pipe] Order realtime TsFile source events and address downgrade review
---
.../event/dml/insertion/TsFileInsertionEvent.java | 5 +
.../agent/task/connection/PipeEventCollector.java | 10 +
.../common/tablet/PipeRawTabletInsertionEvent.java | 31 +-
.../common/tsfile/PipeTsFileInsertionEvent.java | 201 ++-
.../PipeRealtimeDataRegionHybridSource.java | 668 +++++++++-
.../realtime/PipeRealtimeDataRegionSource.java | 35 +
.../PipeRealtimeDataRegionTsFileSource.java | 150 ++-
.../db/pipe/source/PipeRealtimeExtractTest.java | 1277 +++++++++++++++++++-
.../pipe/config/constant/PipeSourceConstant.java | 5 +
9 files changed, 2311 insertions(+), 71 deletions(-)
diff --git
a/iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/event/dml/insertion/TsFileInsertionEvent.java
b/iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/event/dml/insertion/TsFileInsertionEvent.java
index 4c7fffcfba5..594a3c89a64 100644
---
a/iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/event/dml/insertion/TsFileInsertionEvent.java
+++
b/iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/event/dml/insertion/TsFileInsertionEvent.java
@@ -33,6 +33,11 @@ public interface TsFileInsertionEvent extends Event,
AutoCloseable {
* The method is used to convert the {@link TsFileInsertionEvent} into
several {@link
* TabletInsertionEvent}s.
*
+ * <p>The returned iterable represents the lifetime of the conversion.
Processors that retain it
+ * for asynchronous work must eventually exhaust the iterable before the
event can be considered
+ * complete. A processor should not retain the iterable and silently abandon
it, because the
+ * source cannot otherwise determine when all generated tablet events have
finished.
+ *
* @return {@code Iterable<TabletInsertionEvent>} the list of {@link
TabletInsertionEvent}
*/
Iterable<TabletInsertionEvent> toTabletInsertionEvents();
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/connection/PipeEventCollector.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/connection/PipeEventCollector.java
index ac0f6c7bf0b..53efadc77ed 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/connection/PipeEventCollector.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/connection/PipeEventCollector.java
@@ -273,8 +273,18 @@ public class PipeEventCollector implements EventCollector {
((PipeHeartbeatEvent) event).recordConnectorQueueSize(pendingQueue);
}
+ if (event instanceof PipeRawTabletInsertionEvent
+ && ((PipeRawTabletInsertionEvent) event).getSourceEvent()
+ instanceof PipeTsFileInsertionEvent) {
+ // Register before publishing to the queue so a concurrent queue cleanup
can report a
+ // generated tablet as discarded instead of racing with registration.
+ ((PipeTsFileInsertionEvent) ((PipeRawTabletInsertionEvent)
event).getSourceEvent())
+ .registerGeneratedTabletInsertionEvent((PipeRawTabletInsertionEvent)
event);
+ }
+
if (pendingQueue.offer(event)) {
collectInvocationCount.incrementAndGet();
+ return;
}
}
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 dc2ab1d381f..02162bbf3d9 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
@@ -64,6 +64,8 @@ public class PipeRawTabletInsertionEvent extends
PipeInsertionEvent
private final boolean isAligned;
private final EnrichedEvent sourceEvent;
+ private final AtomicBoolean generatedEventRegisteredWithSource = new
AtomicBoolean(false);
+ private final AtomicBoolean generatedEventOutcomeReported = new
AtomicBoolean(false);
private boolean needToReport;
private final PipeTabletMemoryBlock allocatedMemoryBlock;
@@ -118,12 +120,19 @@ public class PipeRawTabletInsertionEvent extends
PipeInsertionEvent
this.allocatedMemoryBlock =
PipeDataNodeResourceManager.memory().forceAllocateForTabletWithRetry(0);
- if (needToReport) {
+ if (needToReport || sourceEvent instanceof PipeTsFileInsertionEvent) {
addOnCommittedHook(
() -> {
- if (shouldReportOnCommit) {
+ // The last generated tablet still owns the progress-index
cleanup. Other generated
+ // tablets only contribute to the transfer completion of their
source TsFile.
+ if (shouldReportOnCommit && needToReport) {
eliminateProgressIndex();
}
+ if (sourceEvent instanceof PipeTsFileInsertionEvent
+ && generatedEventOutcomeReported.compareAndSet(false, true)) {
+ ((PipeTsFileInsertionEvent) sourceEvent)
+ .markGeneratedTabletInsertionEventAsTransferred();
+ }
});
}
}
@@ -306,6 +315,22 @@ public class PipeRawTabletInsertionEvent extends
PipeInsertionEvent
return true;
}
+ /** Marks this generated tablet as accepted by the processor output
collector. */
+ public void markAsGeneratedEventRegisteredWithSource() {
+ generatedEventRegisteredWithSource.set(true);
+ }
+
+ @Override
+ public boolean clearReferenceCount(final String holderMessage) {
+ final boolean cleared = super.clearReferenceCount(holderMessage);
+ if (generatedEventRegisteredWithSource.get()
+ && generatedEventOutcomeReported.compareAndSet(false, true)
+ && sourceEvent instanceof PipeTsFileInsertionEvent) {
+ ((PipeTsFileInsertionEvent)
sourceEvent).markGeneratedTabletInsertionEventAsDiscarded();
+ }
+ return cleared;
+ }
+
protected void eliminateProgressIndex() {
if (sourceEvent instanceof PipeTsFileInsertionEvent) {
((PipeTsFileInsertionEvent) sourceEvent).eliminateProgressIndex();
@@ -391,7 +416,7 @@ public class PipeRawTabletInsertionEvent extends
PipeInsertionEvent
}
public void markAsNeedToReport() {
- if (!needToReport) {
+ if (!needToReport && !(sourceEvent instanceof PipeTsFileInsertionEvent)) {
addOnCommittedHook(
() -> {
if (shouldReportOnCommit) {
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 aebdec7d3d6..9a89c1d367c 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
@@ -124,6 +124,18 @@ public class PipeTsFileInsertionEvent extends
PipeInsertionEvent
private final AtomicInteger parsedTabletInsertionEventCount = new
AtomicInteger(0);
private final AtomicBoolean isTsFileParsingCompleted = new
AtomicBoolean(false);
private final AtomicLong parsedPointCountForCount = new AtomicLong(0);
+ private final AtomicInteger generatedTabletInsertionEventCount = new
AtomicInteger(0);
+ private final AtomicInteger transferredGeneratedTabletInsertionEventCount =
new AtomicInteger(0);
+ private final AtomicInteger discardedGeneratedTabletInsertionEventCount =
new AtomicInteger(0);
+ private final AtomicBoolean generatedTabletInsertionEventsParsingStarted =
+ new AtomicBoolean(false);
+ private final AtomicBoolean generatedTabletInsertionEventsParsingCompleted =
+ new AtomicBoolean(false);
+ private final AtomicBoolean isTsFileEventCommitted = new
AtomicBoolean(false);
+ private final AtomicBoolean isTransferred = new AtomicBoolean(false);
+ private final AtomicBoolean isDiscarded = new AtomicBoolean(false);
+ private final List<Runnable> onTransferredHooks = new ArrayList<>();
+ private final List<Runnable> onDiscardedHooks = new ArrayList<>();
// The point count of the TsFile. Used for metrics on IoTConsensusV2'
receiver side.
// May be updated after it is flushed. Should be negative if not set.
@@ -311,6 +323,7 @@ public class PipeTsFileInsertionEvent extends
PipeInsertionEvent
if (shouldReportOnCommit) {
eliminateProgressIndex();
}
+ markAsTransferredOnCommit();
});
}
@@ -542,6 +555,16 @@ public class PipeTsFileInsertionEvent extends
PipeInsertionEvent
return isSuccessful;
}
+ @Override
+ public boolean clearReferenceCount(final String holderMessage) {
+ final boolean cleared = super.clearReferenceCount(holderMessage);
+ // clearReferenceCount intentionally bypasses the commit queue. Notify the
realtime source even
+ // when the event was already released by a previous holder: its commit
hook may still be
+ // waiting behind an earlier commit, while the explicit discard must
release the in-flight slot.
+ markAsDiscarded();
+ return cleared;
+ }
+
@Override
public void bindProgressIndex(final ProgressIndex overridingProgressIndex) {
this.overridingProgressIndex = overridingProgressIndex;
@@ -597,6 +620,133 @@ public class PipeTsFileInsertionEvent extends
PipeInsertionEvent
}
}
+ /**
+ * Records a tablet generated from this TsFile that has been accepted by the
processor output
+ * collector. The transfer hook must wait for all such tablets, rather than
the first one, before
+ * allowing the region-level downgrade to end.
+ */
+ public void registerGeneratedTabletInsertionEvent() {
+ generatedTabletInsertionEventCount.incrementAndGet();
+ }
+
+ public void registerGeneratedTabletInsertionEvent(final
PipeRawTabletInsertionEvent event) {
+ event.markAsGeneratedEventRegisteredWithSource();
+ registerGeneratedTabletInsertionEvent();
+ }
+
+ /** Marks that a processor has obtained the tablet iterable and may consume
it later. */
+ public void markGeneratedTabletInsertionEventsParsingStarted() {
+ generatedTabletInsertionEventsParsingStarted.set(true);
+ }
+
+ /** Records that one tablet generated from this TsFile has been committed
downstream. */
+ public void markGeneratedTabletInsertionEventAsTransferred() {
+ transferredGeneratedTabletInsertionEventCount.incrementAndGet();
+ markAsTransferredIfGeneratedTabletEventsCompleted();
+ }
+
+ /** Records that one generated tablet was discarded before it could be
committed downstream. */
+ public void markGeneratedTabletInsertionEventAsDiscarded() {
+ discardedGeneratedTabletInsertionEventCount.incrementAndGet();
+ markAsTransferredIfGeneratedTabletEventsCompleted();
+ }
+
+ /** Marks the end of tablet generation for this TsFile. */
+ public void markGeneratedTabletInsertionEventsParsingCompleted() {
+ generatedTabletInsertionEventsParsingCompleted.set(true);
+ markAsTransferredIfGeneratedTabletEventsCompleted();
+ }
+
+ private void markAsTransferredIfGeneratedTabletEventsCompleted() {
+ final boolean generationBoundaryReached =
+ !generatedTabletInsertionEventsParsingStarted.get()
+ || generatedTabletInsertionEventsParsingCompleted.get();
+ // The normal processor path does not retain a tablet iterable unless
parsing has explicitly
+ // been marked as started. In that path, registering the generated
tablets and reaching the
+ // generation boundary is sufficient to establish the transfer boundary
(the source TsFile
+ // commit may be reported separately or may be intentionally skipped).
Once an iterable has
+ // been retained, however, the source TsFile commit must also be observed
before the generated
+ // tablet counts can release the TsFile. This distinction keeps deferred
parsing safe without
+ // delaying the established non-deferred path.
+ if ((isTsFileEventCommitted.get() ||
!generatedTabletInsertionEventsParsingStarted.get())
+ && generationBoundaryReached
+ && transferredGeneratedTabletInsertionEventCount.get()
+ + discardedGeneratedTabletInsertionEventCount.get()
+ >= generatedTabletInsertionEventCount.get()) {
+ markAsTransferred();
+ }
+ }
+
+ private void markAsTransferredOnCommit() {
+ // A processor may retain the iterable returned by
toTabletInsertionEvents() and consume it
+ // after process() returns. In that case the iterable wrapper keeps the
generation boundary open
+ // until its iterator is exhausted. If no processor ever requests the
iterable (for example,
+ // DoNothingProcessor), committing the TsFile itself is the completion
boundary.
+ isTsFileEventCommitted.set(true);
+ markAsTransferredIfGeneratedTabletEventsCompleted();
+ }
+
+ private void markAsTransferred() {
+ final List<Runnable> hooksToRun;
+ synchronized (onTransferredHooks) {
+ if (isDiscarded.get() || !isTransferred.compareAndSet(false, true)) {
+ return;
+ }
+ hooksToRun = new ArrayList<>(onTransferredHooks);
+ onTransferredHooks.clear();
+ onDiscardedHooks.clear();
+ }
+ hooksToRun.forEach(Runnable::run);
+ }
+
+ private void markAsDiscarded() {
+ final List<Runnable> hooksToRun;
+ synchronized (onTransferredHooks) {
+ if (isTransferred.get() || !isDiscarded.compareAndSet(false, true)) {
+ return;
+ }
+ hooksToRun = new ArrayList<>(onDiscardedHooks);
+ onDiscardedHooks.clear();
+ onTransferredHooks.clear();
+ }
+ hooksToRun.forEach(Runnable::run);
+ }
+
+ /**
+ * Adds a hook that is invoked after this TsFile, or the last tablet
generated from it, is
+ * committed downstream.
+ */
+ public void addOnTransferredHook(final Runnable hook) {
+ boolean runHook = false;
+ synchronized (onTransferredHooks) {
+ if (!isTransferred.get() && !isDiscarded.get()) {
+ onTransferredHooks.add(hook);
+ return;
+ }
+ runHook = isTransferred.get();
+ }
+ if (runHook) {
+ hook.run();
+ }
+ }
+
+ /**
+ * Adds a hook that runs when this TsFile event is explicitly discarded via
clearReferenceCount.
+ */
+ public void addOnDiscardedHook(final Runnable hook) {
+ boolean runHook = false;
+ synchronized (onTransferredHooks) {
+ if (!isTransferred.get() && !isDiscarded.get()) {
+ onDiscardedHooks.add(hook);
+ return;
+ }
+ runHook = isDiscarded.get();
+ }
+ if (runHook) {
+ hook.run();
+ }
+ }
+
public PipeTsFileInsertionEvent bindTsFileDedupScopeID(final String
tsFileDedupScopeID) {
this.tsFileDedupScopeID = tsFileDedupScopeID;
return this;
@@ -867,6 +1017,7 @@ public class PipeTsFileInsertionEvent extends
PipeInsertionEvent
final String callerName,
final PipeProcessorSubtaskExecutionGuard processorExecutionGuard)
throws Exception {
+ markGeneratedTabletInsertionEventsParsingStarted();
synchronized (tabletConsumptionLock) {
final long parserStateGeneration;
synchronized (eventParser) {
@@ -891,6 +1042,7 @@ public class PipeTsFileInsertionEvent extends
PipeInsertionEvent
if (!isCurrentGeneration) {
return;
}
+ markGeneratedTabletInsertionEventsParsingCompleted();
return;
}
@@ -1129,9 +1281,12 @@ public class PipeTsFileInsertionEvent extends
PipeInsertionEvent
public Iterable<TabletInsertionEvent> toTabletInsertionEvents(final long
timeoutMs)
throws PipeException {
+ markGeneratedTabletInsertionEventsParsingStarted();
+
try {
if (!waitForTsFileClose()) {
LOGGER.warn(DataNodePipeMessages.PIPE_SKIPPING_TEMPORARY_TSFILE_S_PARSING_WHICH,
tsFile);
+ markGeneratedTabletInsertionEventsParsingCompleted();
return Collections.emptyList();
}
final long parserStateGeneration;
@@ -1145,9 +1300,10 @@ public class PipeTsFileInsertionEvent extends
PipeInsertionEvent
if (parserStateGeneration != this.parserStateGeneration) {
return Collections.emptyList();
}
- return initEventParser().toTabletInsertionEvents();
+ return
wrapGeneratedTabletInsertionEvents(initEventParser().toTabletInsertionEvents());
}
} catch (final Exception e) {
+ markGeneratedTabletInsertionEventsParsingCompleted();
close();
// close() should be called before re-interrupting the thread
@@ -1173,6 +1329,49 @@ public class PipeTsFileInsertionEvent extends
PipeInsertionEvent
}
}
+ private Iterable<TabletInsertionEvent> wrapGeneratedTabletInsertionEvents(
+ final Iterable<TabletInsertionEvent> iterable) {
+ return () -> {
+ final Iterator<TabletInsertionEvent> iterator;
+ try {
+ iterator = iterable.iterator();
+ } catch (final RuntimeException e) {
+ markGeneratedTabletInsertionEventsParsingCompleted();
+ throw e;
+ }
+ return new Iterator<TabletInsertionEvent>() {
+ @Override
+ public boolean hasNext() {
+ try {
+ final boolean hasNext = iterator.hasNext();
+ if (!hasNext) {
+ markGeneratedTabletInsertionEventsParsingCompleted();
+ }
+ return hasNext;
+ } catch (final RuntimeException e) {
+ markGeneratedTabletInsertionEventsParsingCompleted();
+ throw e;
+ }
+ }
+
+ @Override
+ public TabletInsertionEvent next() {
+ try {
+ return iterator.next();
+ } catch (final RuntimeException e) {
+ markGeneratedTabletInsertionEventsParsingCompleted();
+ throw e;
+ }
+ }
+
+ @Override
+ public void remove() {
+ iterator.remove();
+ }
+ };
+ };
+ }
+
private boolean reserveResource4Parsing(
final PipeProcessorSubtaskExecutionGuard processorExecutionGuard,
final long expectedParserStateGeneration)
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionHybridSource.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionHybridSource.java
index 65d03919ab1..7d3f6be7805 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionHybridSource.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionHybridSource.java
@@ -21,6 +21,7 @@ package org.apache.iotdb.db.pipe.source.dataregion.realtime;
import org.apache.iotdb.commons.exception.pipe.PipeRuntimeNonCriticalException;
import org.apache.iotdb.commons.pipe.config.PipeConfig;
+import org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant;
import org.apache.iotdb.commons.pipe.event.ProgressReportEvent;
import org.apache.iotdb.db.i18n.DataNodePipeMessages;
import org.apache.iotdb.db.pipe.agent.PipeDataNodeAgent;
@@ -33,6 +34,9 @@ import
org.apache.iotdb.db.pipe.metric.overview.PipeDataNodeSinglePipeMetrics;
import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager;
import
org.apache.iotdb.db.pipe.source.dataregion.realtime.assigner.PipeTsFileEpochProgressIndexKeeper;
import org.apache.iotdb.db.pipe.source.dataregion.realtime.epoch.TsFileEpoch;
+import
org.apache.iotdb.pipe.api.customizer.configuration.PipeExtractorRuntimeConfiguration;
+import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameterValidator;
+import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
import org.apache.iotdb.pipe.api.event.Event;
import org.apache.iotdb.pipe.api.event.dml.insertion.TabletInsertionEvent;
import org.apache.iotdb.pipe.api.event.dml.insertion.TsFileInsertionEvent;
@@ -40,8 +44,13 @@ import
org.apache.iotdb.pipe.api.event.dml.insertion.TsFileInsertionEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.ArrayDeque;
+import java.util.Arrays;
import java.util.Collections;
+import java.util.Deque;
import java.util.Optional;
+import java.util.PriorityQueue;
+import java.util.Queue;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
@@ -50,13 +59,69 @@ public class PipeRealtimeDataRegionHybridSource extends
PipeRealtimeDataRegionSo
private static final Logger LOGGER =
LoggerFactory.getLogger(PipeRealtimeDataRegionHybridSource.class);
+ private boolean isRegionLevelDowngradingEnabled =
+
PipeSourceConstant.EXTRACTOR_REALTIME_REGION_LEVEL_DOWNGRADING_DEFAULT_VALUE;
+ private final Object regionLevelDowngradingLock = new Object();
+
private final Set<TsFileEpoch> activeTsFileEpochs =
Collections.newSetFromMap(new ConcurrentHashMap<>());
private final Set<TsFileEpoch> degradedTsFileEpochs =
Collections.newSetFromMap(new ConcurrentHashMap<>());
+ private final Queue<TsFileEpoch> regionLevelDegradedTsFileEpochs =
+ new
PriorityQueue<>(PipeRealtimeDataRegionHybridSource::compareTsFileEpochsByQueryPriority);
+ private final Deque<PipeRealtimeEvent> eventsBeforeRegionLevelDowngrading =
new ArrayDeque<>();
+ private final Deque<PipeRealtimeEvent> regionLevelBufferedEvents = new
ArrayDeque<>();
+
+ private volatile boolean isRegionLevelDegraded = false;
+ private TsFileEpoch regionLevelTailTsFileEpoch = null;
+ private boolean canSupplyEventsBeforeRegionLevelDowngrading = false;
+ private int inFlightTsFileCount = 0;
+
+ @Override
+ public void validate(final PipeParameterValidator validator) throws
Exception {
+ super.validate(validator);
+ validator
+ .validateAttributeValueRange(
+ PipeSourceConstant.EXTRACTOR_REALTIME_REGION_LEVEL_DOWNGRADING_KEY,
+ true,
+ Boolean.TRUE.toString(),
+ Boolean.FALSE.toString())
+ .validateAttributeValueRange(
+ PipeSourceConstant.SOURCE_REALTIME_REGION_LEVEL_DOWNGRADING_KEY,
+ true,
+ Boolean.TRUE.toString(),
+ Boolean.FALSE.toString());
+ }
+
+ @Override
+ public void customize(
+ final PipeParameters parameters, final PipeExtractorRuntimeConfiguration
configuration)
+ throws Exception {
+ super.customize(parameters, configuration);
+ isRegionLevelDowngradingEnabled =
+ parameters.getBooleanOrDefault(
+ Arrays.asList(
+
PipeSourceConstant.EXTRACTOR_REALTIME_REGION_LEVEL_DOWNGRADING_KEY,
+
PipeSourceConstant.SOURCE_REALTIME_REGION_LEVEL_DOWNGRADING_KEY),
+
PipeSourceConstant.EXTRACTOR_REALTIME_REGION_LEVEL_DOWNGRADING_DEFAULT_VALUE);
+ }
@Override
protected void doExtract(final PipeRealtimeEvent event) {
+ if (isRegionLevelDowngradingEnabled) {
+ synchronized (regionLevelDowngradingLock) {
+ if (isClosed.get()) {
+
event.decreaseReferenceCount(PipeRealtimeDataRegionHybridSource.class.getName(),
false);
+ return;
+ }
+ doExtractInternal(event);
+ }
+ return;
+ }
+ doExtractInternal(event);
+ }
+
+ private void doExtractInternal(final PipeRealtimeEvent event) {
final Event eventToExtract = event.getEvent();
if (eventToExtract instanceof TabletInsertionEvent) {
@@ -91,9 +156,46 @@ public class PipeRealtimeDataRegionHybridSource extends
PipeRealtimeDataRegionSo
private void extractTabletInsertion(final PipeRealtimeEvent event) {
markTsFileEpochActive(event.getTsFileEpoch());
+ if (isRegionLevelDowngradingEnabled
+ && isRegionLevelDegraded
+ && degradedTsFileEpochs.contains(event.getTsFileEpoch())) {
+ event.getTsFileEpoch().migrateState(this, currentState ->
TsFileEpoch.State.USING_TSFILE);
+ PipeTsFileEpochProgressIndexKeeper.getInstance()
+ .registerProgressIndex(
+ dataRegionId, getTsFileDedupScopeID(),
event.getTsFileEpoch().getResource());
+
event.decreaseReferenceCount(PipeRealtimeDataRegionHybridSource.class.getName(),
false);
+ return;
+ }
+
TsFileEpoch.State state;
- if (canNotUseTabletAnymore(event)) {
+ if (isRegionLevelDowngradingEnabled && isRegionLevelDegraded) {
+ // Retain only the newest epoch as a realtime tail. Once another epoch
arrives, the previous
+ // tail is downgraded to bound the buffered tablet memory to roughly one
TsFile.
+ prepareRegionLevelTailTsFileEpochUnderLock(event.getTsFileEpoch());
+ if (canNotUseTabletAnymore(event)) {
+ if (regionLevelTailTsFileEpoch == event.getTsFileEpoch()) {
+ degradeRegionLevelTailTsFileEpochUnderLock();
+ bufferPendingEventsForRegionLevelExitUnderLock();
+ rebalanceRegionLevelBufferedEventsUnderLock();
+ }
+ event.getTsFileEpoch().migrateState(this, currentState ->
TsFileEpoch.State.USING_TSFILE);
+ PipeTsFileEpochProgressIndexKeeper.getInstance()
+ .registerProgressIndex(
+ dataRegionId, getTsFileDedupScopeID(),
event.getTsFileEpoch().getResource());
+ markTsFileEpochDegradedFromExtraction(event.getTsFileEpoch());
+
event.decreaseReferenceCount(PipeRealtimeDataRegionHybridSource.class.getName(),
false);
+ return;
+ }
+ event
+ .getTsFileEpoch()
+ .migrateState(
+ this,
+ currentState ->
+ currentState == TsFileEpoch.State.EMPTY
+ ? TsFileEpoch.State.USING_TABLET
+ : currentState);
+ } else if (canNotUseTabletAnymore(event)) {
event.getTsFileEpoch().migrateState(this, curState ->
TsFileEpoch.State.USING_TSFILE);
PipeTsFileEpochProgressIndexKeeper.getInstance()
.registerProgressIndex(
@@ -118,21 +220,25 @@ public class PipeRealtimeDataRegionHybridSource extends
PipeRealtimeDataRegionSo
state = event.getTsFileEpoch().getState(this);
if (state == TsFileEpoch.State.USING_TSFILE || state ==
TsFileEpoch.State.USING_BOTH) {
- markTsFileEpochDegraded(event.getTsFileEpoch());
+ markTsFileEpochDegradedFromExtraction(event.getTsFileEpoch());
}
switch (state) {
case USING_TSFILE:
// Ignore the tablet event.
event.decreaseReferenceCount(PipeRealtimeDataRegionHybridSource.class.getName(),
false);
break;
- case EMPTY:
- case USING_TABLET:
case USING_BOTH:
- // USING_BOTH indicates that there are discarded events previously.
- // In this case, we need to delay the progress report to tsFile event,
to avoid losing data.
- if (state == TsFileEpoch.State.USING_BOTH) {
- event.skipReportOnCommit();
+ if (isRegionLevelDowngradingEnabled && isRegionLevelDegraded) {
+
event.decreaseReferenceCount(PipeRealtimeDataRegionHybridSource.class.getName(),
false);
+ break;
}
+ // USING_BOTH indicates that there are discarded events previously. In
this case, we need
+ // to delay the progress report to the TsFile event, to avoid losing
data.
+ event.skipReportOnCommit();
+ pendingQueue.offer(event);
+ break;
+ case EMPTY:
+ case USING_TABLET:
pendingQueue.offer(event);
break;
default:
@@ -173,11 +279,18 @@ public class PipeRealtimeDataRegionHybridSource extends
PipeRealtimeDataRegionSo
});
final TsFileEpoch.State state = event.getTsFileEpoch().getState(this);
- if (state == TsFileEpoch.State.USING_BOTH) {
- markTsFileEpochDegraded(event.getTsFileEpoch());
+ if (state == TsFileEpoch.State.USING_BOTH
+ || (isRegionLevelDowngradingEnabled
+ && isRegionLevelDegraded
+ && state == TsFileEpoch.State.USING_TSFILE)) {
+ markTsFileEpochDegradedFromExtraction(event.getTsFileEpoch());
}
switch (state) {
case USING_TABLET:
+ if (isRegionLevelDowngradingEnabled && isRegionLevelDegraded) {
+ pendingQueue.offer(event);
+ return;
+ }
// If the state is USING_TABLET, discard the event
PipeTsFileEpochProgressIndexKeeper.getInstance()
.eliminateProgressIndex(
@@ -201,43 +314,289 @@ public class PipeRealtimeDataRegionHybridSource extends
PipeRealtimeDataRegionSo
}
private void markTsFileEpochActive(final TsFileEpoch tsFileEpoch) {
- activeTsFileEpochs.add(tsFileEpoch);
- reportTsFileEpochDegradedStatus();
+ synchronized (regionLevelDowngradingLock) {
+ activeTsFileEpochs.add(tsFileEpoch);
+ reportTsFileEpochDegradedStatusUnderLock();
+ }
}
private void markTsFileEpochDegraded(final TsFileEpoch tsFileEpoch) {
+ markTsFileEpochDegraded(tsFileEpoch, false);
+ }
+
+ private void markTsFileEpochDegradedFromExtraction(final TsFileEpoch
tsFileEpoch) {
+ markTsFileEpochDegraded(tsFileEpoch, true);
+ }
+
+ private void markTsFileEpochDegraded(
+ final TsFileEpoch tsFileEpoch, final boolean
shouldPreservePendingEvents) {
+ synchronized (regionLevelDowngradingLock) {
+ final boolean wasRegionLevelDegraded = isRegionLevelDegraded;
+ if (isRegionLevelDowngradingEnabled
+ && shouldPreservePendingEvents
+ && !wasRegionLevelDegraded) {
+ PipeRealtimeEvent pendingEvent;
+ while ((pendingEvent = (PipeRealtimeEvent) pendingQueue.directPoll())
!= null) {
+ eventsBeforeRegionLevelDowngrading.offerLast(pendingEvent);
+ }
+ canSupplyEventsBeforeRegionLevelDowngrading = true;
+ }
+ if (regionLevelTailTsFileEpoch == tsFileEpoch) {
+ regionLevelTailTsFileEpoch = null;
+ }
+ markTsFileEpochDegradedUnderLock(tsFileEpoch);
+ if (isRegionLevelDowngradingEnabled
+ && shouldPreservePendingEvents
+ && !wasRegionLevelDegraded) {
+
eventsBeforeRegionLevelDowngrading.forEach(this::markTsFileEpochDegradedIfNeededUnderLock);
+ }
+ if (isRegionLevelDowngradingEnabled) {
+ // A downgrade discovered while supplying an event happens after all
remaining pending
+ // events. Track those events as the possible realtime tail before
rebalancing the queues.
+ if (!wasRegionLevelDegraded && !shouldPreservePendingEvents) {
+ bufferPendingEventsAndTrackRegionLevelTailUnderLock();
+ }
+ rebalanceRegionLevelBufferedEventsUnderLock();
+ }
+ }
+ }
+
+ private void markTsFileEpochDegradedUnderLock(final TsFileEpoch tsFileEpoch)
{
activeTsFileEpochs.add(tsFileEpoch);
- degradedTsFileEpochs.add(tsFileEpoch);
- reportTsFileEpochDegradedStatus();
+ if (degradedTsFileEpochs.add(tsFileEpoch) &&
isRegionLevelDowngradingEnabled) {
+ regionLevelDegradedTsFileEpochs.offer(tsFileEpoch);
+ }
+ if (isRegionLevelDowngradingEnabled) {
+ isRegionLevelDegraded = true;
+ }
+ reportTsFileEpochDegradedStatusUnderLock();
+ }
+
+ private boolean shouldOrderTsFileEventUnderLock(final PipeRealtimeEvent
event) {
+ if (!(event.getEvent() instanceof TsFileInsertionEvent)) {
+ return false;
+ }
+
+ final TsFileEpoch.State state = event.getTsFileEpoch().getState(this);
+ return degradedTsFileEpochs.contains(event.getTsFileEpoch())
+ || state == TsFileEpoch.State.USING_TSFILE
+ || state == TsFileEpoch.State.USING_BOTH;
+ }
+
+ private void markTsFileEpochDegradedIfNeededUnderLock(final
PipeRealtimeEvent event) {
+ if (!shouldOrderTsFileEventUnderLock(event)) {
+ return;
+ }
+
+ if (regionLevelTailTsFileEpoch == event.getTsFileEpoch()) {
+ regionLevelTailTsFileEpoch = null;
+ }
+ markTsFileEpochDegradedUnderLock(event.getTsFileEpoch());
+ }
+
+ private void prepareRegionLevelTailTsFileEpochUnderLock(final TsFileEpoch
tsFileEpoch) {
+ if (regionLevelTailTsFileEpoch == tsFileEpoch) {
+ return;
+ }
+
+ if (regionLevelTailTsFileEpoch != null) {
+ degradeRegionLevelTailTsFileEpochUnderLock();
+ bufferPendingEventsForRegionLevelExitUnderLock();
+ rebalanceRegionLevelBufferedEventsUnderLock();
+ }
+ regionLevelTailTsFileEpoch = tsFileEpoch;
+ }
+
+ private void degradeRegionLevelTailTsFileEpochUnderLock() {
+ final TsFileEpoch tsFileEpoch = regionLevelTailTsFileEpoch;
+ if (tsFileEpoch == null) {
+ return;
+ }
+
+ degradeTsFileEpochUnderLock(tsFileEpoch);
+ }
+
+ private void degradeTsFileEpochUnderLock(final TsFileEpoch tsFileEpoch) {
+ if (regionLevelTailTsFileEpoch == tsFileEpoch) {
+ regionLevelTailTsFileEpoch = null;
+ }
+ tsFileEpoch.migrateState(this, state -> TsFileEpoch.State.USING_TSFILE);
+ PipeTsFileEpochProgressIndexKeeper.getInstance()
+ .registerProgressIndex(dataRegionId, getTsFileDedupScopeID(),
tsFileEpoch.getResource());
+ markTsFileEpochDegradedUnderLock(tsFileEpoch);
+ }
+
+ private void degradeLowerPriorityActiveTsFileEpochsUnderLock() {
+ final TsFileEpoch nextDegradedTsFileEpoch =
regionLevelDegradedTsFileEpochs.peek();
+ if (nextDegradedTsFileEpoch == null) {
+ return;
+ }
+
+ // An active epoch may be discovered as degraded only when its queued
tablet is supplied. Make
+ // every known lower-priority epoch enter the ordered TsFile queue before
allowing the current
+ // head to pass, including the realtime tail and epochs whose TsFile has
not been flushed yet.
+ for (final TsFileEpoch activeTsFileEpoch : activeTsFileEpochs) {
+ if (!degradedTsFileEpochs.contains(activeTsFileEpoch)
+ && compareTsFileEpochsByQueryPriority(activeTsFileEpoch,
nextDegradedTsFileEpoch) < 0) {
+ degradeTsFileEpochUnderLock(activeTsFileEpoch);
+ }
+ }
+ }
+
+ private void bufferPendingEventsAndTrackRegionLevelTailUnderLock() {
+ PipeRealtimeEvent pendingEvent;
+ while ((pendingEvent = (PipeRealtimeEvent) pendingQueue.directPoll()) !=
null) {
+ if (pendingEvent.getEvent() instanceof TabletInsertionEvent
+ && !degradedTsFileEpochs.contains(pendingEvent.getTsFileEpoch())) {
+ if (regionLevelTailTsFileEpoch != null
+ && regionLevelTailTsFileEpoch != pendingEvent.getTsFileEpoch()) {
+ degradeRegionLevelTailTsFileEpochUnderLock();
+ }
+ regionLevelTailTsFileEpoch = pendingEvent.getTsFileEpoch();
+ }
+ regionLevelBufferedEvents.offerLast(pendingEvent);
+ }
+ }
+
+ private void rebalanceRegionLevelBufferedEventsUnderLock() {
+ bufferPendingEventsForRegionLevelExitUnderLock();
+ degradeLowerPriorityActiveTsFileEpochsUnderLock();
+
+ final TsFileEpoch nextDegradedTsFileEpoch =
regionLevelDegradedTsFileEpochs.peek();
+ final Deque<PipeRealtimeEvent> retainedEvents = new ArrayDeque<>();
+ boolean nextDegradedTsFileEventPromoted = false;
+ PipeRealtimeEvent bufferedEvent;
+ while ((bufferedEvent = regionLevelBufferedEvents.pollFirst()) != null) {
+ if (bufferedEvent.getEvent() instanceof TabletInsertionEvent
+ && degradedTsFileEpochs.contains(bufferedEvent.getTsFileEpoch())) {
+ bufferedEvent.decreaseReferenceCount(
+ PipeRealtimeDataRegionHybridSource.class.getName(), false);
+ } else if (!nextDegradedTsFileEventPromoted
+ && bufferedEvent.getEvent() instanceof TsFileInsertionEvent
+ && bufferedEvent.getTsFileEpoch() == nextDegradedTsFileEpoch) {
+ pendingQueue.offer(bufferedEvent);
+ nextDegradedTsFileEventPromoted = true;
+ } else {
+ retainedEvents.offerLast(bufferedEvent);
+ }
+ }
+ regionLevelBufferedEvents.addAll(retainedEvents);
}
private void clearTsFileEpoch(final TsFileEpoch tsFileEpoch) {
+ synchronized (regionLevelDowngradingLock) {
+ clearTsFileEpochUnderLock(tsFileEpoch);
+ }
+ }
+
+ private void clearTsFileEpochAfterCommit(final TsFileEpoch tsFileEpoch) {
+ synchronized (regionLevelDowngradingLock) {
+ if (isClosed.get()) {
+ return;
+ }
+ if (inFlightTsFileCount > 0) {
+ --inFlightTsFileCount;
+ }
+ clearTsFileEpochUnderLock(tsFileEpoch);
+
+ // The decision whether newer writes can resume the realtime path must
be made at the exact
+ // point when the last currently degraded TsFile is committed.
Otherwise, writes arriving
+ // between the commit and the next supply call would still be
unnecessarily downgraded.
+ if (isRegionLevelDegraded
+ && inFlightTsFileCount == 0
+ && degradedTsFileEpochs.isEmpty()
+ && eventsBeforeRegionLevelDowngrading.isEmpty()) {
+ bufferPendingEventsForRegionLevelExitUnderLock();
+ tryExitRegionLevelDowngrading(false);
+ }
+ }
+ }
+
+ private void bufferPendingEventsForRegionLevelExitUnderLock() {
+ PipeRealtimeEvent event;
+ while ((event = (PipeRealtimeEvent) pendingQueue.directPoll()) != null) {
+ markTsFileEpochDegradedIfNeededUnderLock(event);
+ regionLevelBufferedEvents.offerLast(event);
+ }
+ }
+
+ private void clearTsFileEpochUnderLock(final TsFileEpoch tsFileEpoch) {
activeTsFileEpochs.remove(tsFileEpoch);
degradedTsFileEpochs.remove(tsFileEpoch);
- reportTsFileEpochDegradedStatus();
+ regionLevelDegradedTsFileEpochs.remove(tsFileEpoch);
+ if (regionLevelTailTsFileEpoch == tsFileEpoch) {
+ regionLevelTailTsFileEpoch = null;
+ }
+ if (isRegionLevelDowngradingEnabled && isRegionLevelDegraded &&
inFlightTsFileCount == 0) {
+ rebalanceRegionLevelBufferedEventsUnderLock();
+ }
+ reportTsFileEpochDegradedStatusUnderLock();
}
- private void reportTsFileEpochDegradedStatus() {
- if (activeTsFileEpochs.isEmpty()) {
+ private void clearRegionLevelBufferedEventsUnderLock() {
+ clearBufferedEventsUnderLock(eventsBeforeRegionLevelDowngrading);
+ clearBufferedEventsUnderLock(regionLevelBufferedEvents);
+ }
+
+ private void clearBufferedEventsUnderLock(final Deque<PipeRealtimeEvent>
events) {
+ PipeRealtimeEvent event;
+ while ((event = events.pollFirst()) != null) {
+
event.clearReferenceCount(PipeRealtimeDataRegionHybridSource.class.getName());
+ }
+ }
+
+ private void reportTsFileEpochDegradedStatusUnderLock() {
+ if (isRegionLevelDowngradingEnabled && isRegionLevelDegraded) {
+ PipeDataNodeAgent.task()
+ .setPipeTsFileEpochDegraded(pipeName, creationTime, dataRegionId,
true);
+ } else if (activeTsFileEpochs.isEmpty()) {
PipeDataNodeAgent.task().clearPipeTsFileEpochDegraded(pipeName,
creationTime, dataRegionId);
} else {
PipeDataNodeAgent.task()
.setPipeTsFileEpochDegraded(
- pipeName, creationTime, dataRegionId,
!degradedTsFileEpochs.isEmpty());
+ pipeName,
+ creationTime,
+ dataRegionId,
+ isRegionLevelDowngradingEnabled ? false :
!degradedTsFileEpochs.isEmpty());
}
}
@Override
public void close() throws Exception {
try {
+ // Do not hold regionLevelDowngradingLock while waiting for the assigner
to stop. An event
+ // already being assigned may need the same lock to finish extraction.
super.close();
} finally {
- activeTsFileEpochs.clear();
- degradedTsFileEpochs.clear();
- PipeDataNodeAgent.task().clearPipeTsFileEpochDegraded(pipeName,
creationTime, dataRegionId);
+ synchronized (regionLevelDowngradingLock) {
+ clearRegionLevelBufferedEventsUnderLock();
+ activeTsFileEpochs.clear();
+ degradedTsFileEpochs.clear();
+ regionLevelDegradedTsFileEpochs.clear();
+ isRegionLevelDegraded = false;
+ regionLevelTailTsFileEpoch = null;
+ canSupplyEventsBeforeRegionLevelDowngrading = false;
+ inFlightTsFileCount = 0;
+ PipeDataNodeAgent.task().clearPipeTsFileEpochDegraded(pipeName,
creationTime, dataRegionId);
+ }
}
}
+ @Override
+ protected void extractProgressReportEvent(final PipeRealtimeEvent event) {
+ if (isRegionLevelDowngradingEnabled) {
+ synchronized (regionLevelDowngradingLock) {
+ if (isClosed.get()) {
+
event.decreaseReferenceCount(PipeRealtimeDataRegionHybridSource.class.getName(),
false);
+ return;
+ }
+ super.extractProgressReportEvent(event);
+ }
+ return;
+ }
+ super.extractProgressReportEvent(event);
+ }
+
// If the insertNode's memory has reached the dangerous threshold, we should
not extract any
// tablets.
private boolean canNotUseTabletAnymore(final PipeRealtimeEvent event) {
@@ -278,47 +637,206 @@ public class PipeRealtimeDataRegionHybridSource extends
PipeRealtimeDataRegionSo
@Override
public Event supply() {
+ if (isRegionLevelDowngradingEnabled) {
+ synchronized (regionLevelDowngradingLock) {
+ return isRegionLevelDegraded ? supplyRegionLevelDegradedInternal() :
supplyInternal();
+ }
+ }
+ return supplyInternal();
+ }
+
+ private Event supplyInternal() {
PipeRealtimeEvent realtimeEvent = (PipeRealtimeEvent)
pendingQueue.directPoll();
while (realtimeEvent != null) {
- Event suppliedEvent;
+ final Event suppliedEvent = supplyExtractedEvent(realtimeEvent);
+ if (suppliedEvent != null) {
+ return suppliedEvent;
+ }
- // Used to judge the type of the event, not directly for supplying.
+ if (isRegionLevelDowngradingEnabled && isRegionLevelDegraded) {
+ return supplyRegionLevelDegradedInternal();
+ }
+
+ realtimeEvent = (PipeRealtimeEvent) pendingQueue.directPoll();
+ }
+
+ // Means the pending queue is empty.
+ return null;
+ }
+
+ private Event supplyExtractedEvent(final PipeRealtimeEvent realtimeEvent) {
+ Event suppliedEvent;
+
+ // Used to judge the type of the event, not directly for supplying.
+ final Event eventToSupply = realtimeEvent.getEvent();
+ if (eventToSupply instanceof TabletInsertionEvent) {
+ suppliedEvent = supplyTabletInsertion(realtimeEvent);
+ } else if (eventToSupply instanceof TsFileInsertionEvent) {
+ suppliedEvent = supplyTsFileInsertion(realtimeEvent);
+ } else if (eventToSupply instanceof PipeHeartbeatEvent) {
+ suppliedEvent = supplyHeartbeat(realtimeEvent);
+ } else if (eventToSupply instanceof PipeDeleteDataNodeEvent
+ || eventToSupply instanceof ProgressReportEvent) {
+ suppliedEvent = supplyDirectly(realtimeEvent);
+ } else {
+ throw new UnsupportedOperationException(
+ String.format(
+ DataNodePipeMessages
+
.PIPE_EXCEPTION_UNSUPPORTED_EVENT_TYPE_S_FOR_HYBRID_REALTIME_EXTRACTOR_S_474BAAC2,
+ eventToSupply.getClass(),
+ this));
+ }
+
+
realtimeEvent.decreaseReferenceCount(PipeRealtimeDataRegionHybridSource.class.getName(),
false);
+
+ if (suppliedEvent != null) {
+ suppliedEvent = assignReplicateIndexIfNeeded(realtimeEvent,
suppliedEvent);
+ maySkipIndex4Event(realtimeEvent);
+ }
+ return suppliedEvent;
+ }
+
+ private Event supplyRegionLevelDegradedInternal() {
+ // Once region-level downgrading starts, only one TsFile is allowed to be
in flight. Events of
+ // newer epochs are buffered until all currently degraded TsFiles are
committed downstream.
+ if (inFlightTsFileCount > 0) {
+ return null;
+ }
+
+ final Event eventBeforeDowngrading =
supplyEventsBeforeRegionLevelDowngradingInternal();
+ if (eventBeforeDowngrading != null) {
+ return eventBeforeDowngrading;
+ }
+
+ PipeRealtimeEvent realtimeEvent = (PipeRealtimeEvent)
pendingQueue.directPoll();
+ while (realtimeEvent != null) {
final Event eventToSupply = realtimeEvent.getEvent();
if (eventToSupply instanceof TabletInsertionEvent) {
- suppliedEvent = supplyTabletInsertion(realtimeEvent);
+ if (degradedTsFileEpochs.contains(realtimeEvent.getTsFileEpoch())) {
+ realtimeEvent.decreaseReferenceCount(
+ PipeRealtimeDataRegionHybridSource.class.getName(), false);
+ } else {
+ regionLevelBufferedEvents.offerLast(realtimeEvent);
+ }
} else if (eventToSupply instanceof TsFileInsertionEvent) {
- suppliedEvent = supplyTsFileInsertion(realtimeEvent);
- } else if (eventToSupply instanceof PipeHeartbeatEvent) {
- suppliedEvent = supplyHeartbeat(realtimeEvent);
- } else if (eventToSupply instanceof PipeDeleteDataNodeEvent
- || eventToSupply instanceof ProgressReportEvent) {
- suppliedEvent = supplyDirectly(realtimeEvent);
+ final TsFileEpoch.State state =
realtimeEvent.getTsFileEpoch().getState(this);
+ if (degradedTsFileEpochs.contains(realtimeEvent.getTsFileEpoch())
+ || state == TsFileEpoch.State.USING_TSFILE
+ || state == TsFileEpoch.State.USING_BOTH) {
+ markTsFileEpochDegraded(realtimeEvent.getTsFileEpoch());
+ if (regionLevelDegradedTsFileEpochs.peek() !=
realtimeEvent.getTsFileEpoch()) {
+ regionLevelBufferedEvents.offerLast(realtimeEvent);
+ realtimeEvent = (PipeRealtimeEvent) pendingQueue.directPoll();
+ continue;
+ }
+ final Event suppliedEvent = supplyExtractedEvent(realtimeEvent);
+ if (suppliedEvent != null) {
+ return suppliedEvent;
+ }
+ } else {
+ regionLevelBufferedEvents.offerLast(realtimeEvent);
+ }
} else {
- throw new UnsupportedOperationException(
- String.format(
- DataNodePipeMessages
-
.PIPE_EXCEPTION_UNSUPPORTED_EVENT_TYPE_S_FOR_HYBRID_REALTIME_EXTRACTOR_S_474BAAC2,
- eventToSupply.getClass(),
- this));
+ regionLevelBufferedEvents.offerLast(realtimeEvent);
}
- realtimeEvent.decreaseReferenceCount(
- PipeRealtimeDataRegionHybridSource.class.getName(), false);
+ realtimeEvent = (PipeRealtimeEvent) pendingQueue.directPoll();
+ }
+
+ return tryExitRegionLevelDowngrading(true);
+ }
- if (suppliedEvent != null) {
- suppliedEvent = assignReplicateIndexIfNeeded(realtimeEvent,
suppliedEvent);
- maySkipIndex4Event(realtimeEvent);
- return suppliedEvent;
+ private Event supplyEventsBeforeRegionLevelDowngradingInternal() {
+ PipeRealtimeEvent realtimeEvent;
+ while ((realtimeEvent = eventsBeforeRegionLevelDowngrading.pollFirst()) !=
null) {
+ final Event eventToSupply = realtimeEvent.getEvent();
+
+ if (canSupplyEventsBeforeRegionLevelDowngrading) {
+ if (eventToSupply instanceof TabletInsertionEvent
+ && degradedTsFileEpochs.contains(realtimeEvent.getTsFileEpoch())) {
+ realtimeEvent.decreaseReferenceCount(
+ PipeRealtimeDataRegionHybridSource.class.getName(), false);
+ continue;
+ }
+
+ if (shouldOrderTsFileEventUnderLock(realtimeEvent)) {
+ markTsFileEpochDegradedIfNeededUnderLock(realtimeEvent);
+ regionLevelBufferedEvents.offerLast(realtimeEvent);
+ continue;
+ }
+
+ final Event suppliedEvent = supplyExtractedEvent(realtimeEvent);
+ if (suppliedEvent != null) {
+ return suppliedEvent;
+ }
+ if (eventToSupply instanceof TabletInsertionEvent
+ && degradedTsFileEpochs.contains(realtimeEvent.getTsFileEpoch())) {
+ canSupplyEventsBeforeRegionLevelDowngrading = false;
+ }
+ continue;
}
- realtimeEvent = (PipeRealtimeEvent) pendingQueue.directPoll();
+ if (eventToSupply instanceof TabletInsertionEvent) {
+ if (degradedTsFileEpochs.contains(realtimeEvent.getTsFileEpoch())) {
+ realtimeEvent.decreaseReferenceCount(
+ PipeRealtimeDataRegionHybridSource.class.getName(), false);
+ } else {
+ regionLevelBufferedEvents.offerLast(realtimeEvent);
+ }
+ } else if (eventToSupply instanceof TsFileInsertionEvent) {
+ final TsFileEpoch.State state =
realtimeEvent.getTsFileEpoch().getState(this);
+ if (degradedTsFileEpochs.contains(realtimeEvent.getTsFileEpoch())
+ || state == TsFileEpoch.State.USING_TSFILE
+ || state == TsFileEpoch.State.USING_BOTH) {
+ markTsFileEpochDegraded(realtimeEvent.getTsFileEpoch());
+ if (regionLevelDegradedTsFileEpochs.peek() !=
realtimeEvent.getTsFileEpoch()) {
+ regionLevelBufferedEvents.offerLast(realtimeEvent);
+ continue;
+ }
+ final Event suppliedEvent = supplyExtractedEvent(realtimeEvent);
+ if (suppliedEvent != null) {
+ return suppliedEvent;
+ }
+ } else {
+ regionLevelBufferedEvents.offerLast(realtimeEvent);
+ }
+ } else {
+ regionLevelBufferedEvents.offerLast(realtimeEvent);
+ }
}
- // Means the pending queue is empty.
+ canSupplyEventsBeforeRegionLevelDowngrading = false;
+ rebalanceRegionLevelBufferedEventsUnderLock();
return null;
}
+ private Event tryExitRegionLevelDowngrading(final boolean
shouldSupplyAfterTransition) {
+ if (!degradedTsFileEpochs.isEmpty()) {
+ // Some degraded epochs are still waiting for their TsFile to be flushed.
+ return null;
+ }
+
+ if (regionLevelTailTsFileEpoch != null
+ && regionLevelBufferedEvents.stream()
+ .filter(event -> event.getTsFileEpoch() ==
regionLevelTailTsFileEpoch)
+ .filter(event -> event.getEvent() instanceof TabletInsertionEvent)
+ .anyMatch(event -> event.getEvent().isReleased())) {
+ degradeRegionLevelTailTsFileEpochUnderLock();
+ rebalanceRegionLevelBufferedEventsUnderLock();
+ return shouldSupplyAfterTransition ? supplyRegionLevelDegradedInternal()
: null;
+ }
+
+ isRegionLevelDegraded = false;
+ regionLevelTailTsFileEpoch = null;
+ PipeRealtimeEvent bufferedEvent;
+ while ((bufferedEvent = regionLevelBufferedEvents.pollFirst()) != null) {
+ pendingQueue.offer(bufferedEvent);
+ }
+ reportTsFileEpochDegradedStatusUnderLock();
+ return shouldSupplyAfterTransition ? supplyInternal() : null;
+ }
+
private Event supplyTabletInsertion(final PipeRealtimeEvent event) {
if
(event.increaseReferenceCount(PipeRealtimeDataRegionHybridSource.class.getName()))
{
return event.getEvent();
@@ -326,7 +844,14 @@ public class PipeRealtimeDataRegionHybridSource extends
PipeRealtimeDataRegionSo
// If the event's reference count can not be increased, it means the
data represented by
// this event is not reliable anymore. but the data represented by this
event
// has been carried by the following tsfile event, so we can just
discard this event.
- event.getTsFileEpoch().migrateState(this, s ->
TsFileEpoch.State.USING_BOTH);
+ event
+ .getTsFileEpoch()
+ .migrateState(
+ this,
+ state ->
+ isRegionLevelDowngradingEnabled
+ ? TsFileEpoch.State.USING_TSFILE
+ : TsFileEpoch.State.USING_BOTH);
markTsFileEpochDegraded(event.getTsFileEpoch());
LOGGER.warn(DataNodePipeMessages.DISCARD_TABLET_EVENT_BECAUSE_IT_IS_NOT,
event);
return null;
@@ -334,27 +859,44 @@ public class PipeRealtimeDataRegionHybridSource extends
PipeRealtimeDataRegionSo
}
private Event supplyTsFileInsertion(final PipeRealtimeEvent event) {
- try {
- if
(event.increaseReferenceCount(PipeRealtimeDataRegionHybridSource.class.getName()))
{
- return event.getEvent();
+ if (isRegionLevelDowngradingEnabled
+ && event.getTsFileEpoch().getState(this) ==
TsFileEpoch.State.USING_TABLET) {
+ PipeTsFileEpochProgressIndexKeeper.getInstance()
+ .eliminateProgressIndex(
+ dataRegionId, getTsFileDedupScopeID(),
event.getTsFileEpoch().getFilePath());
+ clearTsFileEpoch(event.getTsFileEpoch());
+ return null;
+ }
+
+ if
(event.increaseReferenceCount(PipeRealtimeDataRegionHybridSource.class.getName()))
{
+ if (isRegionLevelDowngradingEnabled) {
+ ++inFlightTsFileCount;
+ final PipeTsFileInsertionEvent tsFileInsertionEvent =
+ (PipeTsFileInsertionEvent) event.getEvent();
+ final Runnable clearTsFileEpochHook =
+ () -> clearTsFileEpochAfterCommit(event.getTsFileEpoch());
+ tsFileInsertionEvent.addOnTransferredHook(clearTsFileEpochHook);
+ tsFileInsertionEvent.addOnDiscardedHook(clearTsFileEpochHook);
} else {
- // If the event's reference count can not be increased, it means the
data represented by
- // this event is not reliable anymore. the data has been lost. we
simply discard this
- // event and report the exception to PipeRuntimeAgent.
- final String errorMessage =
- String.format(
-
DataNodePipeMessages.EVENT_CAN_NOT_BE_SUPPLIED_BECAUSE_DATA_IS_LOST,
- event.getEvent());
- LOGGER.error(errorMessage);
- PipeDataNodeAgent.runtime()
- .report(pipeTaskMeta, new
PipeRuntimeNonCriticalException(errorMessage));
- PipeTsFileEpochProgressIndexKeeper.getInstance()
- .eliminateProgressIndex(
- dataRegionId, getTsFileDedupScopeID(),
event.getTsFileEpoch().getFilePath());
- return null;
+ clearTsFileEpoch(event.getTsFileEpoch());
}
- } finally {
+ return event.getEvent();
+ } else {
+ // If the event's reference count can not be increased, it means the
data represented by
+ // this event is not reliable anymore. the data has been lost. we simply
discard this
+ // event and report the exception to PipeRuntimeAgent.
+ final String errorMessage =
+ String.format(
+
DataNodePipeMessages.EVENT_CAN_NOT_BE_SUPPLIED_BECAUSE_DATA_IS_LOST,
+ event.getEvent());
+ LOGGER.error(errorMessage);
+ PipeDataNodeAgent.runtime()
+ .report(pipeTaskMeta, new
PipeRuntimeNonCriticalException(errorMessage));
+ PipeTsFileEpochProgressIndexKeeper.getInstance()
+ .eliminateProgressIndex(
+ dataRegionId, getTsFileDedupScopeID(),
event.getTsFileEpoch().getFilePath());
clearTsFileEpoch(event.getTsFileEpoch());
+ return null;
}
}
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionSource.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionSource.java
index 92ad93e09f1..a678fd0514a 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionSource.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionSource.java
@@ -46,10 +46,13 @@ import
org.apache.iotdb.db.pipe.metric.source.PipeDataRegionEventCounter;
import
org.apache.iotdb.db.pipe.processor.iotconsensusv2.IoTConsensusV2Processor;
import org.apache.iotdb.db.pipe.source.dataregion.DataRegionListeningFilter;
import
org.apache.iotdb.db.pipe.source.dataregion.realtime.assigner.PipeTsFileEpochProgressIndexKeeper;
+import org.apache.iotdb.db.pipe.source.dataregion.realtime.epoch.TsFileEpoch;
import
org.apache.iotdb.db.pipe.source.dataregion.realtime.listener.PipeInsertionDataNodeListener;
import
org.apache.iotdb.db.pipe.source.dataregion.realtime.listener.PipeTimePartitionListener;
import org.apache.iotdb.db.storageengine.StorageEngine;
import org.apache.iotdb.db.storageengine.dataregion.DataRegion;
+import
org.apache.iotdb.db.storageengine.dataregion.read.reader.common.MergeReaderPriority;
+import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
import org.apache.iotdb.pipe.api.PipeExtractor;
import
org.apache.iotdb.pipe.api.customizer.configuration.PipeExtractorRuntimeConfiguration;
import
org.apache.iotdb.pipe.api.customizer.configuration.PipeRuntimeEnvironment;
@@ -417,6 +420,38 @@ public abstract class PipeRealtimeDataRegionSource
implements PipeExtractor {
protected abstract void doExtract(final PipeRealtimeEvent event);
+ protected static int compareTsFileEpochsByQueryPriority(
+ final TsFileEpoch epoch1, final TsFileEpoch epoch2) {
+ final TsFileResource resource1 = epoch1.getResource();
+ final TsFileResource resource2 = epoch2.getResource();
+
+ final int queryPriorityComparison =
+ new MergeReaderPriority(
+ resource1.getTsFileID().getTimestamp(),
+ resource1.getVersion(),
+ 0,
+ resource1.isSeq())
+ .compareTo(
+ new MergeReaderPriority(
+ resource2.getTsFileID().getTimestamp(),
+ resource2.getVersion(),
+ 0,
+ resource2.isSeq()));
+ if (queryPriorityComparison != 0) {
+ return queryPriorityComparison;
+ }
+
+ // Equal query priority does not imply the same file: this file-level
comparison omits
+ // compaction versions and directories, and fixes the chunk offset at
zero. Use deterministic
+ // tie-breakers for distinct files with the same seq/unseq flag, version
and timestamp.
+ final int compactionVersionComparison =
+ Long.compare(
+ resource1.getTsFileID().compactionVersion,
resource2.getTsFileID().compactionVersion);
+ return compactionVersionComparison != 0
+ ? compactionVersionComparison
+ : resource1.getTsFilePath().compareTo(resource2.getTsFilePath());
+ }
+
protected void extractHeartbeat(final PipeRealtimeEvent event) {
// Record the pending queue size before trying to put heartbeatEvent into
queue
((PipeHeartbeatEvent)
event.getEvent()).recordExtractorQueueSize(pendingQueue);
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionTsFileSource.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionTsFileSource.java
index 73bef31d85b..1e7a000c8c8 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionTsFileSource.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionTsFileSource.java
@@ -20,27 +20,91 @@
package org.apache.iotdb.db.pipe.source.dataregion.realtime;
import org.apache.iotdb.commons.exception.pipe.PipeRuntimeNonCriticalException;
+import org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant;
import org.apache.iotdb.commons.pipe.event.ProgressReportEvent;
import org.apache.iotdb.db.i18n.DataNodePipeMessages;
import org.apache.iotdb.db.pipe.agent.PipeDataNodeAgent;
import org.apache.iotdb.db.pipe.event.common.deletion.PipeDeleteDataNodeEvent;
import org.apache.iotdb.db.pipe.event.common.heartbeat.PipeHeartbeatEvent;
+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.source.dataregion.realtime.assigner.PipeTsFileEpochProgressIndexKeeper;
import org.apache.iotdb.db.pipe.source.dataregion.realtime.epoch.TsFileEpoch;
+import
org.apache.iotdb.pipe.api.customizer.configuration.PipeExtractorRuntimeConfiguration;
+import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameterValidator;
+import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
import org.apache.iotdb.pipe.api.event.Event;
import org.apache.iotdb.pipe.api.event.dml.insertion.TsFileInsertionEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.ArrayDeque;
+import java.util.Arrays;
+import java.util.Deque;
+import java.util.PriorityQueue;
+import java.util.Queue;
+
public class PipeRealtimeDataRegionTsFileSource extends
PipeRealtimeDataRegionSource {
private static final Logger LOGGER =
LoggerFactory.getLogger(PipeRealtimeDataRegionTsFileSource.class);
+ private boolean isRegionLevelDowngradingEnabled =
+
PipeSourceConstant.EXTRACTOR_REALTIME_REGION_LEVEL_DOWNGRADING_DEFAULT_VALUE;
+ private final Object tsFileOrderingLock = new Object();
+ private final Queue<PipeRealtimeEvent> orderedTsFileEvents =
+ new PriorityQueue<>(
+ (event1, event2) ->
+ compareTsFileEpochsByQueryPriority(event1.getTsFileEpoch(),
event2.getTsFileEpoch()));
+ private final Deque<PipeRealtimeEvent> bufferedEvents = new ArrayDeque<>();
+ private TsFileEpoch inFlightTsFileEpoch;
+
+ @Override
+ public void validate(final PipeParameterValidator validator) throws
Exception {
+ super.validate(validator);
+ validator
+ .validateAttributeValueRange(
+ PipeSourceConstant.EXTRACTOR_REALTIME_REGION_LEVEL_DOWNGRADING_KEY,
+ true,
+ Boolean.TRUE.toString(),
+ Boolean.FALSE.toString())
+ .validateAttributeValueRange(
+ PipeSourceConstant.SOURCE_REALTIME_REGION_LEVEL_DOWNGRADING_KEY,
+ true,
+ Boolean.TRUE.toString(),
+ Boolean.FALSE.toString());
+ }
+
@Override
- protected void doExtract(PipeRealtimeEvent event) {
+ public void customize(
+ final PipeParameters parameters, final PipeExtractorRuntimeConfiguration
configuration)
+ throws Exception {
+ super.customize(parameters, configuration);
+ isRegionLevelDowngradingEnabled =
+ parameters.getBooleanOrDefault(
+ Arrays.asList(
+
PipeSourceConstant.EXTRACTOR_REALTIME_REGION_LEVEL_DOWNGRADING_KEY,
+
PipeSourceConstant.SOURCE_REALTIME_REGION_LEVEL_DOWNGRADING_KEY),
+
PipeSourceConstant.EXTRACTOR_REALTIME_REGION_LEVEL_DOWNGRADING_DEFAULT_VALUE);
+ }
+
+ @Override
+ protected void doExtract(final PipeRealtimeEvent event) {
+ if (isRegionLevelDowngradingEnabled) {
+ synchronized (tsFileOrderingLock) {
+ if (isClosed.get()) {
+
event.decreaseReferenceCount(PipeRealtimeDataRegionTsFileSource.class.getName(),
false);
+ return;
+ }
+ doExtractInternal(event);
+ }
+ return;
+ }
+ doExtractInternal(event);
+ }
+
+ private void doExtractInternal(final PipeRealtimeEvent event) {
if (event.getEvent() instanceof PipeHeartbeatEvent) {
extractHeartbeat(event);
return;
@@ -76,9 +140,72 @@ public class PipeRealtimeDataRegionTsFileSource extends
PipeRealtimeDataRegionSo
return false;
}
+ @Override
+ protected void extractProgressReportEvent(final PipeRealtimeEvent event) {
+ if (isRegionLevelDowngradingEnabled) {
+ synchronized (tsFileOrderingLock) {
+ if (isClosed.get()) {
+
event.decreaseReferenceCount(PipeRealtimeDataRegionTsFileSource.class.getName(),
false);
+ return;
+ }
+ super.extractProgressReportEvent(event);
+ }
+ return;
+ }
+ super.extractProgressReportEvent(event);
+ }
+
+ @Override
+ public void close() throws Exception {
+ try {
+ super.close();
+ } finally {
+ synchronized (tsFileOrderingLock) {
+ PipeRealtimeEvent event;
+ while ((event = orderedTsFileEvents.poll()) != null) {
+
event.clearReferenceCount(PipeRealtimeDataRegionTsFileSource.class.getName());
+ }
+ while ((event = bufferedEvents.pollFirst()) != null) {
+
event.clearReferenceCount(PipeRealtimeDataRegionTsFileSource.class.getName());
+ }
+ inFlightTsFileEpoch = null;
+ }
+ }
+ }
+
@Override
public Event supply() {
- PipeRealtimeEvent realtimeEvent = (PipeRealtimeEvent)
pendingQueue.directPoll();
+ if (isRegionLevelDowngradingEnabled) {
+ synchronized (tsFileOrderingLock) {
+ if (isClosed.get() || inFlightTsFileEpoch != null) {
+ return null;
+ }
+
+ // Include every observed, unsent TsFile before selecting the next
one. Non-file events
+ // stay behind the files so they cannot advance progress past an
unfinished file.
+ PipeRealtimeEvent pendingEvent;
+ while ((pendingEvent = (PipeRealtimeEvent) pendingQueue.directPoll())
!= null) {
+ if (pendingEvent.getEvent() instanceof TsFileInsertionEvent) {
+ orderedTsFileEvents.offer(pendingEvent);
+ } else {
+ bufferedEvents.offerLast(pendingEvent);
+ }
+ }
+ return supplyInternal();
+ }
+ }
+ return supplyInternal();
+ }
+
+ private PipeRealtimeEvent pollNextEvent() {
+ if (!isRegionLevelDowngradingEnabled) {
+ return (PipeRealtimeEvent) pendingQueue.directPoll();
+ }
+ return orderedTsFileEvents.isEmpty() ? bufferedEvents.pollFirst() :
orderedTsFileEvents.poll();
+ }
+
+ private Event supplyInternal() {
+ PipeRealtimeEvent realtimeEvent = pollNextEvent();
while (realtimeEvent != null) {
Event suppliedEvent = null;
@@ -90,6 +217,15 @@ public class PipeRealtimeDataRegionTsFileSource extends
PipeRealtimeDataRegionSo
suppliedEvent = supplyDirectly(realtimeEvent);
} else if (realtimeEvent.increaseReferenceCount(
PipeRealtimeDataRegionTsFileSource.class.getName())) {
+ if (isRegionLevelDowngradingEnabled) {
+ inFlightTsFileEpoch = realtimeEvent.getTsFileEpoch();
+ final TsFileEpoch suppliedTsFileEpoch = inFlightTsFileEpoch;
+ final PipeTsFileInsertionEvent tsFileInsertionEvent =
+ (PipeTsFileInsertionEvent) realtimeEvent.getEvent();
+ final Runnable onCompletedHook = () ->
clearInFlightTsFileEpoch(suppliedTsFileEpoch);
+ tsFileInsertionEvent.addOnTransferredHook(onCompletedHook);
+ tsFileInsertionEvent.addOnDiscardedHook(onCompletedHook);
+ }
suppliedEvent = realtimeEvent.getEvent();
} else {
// if the event's reference count can not be increased, it means the
data represented by
@@ -118,10 +254,18 @@ public class PipeRealtimeDataRegionTsFileSource extends
PipeRealtimeDataRegionSo
return suppliedEvent;
}
- realtimeEvent = (PipeRealtimeEvent) pendingQueue.directPoll();
+ realtimeEvent = pollNextEvent();
}
// means the pending queue is empty.
return null;
}
+
+ private void clearInFlightTsFileEpoch(final TsFileEpoch tsFileEpoch) {
+ synchronized (tsFileOrderingLock) {
+ if (inFlightTsFileEpoch == tsFileEpoch) {
+ inFlightTsFileEpoch = null;
+ }
+ }
+ }
}
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/PipeRealtimeExtractTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/PipeRealtimeExtractTest.java
index fc4a15a885b..da19dd72426 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/PipeRealtimeExtractTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/PipeRealtimeExtractTest.java
@@ -19,6 +19,7 @@
package org.apache.iotdb.db.pipe.source;
+import org.apache.iotdb.commons.conf.CommonDescriptor;
import org.apache.iotdb.commons.conf.IoTDBConstant;
import org.apache.iotdb.commons.consensus.index.impl.MinimumProgressIndex;
import org.apache.iotdb.commons.path.PartialPath;
@@ -29,6 +30,7 @@ import
org.apache.iotdb.commons.pipe.agent.task.meta.PipeRuntimeMeta;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeStaticMeta;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTemporaryMetaInAgent;
+import
org.apache.iotdb.commons.pipe.agent.task.progress.PipeEventCommitManager;
import org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant;
import
org.apache.iotdb.commons.pipe.config.plugin.configuraion.PipeTaskRuntimeConfiguration;
import
org.apache.iotdb.commons.pipe.config.plugin.env.PipeTaskSourceRuntimeEnvironment;
@@ -38,12 +40,16 @@ import
org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId;
import org.apache.iotdb.commons.utils.FileUtils;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.pipe.agent.PipeDataNodeAgent;
+import
org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent;
+import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent;
import org.apache.iotdb.db.pipe.event.realtime.PipeRealtimeEvent;
import org.apache.iotdb.db.pipe.event.realtime.PipeRealtimeEventFactory;
import
org.apache.iotdb.db.pipe.source.dataregion.realtime.PipeRealtimeDataRegionHybridSource;
import
org.apache.iotdb.db.pipe.source.dataregion.realtime.PipeRealtimeDataRegionLogSource;
import
org.apache.iotdb.db.pipe.source.dataregion.realtime.PipeRealtimeDataRegionSource;
import
org.apache.iotdb.db.pipe.source.dataregion.realtime.PipeRealtimeDataRegionTsFileSource;
+import
org.apache.iotdb.db.pipe.source.dataregion.realtime.assigner.PipeTsFileEpochProgressIndexKeeper;
+import org.apache.iotdb.db.pipe.source.dataregion.realtime.epoch.TsFileEpoch;
import
org.apache.iotdb.db.pipe.source.dataregion.realtime.listener.PipeInsertionDataNodeListener;
import
org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowNode;
import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
@@ -56,6 +62,8 @@ import
org.apache.iotdb.pipe.api.event.dml.insertion.TsFileInsertionEvent;
import org.apache.tsfile.common.constant.TsFileConstant;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.file.metadata.IDeviceID;
+import org.apache.tsfile.write.record.Tablet;
+import org.apache.tsfile.write.schema.MeasurementSchema;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
@@ -68,8 +76,10 @@ import java.io.IOException;
import java.lang.reflect.Field;
import java.nio.file.Files;
import java.util.Arrays;
+import java.util.Collections;
import java.util.HashMap;
import java.util.List;
+import java.util.Set;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@@ -99,11 +109,14 @@ public class PipeRealtimeExtractTest {
private ExecutorService writeService;
private ExecutorService listenerService;
private int dataNodeId;
+ private double pipeTotalFloatingMemoryProportion;
@Before
public void setUp() throws Exception {
dataNodeId = IoTDBDescriptor.getInstance().getConfig().getDataNodeId();
IoTDBDescriptor.getInstance().getConfig().setDataNodeId(0);
+ pipeTotalFloatingMemoryProportion =
+
CommonDescriptor.getInstance().getConfig().getPipeTotalFloatingMemoryProportion();
removeTestPipeMeta();
writeService = Executors.newFixedThreadPool(2);
listenerService = Executors.newFixedThreadPool(4);
@@ -120,6 +133,9 @@ public class PipeRealtimeExtractTest {
@After
public void tearDown() throws Exception {
IoTDBDescriptor.getInstance().getConfig().setDataNodeId(dataNodeId);
+ CommonDescriptor.getInstance()
+ .getConfig()
+
.setPipeTotalFloatingMemoryProportion(pipeTotalFloatingMemoryProportion);
writeService.shutdownNow();
listenerService.shutdownNow();
FileUtils.deleteFileOrDirectory(tmpDir);
@@ -379,6 +395,1163 @@ public class PipeRealtimeExtractTest {
Assert.assertNull(getGlobalTsFileEpochDegraded());
}
+ @Test
+ public void testHybridSourceRegionLevelDowngradingIsPipeSpecific() throws
Exception {
+ try (final PipeRealtimeDataRegionHybridSource disabledExtractor =
+ new PipeRealtimeDataRegionHybridSource();
+ final PipeRealtimeDataRegionHybridSource enabledExtractor =
+ new PipeRealtimeDataRegionHybridSource()) {
+ final PipeParameters disabledParameters =
+ new PipeParameters(
+ new HashMap<String, String>() {
+ {
+ put(PipeSourceConstant.EXTRACTOR_PATTERN_KEY, pattern1);
+ }
+ });
+ final PipeParameters enabledParameters =
+ new PipeParameters(
+ new HashMap<String, String>() {
+ {
+ put(PipeSourceConstant.EXTRACTOR_PATTERN_KEY, pattern1);
+ put(
+
PipeSourceConstant.EXTRACTOR_REALTIME_REGION_LEVEL_DOWNGRADING_KEY,
+ Boolean.TRUE.toString());
+ }
+ });
+
+ final PipeTaskRuntimeConfiguration disabledConfiguration =
+ new PipeTaskRuntimeConfiguration(
+ new PipeTaskSourceRuntimeEnvironment(
+ "region-level-downgrading-disabled",
+ TEST_PIPE_CREATION_TIME,
+ dataRegion1,
+ new PipeTaskMeta(MinimumProgressIndex.INSTANCE, 1)));
+ final PipeTaskRuntimeConfiguration enabledConfiguration =
+ new PipeTaskRuntimeConfiguration(
+ new PipeTaskSourceRuntimeEnvironment(
+ "region-level-downgrading-enabled",
+ TEST_PIPE_CREATION_TIME,
+ dataRegion1,
+ new PipeTaskMeta(MinimumProgressIndex.INSTANCE, 1)));
+
+ disabledExtractor.validate(new
PipeParameterValidator(disabledParameters));
+ disabledExtractor.customize(disabledParameters, disabledConfiguration);
+ enabledExtractor.validate(new PipeParameterValidator(enabledParameters));
+ enabledExtractor.customize(enabledParameters, enabledConfiguration);
+
+ Assert.assertFalse(isRegionLevelDowngradingEnabled(disabledExtractor));
+ Assert.assertTrue(isRegionLevelDowngradingEnabled(enabledExtractor));
+ }
+ }
+
+ @Test
+ public void testHybridSourceRegionLevelDowngradingWaitsForTsFileCommit()
throws Exception {
+ registerTestPipeMeta();
+
+ final PipeEventCommitManager commitManager =
PipeEventCommitManager.getInstance();
+ commitManager.register(TEST_PIPE_NAME, TEST_PIPE_CREATION_TIME,
dataRegion1, "test");
+ try (final PipeRealtimeDataRegionHybridSource extractor =
+ new PipeRealtimeDataRegionHybridSource()) {
+ final PipeParameters parameters =
+ new PipeParameters(
+ new HashMap<String, String>() {
+ {
+ put(PipeSourceConstant.EXTRACTOR_PATTERN_KEY, pattern1);
+ put(
+
PipeSourceConstant.SOURCE_REALTIME_REGION_LEVEL_DOWNGRADING_KEY,
+ Boolean.TRUE.toString());
+ }
+ });
+ final PipeTaskMeta pipeTaskMeta = new
PipeTaskMeta(MinimumProgressIndex.INSTANCE, 1);
+ final PipeTaskRuntimeConfiguration configuration =
+ new PipeTaskRuntimeConfiguration(
+ new PipeTaskSourceRuntimeEnvironment(
+ TEST_PIPE_NAME, TEST_PIPE_CREATION_TIME, dataRegion1,
pipeTaskMeta));
+
+ extractor.validate(new PipeParameterValidator(parameters));
+ extractor.customize(parameters, configuration);
+
+ final TsFileResource firstResource = createTsFileResource(dataRegion1,
"101-101-0-0.tsfile");
+ final PipeRealtimeEvent firstTabletEvent =
+ bindToTestPipe(
+ PipeRealtimeEventFactory.createRealtimeEvent(
+ false,
+ "root.sg",
+ createInsertRowNode("first-degraded-tablet", "a"),
+ firstResource),
+ extractor,
+ pipeTaskMeta);
+
+
Assert.assertTrue(firstTabletEvent.increaseReferenceCount(TEST_REFERENCE_HOLDER));
+ extractor.extract(firstTabletEvent);
+ Assert.assertEquals(Boolean.FALSE, getGlobalTsFileEpochDegraded());
+
+ firstTabletEvent.clearReferenceCount(TEST_REFERENCE_HOLDER);
+
+ // Queue a tablet from another epoch before the first epoch triggers
region-level
+ // downgrading. It should be buffered while the degraded TsFile is being
sent.
+ final TsFileResource secondResource = createTsFileResource(dataRegion1,
"102-102-0-0.tsfile");
+ final PipeRealtimeEvent secondTabletEvent =
+ bindToTestPipe(
+ PipeRealtimeEventFactory.createRealtimeEvent(
+ false,
+ "root.sg",
+ createInsertRowNode("second-degraded-tablet", "a"),
+ secondResource),
+ extractor,
+ pipeTaskMeta);
+
Assert.assertTrue(secondTabletEvent.increaseReferenceCount(TEST_REFERENCE_HOLDER));
+ extractor.extract(secondTabletEvent);
+ Assert.assertEquals(
+ TsFileEpoch.State.USING_TABLET,
secondTabletEvent.getTsFileEpoch().getState(extractor));
+
+ Assert.assertNull(extractor.supply());
+ Assert.assertEquals(Boolean.TRUE, getGlobalTsFileEpochDegraded());
+ Assert.assertEquals(
+ TsFileEpoch.State.USING_TABLET,
secondTabletEvent.getTsFileEpoch().getState(extractor));
+ Assert.assertFalse(secondTabletEvent.getEvent().isReleased());
+
+ // Simulate that the buffered tablet is evicted before the previous
degraded TsFile is
+ // committed. The second epoch should then continue region-level
downgrading with its TsFile.
+ secondTabletEvent.clearReferenceCount(TEST_REFERENCE_HOLDER);
+
+ final PipeRealtimeEvent firstTsFileEvent =
+ bindToTestPipe(
+ PipeRealtimeEventFactory.createRealtimeEvent(false, "root.sg",
firstResource, false),
+ extractor,
+ pipeTaskMeta);
+
Assert.assertTrue(firstTsFileEvent.increaseReferenceCount(TEST_REFERENCE_HOLDER));
+ extractor.extract(firstTsFileEvent);
+ Assert.assertEquals(
+ TsFileEpoch.State.USING_TSFILE,
firstTsFileEvent.getTsFileEpoch().getState(extractor));
+
+ final Event firstSuppliedTsFile = extractor.supply();
+ Assert.assertTrue(firstSuppliedTsFile instanceof TsFileInsertionEvent);
+ Assert.assertEquals(Boolean.TRUE, getGlobalTsFileEpochDegraded());
+
+ final PipeRealtimeEvent secondTsFileEvent =
+ bindToTestPipe(
+ PipeRealtimeEventFactory.createRealtimeEvent(false, "root.sg",
secondResource, false),
+ extractor,
+ pipeTaskMeta);
+
Assert.assertTrue(secondTsFileEvent.increaseReferenceCount(TEST_REFERENCE_HOLDER));
+ extractor.extract(secondTsFileEvent);
+
+ // The second TsFile stays in the source until the first TsFile is
committed downstream.
+ Assert.assertNull(extractor.supply());
+ final PipeTsFileInsertionEvent suppliedFirstTsFile =
+ (PipeTsFileInsertionEvent) firstSuppliedTsFile;
+ suppliedFirstTsFile.registerGeneratedTabletInsertionEvent();
+ suppliedFirstTsFile.registerGeneratedTabletInsertionEvent();
+ suppliedFirstTsFile.markGeneratedTabletInsertionEventsParsingCompleted();
+ final PipeRawTabletInsertionEvent firstGeneratedTabletEvent =
+ createGeneratedTabletEvent(suppliedFirstTsFile, pipeTaskMeta,
"first-generated");
+ final PipeRawTabletInsertionEvent secondGeneratedTabletEvent =
+ createGeneratedTabletEvent(suppliedFirstTsFile, pipeTaskMeta,
"second-generated");
+
Assert.assertTrue(firstGeneratedTabletEvent.increaseReferenceCount(TEST_REFERENCE_HOLDER));
+
Assert.assertTrue(secondGeneratedTabletEvent.increaseReferenceCount(TEST_REFERENCE_HOLDER));
+
Assert.assertTrue(suppliedFirstTsFile.decreaseReferenceCount(TEST_REFERENCE_HOLDER,
false));
+ commitSuppliedEvent(firstGeneratedTabletEvent, commitManager);
+ Assert.assertEquals(Boolean.TRUE, getGlobalTsFileEpochDegraded());
+ Assert.assertEquals(2, getActiveTsFileEpochCount(extractor));
+ Assert.assertEquals(1, getInFlightTsFileCount(extractor));
+ commitSuppliedEvent(secondGeneratedTabletEvent, commitManager);
+ Assert.assertEquals(1, getActiveTsFileEpochCount(extractor));
+ Assert.assertEquals(0, getInFlightTsFileCount(extractor));
+ Assert.assertEquals(Boolean.TRUE, getGlobalTsFileEpochDegraded());
+
+ final Event secondSuppliedTsFile = extractor.supply();
+ Assert.assertTrue(secondSuppliedTsFile instanceof TsFileInsertionEvent);
+ Assert.assertEquals(Boolean.TRUE, getGlobalTsFileEpochDegraded());
+
+ commitSuppliedEvent(secondSuppliedTsFile, commitManager);
+ Assert.assertEquals(0, getActiveTsFileEpochCount(extractor));
+ Assert.assertEquals(0, getInFlightTsFileCount(extractor));
+ Assert.assertNull(getGlobalTsFileEpochDegraded());
+ Assert.assertNull(extractor.supply());
+ } finally {
+ commitManager.deregister(TEST_PIPE_NAME, TEST_PIPE_CREATION_TIME,
dataRegion1);
+ }
+ }
+
+ @Test
+ public void testGeneratedTabletTransferWaitsForAllTabletCommits() throws
Exception {
+ registerTestPipeMeta();
+
+ final PipeEventCommitManager commitManager =
PipeEventCommitManager.getInstance();
+ final String dedupScopeId = "generated-tablet-transfer-test";
+ commitManager.register(TEST_PIPE_NAME, TEST_PIPE_CREATION_TIME,
dataRegion1, "test");
+ try {
+ final PipeTaskMeta pipeTaskMeta = new
PipeTaskMeta(MinimumProgressIndex.INSTANCE, 1);
+ final TsFileResource resource = createTsFileResource(dataRegion1,
"110-110-0-0.tsfile");
+ final PipeTsFileInsertionEvent tsFileEvent =
+ new PipeTsFileInsertionEvent(false, "root.sg", resource, false)
+ .shallowCopySelfAndBindPipeTaskMetaForProgressReport(
+ TEST_PIPE_NAME,
+ TEST_PIPE_CREATION_TIME,
+ pipeTaskMeta,
+ null,
+ null,
+ null,
+ null,
+ null,
+ true,
+ Long.MIN_VALUE,
+ Long.MAX_VALUE);
+ tsFileEvent.bindTsFileDedupScopeID(dedupScopeId);
+ PipeTsFileEpochProgressIndexKeeper.getInstance()
+ .registerProgressIndex(dataRegion1, dedupScopeId, resource);
+
+ final AtomicBoolean transferred = new AtomicBoolean(false);
+ tsFileEvent.addOnTransferredHook(() -> transferred.set(true));
+ tsFileEvent.registerGeneratedTabletInsertionEvent();
+ tsFileEvent.registerGeneratedTabletInsertionEvent();
+
+ // The default PipeProcessor path iterates toTabletInsertionEvents()
directly and does not
+ // report parser completion. The source TsFile commit is still the
boundary before generated
+ // tablet commits.
+ tsFileEvent.skipReportOnCommit();
+ tsFileEvent.getOnCommittedHooks().forEach(Runnable::run);
+
+ final PipeRawTabletInsertionEvent firstGeneratedTabletEvent =
+ createGeneratedTabletEvent(tsFileEvent, pipeTaskMeta, "first",
false);
+ final PipeRawTabletInsertionEvent secondGeneratedTabletEvent =
+ createGeneratedTabletEvent(tsFileEvent, pipeTaskMeta, "second",
true);
+
Assert.assertTrue(firstGeneratedTabletEvent.increaseReferenceCount(TEST_REFERENCE_HOLDER));
+
Assert.assertTrue(secondGeneratedTabletEvent.increaseReferenceCount(TEST_REFERENCE_HOLDER));
+
+ commitSuppliedEvent(firstGeneratedTabletEvent, commitManager);
+ Assert.assertFalse(transferred.get());
+ Assert.assertTrue(
+ PipeTsFileEpochProgressIndexKeeper.getInstance()
+ .containsTsFile(dataRegion1, dedupScopeId,
resource.getTsFilePath()));
+
+ commitSuppliedEvent(secondGeneratedTabletEvent, commitManager);
+ Assert.assertTrue(transferred.get());
+ Assert.assertFalse(
+ PipeTsFileEpochProgressIndexKeeper.getInstance()
+ .containsTsFile(dataRegion1, dedupScopeId,
resource.getTsFilePath()));
+ } finally {
+ commitManager.deregister(TEST_PIPE_NAME, TEST_PIPE_CREATION_TIME,
dataRegion1);
+ PipeTsFileEpochProgressIndexKeeper.getInstance()
+ .clearProgressIndex(dataRegion1, dedupScopeId);
+ }
+ }
+
+ @Test
+ public void testGeneratedTabletTransferWaitsForDeferredGeneration() throws
Exception {
+ registerTestPipeMeta();
+
+ final PipeEventCommitManager commitManager =
PipeEventCommitManager.getInstance();
+ commitManager.register(TEST_PIPE_NAME, TEST_PIPE_CREATION_TIME,
dataRegion1, "test");
+ final String dedupScopeId = "deferred-generated-tablet-transfer-test";
+ try {
+ final PipeTaskMeta pipeTaskMeta = new
PipeTaskMeta(MinimumProgressIndex.INSTANCE, 1);
+ final TsFileResource resource = createTsFileResource(dataRegion1,
"111-111-0-0.tsfile");
+ final PipeTsFileInsertionEvent tsFileEvent =
+ new PipeTsFileInsertionEvent(false, "root.sg", resource, false)
+ .shallowCopySelfAndBindPipeTaskMetaForProgressReport(
+ TEST_PIPE_NAME,
+ TEST_PIPE_CREATION_TIME,
+ pipeTaskMeta,
+ null,
+ null,
+ null,
+ null,
+ null,
+ true,
+ Long.MIN_VALUE,
+ Long.MAX_VALUE);
+ tsFileEvent.bindTsFileDedupScopeID(dedupScopeId);
+
+ final AtomicBoolean transferred = new AtomicBoolean(false);
+ tsFileEvent.addOnTransferredHook(() -> transferred.set(true));
+ tsFileEvent.markGeneratedTabletInsertionEventsParsingStarted();
+ tsFileEvent.skipReportOnCommit();
+ tsFileEvent.getOnCommittedHooks().forEach(Runnable::run);
+ Assert.assertFalse(transferred.get());
+
+ tsFileEvent.registerGeneratedTabletInsertionEvent();
+ tsFileEvent.markGeneratedTabletInsertionEventsParsingCompleted();
+ final PipeRawTabletInsertionEvent generatedTabletEvent =
+ createGeneratedTabletEvent(tsFileEvent, pipeTaskMeta, "deferred");
+
Assert.assertTrue(generatedTabletEvent.increaseReferenceCount(TEST_REFERENCE_HOLDER));
+ commitSuppliedEvent(generatedTabletEvent, commitManager);
+ Assert.assertTrue(transferred.get());
+ } finally {
+ commitManager.deregister(TEST_PIPE_NAME, TEST_PIPE_CREATION_TIME,
dataRegion1);
+ PipeTsFileEpochProgressIndexKeeper.getInstance()
+ .clearProgressIndex(dataRegion1, dedupScopeId);
+ }
+ }
+
+ @Test
+ public void
testHybridSourceClearsInFlightTsFileWhenSuppliedEventIsDiscarded() throws
Exception {
+ registerTestPipeMeta();
+
+ final PipeEventCommitManager commitManager =
PipeEventCommitManager.getInstance();
+ commitManager.register(TEST_PIPE_NAME, TEST_PIPE_CREATION_TIME,
dataRegion1, "test");
+ try (final PipeRealtimeDataRegionHybridSource extractor =
+ new PipeRealtimeDataRegionHybridSource()) {
+ final PipeTaskMeta pipeTaskMeta = new
PipeTaskMeta(MinimumProgressIndex.INSTANCE, 1);
+ final PipeTaskRuntimeConfiguration configuration =
+ new PipeTaskRuntimeConfiguration(
+ new PipeTaskSourceRuntimeEnvironment(
+ TEST_PIPE_NAME, TEST_PIPE_CREATION_TIME, dataRegion1,
pipeTaskMeta));
+ final PipeParameters parameters =
+ new PipeParameters(
+ new HashMap<String, String>() {
+ {
+ put(PipeSourceConstant.EXTRACTOR_PATTERN_KEY, pattern1);
+ put(
+
PipeSourceConstant.SOURCE_REALTIME_REGION_LEVEL_DOWNGRADING_KEY,
+ Boolean.TRUE.toString());
+ }
+ });
+ extractor.validate(new PipeParameterValidator(parameters));
+ extractor.customize(parameters, configuration);
+
+ final TsFileResource resource = createTsFileResource(dataRegion1,
"112-112-0-0.tsfile");
+ final PipeRealtimeEvent tabletEvent =
+ bindToTestPipe(
+ PipeRealtimeEventFactory.createRealtimeEvent(
+ false, "root.sg",
createInsertRowNode("discarded-tsfile-tablet", "a"), resource),
+ extractor,
+ pipeTaskMeta);
+
Assert.assertTrue(tabletEvent.increaseReferenceCount(TEST_REFERENCE_HOLDER));
+ extractor.extract(tabletEvent);
+ tabletEvent.clearReferenceCount(TEST_REFERENCE_HOLDER);
+ Assert.assertNull(extractor.supply());
+ Assert.assertEquals(Boolean.TRUE, getGlobalTsFileEpochDegraded());
+
+ final PipeRealtimeEvent tsFileEvent =
+ bindToTestPipe(
+ PipeRealtimeEventFactory.createRealtimeEvent(false, "root.sg",
resource, false),
+ extractor,
+ pipeTaskMeta);
+
Assert.assertTrue(tsFileEvent.increaseReferenceCount(TEST_REFERENCE_HOLDER));
+ extractor.extract(tsFileEvent);
+
+ final Event suppliedTsFile = extractor.supply();
+ Assert.assertTrue(suppliedTsFile instanceof TsFileInsertionEvent);
+ Assert.assertEquals(1, getInFlightTsFileCount(extractor));
+
+ ((EnrichedEvent)
suppliedTsFile).clearReferenceCount(TEST_REFERENCE_HOLDER);
+ Assert.assertEquals(0, getInFlightTsFileCount(extractor));
+ Assert.assertNull(getGlobalTsFileEpochDegraded());
+ } finally {
+ commitManager.deregister(TEST_PIPE_NAME, TEST_PIPE_CREATION_TIME,
dataRegion1);
+ }
+ }
+
+ @Test
+ public void testHybridSourceCompensatesForDiscardedGeneratedTabletEvents()
throws Exception {
+ registerTestPipeMeta();
+
+ final PipeEventCommitManager commitManager =
PipeEventCommitManager.getInstance();
+ commitManager.register(TEST_PIPE_NAME, TEST_PIPE_CREATION_TIME,
dataRegion1, "test");
+ try (final PipeRealtimeDataRegionHybridSource extractor =
+ new PipeRealtimeDataRegionHybridSource()) {
+ final PipeTaskMeta pipeTaskMeta = new
PipeTaskMeta(MinimumProgressIndex.INSTANCE, 1);
+ final PipeTaskRuntimeConfiguration configuration =
+ new PipeTaskRuntimeConfiguration(
+ new PipeTaskSourceRuntimeEnvironment(
+ TEST_PIPE_NAME, TEST_PIPE_CREATION_TIME, dataRegion1,
pipeTaskMeta));
+ final PipeParameters parameters =
+ new PipeParameters(
+ new HashMap<String, String>() {
+ {
+ put(PipeSourceConstant.EXTRACTOR_PATTERN_KEY, pattern1);
+ put(
+
PipeSourceConstant.SOURCE_REALTIME_REGION_LEVEL_DOWNGRADING_KEY,
+ Boolean.TRUE.toString());
+ }
+ });
+ extractor.validate(new PipeParameterValidator(parameters));
+ extractor.customize(parameters, configuration);
+
+ final TsFileResource resource = createTsFileResource(dataRegion1,
"113-113-0-0.tsfile");
+ final PipeRealtimeEvent tabletEvent =
+ bindToTestPipe(
+ PipeRealtimeEventFactory.createRealtimeEvent(
+ false,
+ "root.sg",
+ createInsertRowNode("discarded-generated-tablet", "a"),
+ resource),
+ extractor,
+ pipeTaskMeta);
+
Assert.assertTrue(tabletEvent.increaseReferenceCount(TEST_REFERENCE_HOLDER));
+ extractor.extract(tabletEvent);
+ tabletEvent.clearReferenceCount(TEST_REFERENCE_HOLDER);
+ Assert.assertNull(extractor.supply());
+
+ final PipeRealtimeEvent tsFileRealtimeEvent =
+ bindToTestPipe(
+ PipeRealtimeEventFactory.createRealtimeEvent(false, "root.sg",
resource, false),
+ extractor,
+ pipeTaskMeta);
+
Assert.assertTrue(tsFileRealtimeEvent.increaseReferenceCount(TEST_REFERENCE_HOLDER));
+ extractor.extract(tsFileRealtimeEvent);
+ final PipeTsFileInsertionEvent suppliedTsFile =
(PipeTsFileInsertionEvent) extractor.supply();
+ Assert.assertNotNull(suppliedTsFile);
+ Assert.assertEquals(1, getInFlightTsFileCount(extractor));
+
+ suppliedTsFile.registerGeneratedTabletInsertionEvent();
+ suppliedTsFile.registerGeneratedTabletInsertionEvent();
+ suppliedTsFile.markGeneratedTabletInsertionEventsParsingCompleted();
+ final PipeRawTabletInsertionEvent firstGeneratedTablet =
+ createGeneratedTabletEvent(suppliedTsFile, pipeTaskMeta,
"discarded-first");
+ final PipeRawTabletInsertionEvent secondGeneratedTablet =
+ createGeneratedTabletEvent(suppliedTsFile, pipeTaskMeta,
"discarded-second");
+ firstGeneratedTablet.markAsGeneratedEventRegisteredWithSource();
+ secondGeneratedTablet.markAsGeneratedEventRegisteredWithSource();
+
Assert.assertTrue(firstGeneratedTablet.increaseReferenceCount(TEST_REFERENCE_HOLDER));
+
Assert.assertTrue(secondGeneratedTablet.increaseReferenceCount(TEST_REFERENCE_HOLDER));
+
+ commitManager.enrichWithCommitterKeyAndCommitId(
+ suppliedTsFile, TEST_PIPE_CREATION_TIME, dataRegion1);
+
Assert.assertTrue(suppliedTsFile.decreaseReferenceCount(TEST_REFERENCE_HOLDER,
false));
+ firstGeneratedTablet.clearReferenceCount(TEST_REFERENCE_HOLDER);
+ Assert.assertEquals(1, getInFlightTsFileCount(extractor));
+ secondGeneratedTablet.clearReferenceCount(TEST_REFERENCE_HOLDER);
+
+ Assert.assertEquals(0, getInFlightTsFileCount(extractor));
+ Assert.assertNull(getGlobalTsFileEpochDegraded());
+ } finally {
+ commitManager.deregister(TEST_PIPE_NAME, TEST_PIPE_CREATION_TIME,
dataRegion1);
+ }
+ }
+
+ @Test
+ public void
testHybridSourceRegionLevelDowngradingResumesCompleteBufferedTablets()
+ throws Exception {
+ registerTestPipeMeta();
+
+ final PipeEventCommitManager commitManager =
PipeEventCommitManager.getInstance();
+ commitManager.register(TEST_PIPE_NAME, TEST_PIPE_CREATION_TIME,
dataRegion1, "test");
+ try (final PipeRealtimeDataRegionHybridSource extractor =
+ new PipeRealtimeDataRegionHybridSource()) {
+ final PipeParameters parameters =
+ new PipeParameters(
+ new HashMap<String, String>() {
+ {
+ put(PipeSourceConstant.EXTRACTOR_PATTERN_KEY, pattern1);
+ put(
+
PipeSourceConstant.SOURCE_REALTIME_REGION_LEVEL_DOWNGRADING_KEY,
+ Boolean.TRUE.toString());
+ }
+ });
+ final PipeTaskMeta pipeTaskMeta = new
PipeTaskMeta(MinimumProgressIndex.INSTANCE, 1);
+ final PipeTaskRuntimeConfiguration configuration =
+ new PipeTaskRuntimeConfiguration(
+ new PipeTaskSourceRuntimeEnvironment(
+ TEST_PIPE_NAME, TEST_PIPE_CREATION_TIME, dataRegion1,
pipeTaskMeta));
+
+ extractor.validate(new PipeParameterValidator(parameters));
+ extractor.customize(parameters, configuration);
+
+ final TsFileResource firstResource = createTsFileResource(dataRegion1,
"103-103-0-0.tsfile");
+ final PipeRealtimeEvent firstTabletEvent =
+ bindToTestPipe(
+ PipeRealtimeEventFactory.createRealtimeEvent(
+ false,
+ "root.sg",
+ createInsertRowNode("first-degraded-tablet", "a"),
+ firstResource),
+ extractor,
+ pipeTaskMeta);
+
Assert.assertTrue(firstTabletEvent.increaseReferenceCount(TEST_REFERENCE_HOLDER));
+ extractor.extract(firstTabletEvent);
+ firstTabletEvent.clearReferenceCount(TEST_REFERENCE_HOLDER);
+
+ final TsFileResource secondResource = createTsFileResource(dataRegion1,
"104-104-0-0.tsfile");
+ final PipeRealtimeEvent secondTabletEvent =
+ bindToTestPipe(
+ PipeRealtimeEventFactory.createRealtimeEvent(
+ false,
+ "root.sg",
+ createInsertRowNode("fully-buffered-tablet", "a"),
+ secondResource),
+ extractor,
+ pipeTaskMeta);
+
Assert.assertTrue(secondTabletEvent.increaseReferenceCount(TEST_REFERENCE_HOLDER));
+ extractor.extract(secondTabletEvent);
+
+ Assert.assertNull(extractor.supply());
+ Assert.assertEquals(Boolean.TRUE, getGlobalTsFileEpochDegraded());
+ Assert.assertEquals(
+ TsFileEpoch.State.USING_TABLET,
secondTabletEvent.getTsFileEpoch().getState(extractor));
+ Assert.assertFalse(secondTabletEvent.getEvent().isReleased());
+
+ final PipeRealtimeEvent firstTsFileEvent =
+ bindToTestPipe(
+ PipeRealtimeEventFactory.createRealtimeEvent(false, "root.sg",
firstResource, false),
+ extractor,
+ pipeTaskMeta);
+
Assert.assertTrue(firstTsFileEvent.increaseReferenceCount(TEST_REFERENCE_HOLDER));
+ extractor.extract(firstTsFileEvent);
+ final Event firstSuppliedTsFile = extractor.supply();
+ Assert.assertTrue(firstSuppliedTsFile instanceof TsFileInsertionEvent);
+
+ commitLastGeneratedTabletEvent(
+ (PipeTsFileInsertionEvent) firstSuppliedTsFile, commitManager,
pipeTaskMeta);
+ Assert.assertEquals(Boolean.FALSE, getGlobalTsFileEpochDegraded());
+
+ // The latest TsFile is still open. Since all of its requests survived
in memory at the
+ // commit boundary above, later writes of the same TsFile should
immediately continue on the
+ // realtime path instead of waiting for another flush.
+ final PipeRealtimeEvent newRealtimeTabletEvent =
+ bindToTestPipe(
+ PipeRealtimeEventFactory.createRealtimeEvent(
+ false,
+ "root.sg",
+ createInsertRowNode("new-realtime-tablet", "a"),
+ secondResource),
+ extractor,
+ pipeTaskMeta);
+
Assert.assertTrue(newRealtimeTabletEvent.increaseReferenceCount(TEST_REFERENCE_HOLDER));
+ extractor.extract(newRealtimeTabletEvent);
+
+ final Event resumedTabletEvent = extractor.supply();
+ Assert.assertTrue(resumedTabletEvent instanceof TabletInsertionEvent);
+ Assert.assertSame(secondTabletEvent.getEvent(), resumedTabletEvent);
+ Assert.assertEquals(Boolean.FALSE, getGlobalTsFileEpochDegraded());
+ commitSuppliedEvent(resumedTabletEvent, commitManager);
+
+ final Event newSuppliedTabletEvent = extractor.supply();
+ Assert.assertTrue(newSuppliedTabletEvent instanceof
TabletInsertionEvent);
+ Assert.assertSame(newRealtimeTabletEvent.getEvent(),
newSuppliedTabletEvent);
+ Assert.assertEquals(Boolean.FALSE, getGlobalTsFileEpochDegraded());
+ commitSuppliedEvent(newSuppliedTabletEvent, commitManager);
+
+ final PipeRealtimeEvent secondTsFileEvent =
+ bindToTestPipe(
+ PipeRealtimeEventFactory.createRealtimeEvent(false, "root.sg",
secondResource, false),
+ extractor,
+ pipeTaskMeta);
+
Assert.assertTrue(secondTsFileEvent.increaseReferenceCount(TEST_REFERENCE_HOLDER));
+ extractor.extract(secondTsFileEvent);
+
+ // The second TsFile is no longer needed because all of its tablets
survived buffering.
+ Assert.assertNull(extractor.supply());
+ Assert.assertNull(getGlobalTsFileEpochDegraded());
+ } finally {
+ commitManager.deregister(TEST_PIPE_NAME, TEST_PIPE_CREATION_TIME,
dataRegion1);
+ }
+ }
+
+ @Test
+ public void testHybridSourceRegionLevelDowngradingOnlyCachesLatestTsFile()
throws Exception {
+ registerTestPipeMeta();
+
+ final PipeEventCommitManager commitManager =
PipeEventCommitManager.getInstance();
+ commitManager.register(TEST_PIPE_NAME, TEST_PIPE_CREATION_TIME,
dataRegion1, "test");
+ try (final PipeRealtimeDataRegionHybridSource extractor =
+ new PipeRealtimeDataRegionHybridSource()) {
+ final PipeParameters parameters =
+ new PipeParameters(
+ new HashMap<String, String>() {
+ {
+ put(PipeSourceConstant.EXTRACTOR_PATTERN_KEY, pattern1);
+ put(
+
PipeSourceConstant.SOURCE_REALTIME_REGION_LEVEL_DOWNGRADING_KEY,
+ Boolean.TRUE.toString());
+ }
+ });
+ final PipeTaskMeta pipeTaskMeta = new
PipeTaskMeta(MinimumProgressIndex.INSTANCE, 1);
+ final PipeTaskRuntimeConfiguration configuration =
+ new PipeTaskRuntimeConfiguration(
+ new PipeTaskSourceRuntimeEnvironment(
+ TEST_PIPE_NAME, TEST_PIPE_CREATION_TIME, dataRegion1,
pipeTaskMeta));
+
+ extractor.validate(new PipeParameterValidator(parameters));
+ extractor.customize(parameters, configuration);
+
+ final TsFileResource firstResource = createTsFileResource(dataRegion1,
"107-107-0-0.tsfile");
+ final PipeRealtimeEvent firstTabletEvent =
+ bindToTestPipe(
+ PipeRealtimeEventFactory.createRealtimeEvent(
+ false,
+ "root.sg",
+ createInsertRowNode("first-degraded-tablet", "a"),
+ firstResource),
+ extractor,
+ pipeTaskMeta);
+
Assert.assertTrue(firstTabletEvent.increaseReferenceCount(TEST_REFERENCE_HOLDER));
+ extractor.extract(firstTabletEvent);
+ firstTabletEvent.clearReferenceCount(TEST_REFERENCE_HOLDER);
+
+ Assert.assertNull(extractor.supply());
+ Assert.assertEquals(Boolean.TRUE, getGlobalTsFileEpochDegraded());
+
+ final TsFileResource secondResource = createTsFileResource(dataRegion1,
"108-108-0-0.tsfile");
+ final PipeRealtimeEvent secondTabletEvent =
+ bindToTestPipe(
+ PipeRealtimeEventFactory.createRealtimeEvent(
+ false,
+ "root.sg",
+ createInsertRowNode("second-buffered-tablet", "a"),
+ secondResource),
+ extractor,
+ pipeTaskMeta);
+
Assert.assertTrue(secondTabletEvent.increaseReferenceCount(TEST_REFERENCE_HOLDER));
+ extractor.extract(secondTabletEvent);
+
+ final PipeRealtimeEvent secondTsFileEvent =
+ bindToTestPipe(
+ PipeRealtimeEventFactory.createRealtimeEvent(false, "root.sg",
secondResource, false),
+ extractor,
+ pipeTaskMeta);
+
Assert.assertTrue(secondTsFileEvent.increaseReferenceCount(TEST_REFERENCE_HOLDER));
+ extractor.extract(secondTsFileEvent);
+
+ final TsFileResource thirdResource = createTsFileResource(dataRegion1,
"109-109-0-0.tsfile");
+ final PipeRealtimeEvent thirdTabletEvent =
+ bindToTestPipe(
+ PipeRealtimeEventFactory.createRealtimeEvent(
+ false,
+ "root.sg",
+ createInsertRowNode("latest-buffered-tablet", "a"),
+ thirdResource),
+ extractor,
+ pipeTaskMeta);
+
Assert.assertTrue(thirdTabletEvent.increaseReferenceCount(TEST_REFERENCE_HOLDER));
+ extractor.extract(thirdTabletEvent);
+
+ // Once a newer epoch appears, the former tail is downgraded even if all
its tablets are
+ // still available. This bounds the region-level cache to the latest
TsFile.
+ Assert.assertEquals(
+ TsFileEpoch.State.USING_TSFILE,
secondTabletEvent.getTsFileEpoch().getState(extractor));
+ Assert.assertTrue(secondTabletEvent.getEvent().isReleased());
+ Assert.assertEquals(
+ TsFileEpoch.State.USING_TABLET,
thirdTabletEvent.getTsFileEpoch().getState(extractor));
+ Assert.assertFalse(thirdTabletEvent.getEvent().isReleased());
+
+ // Extract the first TsFile after the second one to verify that query
priority, rather than
+ // flush completion order, decides which file can pass downstream.
+ final PipeRealtimeEvent firstTsFileEvent =
+ bindToTestPipe(
+ PipeRealtimeEventFactory.createRealtimeEvent(false, "root.sg",
firstResource, false),
+ extractor,
+ pipeTaskMeta);
+
Assert.assertTrue(firstTsFileEvent.increaseReferenceCount(TEST_REFERENCE_HOLDER));
+ extractor.extract(firstTsFileEvent);
+
+ final Event firstSuppliedTsFile = extractor.supply();
+ Assert.assertTrue(firstSuppliedTsFile instanceof TsFileInsertionEvent);
+ Assert.assertSame(firstTsFileEvent.getEvent(), firstSuppliedTsFile);
+ commitSuppliedEvent(firstSuppliedTsFile, commitManager);
+
+ final Event secondSuppliedTsFile = extractor.supply();
+ Assert.assertTrue(secondSuppliedTsFile instanceof TsFileInsertionEvent);
+ Assert.assertSame(secondTsFileEvent.getEvent(), secondSuppliedTsFile);
+ commitSuppliedEvent(secondSuppliedTsFile, commitManager);
+
+ Assert.assertEquals(Boolean.FALSE, getGlobalTsFileEpochDegraded());
+ final Event resumedLatestTablet = extractor.supply();
+ Assert.assertTrue(resumedLatestTablet instanceof TabletInsertionEvent);
+ Assert.assertSame(thirdTabletEvent.getEvent(), resumedLatestTablet);
+ commitSuppliedEvent(resumedLatestTablet, commitManager);
+
+ final PipeRealtimeEvent thirdTsFileEvent =
+ bindToTestPipe(
+ PipeRealtimeEventFactory.createRealtimeEvent(false, "root.sg",
thirdResource, false),
+ extractor,
+ pipeTaskMeta);
+
Assert.assertTrue(thirdTsFileEvent.increaseReferenceCount(TEST_REFERENCE_HOLDER));
+ extractor.extract(thirdTsFileEvent);
+
+ Assert.assertNull(extractor.supply());
+ Assert.assertNull(getGlobalTsFileEpochDegraded());
+ } finally {
+ commitManager.deregister(TEST_PIPE_NAME, TEST_PIPE_CREATION_TIME,
dataRegion1);
+ }
+ }
+
+ @Test
+ public void testHybridSourceRegionLevelDowngradingFollowsQueryPriority()
throws Exception {
+ registerTestPipeMeta();
+
+ final PipeEventCommitManager commitManager =
PipeEventCommitManager.getInstance();
+ commitManager.register(TEST_PIPE_NAME, TEST_PIPE_CREATION_TIME,
dataRegion1, "test");
+ try (final PipeRealtimeDataRegionHybridSource extractor =
+ new PipeRealtimeDataRegionHybridSource()) {
+ final PipeParameters parameters =
+ new PipeParameters(
+ new HashMap<String, String>() {
+ {
+ put(PipeSourceConstant.EXTRACTOR_PATTERN_KEY, pattern1);
+ put(
+
PipeSourceConstant.SOURCE_REALTIME_REGION_LEVEL_DOWNGRADING_KEY,
+ Boolean.TRUE.toString());
+ }
+ });
+ final PipeTaskMeta pipeTaskMeta = new
PipeTaskMeta(MinimumProgressIndex.INSTANCE, 1);
+ final PipeTaskRuntimeConfiguration configuration =
+ new PipeTaskRuntimeConfiguration(
+ new PipeTaskSourceRuntimeEnvironment(
+ TEST_PIPE_NAME, TEST_PIPE_CREATION_TIME, dataRegion1,
pipeTaskMeta));
+
+ extractor.validate(new PipeParameterValidator(parameters));
+ extractor.customize(parameters, configuration);
+
+ final TsFileResource loadedUnseqResource =
+ createTsFileResource(dataRegion1, "100-3-0-0.tsfile");
+ loadedUnseqResource.setSeq(false);
+ final PipeRealtimeEvent loadedUnseqTsFileEvent =
+ bindToTestPipe(
+ PipeRealtimeEventFactory.createRealtimeEvent(
+ false, "root.sg", loadedUnseqResource, true),
+ extractor,
+ pipeTaskMeta);
+
Assert.assertTrue(loadedUnseqTsFileEvent.increaseReferenceCount(TEST_REFERENCE_HOLDER));
+ extractor.extract(loadedUnseqTsFileEvent);
+
+ final TsFileResource firstUnseqResource =
+ createTsFileResource(dataRegion1, "200-2-0-0.tsfile");
+ firstUnseqResource.setSeq(false);
+ final PipeRealtimeEvent firstUnseqTabletEvent =
+ bindToTestPipe(
+ PipeRealtimeEventFactory.createRealtimeEvent(
+ false,
+ "root.sg",
+ createInsertRowNode("first-unseq-degraded-tablet", "a"),
+ firstUnseqResource),
+ extractor,
+ pipeTaskMeta);
+
Assert.assertTrue(firstUnseqTabletEvent.increaseReferenceCount(TEST_REFERENCE_HOLDER));
+
+
CommonDescriptor.getInstance().getConfig().setPipeTotalFloatingMemoryProportion(0);
+ try {
+ extractor.extract(firstUnseqTabletEvent);
+ } finally {
+ CommonDescriptor.getInstance()
+ .getConfig()
+
.setPipeTotalFloatingMemoryProportion(pipeTotalFloatingMemoryProportion);
+ }
+ Assert.assertEquals(
+ TsFileEpoch.State.USING_TSFILE,
+ firstUnseqTabletEvent.getTsFileEpoch().getState(extractor));
+ firstUnseqTabletEvent.clearReferenceCount(TEST_REFERENCE_HOLDER);
+
+ final TsFileResource seqResource = createTsFileResource(dataRegion1,
"300-10-0-0.tsfile");
+ seqResource.setSeq(true);
+ final PipeRealtimeEvent seqTabletEvent =
+ bindToTestPipe(
+ PipeRealtimeEventFactory.createRealtimeEvent(
+ false,
+ "root.sg",
+ createInsertRowNode("lower-priority-seq-tail", "a"),
+ seqResource),
+ extractor,
+ pipeTaskMeta);
+
Assert.assertTrue(seqTabletEvent.increaseReferenceCount(TEST_REFERENCE_HOLDER));
+ extractor.extract(seqTabletEvent);
+
+ final PipeRealtimeEvent seqTsFileEvent =
+ bindToTestPipe(
+ PipeRealtimeEventFactory.createRealtimeEvent(false, "root.sg",
seqResource, false),
+ extractor,
+ pipeTaskMeta);
+
Assert.assertTrue(seqTsFileEvent.increaseReferenceCount(TEST_REFERENCE_HOLDER));
+ extractor.extract(seqTsFileEvent);
+
+ final PipeRealtimeEvent firstUnseqTsFileEvent =
+ bindToTestPipe(
+ PipeRealtimeEventFactory.createRealtimeEvent(
+ false, "root.sg", firstUnseqResource, false),
+ extractor,
+ pipeTaskMeta);
+
Assert.assertTrue(firstUnseqTsFileEvent.increaseReferenceCount(TEST_REFERENCE_HOLDER));
+ extractor.extract(firstUnseqTsFileEvent);
+
+ // Seq has lower query priority than every unseq file, so the previously
undegraded realtime
+ // tail must be promoted and sent before either unseq file. Within
unseq, version 2 precedes
+ // version 3 even though the loaded version-3 event was queued before
downgrading started.
+ final Event firstSuppliedTsFile = extractor.supply();
+ Assert.assertTrue(firstSuppliedTsFile instanceof TsFileInsertionEvent);
+ Assert.assertSame(seqTsFileEvent.getEvent(), firstSuppliedTsFile);
+ seqTabletEvent.clearReferenceCount(TEST_REFERENCE_HOLDER);
+ commitSuppliedEvent(firstSuppliedTsFile, commitManager);
+
+ final Event secondSuppliedTsFile = extractor.supply();
+ Assert.assertTrue(secondSuppliedTsFile instanceof TsFileInsertionEvent);
+ Assert.assertSame(firstUnseqTsFileEvent.getEvent(),
secondSuppliedTsFile);
+ commitSuppliedEvent(secondSuppliedTsFile, commitManager);
+
+ final Event thirdSuppliedTsFile = extractor.supply();
+ Assert.assertTrue(thirdSuppliedTsFile instanceof TsFileInsertionEvent);
+ Assert.assertSame(loadedUnseqTsFileEvent.getEvent(),
thirdSuppliedTsFile);
+ commitSuppliedEvent(thirdSuppliedTsFile, commitManager);
+
+ Assert.assertNull(getGlobalTsFileEpochDegraded());
+ Assert.assertNull(extractor.supply());
+ } finally {
+ commitManager.deregister(TEST_PIPE_NAME, TEST_PIPE_CREATION_TIME,
dataRegion1);
+ }
+ }
+
+ @Test
+ public void
testHybridSourceRegionLevelDowngradingPreservesPreviouslyQueuedEvents()
+ throws Exception {
+ registerTestPipeMeta();
+
+ final PipeEventCommitManager commitManager =
PipeEventCommitManager.getInstance();
+ commitManager.register(TEST_PIPE_NAME, TEST_PIPE_CREATION_TIME,
dataRegion1, "test");
+ try (final PipeRealtimeDataRegionHybridSource extractor =
+ new PipeRealtimeDataRegionHybridSource()) {
+ final PipeParameters parameters =
+ new PipeParameters(
+ new HashMap<String, String>() {
+ {
+ put(PipeSourceConstant.EXTRACTOR_PATTERN_KEY, pattern1);
+ put(
+
PipeSourceConstant.SOURCE_REALTIME_REGION_LEVEL_DOWNGRADING_KEY,
+ Boolean.TRUE.toString());
+ }
+ });
+ final PipeTaskMeta pipeTaskMeta = new
PipeTaskMeta(MinimumProgressIndex.INSTANCE, 1);
+ final PipeTaskRuntimeConfiguration configuration =
+ new PipeTaskRuntimeConfiguration(
+ new PipeTaskSourceRuntimeEnvironment(
+ TEST_PIPE_NAME, TEST_PIPE_CREATION_TIME, dataRegion1,
pipeTaskMeta));
+
+ extractor.validate(new PipeParameterValidator(parameters));
+ extractor.customize(parameters, configuration);
+
+ final TsFileResource olderResource = createTsFileResource(dataRegion1,
"105-105-0-0.tsfile");
+ final PipeRealtimeEvent olderTabletEvent =
+ bindToTestPipe(
+ PipeRealtimeEventFactory.createRealtimeEvent(
+ false,
+ "root.sg",
+ createInsertRowNode("queued-before-downgrading", "a"),
+ olderResource),
+ extractor,
+ pipeTaskMeta);
+
Assert.assertTrue(olderTabletEvent.increaseReferenceCount(TEST_REFERENCE_HOLDER));
+ extractor.extract(olderTabletEvent);
+
+ // Seal the older epoch while leaving its tablet queued in the source.
+ final PipeRealtimeEvent olderTsFileEvent =
+ bindToTestPipe(
+ PipeRealtimeEventFactory.createRealtimeEvent(false, "root.sg",
olderResource, false),
+ extractor,
+ pipeTaskMeta);
+
Assert.assertTrue(olderTsFileEvent.increaseReferenceCount(TEST_REFERENCE_HOLDER));
+ extractor.extract(olderTsFileEvent);
+
+ final TsFileResource degradedResource =
+ createTsFileResource(dataRegion1, "106-106-0-0.tsfile");
+ final PipeRealtimeEvent degradedTabletEvent =
+ bindToTestPipe(
+ PipeRealtimeEventFactory.createRealtimeEvent(
+ false,
+ "root.sg",
+ createInsertRowNode("trigger-region-downgrading", "a"),
+ degradedResource),
+ extractor,
+ pipeTaskMeta);
+
Assert.assertTrue(degradedTabletEvent.increaseReferenceCount(TEST_REFERENCE_HOLDER));
+
+
CommonDescriptor.getInstance().getConfig().setPipeTotalFloatingMemoryProportion(0);
+ try {
+ extractor.extract(degradedTabletEvent);
+ } finally {
+ CommonDescriptor.getInstance()
+ .getConfig()
+
.setPipeTotalFloatingMemoryProportion(pipeTotalFloatingMemoryProportion);
+ }
+ Assert.assertEquals(
+ TsFileEpoch.State.USING_TSFILE,
degradedTabletEvent.getTsFileEpoch().getState(extractor));
+ Assert.assertEquals(Boolean.TRUE, getGlobalTsFileEpochDegraded());
+
+ final PipeRealtimeEvent degradedTsFileEvent =
+ bindToTestPipe(
+ PipeRealtimeEventFactory.createRealtimeEvent(
+ false, "root.sg", degradedResource, false),
+ extractor,
+ pipeTaskMeta);
+
Assert.assertTrue(degradedTsFileEvent.increaseReferenceCount(TEST_REFERENCE_HOLDER));
+ extractor.extract(degradedTsFileEvent);
+
+ // The tablet that was already queued before downgrading must not be
overtaken by the later
+ // degraded TsFile.
+ final Event firstSuppliedEvent = extractor.supply();
+ Assert.assertTrue(firstSuppliedEvent instanceof TabletInsertionEvent);
+ Assert.assertSame(olderTabletEvent.getEvent(), firstSuppliedEvent);
+ commitSuppliedEvent(firstSuppliedEvent, commitManager);
+
+ final Event secondSuppliedEvent = extractor.supply();
+ Assert.assertTrue(secondSuppliedEvent instanceof TsFileInsertionEvent);
+ Assert.assertSame(degradedTsFileEvent.getEvent(), secondSuppliedEvent);
+ commitSuppliedEvent(secondSuppliedEvent, commitManager);
+
+ Assert.assertNull(getGlobalTsFileEpochDegraded());
+ Assert.assertNull(extractor.supply());
+ } finally {
+ commitManager.deregister(TEST_PIPE_NAME, TEST_PIPE_CREATION_TIME,
dataRegion1);
+ }
+ }
+
+ @Test
+ public void testTsFileSourceRegionLevelDowngradingUsesPipeParameters()
throws Exception {
+ for (final String key :
+ Arrays.asList(
+ null,
+ PipeSourceConstant.SOURCE_REALTIME_REGION_LEVEL_DOWNGRADING_KEY,
+
PipeSourceConstant.EXTRACTOR_REALTIME_REGION_LEVEL_DOWNGRADING_KEY)) {
+ final PipeTaskMeta pipeTaskMeta = new
PipeTaskMeta(MinimumProgressIndex.INSTANCE, 1);
+ try (final PipeRealtimeDataRegionTsFileSource extractor =
+ createTsFileSourceForTest(key, pipeTaskMeta)) {
+ final String prefix = key == null ? "120" : key.startsWith("source") ?
"121" : "122";
+ final PipeRealtimeEvent firstEvent =
+ extractTsFileForTest(
+ extractor,
+ pipeTaskMeta,
+ createTsFileResource(dataRegion1, prefix + "-1-0-0.tsfile"),
+ false);
+ final PipeRealtimeEvent secondEvent =
+ extractTsFileForTest(
+ extractor,
+ pipeTaskMeta,
+ createTsFileResource(dataRegion1, prefix + "-2-0-0.tsfile"),
+ false);
+ Assert.assertSame(firstEvent.getEvent(), extractor.supply());
+ if (key == null) {
+ Assert.assertSame(secondEvent.getEvent(), extractor.supply());
+ } else {
+ Assert.assertNull(extractor.supply());
+ }
+ firstEvent.clearReferenceCount(TEST_REFERENCE_HOLDER);
+ if (key != null) {
+ Assert.assertSame(secondEvent.getEvent(), extractor.supply());
+ }
+ secondEvent.clearReferenceCount(TEST_REFERENCE_HOLDER);
+ }
+ }
+ }
+
+ @Test
+ public void testTsFileSourceRegionLevelDowngradingFollowsQueryPriority()
throws Exception {
+ final PipeEventCommitManager commitManager =
PipeEventCommitManager.getInstance();
+ commitManager.register(TEST_PIPE_NAME, TEST_PIPE_CREATION_TIME,
dataRegion1, "test");
+ final PipeTaskMeta pipeTaskMeta = new
PipeTaskMeta(MinimumProgressIndex.INSTANCE, 1);
+ try (final PipeRealtimeDataRegionTsFileSource extractor =
+ createTsFileSourceForTest(
+ PipeSourceConstant.SOURCE_REALTIME_REGION_LEVEL_DOWNGRADING_KEY,
pipeTaskMeta)) {
+ final TsFileResource[] resources = {
+ createTsFileResource(dataRegion1, "300-10-0-0.tsfile"),
+ createTsFileResource(dataRegion1, "150-2-0-0.tsfile"),
+ createTsFileResource(dataRegion1, "200-2-0-0.tsfile"),
+ createTsFileResource(dataRegion1, "1", "200-2-0-0.tsfile"),
+ createTsFileResource(dataRegion1, "200-2-1-0.tsfile"),
+ createTsFileResource(dataRegion1, "100-3-0-0.tsfile")
+ };
+ final PipeRealtimeEvent[] events = new
PipeRealtimeEvent[resources.length];
+ // Reverse arrival order: seq/unseq, version, timestamp, compaction
version and directory
+ // must all use the same ordering as Hybrid source, including loaded
files.
+ for (int i = resources.length - 1; i >= 0; --i) {
+ resources[i].setSeq(i == 0);
+ events[i] = extractTsFileForTest(extractor, pipeTaskMeta,
resources[i], i == 5);
+ }
+ for (final PipeRealtimeEvent event : events) {
+ final Event suppliedEvent = extractor.supply();
+ Assert.assertSame(event.getEvent(), suppliedEvent);
+ Assert.assertNull(extractor.supply());
+ commitSuppliedEvent(suppliedEvent, commitManager);
+ }
+ Assert.assertNull(extractor.supply());
+ } finally {
+ commitManager.deregister(TEST_PIPE_NAME, TEST_PIPE_CREATION_TIME,
dataRegion1);
+ }
+ }
+
+ @Test
+ public void testTsFileSourceRegionLevelDowngradingWaitsForGeneratedTablets()
throws Exception {
+ final PipeEventCommitManager commitManager =
PipeEventCommitManager.getInstance();
+ commitManager.register(TEST_PIPE_NAME, TEST_PIPE_CREATION_TIME,
dataRegion1, "test");
+ final PipeTaskMeta pipeTaskMeta = new
PipeTaskMeta(MinimumProgressIndex.INSTANCE, 1);
+ try (final PipeRealtimeDataRegionTsFileSource extractor =
+ createTsFileSourceForTest(
+ PipeSourceConstant.SOURCE_REALTIME_REGION_LEVEL_DOWNGRADING_KEY,
pipeTaskMeta)) {
+ extractTsFileForTest(
+ extractor, pipeTaskMeta, createTsFileResource(dataRegion1,
"123-1-0-0.tsfile"), false);
+ final PipeRealtimeEvent progressEvent =
createProgressReportRealtimeEvent();
+
Assert.assertTrue(progressEvent.increaseReferenceCount(TEST_REFERENCE_HOLDER));
+ extractor.extract(progressEvent);
+ final PipeRealtimeEvent secondEvent =
+ extractTsFileForTest(
+ extractor,
+ pipeTaskMeta,
+ createTsFileResource(dataRegion1, "123-2-0-0.tsfile"),
+ false);
+
+ final PipeTsFileInsertionEvent firstTsFile = (PipeTsFileInsertionEvent)
extractor.supply();
+ firstTsFile.markGeneratedTabletInsertionEventsParsingStarted();
+ commitSuppliedEvent(firstTsFile, commitManager);
+ // Committing a TsFile with a retained iterable and no tablets yet must
not release the gate.
+ Assert.assertNull(extractor.supply());
+ final PipeRawTabletInsertionEvent firstTablet =
+ createGeneratedTabletEvent(firstTsFile, pipeTaskMeta, "first",
false);
+ final PipeRawTabletInsertionEvent secondTablet =
+ createGeneratedTabletEvent(firstTsFile, pipeTaskMeta, "second");
+ firstTsFile.registerGeneratedTabletInsertionEvent(firstTablet);
+ firstTsFile.registerGeneratedTabletInsertionEvent(secondTablet);
+
Assert.assertTrue(firstTablet.increaseReferenceCount(TEST_REFERENCE_HOLDER));
+
Assert.assertTrue(secondTablet.increaseReferenceCount(TEST_REFERENCE_HOLDER));
+ commitSuppliedEvent(firstTablet, commitManager);
+ Assert.assertNull(extractor.supply());
+ commitSuppliedEvent(secondTablet, commitManager);
+ Assert.assertNull(extractor.supply());
+ firstTsFile.markGeneratedTabletInsertionEventsParsingCompleted();
+
+ final Event secondTsFile = extractor.supply();
+ Assert.assertSame(secondEvent.getEvent(), secondTsFile);
+ Assert.assertNull(extractor.supply());
+ commitSuppliedEvent(secondTsFile, commitManager);
+ Assert.assertSame(progressEvent.getEvent(), extractor.supply());
+ releaseSuppliedEvent(progressEvent.getEvent());
+ Assert.assertNull(extractor.supply());
+ } finally {
+ commitManager.deregister(TEST_PIPE_NAME, TEST_PIPE_CREATION_TIME,
dataRegion1);
+ }
+ }
+
+ @Test
+ public void
testTsFileSourceRegionLevelDowngradingCompensatesForDiscardedEvents()
+ throws Exception {
+ final PipeEventCommitManager commitManager =
PipeEventCommitManager.getInstance();
+ commitManager.register(TEST_PIPE_NAME, TEST_PIPE_CREATION_TIME,
dataRegion1, "test");
+ final PipeTaskMeta pipeTaskMeta = new
PipeTaskMeta(MinimumProgressIndex.INSTANCE, 1);
+ try (final PipeRealtimeDataRegionTsFileSource extractor =
+ createTsFileSourceForTest(
+ PipeSourceConstant.SOURCE_REALTIME_REGION_LEVEL_DOWNGRADING_KEY,
pipeTaskMeta)) {
+ extractTsFileForTest(
+ extractor, pipeTaskMeta, createTsFileResource(dataRegion1,
"124-1-0-0.tsfile"), false);
+ final PipeRealtimeEvent secondEvent =
+ extractTsFileForTest(
+ extractor,
+ pipeTaskMeta,
+ createTsFileResource(dataRegion1, "124-2-0-0.tsfile"),
+ false);
+ final PipeRealtimeEvent thirdEvent =
+ extractTsFileForTest(
+ extractor,
+ pipeTaskMeta,
+ createTsFileResource(dataRegion1, "124-3-0-0.tsfile"),
+ false);
+
+ final PipeTsFileInsertionEvent firstTsFile = (PipeTsFileInsertionEvent)
extractor.supply();
+ final PipeRawTabletInsertionEvent firstTablet =
+ createGeneratedTabletEvent(firstTsFile, pipeTaskMeta,
"discarded-first", false);
+ final PipeRawTabletInsertionEvent secondTablet =
+ createGeneratedTabletEvent(firstTsFile, pipeTaskMeta,
"discarded-second");
+ firstTsFile.markGeneratedTabletInsertionEventsParsingStarted();
+ firstTsFile.registerGeneratedTabletInsertionEvent(firstTablet);
+ firstTsFile.registerGeneratedTabletInsertionEvent(secondTablet);
+ firstTsFile.markGeneratedTabletInsertionEventsParsingCompleted();
+
Assert.assertTrue(firstTablet.increaseReferenceCount(TEST_REFERENCE_HOLDER));
+
Assert.assertTrue(secondTablet.increaseReferenceCount(TEST_REFERENCE_HOLDER));
+ commitSuppliedEvent(firstTsFile, commitManager);
+ firstTablet.clearReferenceCount(TEST_REFERENCE_HOLDER);
+ firstTablet.clearReferenceCount(TEST_REFERENCE_HOLDER);
+ Assert.assertNull(extractor.supply());
+ secondTablet.clearReferenceCount(TEST_REFERENCE_HOLDER);
+
+ final Event secondTsFile = extractor.supply();
+ Assert.assertSame(secondEvent.getEvent(), secondTsFile);
+ secondTablet.clearReferenceCount(TEST_REFERENCE_HOLDER);
+ Assert.assertNull(extractor.supply());
+ // Direct TsFile discard releases the gate once, without closing the
source.
+ releaseSuppliedEvent(secondTsFile);
+ Assert.assertSame(thirdEvent.getEvent(), extractor.supply());
+ releaseSuppliedEvent(thirdEvent.getEvent());
+ Assert.assertNull(extractor.supply());
+ } finally {
+ commitManager.deregister(TEST_PIPE_NAME, TEST_PIPE_CREATION_TIME,
dataRegion1);
+ }
+ }
+
+ @Test
+ public void
testTsFileSourceRegionLevelDowngradingReleasesBufferedEventsOnClose()
+ throws Exception {
+ final PipeTaskMeta pipeTaskMeta = new
PipeTaskMeta(MinimumProgressIndex.INSTANCE, 1);
+ try (final PipeRealtimeDataRegionTsFileSource extractor =
+ createTsFileSourceForTest(
+ PipeSourceConstant.SOURCE_REALTIME_REGION_LEVEL_DOWNGRADING_KEY,
pipeTaskMeta)) {
+ final PipeRealtimeEvent firstEvent =
+ extractTsFileForTest(
+ extractor,
+ pipeTaskMeta,
+ createTsFileResource(dataRegion1, "125-1-0-0.tsfile"),
+ false);
+ final PipeRealtimeEvent bufferedFile =
+ extractTsFileForTest(
+ extractor,
+ pipeTaskMeta,
+ createTsFileResource(dataRegion1, "125-2-0-0.tsfile"),
+ false);
+ final PipeRealtimeEvent heartbeat =
+ PipeRealtimeEventFactory.createRealtimeEvent(dataRegion1, false);
+
Assert.assertTrue(heartbeat.increaseReferenceCount(TEST_REFERENCE_HOLDER));
+ extractor.extract(heartbeat);
+ Assert.assertSame(firstEvent.getEvent(), extractor.supply());
+ final PipeRealtimeEvent pendingFile =
+ extractTsFileForTest(
+ extractor,
+ pipeTaskMeta,
+ createTsFileResource(dataRegion1, "125-3-0-0.tsfile"),
+ false);
+ extractor.close();
+ Assert.assertEquals(0, bufferedFile.getEvent().getReferenceCount());
+ Assert.assertEquals(0, pendingFile.getEvent().getReferenceCount());
+ Assert.assertEquals(0, heartbeat.getEvent().getReferenceCount());
+ releaseSuppliedEvent(firstEvent.getEvent());
+ final PipeRealtimeEvent lateFile =
+ extractTsFileForTest(
+ extractor,
+ pipeTaskMeta,
+ createTsFileResource(dataRegion1, "125-4-0-0.tsfile"),
+ false);
+ Assert.assertEquals(0, lateFile.getEvent().getReferenceCount());
+ Assert.assertNull(extractor.supply());
+ }
+ }
+
+ private PipeRealtimeDataRegionTsFileSource createTsFileSourceForTest(
+ final String regionLevelDowngradingKey, final PipeTaskMeta pipeTaskMeta)
throws Exception {
+ final HashMap<String, String> attributes = new HashMap<>();
+ attributes.put(PipeSourceConstant.EXTRACTOR_PATTERN_KEY, pattern1);
+ if (regionLevelDowngradingKey != null) {
+ attributes.put(regionLevelDowngradingKey, Boolean.TRUE.toString());
+ }
+ final PipeParameters parameters = new PipeParameters(attributes);
+ final PipeRealtimeDataRegionTsFileSource extractor = new
PipeRealtimeDataRegionTsFileSource();
+ extractor.validate(new PipeParameterValidator(parameters));
+ extractor.customize(
+ parameters,
+ new PipeTaskRuntimeConfiguration(
+ new PipeTaskSourceRuntimeEnvironment(
+ TEST_PIPE_NAME, TEST_PIPE_CREATION_TIME, dataRegion1,
pipeTaskMeta)));
+ return extractor;
+ }
+
+ private PipeRealtimeEvent extractTsFileForTest(
+ final PipeRealtimeDataRegionSource extractor,
+ final PipeTaskMeta pipeTaskMeta,
+ final TsFileResource resource,
+ final boolean isLoaded) {
+ final PipeRealtimeEvent event =
+ bindToTestPipe(
+ PipeRealtimeEventFactory.createRealtimeEvent(false, "root.sg",
resource, isLoaded),
+ extractor,
+ pipeTaskMeta);
+ Assert.assertTrue(event.increaseReferenceCount(TEST_REFERENCE_HOLDER));
+ extractor.extract(event);
+ return event;
+ }
+
private Future<?> write2DataRegion(
final int writeNum, final int dataRegionId, final int startNum) {
final File dataRegionDir =
@@ -469,8 +1642,15 @@ public class PipeRealtimeExtractTest {
private TsFileResource createTsFileResource(final int dataRegionId, final
String fileName)
throws IOException {
+ return createTsFileResource(dataRegionId, "0", fileName);
+ }
+
+ private TsFileResource createTsFileResource(
+ final int dataRegionId, final String timePartition, final String
fileName)
+ throws IOException {
final File dataRegionDir =
- new File(tsFileDir.getPath() + File.separator + dataRegionId +
File.separator + "0");
+ new File(
+ tsFileDir.getPath() + File.separator + dataRegionId +
File.separator + timePartition);
Assert.assertTrue(dataRegionDir.mkdirs() || dataRegionDir.isDirectory());
final File tsFile = new File(dataRegionDir, fileName);
@@ -577,6 +1757,101 @@ public class PipeRealtimeExtractTest {
}
}
+ private PipeRealtimeEvent bindToTestPipe(
+ final PipeRealtimeEvent event,
+ final PipeRealtimeDataRegionSource extractor,
+ final PipeTaskMeta pipeTaskMeta) {
+ return event.shallowCopySelfAndBindPipeTaskMetaForProgressReport(
+ TEST_PIPE_NAME,
+ TEST_PIPE_CREATION_TIME,
+ pipeTaskMeta,
+ extractor.getTreePattern(),
+ extractor.getTablePattern(),
+ String.valueOf(extractor.getUserId()),
+ extractor.getUserName(),
+ extractor.getCliHostname(),
+ extractor.isSkipIfNoPrivileges(),
+ extractor.getRealtimeDataExtractionStartTime(),
+ extractor.getRealtimeDataExtractionEndTime());
+ }
+
+ private void commitSuppliedEvent(final Event event, final
PipeEventCommitManager commitManager) {
+ final EnrichedEvent enrichedEvent = (EnrichedEvent) event;
+ commitManager.enrichWithCommitterKeyAndCommitId(
+ enrichedEvent, TEST_PIPE_CREATION_TIME, dataRegion1);
+
Assert.assertTrue(enrichedEvent.decreaseReferenceCount(TEST_REFERENCE_HOLDER,
true));
+ }
+
+ private void commitLastGeneratedTabletEvent(
+ final PipeTsFileInsertionEvent tsFileEvent,
+ final PipeEventCommitManager commitManager,
+ final PipeTaskMeta pipeTaskMeta) {
+ tsFileEvent.registerGeneratedTabletInsertionEvent();
+ tsFileEvent.markGeneratedTabletInsertionEventsParsingCompleted();
+ final PipeRawTabletInsertionEvent generatedTabletEvent =
+ createGeneratedTabletEvent(tsFileEvent, pipeTaskMeta, "generated");
+
+
Assert.assertTrue(generatedTabletEvent.increaseReferenceCount(TEST_REFERENCE_HOLDER));
+
Assert.assertTrue(tsFileEvent.decreaseReferenceCount(TEST_REFERENCE_HOLDER,
false));
+ commitSuppliedEvent(generatedTabletEvent, commitManager);
+ }
+
+ private PipeRawTabletInsertionEvent createGeneratedTabletEvent(
+ final PipeTsFileInsertionEvent tsFileEvent,
+ final PipeTaskMeta pipeTaskMeta,
+ final String deviceId) {
+ return createGeneratedTabletEvent(tsFileEvent, pipeTaskMeta, deviceId,
true);
+ }
+
+ private PipeRawTabletInsertionEvent createGeneratedTabletEvent(
+ final PipeTsFileInsertionEvent tsFileEvent,
+ final PipeTaskMeta pipeTaskMeta,
+ final String deviceId,
+ final boolean needToReport) {
+ final Tablet tablet =
+ new Tablet(
+ "root.sg.d." + deviceId,
+ Collections.singletonList(new MeasurementSchema("s",
TSDataType.INT32)),
+ 1);
+ return new PipeRawTabletInsertionEvent(
+ false,
+ "root.sg",
+ null,
+ null,
+ tablet,
+ false,
+ TEST_PIPE_NAME,
+ TEST_PIPE_CREATION_TIME,
+ pipeTaskMeta,
+ tsFileEvent,
+ needToReport);
+ }
+
+ private int getActiveTsFileEpochCount(final
PipeRealtimeDataRegionHybridSource extractor)
+ throws Exception {
+ final Field activeTsFileEpochsField =
+
PipeRealtimeDataRegionHybridSource.class.getDeclaredField("activeTsFileEpochs");
+ activeTsFileEpochsField.setAccessible(true);
+ return ((Set<?>) activeTsFileEpochsField.get(extractor)).size();
+ }
+
+ private int getInFlightTsFileCount(final PipeRealtimeDataRegionHybridSource
extractor)
+ throws Exception {
+ final Field inFlightTsFileCountField =
+
PipeRealtimeDataRegionHybridSource.class.getDeclaredField("inFlightTsFileCount");
+ inFlightTsFileCountField.setAccessible(true);
+ return inFlightTsFileCountField.getInt(extractor);
+ }
+
+ private boolean isRegionLevelDowngradingEnabled(
+ final PipeRealtimeDataRegionHybridSource extractor) throws Exception {
+ final Field isRegionLevelDowngradingEnabledField =
+ PipeRealtimeDataRegionHybridSource.class.getDeclaredField(
+ "isRegionLevelDowngradingEnabled");
+ isRegionLevelDowngradingEnabledField.setAccessible(true);
+ return isRegionLevelDowngradingEnabledField.getBoolean(extractor);
+ }
+
private PipeRealtimeEvent createProgressReportRealtimeEvent() {
final ProgressReportEvent progressReportEvent = new
ProgressReportEvent(null, 0, null);
progressReportEvent.bindProgressIndex(MinimumProgressIndex.INSTANCE);
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/PipeSourceConstant.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/PipeSourceConstant.java
index 055553c7e1a..70d452ae36a 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/PipeSourceConstant.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/PipeSourceConstant.java
@@ -142,6 +142,11 @@ public class PipeSourceConstant {
public static final String EXTRACTOR_REALTIME_LOOSE_RANGE_PATH_VALUE =
"path";
public static final String EXTRACTOR_REALTIME_LOOSE_RANGE_ALL_VALUE = "all";
public static final String EXTRACTOR_REALTIME_LOOSE_RANGE_DEFAULT_VALUE = "";
+ public static final String EXTRACTOR_REALTIME_REGION_LEVEL_DOWNGRADING_KEY =
+ "extractor.realtime.region-level-downgrading";
+ public static final String SOURCE_REALTIME_REGION_LEVEL_DOWNGRADING_KEY =
+ "source.realtime.region-level-downgrading";
+ public static final boolean
EXTRACTOR_REALTIME_REGION_LEVEL_DOWNGRADING_DEFAULT_VALUE = false;
public static final String EXTRACTOR_MODE_STREAMING_KEY =
"extractor.mode.streaming";
public static final String SOURCE_MODE_STREAMING_KEY =
"source.mode.streaming";
