This is an automated email from the ASF dual-hosted git repository.
rong pushed a commit to branch rc/1.3.3
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/rc/1.3.3 by this push:
new 7f11cb4b338 Pipe: make progress index immutable (#13571)
7f11cb4b338 is described below
commit 7f11cb4b33882a3a36690bf8a7a9beb34b7d0855
Author: V_Galaxy <[email protected]>
AuthorDate: Sun Sep 22 23:33:59 2024 +0800
Pipe: make progress index immutable (#13571)
In Java, immutable objects have clear requirements:
1. The state of the object cannot be modified after creation.
2. All member variables of the object are final.
3. The object is correctly created (during the creation of the object, the
`this` reference does not escape).
This class is designed to be immutable, meaning that instances of
ProgressIndex can be treated as value objects. Immutability ensures thread
safety, as no external synchronization is required for concurrent access to
instances of this class. It also simplifies reasoning about the state of the
object as it cannot change once created.
However, if a ProgressIndex instance holds any mutable objects, like a Map,
they must be deeply copied during construction or when exposed through
accessors to maintain the immutability contract. This prevents unintended
modifications to the underlying mutable state from affecting other parts of the
program.
---
.../PipeHistoricalDataRegionTsFileExtractor.java | 2 +-
.../realtime/assigner/PipeDataRegionAssigner.java | 2 +-
.../PipeTimePartitionProgressIndexKeeper.java | 2 +-
.../processor/aggregate/AggregateProcessor.java | 14 +++--
.../twostage/plugin/TwoStageCountProcessor.java | 12 ++--
.../plan/node/write/InsertMultiTabletsNode.java | 2 +-
.../plan/planner/plan/node/write/InsertNode.java | 2 +-
.../planner/plan/node/write/InsertRowsNode.java | 4 +-
.../plan/node/write/InsertRowsOfOneDeviceNode.java | 2 +-
.../dataregion/tsfile/TsFileResource.java | 4 +-
.../TsFileResourceProgressIndexTest.java | 64 +++++++++++++---------
.../commons/consensus/index/ProgressIndex.java | 43 ++++++---------
.../consensus/index/impl/HybridProgressIndex.java | 41 ++++++--------
.../consensus/index/impl/IoTProgressIndex.java | 17 +++---
.../consensus/index/impl/MetaProgressIndex.java | 25 +++------
.../consensus/index/impl/MinimumProgressIndex.java | 7 +--
.../consensus/index/impl/RecoverProgressIndex.java | 27 ++++-----
.../consensus/index/impl/SimpleProgressIndex.java | 9 +--
.../consensus/index/impl/StateProgressIndex.java | 29 +++++-----
.../index/impl/TimeWindowStateProgressIndex.java | 47 ++++++++--------
.../commons/pipe/task/meta/PipeMetaDeSerTest.java | 13 ++++-
21 files changed, 168 insertions(+), 200 deletions(-)
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/historical/PipeHistoricalDataRegionTsFileExtractor.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/historical/PipeHistoricalDataRegionTsFileExtractor.java
index 98280f8b9ec..4ecc6fc0c67 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/historical/PipeHistoricalDataRegionTsFileExtractor.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/historical/PipeHistoricalDataRegionTsFileExtractor.java
@@ -263,7 +263,7 @@ public class PipeHistoricalDataRegionTsFileExtractor
implements PipeHistoricalDa
pipeName = environment.getPipeName();
creationTime = environment.getCreationTime();
pipeTaskMeta = environment.getPipeTaskMeta();
- startIndex = environment.getPipeTaskMeta().getProgressIndex().deepCopy();
+ startIndex = environment.getPipeTaskMeta().getProgressIndex();
dataRegionId = environment.getRegionId();
synchronized (DATA_REGION_ID_TO_PIPE_FLUSHED_TIME_MAP) {
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/assigner/PipeDataRegionAssigner.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/assigner/PipeDataRegionAssigner.java
index 88f1d53b9ea..1f2500c1839 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/assigner/PipeDataRegionAssigner.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/assigner/PipeDataRegionAssigner.java
@@ -162,7 +162,7 @@ public class PipeDataRegionAssigner implements Closeable {
if (PipeTimePartitionProgressIndexKeeper.getInstance()
.isProgressIndexAfterOrEquals(
dataRegionId, event.getTimePartitionId(),
event.getProgressIndex())) {
-
event.bindProgressIndex(maxProgressIndexForTsFileInsertionEvent.get().deepCopy());
+ event.bindProgressIndex(maxProgressIndexForTsFileInsertionEvent.get());
LOGGER.warn(
"Data region {} bind {} to event {} because it was flushed
prematurely.",
dataRegionId,
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/assigner/PipeTimePartitionProgressIndexKeeper.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/assigner/PipeTimePartitionProgressIndexKeeper.java
index df563eba6bb..893b832b4cb 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/assigner/PipeTimePartitionProgressIndexKeeper.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/assigner/PipeTimePartitionProgressIndexKeeper.java
@@ -42,7 +42,7 @@ public class PipeTimePartitionProgressIndexKeeper {
timePartitionId,
(k, v) -> {
if (v == null) {
- return new Pair<>(progressIndex.deepCopy(), true);
+ return new Pair<>(progressIndex, true);
}
return new Pair<>(
v.getLeft().updateToMinimumEqualOrIsAfterProgressIndex(progressIndex), true);
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 3e6ed81f713..d24e747cfa9 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
@@ -378,17 +378,21 @@ public class AggregateProcessor implements PipeProcessor {
.set(System.currentTimeMillis());
final AtomicReference<Exception> exception = new AtomicReference<>();
- final TimeWindowStateProgressIndex progressIndex =
- new TimeWindowStateProgressIndex(new ConcurrentHashMap<>());
+ final TimeWindowStateProgressIndex[] progressIndex = {
+ new TimeWindowStateProgressIndex(new ConcurrentHashMap<>())
+ };
final Iterable<TabletInsertionEvent> outputEvents =
tabletInsertionEvent.processRowByRow(
(row, rowCollector) ->
- progressIndex.updateToMinimumEqualOrIsAfterProgressIndex(
- new TimeWindowStateProgressIndex(processRow(row,
rowCollector, exception))));
+ progressIndex[0] =
+ (TimeWindowStateProgressIndex)
+
progressIndex[0].updateToMinimumEqualOrIsAfterProgressIndex(
+ new TimeWindowStateProgressIndex(
+ processRow(row, rowCollector, exception))));
// Must reset progressIndex before collection
- ((EnrichedEvent) tabletInsertionEvent).bindProgressIndex(progressIndex);
+ ((EnrichedEvent) tabletInsertionEvent).bindProgressIndex(progressIndex[0]);
outputEvents.forEach(
event -> {
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 07f952277aa..8a40422af00 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
@@ -172,10 +172,8 @@ public class TwoStageCountProcessor implements
PipeProcessor {
: ((PipeRawTabletInsertionEvent) event).count();
localCount.accumulateAndGet(count, Long::sum);
- localCommitProgressIndex.set(
- localCommitProgressIndex
- .get()
-
.updateToMinimumEqualOrIsAfterProgressIndex(event.getProgressIndex()));
+ localCommitProgressIndex.updateAndGet(
+ index ->
index.updateToMinimumEqualOrIsAfterProgressIndex(event.getProgressIndex()));
}
@Override
@@ -199,10 +197,8 @@ public class TwoStageCountProcessor implements
PipeProcessor {
final long count = event.count(true);
localCount.accumulateAndGet(count, Long::sum);
- localCommitProgressIndex.set(
- localCommitProgressIndex
- .get()
-
.updateToMinimumEqualOrIsAfterProgressIndex(event.getProgressIndex()));
+ localCommitProgressIndex.updateAndGet(
+ index ->
index.updateToMinimumEqualOrIsAfterProgressIndex(event.getProgressIndex()));
}
@Override
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertMultiTabletsNode.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertMultiTabletsNode.java
index a5885604645..430ab19c7c9 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertMultiTabletsNode.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertMultiTabletsNode.java
@@ -289,7 +289,7 @@ public class InsertMultiTabletsNode extends InsertNode {
@Override
public void setProgressIndex(ProgressIndex progressIndex) {
- this.progressIndex = progressIndex.deepCopy();
+ this.progressIndex = progressIndex;
insertTabletNodeList.forEach(node -> node.setProgressIndex(progressIndex));
}
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertNode.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertNode.java
index f667e955324..90c0c90d9dd 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertNode.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertNode.java
@@ -281,7 +281,7 @@ public abstract class InsertNode extends SearchNode
implements ComparableConsens
@Override
public void setProgressIndex(ProgressIndex progressIndex) {
- this.progressIndex = progressIndex.deepCopy();
+ this.progressIndex = progressIndex;
}
// endregion
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertRowsNode.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertRowsNode.java
index f5cd98c8f95..13758064128 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertRowsNode.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertRowsNode.java
@@ -286,7 +286,7 @@ public class InsertRowsNode extends InsertNode implements
WALEntryValue {
@Override
public void setProgressIndex(ProgressIndex progressIndex) {
- this.progressIndex = progressIndex.deepCopy();
+ this.progressIndex = progressIndex;
insertRowNodeList.forEach(insertRowNode ->
insertRowNode.setProgressIndex(progressIndex));
}
@@ -297,7 +297,7 @@ public class InsertRowsNode extends InsertNode implements
WALEntryValue {
this.progressIndex =
(this.progressIndex == null)
- ? progressIndex.deepCopy()
+ ? progressIndex
:
this.progressIndex.updateToMinimumEqualOrIsAfterProgressIndex(progressIndex);
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertRowsOfOneDeviceNode.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertRowsOfOneDeviceNode.java
index 86764946903..0ec16ef69c7 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertRowsOfOneDeviceNode.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertRowsOfOneDeviceNode.java
@@ -331,7 +331,7 @@ public class InsertRowsOfOneDeviceNode extends InsertNode {
@Override
public void setProgressIndex(ProgressIndex progressIndex) {
- this.progressIndex = progressIndex.deepCopy();
+ this.progressIndex = progressIndex;
insertRowNodeList.forEach(insertRowNode ->
insertRowNode.setProgressIndex(progressIndex));
}
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileResource.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileResource.java
index 2d08f7d4a68..bdcdfb2f75b 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileResource.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileResource.java
@@ -1136,7 +1136,7 @@ public class TsFileResource {
maxProgressIndex =
(maxProgressIndex == null
- ? progressIndex.deepCopy()
+ ? progressIndex
:
maxProgressIndex.updateToMinimumEqualOrIsAfterProgressIndex(progressIndex));
PipeTimePartitionProgressIndexKeeper.getInstance()
@@ -1148,7 +1148,7 @@ public class TsFileResource {
return;
}
- maxProgressIndex = progressIndex.deepCopy();
+ maxProgressIndex = progressIndex;
PipeTimePartitionProgressIndexKeeper.getInstance()
.updateProgressIndex(getDataRegionId(), getTimePartition(),
maxProgressIndex);
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 bb36674a7bb..e82ac0e08e5 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
@@ -104,15 +104,26 @@ public class TsFileResourceProgressIndexTest {
public void testProgressIndexRecorder() {
HybridProgressIndex hybridProgressIndex =
new HybridProgressIndex(new SimpleProgressIndex(3, 4));
- hybridProgressIndex.updateToMinimumEqualOrIsAfterProgressIndex(new
SimpleProgressIndex(6, 6));
- hybridProgressIndex.updateToMinimumEqualOrIsAfterProgressIndex(
- new RecoverProgressIndex(1, new SimpleProgressIndex(1, 2)));
- hybridProgressIndex.updateToMinimumEqualOrIsAfterProgressIndex(
- new RecoverProgressIndex(1, new SimpleProgressIndex(1, 3)));
- hybridProgressIndex.updateToMinimumEqualOrIsAfterProgressIndex(
- new RecoverProgressIndex(2, new SimpleProgressIndex(4, 3)));
- hybridProgressIndex.updateToMinimumEqualOrIsAfterProgressIndex(
- new RecoverProgressIndex(3, new SimpleProgressIndex(5, 5)));
+ hybridProgressIndex =
+ (HybridProgressIndex)
+ hybridProgressIndex.updateToMinimumEqualOrIsAfterProgressIndex(
+ new SimpleProgressIndex(6, 6));
+ hybridProgressIndex =
+ (HybridProgressIndex)
+ hybridProgressIndex.updateToMinimumEqualOrIsAfterProgressIndex(
+ new RecoverProgressIndex(1, new SimpleProgressIndex(1, 2)));
+ hybridProgressIndex =
+ (HybridProgressIndex)
+ hybridProgressIndex.updateToMinimumEqualOrIsAfterProgressIndex(
+ new RecoverProgressIndex(1, new SimpleProgressIndex(1, 3)));
+ hybridProgressIndex =
+ (HybridProgressIndex)
+ hybridProgressIndex.updateToMinimumEqualOrIsAfterProgressIndex(
+ new RecoverProgressIndex(2, new SimpleProgressIndex(4, 3)));
+ hybridProgressIndex =
+ (HybridProgressIndex)
+ hybridProgressIndex.updateToMinimumEqualOrIsAfterProgressIndex(
+ new RecoverProgressIndex(3, new SimpleProgressIndex(5, 5)));
Assert.assertTrue(hybridProgressIndex.isAfter(new SimpleProgressIndex(6,
5)));
Assert.assertTrue(
hybridProgressIndex.isAfter(new RecoverProgressIndex(3, new
SimpleProgressIndex(5, 4))));
@@ -191,11 +202,6 @@ public class TsFileResourceProgressIndexTest {
return this.type == that.type && this.val == that.val;
}
- @Override
- public ProgressIndex deepCopy() {
- return new MockProgressIndex(type, val);
- }
-
@Override
public ProgressIndex
updateToMinimumEqualOrIsAfterProgressIndex(ProgressIndex progressIndex) {
if (!(progressIndex instanceof MockProgressIndex)) {
@@ -225,9 +231,11 @@ public class TsFileResourceProgressIndexTest {
final IoTProgressIndex ioTProgressIndex = new IoTProgressIndex(1, 123L);
final RecoverProgressIndex recoverProgressIndex =
new RecoverProgressIndex(1, new SimpleProgressIndex(2, 2));
- final HybridProgressIndex hybridProgressIndex = new
HybridProgressIndex(ioTProgressIndex);
+ HybridProgressIndex hybridProgressIndex = new
HybridProgressIndex(ioTProgressIndex);
-
hybridProgressIndex.updateToMinimumEqualOrIsAfterProgressIndex(recoverProgressIndex);
+ hybridProgressIndex =
+ (HybridProgressIndex)
+
hybridProgressIndex.updateToMinimumEqualOrIsAfterProgressIndex(recoverProgressIndex);
Assert.assertTrue(hybridProgressIndex.isAfter(new IoTProgressIndex(1,
100L)));
Assert.assertTrue(
@@ -331,18 +339,22 @@ public class TsFileResourceProgressIndexTest {
new IoTProgressIndex(
random.nextInt(peerIdRange), (long)
random.nextInt(searchIndexRange)));
if (random.nextInt(2) == 1) {
- hybridProgressIndex.updateToMinimumEqualOrIsAfterProgressIndex(
- new SimpleProgressIndex(
- random.nextInt(rebootTimesRange),
- random.nextInt(memtableFlushOrderIdRange)));
+ hybridProgressIndex =
+ (HybridProgressIndex)
+
hybridProgressIndex.updateToMinimumEqualOrIsAfterProgressIndex(
+ new SimpleProgressIndex(
+ random.nextInt(rebootTimesRange),
+ random.nextInt(memtableFlushOrderIdRange)));
}
if (random.nextInt(2) == 1) {
- hybridProgressIndex.updateToMinimumEqualOrIsAfterProgressIndex(
- new RecoverProgressIndex(
- random.nextInt(dataNodeIdRange),
- new SimpleProgressIndex(
- random.nextInt(rebootTimesRange),
- random.nextInt(memtableFlushOrderIdRange))));
+ hybridProgressIndex =
+ (HybridProgressIndex)
+
hybridProgressIndex.updateToMinimumEqualOrIsAfterProgressIndex(
+ new RecoverProgressIndex(
+ random.nextInt(dataNodeIdRange),
+ new SimpleProgressIndex(
+ random.nextInt(rebootTimesRange),
+
random.nextInt(memtableFlushOrderIdRange))));
}
progressIndexList.add(hybridProgressIndex);
});
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 3160b1c23a1..43f934c5b9b 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
@@ -31,6 +31,7 @@ import java.io.IOException;
import java.io.OutputStream;
import java.nio.ByteBuffer;
import java.util.List;
+import java.util.Map;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
import java.util.stream.LongStream;
@@ -41,6 +42,16 @@ import java.util.stream.LongStream;
* chain. Since strict total order relations can be defined on each of these
causal chains, the
* progress index is considered to be composed of an n-tuple of total order
relations
* (S<sub>1</sub>,S<sub>2</sub>,S<sub>3</sub>,.... ,S<sub>n</sub>).
+ *
+ * <p>This class is designed to be immutable, meaning that instances of {@link
ProgressIndex} can be
+ * treated as value objects. Immutability ensures thread safety, as no
external synchronization is
+ * required for concurrent access to instances of this class. It also
simplifies reasoning about the
+ * state of the object as it cannot change once created.
+ *
+ * <p>However, if a {@link ProgressIndex} instance holds any mutable objects,
like a {@link Map},
+ * they must be deeply copied during construction or when exposed through
accessors to maintain the
+ * immutability contract. This prevents unintended modifications to the
underlying mutable state
+ * from affecting other parts of the program.
*/
public abstract class ProgressIndex {
@@ -95,22 +106,6 @@ public abstract class ProgressIndex {
return super.hashCode();
}
- /**
- * Creates and returns a deep copy of this {@link ProgressIndex} instance.
- *
- * <p>This method performs a deep copy, meaning all nested objects and
fields within this {@link
- * ProgressIndex} are recursively copied, resulting in a new instance that
is independent of the
- * original. Modifications to the copied instance will not affect the
original instance and vice
- * versa.
- *
- * <p>When constructing or updating another {@link ProgressIndex} using an
existing {@link
- * ProgressIndex}, it is recommended to perform a deep copy of the existing
instance to avoid
- * unintended modifications or shared state between the instances.
- *
- * @return a new {@link ProgressIndex} instance that is a deep copy of this
progress index
- */
- public abstract ProgressIndex deepCopy();
-
/**
* Define the isEqualOrAfter relation, A.isEqualOrAfter(B) if and only if
each tuple member in A
* is greater than or equal to B in the corresponding total order relation.
@@ -128,13 +123,11 @@ public abstract class ProgressIndex {
*
A.updateToMinimumIsAfterProgressIndex(B).equals(B.updateToMinimumIsAfterProgressIndex(A))
is
* {@code true}
*
- * <p>Note: this function may modify the caller (this) but will not modify
{@param progressIndex}.
+ * <p>Note: this function will not modify the caller (this) and {@param
progressIndex}.
*
* @param progressIndex the {@link ProgressIndex} to be compared
* @return the minimum {@link ProgressIndex} after the given {@link
ProgressIndex} and this {@link
- * ProgressIndex}, the returned {@link ProgressIndex} will contain deep
copies of all
- * references to the given {@param progressIndex}, ensuring no shared
state between the
- * original and the result
+ * ProgressIndex}.
*/
public abstract ProgressIndex updateToMinimumEqualOrIsAfterProgressIndex(
ProgressIndex progressIndex);
@@ -183,9 +176,7 @@ public abstract class ProgressIndex {
* should be the minimum {@link ProgressIndex} equal or after the first
{@link ProgressIndex}
* and the second {@link ProgressIndex}
* @return the minimum {@link ProgressIndex} after the first {@link
ProgressIndex} and the second
- * {@link ProgressIndex}, the returned {@link ProgressIndex} will
contain deep copies of all
- * references to {@param progressIndex1} and {@param progressIndex2},
ensuring that the result
- * is independent and modifications to it do not affect the original
instances
+ * {@link ProgressIndex}.
*/
protected static ProgressIndex blendProgressIndex(
ProgressIndex progressIndex1, ProgressIndex progressIndex2) {
@@ -193,14 +184,14 @@ public abstract class ProgressIndex {
return MinimumProgressIndex.INSTANCE;
}
if (progressIndex1 == null || progressIndex1 instanceof
MinimumProgressIndex) {
- return progressIndex2 == null ? MinimumProgressIndex.INSTANCE :
progressIndex2.deepCopy();
+ return progressIndex2 == null ? MinimumProgressIndex.INSTANCE :
progressIndex2;
}
if (progressIndex2 == null || progressIndex2 instanceof
MinimumProgressIndex) {
- return progressIndex1.deepCopy(); // progressIndex1 is not null
+ return progressIndex1; // progressIndex1 is not null
}
return progressIndex1 instanceof StateProgressIndex
- ?
progressIndex1.deepCopy().updateToMinimumEqualOrIsAfterProgressIndex(progressIndex2)
+ ?
progressIndex1.updateToMinimumEqualOrIsAfterProgressIndex(progressIndex2)
: new HybridProgressIndex(progressIndex1)
.updateToMinimumEqualOrIsAfterProgressIndex(progressIndex2);
}
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 efa566a9c8d..ac0e50b317c 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
@@ -31,9 +31,9 @@ import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.nio.ByteBuffer;
+import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
-import java.util.Map.Entry;
import java.util.Objects;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.stream.Collectors;
@@ -45,24 +45,19 @@ public class HybridProgressIndex extends ProgressIndex {
private final Map<Short, ProgressIndex> type2Index;
private HybridProgressIndex() {
- this.type2Index = new HashMap<>();
+ this(Collections.emptyMap());
}
public HybridProgressIndex(ProgressIndex progressIndex) {
- short type = progressIndex.getType().getType();
- this.type2Index = new HashMap<>();
- if (ProgressIndexType.HYBRID_PROGRESS_INDEX.getType() != type) {
- this.type2Index.put(type, progressIndex.deepCopy());
- } else {
- for (Entry<Short, ProgressIndex> entry :
- ((HybridProgressIndex) progressIndex).type2Index.entrySet()) {
- this.type2Index.put(entry.getKey(), entry.getValue().deepCopy());
- }
- }
+ this(Collections.singletonMap(progressIndex.getType().getType(),
progressIndex));
+ }
+
+ private HybridProgressIndex(Map<Short, ProgressIndex> type2Index) {
+ this.type2Index = new HashMap<>(type2Index);
}
public Map<Short, ProgressIndex> getType2Index() {
- return ImmutableMap.copyOf(((HybridProgressIndex) deepCopy()).type2Index);
+ return ImmutableMap.copyOf(type2Index);
}
@Override
@@ -175,11 +170,6 @@ public class HybridProgressIndex extends ProgressIndex {
return Objects.hash(type2Index);
}
- @Override
- public ProgressIndex deepCopy() {
- return new HybridProgressIndex(this);
- }
-
@Override
public ProgressIndex
updateToMinimumEqualOrIsAfterProgressIndex(ProgressIndex progressIndex) {
lock.writeLock().lock();
@@ -189,30 +179,33 @@ public class HybridProgressIndex extends ProgressIndex {
}
if (progressIndex instanceof StateProgressIndex) {
- return
progressIndex.deepCopy().updateToMinimumEqualOrIsAfterProgressIndex(this);
+ return progressIndex.updateToMinimumEqualOrIsAfterProgressIndex(this);
}
if (!(progressIndex instanceof HybridProgressIndex)) {
+ final Map<Short, ProgressIndex> type2Index = new
HashMap<>(this.type2Index);
type2Index.compute(
progressIndex.getType().getType(),
(thisK, thisV) ->
(thisV == null
- ? progressIndex.deepCopy()
+ ? progressIndex
:
thisV.updateToMinimumEqualOrIsAfterProgressIndex(progressIndex)));
- return this;
+ return new HybridProgressIndex(type2Index);
}
final HybridProgressIndex thisHybridProgressIndex = this;
final HybridProgressIndex thatHybridProgressIndex =
(HybridProgressIndex) progressIndex;
+ final Map<Short, ProgressIndex> type2Index =
+ new HashMap<>(thisHybridProgressIndex.type2Index);
thatHybridProgressIndex.type2Index.forEach(
(thatK, thatV) ->
- thisHybridProgressIndex.type2Index.compute(
+ type2Index.compute(
thatK,
(thisK, thisV) ->
(thisV == null
- ? thatV.deepCopy()
+ ? thatV
:
thisV.updateToMinimumEqualOrIsAfterProgressIndex(thatV))));
- return this;
+ return new HybridProgressIndex(type2Index);
} finally {
lock.writeLock().unlock();
}
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 a95faea31e8..5c7ffb2cc81 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
@@ -22,7 +22,6 @@ package org.apache.iotdb.commons.consensus.index.impl;
import org.apache.iotdb.commons.consensus.index.ProgressIndex;
import org.apache.iotdb.commons.consensus.index.ProgressIndexType;
-import com.google.common.collect.ImmutableMap;
import org.apache.tsfile.utils.ReadWriteIOUtils;
import javax.annotation.Nonnull;
@@ -31,6 +30,7 @@ import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.nio.ByteBuffer;
+import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
@@ -43,11 +43,11 @@ public class IoTProgressIndex extends ProgressIndex {
private final Map<Integer, Long> peerId2SearchIndex;
private IoTProgressIndex() {
- peerId2SearchIndex = new HashMap<>();
+ this(Collections.emptyMap());
}
public IoTProgressIndex(Integer peerId, Long searchIndex) {
- this(ImmutableMap.of(peerId, searchIndex));
+ this(Collections.singletonMap(peerId, searchIndex));
}
public IoTProgressIndex(Map<Integer, Long> peerId2SearchIndex) {
@@ -159,11 +159,6 @@ public class IoTProgressIndex extends ProgressIndex {
return Objects.hash(peerId2SearchIndex);
}
- @Override
- public ProgressIndex deepCopy() {
- return new IoTProgressIndex(peerId2SearchIndex);
- }
-
@Override
public ProgressIndex
updateToMinimumEqualOrIsAfterProgressIndex(ProgressIndex progressIndex) {
lock.writeLock().lock();
@@ -174,11 +169,13 @@ public class IoTProgressIndex extends ProgressIndex {
final IoTProgressIndex thisIoTProgressIndex = this;
final IoTProgressIndex thatIoTProgressIndex = (IoTProgressIndex)
progressIndex;
+ final Map<Integer, Long> peerId2SearchIndex =
+ new HashMap<>(thisIoTProgressIndex.peerId2SearchIndex);
thatIoTProgressIndex.peerId2SearchIndex.forEach(
(thatK, thatV) ->
- thisIoTProgressIndex.peerId2SearchIndex.compute(
+ peerId2SearchIndex.compute(
thatK, (thisK, thisV) -> (thisV == null ? thatV :
Math.max(thisV, thatV))));
- return this;
+ return new IoTProgressIndex(peerId2SearchIndex);
} finally {
lock.writeLock().unlock();
}
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 5349f83d294..c181345e4ad 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
@@ -37,16 +37,12 @@ public class MetaProgressIndex extends ProgressIndex {
private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
- private long index;
+ private final long index;
public MetaProgressIndex(long index) {
this.index = index;
}
- private MetaProgressIndex() {
- // Empty constructor
- }
-
public long getIndex() {
return index;
}
@@ -134,11 +130,6 @@ public class MetaProgressIndex extends ProgressIndex {
return Objects.hash(index);
}
- @Override
- public ProgressIndex deepCopy() {
- return new MetaProgressIndex(index);
- }
-
@Override
public ProgressIndex
updateToMinimumEqualOrIsAfterProgressIndex(ProgressIndex progressIndex) {
lock.writeLock().lock();
@@ -147,8 +138,10 @@ public class MetaProgressIndex extends ProgressIndex {
return ProgressIndex.blendProgressIndex(this, progressIndex);
}
- this.index = Math.max(this.index, ((MetaProgressIndex)
progressIndex).index);
- return this;
+ final MetaProgressIndex thisMetaProgressIndex = this;
+ final MetaProgressIndex thatMetaProgressIndex = (MetaProgressIndex)
progressIndex;
+ return new MetaProgressIndex(
+ Math.max(thisMetaProgressIndex.index, thatMetaProgressIndex.index));
} finally {
lock.writeLock().unlock();
}
@@ -169,15 +162,11 @@ public class MetaProgressIndex extends ProgressIndex {
}
public static MetaProgressIndex deserializeFrom(ByteBuffer byteBuffer) {
- final MetaProgressIndex metaProgressIndex = new MetaProgressIndex();
- metaProgressIndex.index = ReadWriteIOUtils.readLong(byteBuffer);
- return metaProgressIndex;
+ return new MetaProgressIndex(ReadWriteIOUtils.readLong(byteBuffer));
}
public static MetaProgressIndex deserializeFrom(InputStream stream) throws
IOException {
- final MetaProgressIndex metaProgressIndex = new MetaProgressIndex();
- metaProgressIndex.index = ReadWriteIOUtils.readLong(stream);
- return metaProgressIndex;
+ return new MetaProgressIndex(ReadWriteIOUtils.readLong(stream));
}
@Override
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 cd5b254f74f..e476409a30b 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
@@ -72,14 +72,9 @@ public class MinimumProgressIndex extends ProgressIndex {
return 0;
}
- @Override
- public ProgressIndex deepCopy() {
- return INSTANCE;
- }
-
@Override
public ProgressIndex
updateToMinimumEqualOrIsAfterProgressIndex(ProgressIndex progressIndex) {
- return progressIndex == null ? this : progressIndex.deepCopy();
+ return progressIndex == null ? this : progressIndex;
}
@Override
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 3a743bcb5d5..c2511222eab 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
@@ -31,9 +31,9 @@ import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.nio.ByteBuffer;
+import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
-import java.util.Map.Entry;
import java.util.Objects;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.stream.Collectors;
@@ -45,23 +45,19 @@ public class RecoverProgressIndex extends ProgressIndex {
private final Map<Integer, SimpleProgressIndex> dataNodeId2LocalIndex;
private RecoverProgressIndex() {
- this.dataNodeId2LocalIndex = new HashMap<>();
+ this(Collections.emptyMap());
}
public RecoverProgressIndex(int dataNodeId, SimpleProgressIndex
simpleProgressIndex) {
- this(ImmutableMap.of(dataNodeId, simpleProgressIndex));
+ this(Collections.singletonMap(dataNodeId, simpleProgressIndex));
}
public RecoverProgressIndex(Map<Integer, SimpleProgressIndex>
dataNodeId2LocalIndex) {
- this.dataNodeId2LocalIndex = new HashMap<>();
- for (Entry<Integer, SimpleProgressIndex> entry :
dataNodeId2LocalIndex.entrySet()) {
- this.dataNodeId2LocalIndex.put(
- entry.getKey(), (SimpleProgressIndex) entry.getValue().deepCopy());
- }
+ this.dataNodeId2LocalIndex = new HashMap<>(dataNodeId2LocalIndex);
}
public Map<Integer, SimpleProgressIndex> getDataNodeId2LocalIndex() {
- return ImmutableMap.copyOf(((RecoverProgressIndex)
deepCopy()).dataNodeId2LocalIndex);
+ return ImmutableMap.copyOf(dataNodeId2LocalIndex);
}
@Override
@@ -171,11 +167,6 @@ public class RecoverProgressIndex extends ProgressIndex {
return Objects.hash(dataNodeId2LocalIndex);
}
- @Override
- public ProgressIndex deepCopy() {
- return new RecoverProgressIndex(dataNodeId2LocalIndex);
- }
-
@Override
public ProgressIndex
updateToMinimumEqualOrIsAfterProgressIndex(ProgressIndex progressIndex) {
lock.writeLock().lock();
@@ -186,16 +177,18 @@ public class RecoverProgressIndex extends ProgressIndex {
final RecoverProgressIndex thisRecoverProgressIndex = this;
final RecoverProgressIndex thatRecoverProgressIndex =
(RecoverProgressIndex) progressIndex;
+ final Map<Integer, SimpleProgressIndex> dataNodeId2LocalIndex =
+ new HashMap<>(thisRecoverProgressIndex.dataNodeId2LocalIndex);
thatRecoverProgressIndex.dataNodeId2LocalIndex.forEach(
(thatK, thatV) ->
- thisRecoverProgressIndex.dataNodeId2LocalIndex.compute(
+ dataNodeId2LocalIndex.compute(
thatK,
(thisK, thisV) ->
(thisV == null
- ? (SimpleProgressIndex) thatV.deepCopy()
+ ? thatV
: (SimpleProgressIndex)
thisV.updateToMinimumEqualOrIsAfterProgressIndex(thatV))));
- return this;
+ return new RecoverProgressIndex(dataNodeId2LocalIndex);
} finally {
lock.writeLock().unlock();
}
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 267da7a8b50..209fa21d321 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
@@ -140,11 +140,6 @@ public class SimpleProgressIndex extends ProgressIndex {
return Objects.hash(rebootTimes, memtableFlushOrderId);
}
- @Override
- public ProgressIndex deepCopy() {
- return new SimpleProgressIndex(rebootTimes, memtableFlushOrderId);
- }
-
@Override
public ProgressIndex
updateToMinimumEqualOrIsAfterProgressIndex(ProgressIndex progressIndex) {
lock.writeLock().lock();
@@ -159,7 +154,7 @@ public class SimpleProgressIndex extends ProgressIndex {
return this;
}
if (thisSimpleProgressIndex.rebootTimes <
thatSimpleProgressIndex.rebootTimes) {
- return progressIndex.deepCopy();
+ return progressIndex;
}
// thisSimpleProgressIndex.rebootTimes ==
thatSimpleProgressIndex.rebootTimes
if (thisSimpleProgressIndex.memtableFlushOrderId
@@ -168,7 +163,7 @@ public class SimpleProgressIndex extends ProgressIndex {
}
if (thisSimpleProgressIndex.memtableFlushOrderId
< thatSimpleProgressIndex.memtableFlushOrderId) {
- return progressIndex.deepCopy();
+ return progressIndex;
}
// thisSimpleProgressIndex.memtableFlushOrderId ==
// thatSimpleProgressIndex.memtableFlushOrderId
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 e66dae98d49..9eca0795e51 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
@@ -39,23 +39,23 @@ import java.util.concurrent.locks.ReentrantReadWriteLock;
/**
* NOTE: Currently, {@link StateProgressIndex} does not perform deep copies of
the {@link Binary}
- * during construction or updates, which may lead to unintended shared state
or modifications. This
- * behavior should be reviewed and adjusted as necessary to ensure the
integrity and independence of
- * the progress index instances.
+ * during construction or when exposed through accessors, which may lead to
unintended shared state
+ * or modifications. This behavior should be reviewed and adjusted as
necessary to ensure the
+ * integrity and independence of the progress index instances.
*/
public class StateProgressIndex extends ProgressIndex {
private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
- private long version;
- private Map<String, Binary> state;
- private ProgressIndex innerProgressIndex;
+ private final long version;
+ private final Map<String, Binary> state;
+ private final ProgressIndex innerProgressIndex;
public StateProgressIndex(
long version, Map<String, Binary> state, ProgressIndex
innerProgressIndex) {
this.version = version;
this.state = new HashMap<>(state);
- this.innerProgressIndex = innerProgressIndex.deepCopy();
+ this.innerProgressIndex = innerProgressIndex;
}
public long getVersion() {
@@ -63,9 +63,7 @@ public class StateProgressIndex extends ProgressIndex {
}
public ProgressIndex getInnerProgressIndex() {
- return innerProgressIndex == null
- ? MinimumProgressIndex.INSTANCE
- : innerProgressIndex.deepCopy();
+ return innerProgressIndex == null ? MinimumProgressIndex.INSTANCE :
innerProgressIndex;
}
public Map<String, Binary> getState() {
@@ -167,15 +165,14 @@ public class StateProgressIndex extends ProgressIndex {
return Objects.hash(innerProgressIndex, version);
}
- @Override
- public ProgressIndex deepCopy() {
- return new StateProgressIndex(version, state, innerProgressIndex);
- }
-
@Override
public ProgressIndex
updateToMinimumEqualOrIsAfterProgressIndex(ProgressIndex progressIndex) {
lock.writeLock().lock();
try {
+ long version = this.version;
+ Map<String, Binary> state = new HashMap<>(this.state);
+ ProgressIndex innerProgressIndex = this.innerProgressIndex;
+
innerProgressIndex =
innerProgressIndex.updateToMinimumEqualOrIsAfterProgressIndex(
progressIndex instanceof StateProgressIndex
@@ -186,7 +183,7 @@ public class StateProgressIndex extends ProgressIndex {
version = ((StateProgressIndex) progressIndex).version;
state = ((StateProgressIndex) progressIndex).state;
}
- return this;
+ return new StateProgressIndex(version, state, innerProgressIndex);
} finally {
lock.writeLock().unlock();
}
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 4195f027681..37d6b6fdf38 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
@@ -32,6 +32,7 @@ import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.nio.ByteBuffer;
+import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
@@ -40,16 +41,16 @@ import java.util.stream.Collectors;
/**
* NOTE: Currently, {@link TimeWindowStateProgressIndex} does not perform deep
copies of the {@link
- * ByteBuffer} during construction or updates, which may lead to unintended
shared state or
- * modifications. This behavior should be reviewed and adjusted as necessary
to ensure the integrity
- * and independence of the progress index instances.
+ * ByteBuffer} and {@link Pair} during construction or when exposed through
accessors, which may
+ * lead to unintended shared state or modifications. This behavior should be
reviewed and adjusted
+ * as necessary to ensure the integrity and independence of the progress index
instances.
*/
public class TimeWindowStateProgressIndex extends ProgressIndex {
private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
// Only the byteBuffer is nullable, the timeSeries, pair and timestamp must
not be null
- private Map<String, Pair<Long, ByteBuffer>>
timeSeries2TimestampWindowBufferPairMap;
+ private final Map<String, Pair<Long, ByteBuffer>>
timeSeries2TimestampWindowBufferPairMap;
public TimeWindowStateProgressIndex(
@Nonnull Map<String, Pair<Long, ByteBuffer>>
timeSeries2TimestampWindowBufferPairMap) {
@@ -58,7 +59,7 @@ public class TimeWindowStateProgressIndex extends
ProgressIndex {
}
private TimeWindowStateProgressIndex() {
- // Empty constructor
+ this(Collections.emptyMap());
}
public Map<String, Pair<Long, ByteBuffer>>
getTimeSeries2TimestampWindowBufferPairMap() {
@@ -194,11 +195,6 @@ public class TimeWindowStateProgressIndex extends
ProgressIndex {
return Objects.hash(timeSeries2TimestampWindowBufferPairMap);
}
- @Override
- public ProgressIndex deepCopy() {
- return new
TimeWindowStateProgressIndex(timeSeries2TimestampWindowBufferPairMap);
- }
-
@Override
public ProgressIndex
updateToMinimumEqualOrIsAfterProgressIndex(ProgressIndex progressIndex) {
lock.writeLock().lock();
@@ -207,18 +203,23 @@ public class TimeWindowStateProgressIndex extends
ProgressIndex {
return this;
}
- this.timeSeries2TimestampWindowBufferPairMap.putAll(
- ((TimeWindowStateProgressIndex) progressIndex)
- .timeSeries2TimestampWindowBufferPairMap.entrySet().stream()
- .filter(
- entry ->
-
!this.timeSeries2TimestampWindowBufferPairMap.containsKey(entry.getKey())
- || this.timeSeries2TimestampWindowBufferPairMap
- .get(entry.getKey())
- .getLeft()
- <= entry.getValue().getLeft())
- .collect(Collectors.toMap(Map.Entry::getKey,
Map.Entry::getValue)));
- return this;
+ final TimeWindowStateProgressIndex thisTimeWindowStateProgressIndex =
this;
+ final TimeWindowStateProgressIndex thatTimeWindowStateProgressIndex =
+ (TimeWindowStateProgressIndex) progressIndex;
+ final Map<String, Pair<Long, ByteBuffer>>
timeSeries2TimestampWindowBufferPairMap =
+ new
HashMap<>(thisTimeWindowStateProgressIndex.timeSeries2TimestampWindowBufferPairMap);
+ timeSeries2TimestampWindowBufferPairMap.putAll(
+ thatTimeWindowStateProgressIndex
+ .timeSeries2TimestampWindowBufferPairMap
+ .entrySet()
+ .stream()
+ .filter(
+ entry ->
+
!timeSeries2TimestampWindowBufferPairMap.containsKey(entry.getKey())
+ ||
timeSeries2TimestampWindowBufferPairMap.get(entry.getKey()).getLeft()
+ <= entry.getValue().getLeft())
+ .collect(Collectors.toMap(Map.Entry::getKey,
Map.Entry::getValue)));
+ return new
TimeWindowStateProgressIndex(timeSeries2TimestampWindowBufferPairMap);
} finally {
lock.writeLock().unlock();
}
@@ -238,7 +239,6 @@ public class TimeWindowStateProgressIndex extends
ProgressIndex {
public static TimeWindowStateProgressIndex deserializeFrom(ByteBuffer
byteBuffer) {
final TimeWindowStateProgressIndex timeWindowStateProgressIndex =
new TimeWindowStateProgressIndex();
- timeWindowStateProgressIndex.timeSeries2TimestampWindowBufferPairMap = new
HashMap<>();
final int size = ReadWriteIOUtils.readInt(byteBuffer);
for (int i = 0; i < size; ++i) {
@@ -261,7 +261,6 @@ public class TimeWindowStateProgressIndex extends
ProgressIndex {
throws IOException {
final TimeWindowStateProgressIndex timeWindowStateProgressIndex =
new TimeWindowStateProgressIndex();
- timeWindowStateProgressIndex.timeSeries2TimestampWindowBufferPairMap = new
HashMap<>();
final int size = ReadWriteIOUtils.readInt(stream);
for (int i = 0; i < size; ++i) {
diff --git
a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/task/meta/PipeMetaDeSerTest.java
b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/task/meta/PipeMetaDeSerTest.java
index 419baea4b82..3a0fd3646fe 100644
---
a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/task/meta/PipeMetaDeSerTest.java
+++
b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/task/meta/PipeMetaDeSerTest.java
@@ -68,8 +68,14 @@ public class PipeMetaDeSerTest {
HybridProgressIndex hybridProgressIndex =
new HybridProgressIndex(new SimpleProgressIndex(1, 2));
- hybridProgressIndex.updateToMinimumEqualOrIsAfterProgressIndex(new
SimpleProgressIndex(2, 4));
- hybridProgressIndex.updateToMinimumEqualOrIsAfterProgressIndex(new
IoTProgressIndex(3, 6L));
+ hybridProgressIndex =
+ (HybridProgressIndex)
+ hybridProgressIndex.updateToMinimumEqualOrIsAfterProgressIndex(
+ new SimpleProgressIndex(2, 4));
+ hybridProgressIndex =
+ (HybridProgressIndex)
+ hybridProgressIndex.updateToMinimumEqualOrIsAfterProgressIndex(
+ new IoTProgressIndex(3, 6L));
Map<String, Pair<Long, ByteBuffer>>
timeSeries2TimestampWindowBufferPairMap = new HashMap<>();
ByteBuffer buffer;
@@ -80,6 +86,7 @@ public class PipeMetaDeSerTest {
}
timeSeries2TimestampWindowBufferPairMap.put("root.test.a1", new
Pair<>(123L, buffer));
+ final HybridProgressIndex finalHybridProgressIndex = hybridProgressIndex;
PipeRuntimeMeta pipeRuntimeMeta =
new PipeRuntimeMeta(
new ConcurrentHashMap<Integer, PipeTaskMeta>() {
@@ -87,7 +94,7 @@ public class PipeMetaDeSerTest {
put(123, new PipeTaskMeta(MinimumProgressIndex.INSTANCE, 987));
put(234, new PipeTaskMeta(new IoTProgressIndex(1, 2L), 789));
put(345, new PipeTaskMeta(new SimpleProgressIndex(3, 4), 789));
- put(456, new PipeTaskMeta(hybridProgressIndex, 789));
+ put(456, new PipeTaskMeta(finalHybridProgressIndex, 789));
put(
567,
new PipeTaskMeta(
