diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/agent/task/PipeConfigNodeTaskAgent.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/agent/task/PipeConfigNodeTaskAgent.java index 156a98b79d4a4..3fe0b1757cfb0 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/agent/task/PipeConfigNodeTaskAgent.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/agent/task/PipeConfigNodeTaskAgent.java @@ -50,6 +50,7 @@ import java.util.Collections; import java.util.List; import java.util.Map; +import java.util.Objects; import java.util.Optional; import java.util.concurrent.atomic.AtomicLong; import java.util.function.Consumer; @@ -177,10 +178,13 @@ private void clearConfigRegionListeningQueueIfNecessary( } final ProgressIndex progressIndex = groupId2TaskMetaMap.get(regionId).getProgressIndex(); - if (progressIndex instanceof MetaProgressIndex) { - if (((MetaProgressIndex) progressIndex).getIndex() + 1 - < listeningQueueNewFirstIndex.get()) { - listeningQueueNewFirstIndex.set(((MetaProgressIndex) progressIndex).getIndex() + 1); + final Optional metaProgressIndex = + Objects.isNull(progressIndex) + ? Optional.empty() + : progressIndex.getProgressIndexByType(MetaProgressIndex.class); + if (metaProgressIndex.isPresent()) { + if (metaProgressIndex.get().getIndex() + 1 < listeningQueueNewFirstIndex.get()) { + listeningQueueNewFirstIndex.set(metaProgressIndex.get().getIndex() + 1); } } else { // Do not clear "minimumProgressIndex"s related queues to avoid clearing diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgent.java index 17504ab65573c..54e8f1752eaf3 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgent.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgent.java @@ -240,11 +240,15 @@ private Set clearSchemaRegionListeningQueueIfNecessary( } final ProgressIndex progressIndex = pipeTaskMeta.getProgressIndex(); - if (progressIndex instanceof MetaProgressIndex) { - if (((MetaProgressIndex) progressIndex).getIndex() + 1 + final Optional metaProgressIndex = + Objects.isNull(progressIndex) + ? Optional.empty() + : progressIndex.getProgressIndexByType(MetaProgressIndex.class); + if (metaProgressIndex.isPresent()) { + if (metaProgressIndex.get().getIndex() + 1 < schemaRegionId2ListeningQueueNewFirstIndex.getOrDefault(id, Long.MAX_VALUE)) { schemaRegionId2ListeningQueueNewFirstIndex.put( - id, ((MetaProgressIndex) progressIndex).getIndex() + 1); + id, metaProgressIndex.get().getIndex() + 1); } } else { // Do not clear "minimumProgressIndex"s related queues to avoid clearing diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/aggregate/AggregateProcessor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/aggregate/AggregateProcessor.java index f12de14ebaa4c..6a3ad9a6159cf 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/aggregate/AggregateProcessor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/aggregate/AggregateProcessor.java @@ -331,17 +331,15 @@ public void customize( // Restore window state final ProgressIndex index = pipeTaskMeta.getProgressIndex(); - if (index == MinimumProgressIndex.INSTANCE) { + if (Objects.isNull(index) || index == MinimumProgressIndex.INSTANCE) { return; } - if (!(index instanceof TimeWindowStateProgressIndex)) { - throw new PipeException( - String.format( - "The aggregate processor does not support progressIndexType %s", index.getType())); - } - final TimeWindowStateProgressIndex timeWindowStateProgressIndex = - (TimeWindowStateProgressIndex) index; + index.getProgressIndexByType(TimeWindowStateProgressIndex.class).orElse(null); + // A pipe altered from another processor may not have window state yet. + if (Objects.isNull(timeWindowStateProgressIndex)) { + return; + } for (final Map.Entry> entry : timeWindowStateProgressIndex.getTimeSeries2TimestampWindowBufferPairMap().entrySet()) { final AtomicReference stateReference = diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/twostage/plugin/TwoStageCountProcessor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/twostage/plugin/TwoStageCountProcessor.java index c4a3acc50b33c..55a3b370a05ae 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/twostage/plugin/TwoStageCountProcessor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/twostage/plugin/TwoStageCountProcessor.java @@ -136,13 +136,7 @@ public void customize(PipeParameters parameters, PipeProcessorRuntimeConfigurati outputSeries = parseOutputSeries(parameters); if (Objects.nonNull(pipeTaskMeta) && Objects.nonNull(pipeTaskMeta.getProgressIndex())) { - if (pipeTaskMeta.getProgressIndex() instanceof MinimumProgressIndex) { - pipeTaskMeta.updateProgressIndex( - new StateProgressIndex(Long.MIN_VALUE, new HashMap<>(), MinimumProgressIndex.INSTANCE)); - } - - final StateProgressIndex stateProgressIndex = - (StateProgressIndex) pipeTaskMeta.getProgressIndex(); + final StateProgressIndex stateProgressIndex = initializeStateProgressIndex(pipeTaskMeta); localCommitProgressIndex.set(stateProgressIndex.getInnerProgressIndex()); final Binary localCountState = stateProgressIndex.getState().get(LOCAL_COUNT_STATE_KEY); localCount.set( @@ -173,6 +167,26 @@ static PartialPath parseOutputSeries(final PipeParameters parameters) PipeProcessorConstant.PROCESSOR_OUTPUT_SERIES_KEY, LEGACY_PROCESSOR_OUTPUT_SERIES_KEY)); } + static StateProgressIndex initializeStateProgressIndex(final PipeTaskMeta pipeTaskMeta) { + final ProgressIndex progressIndex = pipeTaskMeta.getProgressIndex(); + if (progressIndex instanceof StateProgressIndex) { + return (StateProgressIndex) progressIndex; + } + + final ProgressIndex updatedProgressIndex = + pipeTaskMeta.updateProgressIndex( + new StateProgressIndex( + Long.MIN_VALUE, Collections.emptyMap(), MinimumProgressIndex.INSTANCE)); + return updatedProgressIndex + .getProgressIndexByType(StateProgressIndex.class) + .orElseThrow( + () -> + new PipeException( + String.format( + "Failed to initialize StateProgressIndex from progress index %s.", + updatedProgressIndex))); + } + @Override public void process(TabletInsertionEvent tabletInsertionEvent, EventCollector eventCollector) throws Exception { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileSource.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileSource.java index 00fd08dc2b7b4..6ebdebb6efda4 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileSource.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileSource.java @@ -398,7 +398,7 @@ public synchronized void start() { originalResourceList.sort( (o1, o2) -> - startIndex instanceof TimeWindowStateProgressIndex + Objects.nonNull(getTimeWindowStateProgressIndex(startIndex)) ? Long.compare(o1.getFileStartTime(), o2.getFileStartTime()) : o1.getMaxProgressIndex().topologicalCompareTo(o2.getMaxProgressIndex())); pendingQueue = new ArrayDeque<>(originalResourceList); @@ -466,9 +466,11 @@ private boolean shouldExtractTsFileResource(final TsFileResource resource) { } private boolean mayTsFileContainUnprocessedData(final TsFileResource resource) { - if (startIndex instanceof TimeWindowStateProgressIndex) { + final TimeWindowStateProgressIndex timeWindowStateProgressIndex = + getTimeWindowStateProgressIndex(startIndex); + if (Objects.nonNull(timeWindowStateProgressIndex)) { // The resource is closed thus the TsFileResource#getFileEndTime() is safe to use - return ((TimeWindowStateProgressIndex) startIndex).getMinTime() <= resource.getFileEndTime(); + return timeWindowStateProgressIndex.getMinTime() <= resource.getFileEndTime(); } if (startIndex instanceof StateProgressIndex) { @@ -488,6 +490,13 @@ private boolean mayTsFileContainUnprocessedData(final TsFileResource resource) { return false; } + private TimeWindowStateProgressIndex getTimeWindowStateProgressIndex( + final ProgressIndex progressIndex) { + return Objects.isNull(progressIndex) + ? null + : progressIndex.getProgressIndexByType(TimeWindowStateProgressIndex.class).orElse(null); + } + private boolean mayTsFileResourceOverlappedWithPattern(final TsFileResource resource) { // Trimming to avoid unnecessary file device getter if (isDbNameCoveredByPattern) { diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/processor/twostage/plugin/TwoStageCountProcessorTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/processor/twostage/plugin/TwoStageCountProcessorTest.java index 2957ffd4e3e55..599ad01bb42c2 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/processor/twostage/plugin/TwoStageCountProcessorTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/processor/twostage/plugin/TwoStageCountProcessorTest.java @@ -19,7 +19,14 @@ package org.apache.iotdb.db.pipe.processor.twostage.plugin; +import org.apache.iotdb.commons.consensus.index.ProgressIndex; +import org.apache.iotdb.commons.consensus.index.impl.HybridProgressIndex; +import org.apache.iotdb.commons.consensus.index.impl.MetaProgressIndex; +import org.apache.iotdb.commons.consensus.index.impl.SimpleProgressIndex; +import org.apache.iotdb.commons.consensus.index.impl.StateProgressIndex; +import org.apache.iotdb.commons.consensus.index.impl.TimeWindowStateProgressIndex; import org.apache.iotdb.commons.path.PartialPath; +import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta; import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameterValidator; import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters; @@ -44,6 +51,54 @@ public void testValidateOutputSeriesSupportsNewAndLegacyKeys() throws Exception validateOutputSeries("processor.output-series", "root.db.d.s2"); } + @Test + public void testInitializeStateProgressIndexFromHybridProgressIndex() { + final MetaProgressIndex metaProgressIndex = new MetaProgressIndex(10L); + final SimpleProgressIndex simpleProgressIndex = new SimpleProgressIndex(1, 2L); + final ProgressIndex hybridProgressIndex = + new HybridProgressIndex(metaProgressIndex) + .updateToMinimumEqualOrIsAfterProgressIndex(simpleProgressIndex); + final PipeTaskMeta pipeTaskMeta = new PipeTaskMeta(hybridProgressIndex, 0); + + final StateProgressIndex stateProgressIndex = + TwoStageCountProcessor.initializeStateProgressIndex(pipeTaskMeta); + + Assert.assertSame(stateProgressIndex, pipeTaskMeta.getProgressIndex()); + Assert.assertEquals( + metaProgressIndex, + stateProgressIndex.getProgressIndexByType(MetaProgressIndex.class).orElse(null)); + Assert.assertEquals( + simpleProgressIndex, + stateProgressIndex.getProgressIndexByType(SimpleProgressIndex.class).orElse(null)); + } + + @Test + public void testInitializeStateProgressIndexFromTimeWindowStateProgressIndex() { + final TimeWindowStateProgressIndex timeWindowStateProgressIndex = + new TimeWindowStateProgressIndex(Collections.emptyMap()); + final PipeTaskMeta pipeTaskMeta = new PipeTaskMeta(timeWindowStateProgressIndex, 0); + + final StateProgressIndex stateProgressIndex = + TwoStageCountProcessor.initializeStateProgressIndex(pipeTaskMeta); + Assert.assertEquals( + timeWindowStateProgressIndex, + stateProgressIndex.getProgressIndexByType(TimeWindowStateProgressIndex.class).orElse(null)); + + final SimpleProgressIndex simpleProgressIndex = new SimpleProgressIndex(1, 2L); + final ProgressIndex updatedProgressIndex = + pipeTaskMeta.updateProgressIndex( + new StateProgressIndex(1L, Collections.emptyMap(), simpleProgressIndex)); + Assert.assertTrue(updatedProgressIndex instanceof StateProgressIndex); + Assert.assertEquals( + timeWindowStateProgressIndex, + updatedProgressIndex + .getProgressIndexByType(TimeWindowStateProgressIndex.class) + .orElse(null)); + Assert.assertEquals( + simpleProgressIndex, + updatedProgressIndex.getProgressIndexByType(SimpleProgressIndex.class).orElse(null)); + } + private PartialPath parseOutputSeries(final String key, final String value) throws Exception { return TwoStageCountProcessor.parseOutputSeries( new PipeParameters(Collections.singletonMap(key, value))); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/TsFileResourceProgressIndexTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/TsFileResourceProgressIndexTest.java index 87b25883fb17a..a0a61e3bbd381 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/TsFileResourceProgressIndexTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/TsFileResourceProgressIndexTest.java @@ -54,6 +54,7 @@ import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.Optional; import java.util.Random; import java.util.stream.IntStream; @@ -213,6 +214,14 @@ public ProgressIndexType getType() { throw new UnsupportedOperationException("method not implemented."); } + @Override + public Optional getProgressIndexByType( + final Class progressIndexClass) { + return progressIndexClass.isInstance(this) + ? Optional.of(progressIndexClass.cast(this)) + : Optional.empty(); + } + @Override public TotalOrderSumTuple getTotalOrderSumTuple() { return new TotalOrderSumTuple((long) val); diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/consensus/index/ProgressIndex.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/consensus/index/ProgressIndex.java index 979eee0c8db3f..83fdac10b5c8e 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/consensus/index/ProgressIndex.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/consensus/index/ProgressIndex.java @@ -34,6 +34,7 @@ import java.nio.ByteBuffer; import java.util.List; import java.util.Map; +import java.util.Optional; import java.util.concurrent.locks.ReentrantReadWriteLock; import java.util.stream.Collectors; import java.util.stream.IntStream; @@ -158,6 +159,15 @@ public abstract ProgressIndex updateToMinimumEqualOrIsAfterProgressIndex( */ public abstract ProgressIndexType getType(); + /** + * Extracts a progress index of the given type from this progress index. + * + *

{@link StateProgressIndex} and {@link HybridProgressIndex} are recursively unwrapped because + * they may contain progress indexes from other causal chains. + */ + public abstract Optional getProgressIndexByType( + Class progressIndexClass); + /** * Get the sum of the tuples of each total order relation of the {@link ProgressIndex}, which is * used for topological sorting of the {@link ProgressIndex}. diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/consensus/index/impl/HybridProgressIndex.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/consensus/index/impl/HybridProgressIndex.java index 2c8895532dcd9..724a150091158 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/consensus/index/impl/HybridProgressIndex.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/consensus/index/impl/HybridProgressIndex.java @@ -36,6 +36,7 @@ import java.util.HashMap; import java.util.Map; import java.util.Objects; +import java.util.Optional; import java.util.concurrent.locks.ReentrantReadWriteLock; import java.util.stream.Collectors; @@ -225,6 +226,30 @@ public ProgressIndexType getType() { return ProgressIndexType.HYBRID_PROGRESS_INDEX; } + @Override + public Optional getProgressIndexByType( + final Class progressIndexClass) { + if (progressIndexClass.isInstance(this)) { + return Optional.of(progressIndexClass.cast(this)); + } + + final Map type2Index = getType2Index(); + // Prefer a direct component over one nested in another composite progress index. + for (final ProgressIndex progressIndex : type2Index.values()) { + if (progressIndexClass.isInstance(progressIndex)) { + return Optional.of(progressIndexClass.cast(progressIndex)); + } + } + for (final ProgressIndex progressIndex : type2Index.values()) { + final Optional extractedProgressIndex = + progressIndex.getProgressIndexByType(progressIndexClass); + if (extractedProgressIndex.isPresent()) { + return extractedProgressIndex; + } + } + return Optional.empty(); + } + @Override public TotalOrderSumTuple getTotalOrderSumTuple() { lock.readLock().lock(); diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/consensus/index/impl/IoTProgressIndex.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/consensus/index/impl/IoTProgressIndex.java index 8f6a24845aa5d..2de6e4cb1351e 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/consensus/index/impl/IoTProgressIndex.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/consensus/index/impl/IoTProgressIndex.java @@ -35,6 +35,7 @@ import java.util.HashMap; import java.util.Map; import java.util.Objects; +import java.util.Optional; import java.util.concurrent.locks.ReentrantReadWriteLock; public class IoTProgressIndex extends ProgressIndex { @@ -197,6 +198,14 @@ public ProgressIndexType getType() { return ProgressIndexType.IOT_PROGRESS_INDEX; } + @Override + public Optional getProgressIndexByType( + final Class progressIndexClass) { + return progressIndexClass.isInstance(this) + ? Optional.of(progressIndexClass.cast(this)) + : Optional.empty(); + } + @Override public TotalOrderSumTuple getTotalOrderSumTuple() { lock.readLock().lock(); diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/consensus/index/impl/MetaProgressIndex.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/consensus/index/impl/MetaProgressIndex.java index 75322152d45c2..80a31a7e4f39a 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/consensus/index/impl/MetaProgressIndex.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/consensus/index/impl/MetaProgressIndex.java @@ -32,6 +32,7 @@ import java.io.OutputStream; import java.nio.ByteBuffer; import java.util.Objects; +import java.util.Optional; import java.util.concurrent.locks.ReentrantReadWriteLock; public class MetaProgressIndex extends ProgressIndex { @@ -153,6 +154,14 @@ public ProgressIndexType getType() { return ProgressIndexType.META_PROGRESS_INDEX; } + @Override + public Optional getProgressIndexByType( + final Class progressIndexClass) { + return progressIndexClass.isInstance(this) + ? Optional.of(progressIndexClass.cast(this)) + : Optional.empty(); + } + @Override public TotalOrderSumTuple getTotalOrderSumTuple() { lock.readLock().lock(); diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/consensus/index/impl/MinimumProgressIndex.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/consensus/index/impl/MinimumProgressIndex.java index e22f82c9fbbc1..b34e97e349339 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/consensus/index/impl/MinimumProgressIndex.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/consensus/index/impl/MinimumProgressIndex.java @@ -28,6 +28,7 @@ import java.io.InputStream; import java.io.OutputStream; import java.nio.ByteBuffer; +import java.util.Optional; public class MinimumProgressIndex extends ProgressIndex { @@ -82,6 +83,14 @@ public ProgressIndexType getType() { return ProgressIndexType.MINIMUM_PROGRESS_INDEX; } + @Override + public Optional getProgressIndexByType( + final Class progressIndexClass) { + return progressIndexClass.isInstance(this) + ? Optional.of(progressIndexClass.cast(this)) + : Optional.empty(); + } + @Override public TotalOrderSumTuple getTotalOrderSumTuple() { return TOTAL_ORDER_SUM_TUPLE; diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/consensus/index/impl/RecoverProgressIndex.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/consensus/index/impl/RecoverProgressIndex.java index 5756594abeb3f..f42ad50f63a6a 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/consensus/index/impl/RecoverProgressIndex.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/consensus/index/impl/RecoverProgressIndex.java @@ -36,6 +36,7 @@ import java.util.HashMap; import java.util.Map; import java.util.Objects; +import java.util.Optional; import java.util.concurrent.locks.ReentrantReadWriteLock; import java.util.stream.Collectors; @@ -202,6 +203,14 @@ public ProgressIndexType getType() { return ProgressIndexType.RECOVER_PROGRESS_INDEX; } + @Override + public Optional getProgressIndexByType( + final Class progressIndexClass) { + return progressIndexClass.isInstance(this) + ? Optional.of(progressIndexClass.cast(this)) + : Optional.empty(); + } + @Override public TotalOrderSumTuple getTotalOrderSumTuple() { lock.readLock().lock(); diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/consensus/index/impl/SimpleProgressIndex.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/consensus/index/impl/SimpleProgressIndex.java index 26d3723725678..dbee0d725a5cf 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/consensus/index/impl/SimpleProgressIndex.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/consensus/index/impl/SimpleProgressIndex.java @@ -32,6 +32,7 @@ import java.io.OutputStream; import java.nio.ByteBuffer; import java.util.Objects; +import java.util.Optional; import java.util.concurrent.locks.ReentrantReadWriteLock; public class SimpleProgressIndex extends ProgressIndex { @@ -179,6 +180,14 @@ public ProgressIndexType getType() { return ProgressIndexType.SIMPLE_PROGRESS_INDEX; } + @Override + public Optional getProgressIndexByType( + final Class progressIndexClass) { + return progressIndexClass.isInstance(this) + ? Optional.of(progressIndexClass.cast(this)) + : Optional.empty(); + } + @Override public TotalOrderSumTuple getTotalOrderSumTuple() { return new TotalOrderSumTuple(memtableFlushOrderId, (long) rebootTimes); diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/consensus/index/impl/StateProgressIndex.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/consensus/index/impl/StateProgressIndex.java index 00a3e4dce8700..96a11989e7f09 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/consensus/index/impl/StateProgressIndex.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/consensus/index/impl/StateProgressIndex.java @@ -36,6 +36,7 @@ import java.util.HashMap; import java.util.Map; import java.util.Objects; +import java.util.Optional; import java.util.concurrent.locks.ReentrantReadWriteLock; /** @@ -196,6 +197,14 @@ public ProgressIndexType getType() { return ProgressIndexType.STATE_PROGRESS_INDEX; } + @Override + public Optional getProgressIndexByType( + final Class progressIndexClass) { + return progressIndexClass.isInstance(this) + ? Optional.of(progressIndexClass.cast(this)) + : getInnerProgressIndex().getProgressIndexByType(progressIndexClass); + } + @Override public TotalOrderSumTuple getTotalOrderSumTuple() { return innerProgressIndex.getTotalOrderSumTuple(); diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/consensus/index/impl/TimeWindowStateProgressIndex.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/consensus/index/impl/TimeWindowStateProgressIndex.java index 02139d2058d83..f98a3a1e512e2 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/consensus/index/impl/TimeWindowStateProgressIndex.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/consensus/index/impl/TimeWindowStateProgressIndex.java @@ -38,6 +38,7 @@ import java.util.HashMap; import java.util.Map; import java.util.Objects; +import java.util.Optional; import java.util.concurrent.locks.ReentrantReadWriteLock; import java.util.stream.Collectors; @@ -209,7 +210,7 @@ public ProgressIndex updateToMinimumEqualOrIsAfterProgressIndex(ProgressIndex pr lock.writeLock().lock(); try { if (!(progressIndex instanceof TimeWindowStateProgressIndex)) { - return this; + return ProgressIndex.blendProgressIndex(this, progressIndex); } final TimeWindowStateProgressIndex thisTimeWindowStateProgressIndex = this; @@ -239,6 +240,14 @@ public ProgressIndexType getType() { return ProgressIndexType.TIME_WINDOW_STATE_PROGRESS_INDEX; } + @Override + public Optional getProgressIndexByType( + final Class progressIndexClass) { + return progressIndexClass.isInstance(this) + ? Optional.of(progressIndexClass.cast(this)) + : Optional.empty(); + } + @Override public TotalOrderSumTuple getTotalOrderSumTuple() { throw new UnsupportedOperationException( diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/source/IoTDBNonDataRegionSource.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/source/IoTDBNonDataRegionSource.java index e8fd5e54e3681..7af52fc0edc15 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/source/IoTDBNonDataRegionSource.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/source/IoTDBNonDataRegionSource.java @@ -20,8 +20,8 @@ package org.apache.iotdb.commons.pipe.source; import org.apache.iotdb.commons.consensus.index.ProgressIndex; +import org.apache.iotdb.commons.consensus.index.impl.HybridProgressIndex; import org.apache.iotdb.commons.consensus.index.impl.MetaProgressIndex; -import org.apache.iotdb.commons.consensus.index.impl.MinimumProgressIndex; import org.apache.iotdb.commons.pipe.datastructure.pattern.IoTDBPipePatternOperations; import org.apache.iotdb.commons.pipe.datastructure.pattern.PipePattern; import org.apache.iotdb.commons.pipe.datastructure.queue.ConcurrentIterableLinkedQueue; @@ -36,6 +36,8 @@ import org.apache.iotdb.pipe.api.exception.PipeException; import org.apache.tsfile.utils.Pair; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import java.util.LinkedList; import java.util.List; @@ -45,6 +47,8 @@ public abstract class IoTDBNonDataRegionSource extends IoTDBSource { + private static final Logger LOGGER = LoggerFactory.getLogger(IoTDBNonDataRegionSource.class); + protected IoTDBPipePatternOperations pipePattern; private List historicalEvents = new LinkedList<>(); @@ -58,6 +62,7 @@ public abstract class IoTDBNonDataRegionSource extends IoTDBSource { // If the extractor is closed, it should not be started again. This is to avoid the case that // the extractor is closed and then be reused by processor. protected final AtomicBoolean hasBeenClosed = new AtomicBoolean(false); + private final AtomicBoolean hasWarnedUnexpectedHybridProgressIndex = new AtomicBoolean(false); protected abstract AbstractPipeListeningQueue getListeningQueue(); @@ -85,14 +90,15 @@ public void start() throws Exception { } final ProgressIndex progressIndex = pipeTaskMeta.getProgressIndex(); + warnIfUnexpectedHybridProgressIndex(progressIndex); + final MetaProgressIndex metaProgressIndex = extractMetaProgressIndex(progressIndex); final long nextIndex = - progressIndex instanceof MinimumProgressIndex + Objects.isNull(metaProgressIndex) // If the index is invalid, the queue is seen as cleared before and thus // needs snapshot re-transferring - || !getListeningQueue() - .isGivenNextIndexValid(((MetaProgressIndex) progressIndex).getIndex() + 1) + || !getListeningQueue().isGivenNextIndexValid(metaProgressIndex.getIndex() + 1) ? getNextIndexAfterSnapshot() - : ((MetaProgressIndex) progressIndex).getIndex() + 1; + : metaProgressIndex.getIndex() + 1; iterator = getListeningQueue().newIterator(nextIndex); super.start(); } @@ -222,10 +228,33 @@ public void close() throws Exception { //////////////////////////// APIs provided for metric framework //////////////////////////// public long getUnTransferredEventCount() { - return !(pipeTaskMeta.getProgressIndex() instanceof MinimumProgressIndex) - ? getListeningQueue().getTailIndex() - - ((MetaProgressIndex) pipeTaskMeta.getProgressIndex()).getIndex() - - 1 + if (Objects.isNull(pipeTaskMeta)) { + return 0L; + } + final ProgressIndex progressIndex = pipeTaskMeta.getProgressIndex(); + warnIfUnexpectedHybridProgressIndex(progressIndex); + final MetaProgressIndex metaProgressIndex = extractMetaProgressIndex(progressIndex); + return Objects.nonNull(metaProgressIndex) + ? getListeningQueue().getTailIndex() - metaProgressIndex.getIndex() - 1 : getListeningQueue().getSize() + historicalEventsCount; } + + private static MetaProgressIndex extractMetaProgressIndex(final ProgressIndex progressIndex) { + return Objects.isNull(progressIndex) + ? null + : progressIndex.getProgressIndexByType(MetaProgressIndex.class).orElse(null); + } + + private void warnIfUnexpectedHybridProgressIndex(final ProgressIndex progressIndex) { + if (Objects.nonNull(progressIndex) + && progressIndex.getProgressIndexByType(HybridProgressIndex.class).isPresent() + && hasWarnedUnexpectedHybridProgressIndex.compareAndSet(false, true)) { + LOGGER.warn( + "Pipe {}@{} encountered an unexpected HybridProgressIndex in {}. Progress index: {}.", + pipeName, + creationTime, + getClass().getSimpleName(), + progressIndex); + } + } } diff --git a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/consensus/index/ProgressIndexTest.java b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/consensus/index/ProgressIndexTest.java new file mode 100644 index 0000000000000..07ee3ef268ea0 --- /dev/null +++ b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/consensus/index/ProgressIndexTest.java @@ -0,0 +1,82 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iotdb.commons.consensus.index; + +import org.apache.iotdb.commons.consensus.index.impl.HybridProgressIndex; +import org.apache.iotdb.commons.consensus.index.impl.MetaProgressIndex; +import org.apache.iotdb.commons.consensus.index.impl.SimpleProgressIndex; +import org.apache.iotdb.commons.consensus.index.impl.StateProgressIndex; +import org.apache.iotdb.commons.consensus.index.impl.TimeWindowStateProgressIndex; + +import org.junit.Assert; +import org.junit.Test; + +import java.util.Collections; + +public class ProgressIndexTest { + + @Test + public void testGetProgressIndexByTypeFromStateWrappedHybridProgressIndex() { + final MetaProgressIndex metaProgressIndex = new MetaProgressIndex(10L); + final SimpleProgressIndex simpleProgressIndex = new SimpleProgressIndex(1, 2L); + final ProgressIndex hybridProgressIndex = + new HybridProgressIndex(metaProgressIndex) + .updateToMinimumEqualOrIsAfterProgressIndex(simpleProgressIndex); + final StateProgressIndex stateProgressIndex = + new StateProgressIndex(1L, Collections.emptyMap(), hybridProgressIndex); + + Assert.assertEquals( + metaProgressIndex, + stateProgressIndex.getProgressIndexByType(MetaProgressIndex.class).orElse(null)); + Assert.assertEquals( + simpleProgressIndex, + stateProgressIndex.getProgressIndexByType(SimpleProgressIndex.class).orElse(null)); + Assert.assertSame( + hybridProgressIndex, + stateProgressIndex.getProgressIndexByType(HybridProgressIndex.class).orElse(null)); + Assert.assertFalse( + stateProgressIndex.getProgressIndexByType(TimeWindowStateProgressIndex.class).isPresent()); + } + + @Test + public void testTimeWindowStateProgressIndexBlendsWithOtherProgressIndexTypes() { + final TimeWindowStateProgressIndex timeWindowStateProgressIndex = + new TimeWindowStateProgressIndex(Collections.emptyMap()); + final SimpleProgressIndex simpleProgressIndex = new SimpleProgressIndex(1, 2L); + + final ProgressIndex blendedProgressIndex = + timeWindowStateProgressIndex.updateToMinimumEqualOrIsAfterProgressIndex( + simpleProgressIndex); + Assert.assertTrue(blendedProgressIndex instanceof HybridProgressIndex); + Assert.assertEquals( + timeWindowStateProgressIndex, + blendedProgressIndex + .getProgressIndexByType(TimeWindowStateProgressIndex.class) + .orElse(null)); + Assert.assertEquals( + simpleProgressIndex, + blendedProgressIndex.getProgressIndexByType(SimpleProgressIndex.class).orElse(null)); + + final ProgressIndex reverseBlendedProgressIndex = + simpleProgressIndex.updateToMinimumEqualOrIsAfterProgressIndex( + timeWindowStateProgressIndex); + Assert.assertEquals(blendedProgressIndex, reverseBlendedProgressIndex); + } +} diff --git a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/source/IoTDBNonDataRegionSourceTest.java b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/source/IoTDBNonDataRegionSourceTest.java new file mode 100644 index 0000000000000..55f10b0828573 --- /dev/null +++ b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/source/IoTDBNonDataRegionSourceTest.java @@ -0,0 +1,135 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iotdb.commons.pipe.source; + +import org.apache.iotdb.commons.consensus.index.ProgressIndex; +import org.apache.iotdb.commons.consensus.index.impl.HybridProgressIndex; +import org.apache.iotdb.commons.consensus.index.impl.MetaProgressIndex; +import org.apache.iotdb.commons.consensus.index.impl.SimpleProgressIndex; +import org.apache.iotdb.commons.consensus.index.impl.StateProgressIndex; +import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta; +import org.apache.iotdb.commons.pipe.datastructure.queue.listening.AbstractPipeListeningQueue; +import org.apache.iotdb.commons.pipe.event.PipeSnapshotEvent; +import org.apache.iotdb.commons.pipe.event.PipeWritePlanEvent; + +import org.junit.Assert; +import org.junit.Test; +import org.mockito.Mockito; + +import java.util.Collections; +import java.util.Optional; + +public class IoTDBNonDataRegionSourceTest { + + @Test + public void testStartWithStateWrappedHybridProgressIndex() throws Exception { + final AbstractPipeListeningQueue listeningQueue = + Mockito.mock(AbstractPipeListeningQueue.class); + Mockito.when(listeningQueue.isGivenNextIndexValid(11L)).thenReturn(true); + + final ProgressIndex hybridProgressIndex = + new HybridProgressIndex(new MetaProgressIndex(10L)) + .updateToMinimumEqualOrIsAfterProgressIndex(new SimpleProgressIndex(1, 2L)); + final StateProgressIndex stateProgressIndex = + new StateProgressIndex(1L, Collections.emptyMap(), hybridProgressIndex); + final TestNonDataRegionSource source = + new TestNonDataRegionSource(listeningQueue, new PipeTaskMeta(stateProgressIndex, 0)); + + source.start(); + + Mockito.verify(listeningQueue).newIterator(11L); + } + + @Test + public void testGetUnTransferredEventCountWithHybridProgressIndex() { + final AbstractPipeListeningQueue listeningQueue = + Mockito.mock(AbstractPipeListeningQueue.class); + Mockito.when(listeningQueue.getTailIndex()).thenReturn(20L); + + final ProgressIndex hybridProgressIndex = + new HybridProgressIndex(new MetaProgressIndex(10L)) + .updateToMinimumEqualOrIsAfterProgressIndex(new SimpleProgressIndex(1, 2L)); + final TestNonDataRegionSource source = + new TestNonDataRegionSource(listeningQueue, new PipeTaskMeta(hybridProgressIndex, 0)); + + Assert.assertEquals(9L, source.getUnTransferredEventCount()); + } + + @Test + public void testGetUnTransferredEventCountWithHybridProgressIndexWithoutMetaIndex() { + final AbstractPipeListeningQueue listeningQueue = + Mockito.mock(AbstractPipeListeningQueue.class); + Mockito.when(listeningQueue.getSize()).thenReturn(7L); + + final TestNonDataRegionSource source = + new TestNonDataRegionSource( + listeningQueue, + new PipeTaskMeta(new HybridProgressIndex(new SimpleProgressIndex(1, 2L)), 0)); + + Assert.assertEquals(7L, source.getUnTransferredEventCount()); + } + + private static final class TestNonDataRegionSource extends IoTDBNonDataRegionSource { + + private final AbstractPipeListeningQueue listeningQueue; + + private TestNonDataRegionSource( + final AbstractPipeListeningQueue listeningQueue, final PipeTaskMeta pipeTaskMeta) { + this.listeningQueue = listeningQueue; + this.pipeTaskMeta = pipeTaskMeta; + } + + @Override + protected AbstractPipeListeningQueue getListeningQueue() { + return listeningQueue; + } + + @Override + protected boolean needTransferSnapshot() { + return false; + } + + @Override + protected void triggerSnapshot() { + // Do nothing + } + + @Override + protected long getMaxBlockingTimeMs() { + return 0L; + } + + @Override + protected Optional trimRealtimeEventByPipePattern( + final PipeWritePlanEvent event) { + return Optional.of(event); + } + + @Override + protected boolean isTypeListened(final PipeWritePlanEvent event) { + return true; + } + + @Override + protected void confineHistoricalEventTransferTypes(final PipeSnapshotEvent event) { + // Do nothing + } + } +}