This is an automated email from the ASF dual-hosted git repository.
yujun777 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 1c5e867d849 [refactor](stream) Consolidate OLAP table stream partition
state (#70242)
1c5e867d849 is described below
commit 1c5e867d84958f2299d3d20fb2b7b5efbd8bc160
Author: yujun <[email protected]>
AuthorDate: Thu Nov 5 16:40:29 2026 +0800
[refactor](stream) Consolidate OLAP table stream partition state (#70242)
OLAP table streams track consumption separately for each base-table
partition. Each partition can carry a consumed offset, its last
consumption time, and a historical snapshot TSO until initial rows are
consumed.
---
.../doris/catalog/stream/OlapTableStream.java | 120 ++++++++++-----------
.../catalog/stream/OlapTableStreamUpdate.java | 30 +++---
.../stream/OlapTableStreamConsumptionTest.java | 4 -
.../stream/OlapTableStreamSnapshotReadTest.java | 6 +-
.../catalog/stream/OlapTableStreamUpdateTest.java | 77 +++++++++++--
.../stream/TableStreamManagerCleanupTest.java | 54 ++++------
.../InsertIntoTableCommandTableStreamTest.java | 30 ++----
.../TableStreamOffsetTransactionTest.java | 18 ++--
8 files changed, 186 insertions(+), 153 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/catalog/stream/OlapTableStream.java
b/fe/fe-core/src/main/java/org/apache/doris/catalog/stream/OlapTableStream.java
index 48c3ca5a438..9c3deb00e71 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/catalog/stream/OlapTableStream.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/catalog/stream/OlapTableStream.java
@@ -47,14 +47,37 @@ import java.util.function.Predicate;
public class OlapTableStream extends BaseTableStream {
- @SerializedName("po")
- private Map<Long, Long> partitionOffset;
+ @SerializedName("ps")
+ private Map<Long, PartitionStreamState> partitionStates = new HashMap<>();
+
+ // Nullable offset and historicalTso distinguish an unconsumed snapshot
from an incremental baseline.
+ // State is replaced as a whole when a consumption commits.
+ public static class PartitionStreamState {
+ @SerializedName("o")
+ private final Long offset;
+ @SerializedName("ct")
+ private final long consumptionTime;
+ @SerializedName("ht")
+ private final Long historicalTso;
+
+ public PartitionStreamState(Long offset, long consumptionTime, Long
historicalTso) {
+ this.offset = offset;
+ this.consumptionTime = consumptionTime;
+ this.historicalTso = historicalTso;
+ }
- @SerializedName("pct")
- private Map<Long, Long> partitionConsumptionTime;
+ public Long getOffset() {
+ return offset;
+ }
- @SerializedName("hpt")
- private Map<Long, Long> historicalPartitionTSO;
+ public long getConsumptionTime() {
+ return consumptionTime;
+ }
+
+ public Long getHistoricalTso() {
+ return historicalTso;
+ }
+ }
// for persist
public OlapTableStream() {
@@ -64,9 +87,6 @@ public class OlapTableStream extends BaseTableStream {
public OlapTableStream(long id, String streamName, TableIf baseTable) {
super(id, streamName, baseTable);
Preconditions.checkArgument(baseTable instanceof OlapTable);
- this.partitionOffset = new HashMap<>();
- this.partitionConsumptionTime = new HashMap<>();
- this.historicalPartitionTSO = new HashMap<>();
}
public OlapTableStream(String streamName, TableIf baseTable) {
@@ -137,15 +157,13 @@ public class OlapTableStream extends BaseTableStream {
if (!showInitialRows) {
// set partition offset
baseTable.getPartitions()
- .forEach(p -> partitionOffset.put(p.getId(), p.getTso()));
+ .forEach(p -> partitionStates.put(p.getId(), new
PartitionStreamState(p.getTso(), -1, null)));
} else {
baseTable.getPartitions()
.stream()
.filter(p -> p.getVisibleVersion() >
Partition.PARTITION_INIT_VERSION)
- .forEach(p -> {
- historicalPartitionTSO.put(p.getId(),
p.getTso());
- }
- );
+ .forEach(p -> partitionStates.put(p.getId(),
+ new PartitionStreamState(null, -1, p.getTso())));
}
}
@@ -167,13 +185,7 @@ public class OlapTableStream extends BaseTableStream {
if (table.readLockIfExist()) {
try {
for (Partition partition : table.getPartitions()) {
- long partitionId = partition.getId();
- boolean hasOffset =
partitionOffset.containsKey(partitionId);
- appendConsumptionRow(dataBatch, qualifiedDbName, name, id,
partition.getName(), hasOffset,
- hasOffset ? partitionOffset.get(partitionId) : 0,
- hasOffset ? partition.getTso() : 0,
- !hasOffset && partition.hasData(),
- partitionConsumptionTime.getOrDefault(partitionId,
-1L));
+ snapshotPartition(partition).appendRow(dataBatch,
qualifiedDbName, name, id);
}
} finally {
table.readUnlock();
@@ -211,13 +223,14 @@ public class OlapTableStream extends BaseTableStream {
private StreamConsumptionUnitSnapshot snapshotPartition(Partition
partition) {
long partitionId = partition.getId();
- boolean hasOffset = partitionOffset.containsKey(partitionId);
+ PartitionStreamState state = partitionStates.get(partitionId);
+ boolean hasOffset = state != null && state.getOffset() != null;
return new StreamConsumptionUnitSnapshot(
partition.getName(), hasOffset,
- hasOffset ? partitionOffset.get(partitionId) : 0,
+ hasOffset ? state.getOffset() : 0,
hasOffset ? partition.getTso() : 0,
!hasOffset && partition.hasData(),
- partitionConsumptionTime.getOrDefault(partitionId, -1L));
+ hasOffset ? state.getConsumptionTime() : -1);
}
private static void appendConsumptionRow(List<TRow> dataBatch, String
dbName, String streamName,
@@ -270,33 +283,29 @@ public class OlapTableStream extends BaseTableStream {
public boolean hasData(Partition partition) {
// if all available visible data has been consumed, return false
- return (!partitionOffset.containsKey(partition.getId())
- ||
!partitionOffset.get(partition.getId()).equals(partition.getTso()))
+ PartitionStreamState state = partitionStates.get(partition.getId());
+ return (state == null || state.getOffset() == null ||
!state.getOffset().equals(partition.getTso()))
&& partition.hasData();
}
public boolean hasHistoricalData(long partitionId) {
- return historicalPartitionTSO.containsKey(partitionId);
+ PartitionStreamState state = partitionStates.get(partitionId);
+ return state != null && state.getHistoricalTso() != null;
}
public boolean hasConsumedData(long partitionId) {
- // A partition that was empty at stream creation is recorded with the
sentinel offset -1
- // (see initializeLocalOffsets); a real committed TSO is always
positive (its physical part
- // is non-zero). So only a positive recorded offset counts as a real
consumption baseline.
- // This keeps empty partitions out of the snapshot scan instead of
letting them fall back to
- // the live partition TSO and leak post-snapshot rows.
- Long offset = partitionOffset.get(partitionId);
- return offset != null && offset > 0;
+ // An empty partition has the sentinel offset -1; only a positive
committed TSO
+ // counts as a consumption baseline for snapshot reads.
+ PartitionStreamState state = partitionStates.get(partitionId);
+ return state != null && state.getOffset() != null && state.getOffset()
> 0;
}
public Pair<Long, Long> getStreamUpdate(Long partitionId) {
- // if partition has historical data, return <historical tso, current
tso>
- // otherwise, return <current consumed tso, current tso>
- Long left = partitionOffset.get(partitionId);
- if (historicalPartitionTSO.containsKey(partitionId)) {
- left = historicalPartitionTSO.get(partitionId);
- }
- return Pair.of(left,
getBaseTableNullable().getPartition(partitionId).getTso());
+ // Historical data takes precedence over the consumed offset until it
is consumed.
+ PartitionStreamState state = partitionStates.get(partitionId);
+ Long startTso = state == null ? null
+ : state.getHistoricalTso() != null ? state.getHistoricalTso()
: state.getOffset();
+ return Pair.of(startTso,
getBaseTableNullable().getPartition(partitionId).getTso());
}
@Override
@@ -304,19 +313,14 @@ public class OlapTableStream extends BaseTableStream {
throws UserException {
Preconditions.checkArgument(update instanceof OlapTableStreamUpdate);
// check valid
- ((OlapTableStreamUpdate) update).checkPartitionOffset(getDBName(),
getName(), historicalPartitionTSO,
- partitionOffset);
+ ((OlapTableStreamUpdate) update).checkPartitionOffset(getDBName(),
getName(), partitionStates);
}
@Override
public void unprotectedUpdateStreamUpdate(AbstractTableStreamUpdate
update, Long ts) {
Map<Long, Long> next = ((OlapTableStreamUpdate) update).getNext();
for (Map.Entry<Long, Long> entry : next.entrySet()) {
- if (historicalPartitionTSO.containsKey(entry.getKey())) {
- historicalPartitionTSO.remove(entry.getKey());
- }
- partitionOffset.put(entry.getKey(), entry.getValue());
- partitionConsumptionTime.put(entry.getKey(), ts);
+ partitionStates.put(entry.getKey(), new
PartitionStreamState(entry.getValue(), ts, null));
}
}
@@ -324,23 +328,11 @@ public class OlapTableStream extends BaseTableStream {
Preconditions.checkState(isWriteLockHeldByCurrentThread(),
"unprotectedCollectStalePartitionOffsetIds must be called with
write lock held");
Set<Long> stalePartitionIds = new HashSet<>();
- for (Long partitionId : partitionOffset.keySet()) {
+ for (Long partitionId : partitionStates.keySet()) {
if (!validPartitionIds.contains(partitionId)) {
stalePartitionIds.add(partitionId);
}
}
- for (Long partitionId : partitionConsumptionTime.keySet()) {
- if (!validPartitionIds.contains(partitionId)) {
- stalePartitionIds.add(partitionId);
- }
- }
- if (historicalPartitionTSO != null) {
- for (Long partitionId : historicalPartitionTSO.keySet()) {
- if (!validPartitionIds.contains(partitionId)) {
- stalePartitionIds.add(partitionId);
- }
- }
- }
return stalePartitionIds;
}
@@ -348,11 +340,7 @@ public class OlapTableStream extends BaseTableStream {
Preconditions.checkState(isWriteLockHeldByCurrentThread(),
"unprotectedPrunePartitionOffsets must be called with write
lock held");
for (Long partitionId : partitionIds) {
- partitionOffset.remove(partitionId);
- partitionConsumptionTime.remove(partitionId);
- if (historicalPartitionTSO != null) {
- historicalPartitionTSO.remove(partitionId);
- }
+ partitionStates.remove(partitionId);
}
}
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/catalog/stream/OlapTableStreamUpdate.java
b/fe/fe-core/src/main/java/org/apache/doris/catalog/stream/OlapTableStreamUpdate.java
index 2357ae4b6e3..677def19e89 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/catalog/stream/OlapTableStreamUpdate.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/catalog/stream/OlapTableStreamUpdate.java
@@ -17,6 +17,7 @@
package org.apache.doris.catalog.stream;
+import org.apache.doris.catalog.stream.OlapTableStream.PartitionStreamState;
import org.apache.doris.common.UserException;
import org.apache.doris.transaction.TransactionCommitFailedException;
@@ -59,32 +60,29 @@ public class OlapTableStreamUpdate extends
AbstractTableStreamUpdate {
this.prev.putAll(((OlapTableStreamUpdate) other).getPrev());
}
- public void checkPartitionOffset(String dbName, String streamName,
Map<Long, Long> historicalPartitionTSO,
- Map<Long, Long> partitionOffset)
+ public void checkPartitionOffset(String dbName, String streamName,
Map<Long, PartitionStreamState> partitionStates)
throws UserException {
for (Map.Entry<Long, Long> entry : prev.entrySet()) {
- if (historicalPartitionTSO.containsKey(entry.getKey())) {
- if
(!historicalPartitionTSO.get(entry.getKey()).equals(-entry.getValue())) {
+ PartitionStreamState state = partitionStates.get(entry.getKey());
+ if (state != null && state.getHistoricalTso() != null) {
+ if (!state.getHistoricalTso().equals(-entry.getValue())) {
throw new TransactionCommitFailedException("history offset
not consumed: "
+ dbName + '-' + streamName + '-' + entry.getKey()
+ '-' + entry.getValue()
- + " vs " +
historicalPartitionTSO.get(entry.getKey()));
+ + " vs " + state.getHistoricalTso());
}
- } else if (partitionOffset.containsKey(entry.getKey())
- &&
!partitionOffset.get(entry.getKey()).equals(entry.getValue())) {
+ } else if (state != null && state.getOffset() != null
+ && !state.getOffset().equals(entry.getValue())) {
throw new TransactionCommitFailedException("target offset
already consumed: "
+ dbName + '-' + streamName + '-' + entry.getKey() +
'-' + entry.getValue()
- + " vs " + partitionOffset.get(entry.getKey()));
+ + " vs " + state.getOffset());
}
}
for (Map.Entry<Long, Long> entry : next.entrySet()) {
- if (partitionOffset.containsKey(entry.getKey())) {
- if (!prev.containsKey(entry.getKey())) {
- throw new TransactionCommitFailedException(
- "previous version missing for partition=" +
entry.getKey() + ", db=" + dbName
- + ", stream=" + streamName);
- }
- } else {
- // the new partition id may not be updated in time, so key in
table stream not exist is also valid
+ PartitionStreamState state = partitionStates.get(entry.getKey());
+ if (state != null && state.getOffset() != null &&
!prev.containsKey(entry.getKey())) {
+ throw new TransactionCommitFailedException(
+ "previous version missing for partition=" +
entry.getKey() + ", db=" + dbName
+ + ", stream=" + streamName);
}
}
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/catalog/stream/OlapTableStreamConsumptionTest.java
b/fe/fe-core/src/test/java/org/apache/doris/catalog/stream/OlapTableStreamConsumptionTest.java
index 5b34e8854ba..f27d8e4cc63 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/catalog/stream/OlapTableStreamConsumptionTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/catalog/stream/OlapTableStreamConsumptionTest.java
@@ -19,7 +19,6 @@ package org.apache.doris.catalog.stream;
import org.apache.doris.catalog.OlapTable;
import org.apache.doris.catalog.Partition;
-import org.apache.doris.common.jmockit.Deencapsulation;
import org.apache.doris.thrift.TRow;
import org.junit.jupiter.api.Assertions;
@@ -27,7 +26,6 @@ import org.junit.jupiter.api.Test;
import org.mockito.Mockito;
import java.util.ArrayList;
-import java.util.HashMap;
import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -55,8 +53,6 @@ public class OlapTableStreamConsumptionTest {
OlapTableStream stream = Mockito.spy(new OlapTableStream());
Mockito.doReturn(table).when(stream).getBaseTableNullable();
- Deencapsulation.setField(stream, "partitionOffset", new HashMap<Long,
Long>());
- Deencapsulation.setField(stream, "partitionConsumptionTime", new
HashMap<Long, Long>());
List<TRow> rows = new ArrayList<>();
stream.fillTableStreamConsumptionInfo(rows, unit -> {
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/catalog/stream/OlapTableStreamSnapshotReadTest.java
b/fe/fe-core/src/test/java/org/apache/doris/catalog/stream/OlapTableStreamSnapshotReadTest.java
index 5fa29bb37c2..1abb876b918 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/catalog/stream/OlapTableStreamSnapshotReadTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/catalog/stream/OlapTableStreamSnapshotReadTest.java
@@ -20,6 +20,7 @@ package org.apache.doris.catalog.stream;
import org.apache.doris.catalog.KeysType;
import org.apache.doris.catalog.OlapTable;
import org.apache.doris.catalog.Partition;
+import org.apache.doris.catalog.stream.OlapTableStream.PartitionStreamState;
import org.apache.doris.common.jmockit.Deencapsulation;
import com.google.common.collect.Lists;
@@ -124,8 +125,9 @@ public class OlapTableStreamSnapshotReadTest {
Mockito.doReturn(table).when(stream).getBaseTableNullable();
// The offsets have to be in place before the wrapper is built: it
reads them into the map the read of
// a snapshot is built from.
- Deencapsulation.setField(stream, "partitionOffset", new
HashMap<>(offsets));
- Deencapsulation.setField(stream, "historicalPartitionTSO", new
HashMap<Long, Long>());
+ Map<Long, PartitionStreamState> partitionStates = new HashMap<>();
+ offsets.forEach((pid, offset) -> partitionStates.put(pid, new
PartitionStreamState(offset, -1, null)));
+ Deencapsulation.setField(stream, "partitionStates", partitionStates);
return new Fixture(stream, new OlapTableStreamWrapper(stream, table,
Lists.newArrayList(byId.keySet())));
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/catalog/stream/OlapTableStreamUpdateTest.java
b/fe/fe-core/src/test/java/org/apache/doris/catalog/stream/OlapTableStreamUpdateTest.java
index ae6ba8d52a3..3fa8a78f767 100755
---
a/fe/fe-core/src/test/java/org/apache/doris/catalog/stream/OlapTableStreamUpdateTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/catalog/stream/OlapTableStreamUpdateTest.java
@@ -17,34 +17,41 @@
package org.apache.doris.catalog.stream;
+import org.apache.doris.catalog.stream.OlapTableStream.PartitionStreamState;
+import org.apache.doris.common.jmockit.Deencapsulation;
import org.apache.doris.transaction.TransactionCommitFailedException;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
+import java.util.Set;
public class OlapTableStreamUpdateTest {
@Test
public void testCheckPartitionOffsetForHistoricalConsumeOk() throws
Exception {
- Map<Long, Long> historicalPartitionTSO = Collections.singletonMap(1L,
100L);
- Map<Long, Long> partitionOffset = Collections.emptyMap();
+ Map<Long, PartitionStreamState> partitionStates =
+ Collections.singletonMap(1L, new PartitionStreamState(null,
-1, 100L));
// history partition prev offset is encoded as negative (see
StreamConsumptionInfoExtractor)
Map<Long, Long> prev = Collections.singletonMap(1L, -100L);
Map<Long, Long> next = Collections.emptyMap();
OlapTableStreamUpdate update = new OlapTableStreamUpdate(prev, next);
- update.checkPartitionOffset("test_db", "s1", historicalPartitionTSO,
partitionOffset);
+ update.checkPartitionOffset("test_db", "s1", partitionStates);
}
@Test
public void testCheckPartitionOffsetForHistoricalConsumeConflict() {
- Map<Long, Long> historicalPartitionTSO = Collections.singletonMap(1L,
99L);
- Map<Long, Long> partitionOffset = Collections.emptyMap();
+ Map<Long, PartitionStreamState> partitionStates =
+ Collections.singletonMap(1L, new PartitionStreamState(null,
-1, 99L));
// history partition prev offset is encoded as negative; -(-100)=100
!= 99 -> conflict
Map<Long, Long> prev = Collections.singletonMap(1L, -100L);
@@ -52,7 +59,65 @@ public class OlapTableStreamUpdateTest {
OlapTableStreamUpdate update = new OlapTableStreamUpdate(prev, next);
TransactionCommitFailedException exception =
Assertions.assertThrows(TransactionCommitFailedException.class,
- () -> update.checkPartitionOffset("test_db", "s1",
historicalPartitionTSO, partitionOffset));
+ () -> update.checkPartitionOffset("test_db", "s1",
partitionStates));
Assertions.assertTrue(exception.getMessage().contains("history offset
not consumed"));
}
+
+ @Test
+ public void testIncrementalOffsetValidation() throws Exception {
+ Map<Long, PartitionStreamState> states = Map.of(1L, new
PartitionStreamState(100L, 10, null));
+ new OlapTableStreamUpdate(Map.of(1L, 100L), Map.of(1L, 200L))
+ .checkPartitionOffset("test_db", "s1", states);
+ Assertions.assertThrows(TransactionCommitFailedException.class,
+ () -> new OlapTableStreamUpdate(Map.of(1L, 99L), Map.of(1L,
200L))
+ .checkPartitionOffset("test_db", "s1", states));
+ Assertions.assertThrows(TransactionCommitFailedException.class,
+ () -> new OlapTableStreamUpdate(Collections.emptyMap(),
Map.of(1L, 200L))
+ .checkPartitionOffset("test_db", "s1", states));
+ // A newly added partition has no recorded offset and can be consumed
without prev.
+ new OlapTableStreamUpdate(Collections.emptyMap(), Map.of(2L, 200L))
+ .checkPartitionOffset("test_db", "s1", states);
+ // A history-only state must not be mistaken for an incremental offset.
+ new OlapTableStreamUpdate(Collections.emptyMap(), Map.of(2L, 200L))
+ .checkPartitionOffset("test_db", "s1", Map.of(2L, new
PartitionStreamState(null, -1, 100L)));
+ }
+
+ @Test
+ public void testPartitionStatePersistenceAndConsumption() throws Exception
{
+ OlapTableStream stream = new OlapTableStream();
+ Map<Long, PartitionStreamState> states = new HashMap<>();
+ states.put(1L, new PartitionStreamState(null, -1, 100L));
+ states.put(2L, new PartitionStreamState(200L, 20, null));
+ states.put(3L, new PartitionStreamState(-1L, -1, null));
+ Deencapsulation.setField(stream, "partitionStates", states);
+ ByteArrayOutputStream bytes = new ByteArrayOutputStream();
+ stream.write(new DataOutputStream(bytes));
+ OlapTableStream restored = OlapTableStream.read(new DataInputStream(
+ new ByteArrayInputStream(bytes.toByteArray())));
+ Assertions.assertTrue(restored.hasHistoricalData(1L));
+ Assertions.assertFalse(restored.hasConsumedData(1L));
+ Assertions.assertTrue(restored.hasConsumedData(2L));
+ Assertions.assertFalse(restored.hasConsumedData(3L));
+ Assertions.assertFalse(restored.hasConsumedData(4L));
+
+ Map<Long, PartitionStreamState> restoredStates =
Deencapsulation.getField(restored, "partitionStates");
+ Assertions.assertNull(restoredStates.get(1L).getOffset());
+ Assertions.assertEquals(100L,
restoredStates.get(1L).getHistoricalTso());
+ Assertions.assertEquals(20,
restoredStates.get(2L).getConsumptionTime());
+ restored.writeLock();
+ try {
+ OlapTableStreamUpdate update = new
OlapTableStreamUpdate(Map.of(1L, -100L), Map.of(1L, 150L));
+ update.checkPartitionOffset("test_db", "s1", restoredStates);
+ restored.unprotectedUpdateStreamUpdate(update, 30L);
+ Assertions.assertFalse(restored.hasHistoricalData(1L));
+ Assertions.assertEquals(150L, restoredStates.get(1L).getOffset());
+ Assertions.assertEquals(30,
restoredStates.get(1L).getConsumptionTime());
+ Assertions.assertEquals(Set.of(1L, 3L),
+
restored.unprotectedCollectStalePartitionOffsetIds(Set.of(2L)));
+ restored.unprotectedPrunePartitionOffsets(Set.of(1L, 3L));
+ Assertions.assertEquals(Set.of(2L), restoredStates.keySet());
+ } finally {
+ restored.writeUnlock();
+ }
+ }
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/catalog/stream/TableStreamManagerCleanupTest.java
b/fe/fe-core/src/test/java/org/apache/doris/catalog/stream/TableStreamManagerCleanupTest.java
index 511374e9cc2..a0c24c3ac7e 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/catalog/stream/TableStreamManagerCleanupTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/catalog/stream/TableStreamManagerCleanupTest.java
@@ -20,6 +20,7 @@ package org.apache.doris.catalog.stream;
import org.apache.doris.catalog.Database;
import org.apache.doris.catalog.Env;
import org.apache.doris.catalog.OlapTable;
+import org.apache.doris.catalog.stream.OlapTableStream.PartitionStreamState;
import org.apache.doris.cloud.proto.Cloud;
import org.apache.doris.common.Config;
import org.apache.doris.common.FeConstants;
@@ -32,6 +33,7 @@ import org.apache.doris.persist.DropInfo;
import org.apache.doris.persist.EditLog;
import org.apache.doris.persist.RecoverInfo;
import org.apache.doris.persist.TableStreamCleanupInfo;
+import org.apache.doris.persist.gson.GsonUtils;
import org.apache.doris.utframe.TestWithFeService;
import org.junit.jupiter.api.Assertions;
@@ -174,8 +176,8 @@ public class TableStreamManagerCleanupTest extends
TestWithFeService {
Assertions.assertTrue(executor.awaitTermination(10,
TimeUnit.SECONDS));
}
- Map<Long, Long> leaderOffsets = new HashMap<>(
- Deencapsulation.getField(context.stream, "partitionOffset"));
+ String leaderStates = GsonUtils.GSON.toJson(
+ Deencapsulation.<Map<Long,
PartitionStreamState>>getField(context.stream, "partitionStates"));
setPartitionState(context.stream, keptPartitionId, removedPartitionId);
for (Object journal : journalOrder) {
if (journal instanceof OlapTableStreamUpdate) {
@@ -186,8 +188,9 @@ public class TableStreamManagerCleanupTest extends
TestWithFeService {
}
}
- Map<Long, Long> replayedOffsets =
Deencapsulation.getField(context.stream, "partitionOffset");
- Assertions.assertEquals(leaderOffsets, replayedOffsets);
+ Map<Long, PartitionStreamState> replayedStates =
+ Deencapsulation.getField(context.stream, "partitionStates");
+ Assertions.assertEquals(leaderStates,
GsonUtils.GSON.toJson(replayedStates));
}
@Test
@@ -249,8 +252,8 @@ public class TableStreamManagerCleanupTest extends
TestWithFeService {
}
}
- Map<Long, Long> leaderOffsets = new HashMap<>(
- Deencapsulation.getField(context.stream, "partitionOffset"));
+ String leaderStates = GsonUtils.GSON.toJson(
+ Deencapsulation.<Map<Long,
PartitionStreamState>>getField(context.stream, "partitionStates"));
setPartitionState(context.stream, keptPartitionId, removedPartitionId);
Database db = (Database)
Env.getCurrentInternalCatalog().getDbOrMetaException("test_stream_cleanup");
for (Object journal : journalOrder) {
@@ -266,9 +269,10 @@ public class TableStreamManagerCleanupTest extends
TestWithFeService {
}
}
- Map<Long, Long> replayedOffsets =
Deencapsulation.getField(context.stream, "partitionOffset");
- Assertions.assertEquals(leaderOffsets, replayedOffsets);
- Assertions.assertTrue(replayedOffsets.containsKey(removedPartitionId));
+ Map<Long, PartitionStreamState> replayedStates =
+ Deencapsulation.getField(context.stream, "partitionStates");
+ Assertions.assertEquals(leaderStates,
GsonUtils.GSON.toJson(replayedStates));
+ Assertions.assertTrue(replayedStates.containsKey(removedPartitionId));
}
@Test
@@ -389,32 +393,20 @@ public class TableStreamManagerCleanupTest extends
TestWithFeService {
}
private void setPartitionState(OlapTableStream stream, long
keptPartitionId, long removedPartitionId) {
- Map<Long, Long> partitionOffset = new HashMap<>();
- partitionOffset.put(keptPartitionId, 11L);
- partitionOffset.put(removedPartitionId, 22L);
- Map<Long, Long> partitionConsumptionTime = new HashMap<>();
- partitionConsumptionTime.put(keptPartitionId, 111L);
- partitionConsumptionTime.put(removedPartitionId, 222L);
- Map<Long, Long> historicalPartitionTSO = new HashMap<>();
- historicalPartitionTSO.put(keptPartitionId, 1001L);
- historicalPartitionTSO.put(removedPartitionId, 2002L);
- Deencapsulation.setField(stream, "partitionOffset", partitionOffset);
- Deencapsulation.setField(stream, "partitionConsumptionTime",
partitionConsumptionTime);
- Deencapsulation.setField(stream, "historicalPartitionTSO",
historicalPartitionTSO);
+ Map<Long, PartitionStreamState> partitionStates = new HashMap<>();
+ partitionStates.put(keptPartitionId, new PartitionStreamState(11L,
111L, 1001L));
+ partitionStates.put(removedPartitionId, new PartitionStreamState(22L,
222L, 2002L));
+ Deencapsulation.setField(stream, "partitionStates", partitionStates);
}
private void assertPartitionState(OlapTableStream stream, long
keptPartitionId, long removedPartitionId,
boolean removedExpected) {
- Map<Long, Long> partitionOffset = Deencapsulation.getField(stream,
"partitionOffset");
- Map<Long, Long> partitionConsumptionTime =
Deencapsulation.getField(stream, "partitionConsumptionTime");
- Map<Long, Long> historicalPartitionTSO =
Deencapsulation.getField(stream, "historicalPartitionTSO");
-
- Assertions.assertTrue(partitionOffset.containsKey(keptPartitionId));
-
Assertions.assertTrue(partitionConsumptionTime.containsKey(keptPartitionId));
-
Assertions.assertTrue(historicalPartitionTSO.containsKey(keptPartitionId));
- Assertions.assertEquals(!removedExpected,
partitionOffset.containsKey(removedPartitionId));
- Assertions.assertEquals(!removedExpected,
partitionConsumptionTime.containsKey(removedPartitionId));
- Assertions.assertEquals(!removedExpected,
historicalPartitionTSO.containsKey(removedPartitionId));
+ Map<Long, PartitionStreamState> partitionStates =
Deencapsulation.getField(stream, "partitionStates");
+ PartitionStreamState kept = partitionStates.get(keptPartitionId);
+ Assertions.assertEquals(11L, kept.getOffset());
+ Assertions.assertEquals(111L, kept.getConsumptionTime());
+ Assertions.assertEquals(1001L, kept.getHistoricalTso());
+ Assertions.assertEquals(!removedExpected,
partitionStates.containsKey(removedPartitionId));
}
private static void await(BooleanSupplier condition) throws
InterruptedException {
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertIntoTableCommandTableStreamTest.java
b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertIntoTableCommandTableStreamTest.java
index 725107dcf67..3e8a4d05156 100755
---
a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertIntoTableCommandTableStreamTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertIntoTableCommandTableStreamTest.java
@@ -30,6 +30,7 @@ import org.apache.doris.catalog.Tablet;
import org.apache.doris.catalog.stream.AbstractTableStreamUpdate;
import org.apache.doris.catalog.stream.CloudOlapTableStreamUpdate;
import org.apache.doris.catalog.stream.OlapTableStream;
+import org.apache.doris.catalog.stream.OlapTableStream.PartitionStreamState;
import org.apache.doris.catalog.stream.OlapTableStreamUpdate;
import org.apache.doris.catalog.stream.OlapTableStreamWrapper;
import org.apache.doris.catalog.stream.TableStreamUpdateInfo;
@@ -549,12 +550,9 @@ public class InsertIntoTableCommandTableStreamTest extends
TestWithFeService {
Assertions.assertFalse(stream.hasHistoricalData(pid),
"no historical offset should be present after advance
without prior history");
}
- // partitionConsumptionTime is private; verify via reflection on the
same field name.
- @SuppressWarnings("unchecked")
- Map<Long, Long> consumptionTime = (Map<Long, Long>)
Deencapsulation.getField(stream,
- "partitionConsumptionTime");
+ Map<Long, PartitionStreamState> partitionStates =
Deencapsulation.getField(stream, "partitionStates");
for (Long pid : next.keySet()) {
- Assertions.assertEquals(Long.valueOf(ts), consumptionTime.get(pid),
+ Assertions.assertEquals(Long.valueOf(ts),
partitionStates.get(pid).getConsumptionTime(),
"partitionConsumptionTime must be recorded with the commit
ts");
}
}
@@ -571,9 +569,7 @@ public class InsertIntoTableCommandTableStreamTest extends
TestWithFeService {
+ "properties('show_initial_rows' = 'false')");
OlapTableStream stream = (OlapTableStream)
db.getTableOrMetaException("s_history");
- @SuppressWarnings("unchecked")
- Map<Long, Long> historicalPartitionTSO = (Map<Long, Long>)
Deencapsulation.getField(stream,
- "historicalPartitionTSO");
+ Map<Long, PartitionStreamState> partitionStates =
Deencapsulation.getField(stream, "partitionStates");
Map<Long, Long> prev = new HashMap<>();
Map<Long, Long> next = new HashMap<>();
@@ -582,7 +578,8 @@ public class InsertIntoTableCommandTableStreamTest extends
TestWithFeService {
long pid = partition.getId();
long histTso = 8000L + seed;
long nextTso = 9999L + seed;
- historicalPartitionTSO.put(pid, histTso);
+ partitionStates.put(pid, new PartitionStreamState(
+ partitionStates.get(pid).getOffset(), -1, histTso));
prev.put(pid, histTso);
next.put(pid, nextTso);
seed++;
@@ -597,18 +594,15 @@ public class InsertIntoTableCommandTableStreamTest
extends TestWithFeService {
baseTable.writeUnlock();
}
- @SuppressWarnings("unchecked")
- Map<Long, Long> consumptionTime = (Map<Long, Long>)
Deencapsulation.getField(stream,
- "partitionConsumptionTime");
for (Map.Entry<Long, Long> entry : next.entrySet()) {
long pid = entry.getKey();
Assertions.assertFalse(stream.hasHistoricalData(pid),
"historicalPartitionTSO must be cleared after commit");
- Assertions.assertFalse(historicalPartitionTSO.containsKey(pid),
- "historicalPartitionTSO must be cleared after commit");
+ Assertions.assertNull(partitionStates.get(pid).getHistoricalTso(),
+ "historical TSO must be cleared after commit");
Assertions.assertEquals(entry.getValue(),
stream.getStreamUpdate(pid).first,
"partitionOffset must be advanced to update.next after
history consumed");
- Assertions.assertEquals(Long.valueOf(ts), consumptionTime.get(pid),
+ Assertions.assertEquals(Long.valueOf(ts),
partitionStates.get(pid).getConsumptionTime(),
"partitionConsumptionTime must be recorded with the commit
ts");
}
}
@@ -625,9 +619,7 @@ public class InsertIntoTableCommandTableStreamTest extends
TestWithFeService {
// s1 was created with show_initial_rows=true. Seed
historicalPartitionTSO for one
// partition to also cover the history branch.
OlapTableStream stream = (OlapTableStream)
db.getTableOrMetaException("s1");
- @SuppressWarnings("unchecked")
- Map<Long, Long> historicalPartitionTSO = (Map<Long, Long>)
Deencapsulation.getField(stream,
- "historicalPartitionTSO");
+ Map<Long, PartitionStreamState> partitionStates =
Deencapsulation.getField(stream, "partitionStates");
// Bump partition versions so partition.getTso() differs from any
initial seed and
// is a meaningful upper bound. Pass tso explicitly (third arg) to
simulate a
@@ -650,7 +642,7 @@ public class InsertIntoTableCommandTableStreamTest extends
TestWithFeService {
Partition historyPartition =
baseTable.getPartitions().iterator().next();
long historyPid = historyPartition.getId();
long historyTso = 6789L;
- historicalPartitionTSO.put(historyPid, historyTso);
+ partitionStates.put(historyPid, new PartitionStreamState(null, -1,
historyTso));
String sql = "insert into test_stream.tbl_target select * from
test_stream.s1";
LogicalPlan logicalPlan = parser.parseSingle(sql);
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/transaction/TableStreamOffsetTransactionTest.java
b/fe/fe-core/src/test/java/org/apache/doris/transaction/TableStreamOffsetTransactionTest.java
index 320b49ff9f3..b11d930f331 100755
---
a/fe/fe-core/src/test/java/org/apache/doris/transaction/TableStreamOffsetTransactionTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/transaction/TableStreamOffsetTransactionTest.java
@@ -21,6 +21,7 @@ import org.apache.doris.catalog.Database;
import org.apache.doris.catalog.Env;
import org.apache.doris.catalog.OlapTable;
import org.apache.doris.catalog.stream.OlapTableStream;
+import org.apache.doris.catalog.stream.OlapTableStream.PartitionStreamState;
import org.apache.doris.catalog.stream.OlapTableStreamUpdate;
import org.apache.doris.catalog.stream.TableStreamUpdateInfo;
import org.apache.doris.common.Config;
@@ -92,8 +93,10 @@ public class TableStreamOffsetTransactionTest extends
TestWithFeService {
}
// committed offset must be advanced to the historical TSO snapshot
captured at creation
Map<Long, Long> expectedPartitionOffset = new
HashMap<>(historicalPartitionTSO);
- Deencapsulation.setField(stream, "historicalPartitionTSO",
historicalPartitionTSO);
- Deencapsulation.setField(stream, "partitionOffset", new HashMap<Long,
Long>());
+ Map<Long, PartitionStreamState> partitionStates = new HashMap<>();
+ historicalPartitionTSO.forEach((pid, tso) ->
+ partitionStates.put(pid, new PartitionStreamState(null, -1,
tso)));
+ Deencapsulation.setField(stream, "partitionStates", partitionStates);
// historical consume: prev is the historical TSO snapshot encoded as
negative
// (see StreamConsumptionInfoExtractor), next points to the historical
TSO snapshot
@@ -120,14 +123,11 @@ public class TableStreamOffsetTransactionTest extends
TestWithFeService {
long commitTime = System.currentTimeMillis();
Deencapsulation.invoke(dbTxnMgr, "updateStreamOffset",
transactionState, commitTime);
- Map<Long, Long> updatedHistoricalPartitionTSO =
Deencapsulation.getField(stream, "historicalPartitionTSO");
- Map<Long, Long> updatedPartitionOffset =
Deencapsulation.getField(stream, "partitionOffset");
- Map<Long, Long> partitionConsumptionTime =
Deencapsulation.getField(stream, "partitionConsumptionTime");
-
for (Long pid : partitionIds) {
-
Assertions.assertFalse(updatedHistoricalPartitionTSO.containsKey(pid));
- Assertions.assertEquals(expectedPartitionOffset.get(pid),
updatedPartitionOffset.get(pid));
- Assertions.assertEquals(commitTime,
partitionConsumptionTime.get(pid));
+ PartitionStreamState state = partitionStates.get(pid);
+ Assertions.assertNull(state.getHistoricalTso());
+ Assertions.assertEquals(expectedPartitionOffset.get(pid),
state.getOffset());
+ Assertions.assertEquals(commitTime, state.getConsumptionTime());
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
