This is an automated email from the ASF dual-hosted git repository.
gavinchou 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 addf0c82974 [feature](ivm) Add the per-partition refresh state and its
journal channel (#68193)
addf0c82974 is described below
commit addf0c829747e08927f593179cb0e08451088460
Author: yujun <[email protected]>
AuthorDate: Mon Sep 21 14:37:32 2026 +0800
[feature](ivm) Add the per-partition refresh state and its journal channel
(#68193)
### What problem does this PR solve?
Trace issue: https://github.com/apache/doris/issues/65418
**This PR adds no behaviour of its own. It adds the state that the
following PRs need, and the channel
that persists it, so that they can be reviewed as logic alone.**
An IVM materialized view has to invalidate the MV partitions that a
base-table change really affected.
A partition drop / truncate / replace / recover changes the base table
through metadata and emits no
row binlog, so the affected MV partitions must be rebuilt; today the
only answer the MV has is "rebuild
all of them", which throws away partitions that are still correct.
Deciding per partition needs a per-partition answer to two questions:
- which generation of data does this MV partition currently hold?
- which generation must it hold?
That pair is `MTMVPartitionState { refreshEpoch, latestEpoch }`, one
entry per MV partition, keyed by
partition name. `latestEpoch` is the requirement, `refreshEpoch` is the
reality, and a partition whose
requirement is ahead of its reality is dirty: it holds rows read before
a change that left no binlog,
so it can no longer be maintained incrementally and has to be rebuilt.
The requirement has to survive a
restart, because an invalidation that only lives in memory is lost the
moment the FE restarts, and a
partition that is then refreshed incrementally keeps the stale rows
forever with no error anywhere.
So this PR adds
1. the state itself (`MTMVPartitionState`, plus a copy helper for taking
a detached snapshot), and
2. the channel that carries it into the journal and back: a field on the
MV, its own field on the alter
record, a dedicated alter op with its replay branch, and the replay
handling of the task result.
Nothing in the FE decides anything from the state yet, and nothing but a
replay ever writes it, so every
MV behaves exactly as before. That is deliberate: it makes this step
independently mergeable and
independently testable, which is what the PR that starts using the state
needs underneath it.
### Scope
| | |
| --- | --- |
| Adds | `MTMVPartitionState`, the `partitionStates` field, the alter
record field, the new alter op and its replay branch |
| Does not touch | any criterion, routing or invalidation decision;
`IvmInfo`; the refresh path |
| Field is shared, behaviour is not | the field sits on the MV, so both
kinds of MV carry it; only an IVM MV ever populates it, and the journal
of a non-IVM MV stays byte-for-byte what it was |
| Compatibility | an image written before this PR has no such field and
loads as an empty map; an ADD_TASK journal written before it applies
nothing on replay instead of clearing what is there |
The state is on the MV rather than inside `IvmInfo`, and the alter op is
its own rather than riding on
`ALTER_IVM_INFO`, whose branch only swaps the `IvmInfo` object. Both are
structural: the same state is
meant to serve a non-IVM MV later, and its journal payload must not be
reconstructed as a side effect of
replaying some other op.
### Key changes
- Add `MTMVPartitionState`, a persisted `refreshEpoch` / `latestEpoch`
pair keyed by MV partition name, and `MTMVPartitionState.copyOf` for
taking a detached snapshot of a state map. A partition gets a new id on
every refresh, so the name is the only identity it can have.
- Add `MTMV.partitionStates` with its getter and its replay setter;
`gsonPostProcess` initializes it, so an image written before the field
existed and a non-IVM MV both load as an empty map.
- Carry the state in the ADD_TASK payload under the same condition as
`ivmInfo`, which keeps the journal of a non-IVM MV byte-for-byte
unchanged, and apply it on replay only when the field is present, so an
old journal applies nothing rather than clearing the state.
- Add `MTMVAlterOpType.ALTER_PARTITION_STATES` and its
`Alter.processAlterMTMV` branch.
---
.../main/java/org/apache/doris/alter/Alter.java | 4 +
.../main/java/org/apache/doris/catalog/MTMV.java | 71 +++++++++
.../org/apache/doris/mtmv/MTMVAlterOpType.java | 3 +-
.../org/apache/doris/mtmv/MTMVPartitionState.java | 98 ++++++++++++
.../java/org/apache/doris/persist/AlterMTMV.java | 11 ++
.../java/org/apache/doris/mtmv/AlterMTMVTest.java | 89 +++++++++++
.../test/java/org/apache/doris/mtmv/MTMVTest.java | 167 +++++++++++++++++++++
7 files changed, 442 insertions(+), 1 deletion(-)
diff --git a/fe/fe-core/src/main/java/org/apache/doris/alter/Alter.java
b/fe/fe-core/src/main/java/org/apache/doris/alter/Alter.java
index 8c30d35a733..28613904211 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/alter/Alter.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/alter/Alter.java
@@ -1344,6 +1344,10 @@ public class Alter {
// Live IVM changes are journaled inside MTMV; this branch
applies the journal snapshot.
mtmv.alterIvmInfo(alterMTMV.getIvmInfo());
break;
+ case ALTER_PARTITION_STATES:
+ // Replay only, like ALTER_IVM_INFO: a live change
journals itself from inside MTMV.
+ mtmv.alterPartitionStates(alterMTMV.getPartitionStates());
+ break;
default:
throw new RuntimeException("Unknown type value: " +
alterMTMV.getOpType());
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/MTMV.java
b/fe/fe-core/src/main/java/org/apache/doris/catalog/MTMV.java
index b162a2b2d51..53efb6b4c62 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/catalog/MTMV.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/MTMV.java
@@ -38,6 +38,7 @@ import org.apache.doris.mtmv.MTMVJobManager;
import org.apache.doris.mtmv.MTMVPartitionExpander;
import org.apache.doris.mtmv.MTMVPartitionInfo;
import org.apache.doris.mtmv.MTMVPartitionInfo.MTMVPartitionType;
+import org.apache.doris.mtmv.MTMVPartitionState;
import org.apache.doris.mtmv.MTMVPartitionUtil;
import org.apache.doris.mtmv.MTMVPlanUtil;
import org.apache.doris.mtmv.MTMVPropertyUtil;
@@ -69,6 +70,7 @@ import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import java.io.IOException;
+import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
@@ -103,6 +105,21 @@ public class MTMV extends OlapTable {
private MTMVRefreshSnapshot refreshSnapshot;
@SerializedName("ii")
private IvmInfo ivmInfo;
+ /**
+ * The refresh epoch of every MV partition, keyed by MV partition name.
+ *
+ * <p>Deliberately on MTMV rather than inside {@link IvmInfo}: the field
is shared, the behaviour is
+ * not. Both kinds of MV carry it, but only an IVM MV ever populates it --
alignment, invalidation,
+ * the ADD_TASK payload and ALTER_PARTITION_STATES are all no-ops for a
non-IVM MV, so for one an
+ * empty map is the complete answer.
+ *
+ * <p>Null means the same thing -- no state -- and has three causes: an
image written before the
+ * field existed, a non-IVM MV, and a live MV that has not been aligned
yet. Only
+ * {@link #gsonPostProcess()} turns it into an empty map, on load; nothing
else needs to, because a
+ * reader treats the two the same.
+ */
+ @SerializedName("pst")
+ private Map<String, MTMVPartitionState> partitionStates;
// Should update after every fresh, not persist
// Cache with SessionVarGuardExpr: used when query session variables
differ from MV creation variables
private MTMVCache cacheWithGuard;
@@ -290,6 +307,11 @@ public class MTMV extends OlapTable {
// Replay the final IVM state; ADD_TASK does not change
schemaChangeVersion.
ivmInfo = new IvmInfo(alterMTMV.getIvmInfo());
}
+ if (isReplay && alterMTMV.getPartitionStates() != null) {
+ // A journal written before the field existed carries no state
at all: leave the
+ // partition states alone rather than clearing them.
+ partitionStates =
MTMVPartitionState.copyOf(alterMTMV.getPartitionStates());
+ }
if (task.getStatus() == TaskStatus.SUCCESS) {
this.status.setState(MTMVState.NORMAL);
this.status.setSchemaChangeDetail(null);
@@ -323,6 +345,10 @@ public class MTMV extends OlapTable {
}
if (ivmInfo.isEnableIvm()) {
alterMTMV.setIvmInfo(ivmInfo);
+ // Same condition as ivmInfo, so the journal of a non-IVM MV
stays byte-for-byte what it
+ // was. The map is null until the states are first aligned,
and a payload without the
+ // member means the same as one carrying an empty map.
+ alterMTMV.setPartitionStates(partitionStates);
}
editLogItem = submitAlterLog(alterMTMV);
} finally {
@@ -598,6 +624,46 @@ public class MTMV extends OlapTable {
}
}
+ /**
+ * A snapshot of the partition states, taken under the MV read lock.
+ *
+ * <p>The caller gets its own map and its own state objects, not the ones
the MV owns: handing those
+ * out would let a caller add or change an entry while {@link
#addTaskResult} copies the same map
+ * into the journal, and a replay that replaces the field would leave the
caller's reference
+ * pointing at state that is no longer the MV's. Changing the states is
the MV's own job, under its
+ * write lock.
+ *
+ * <p>A missing map -- an image written before the field existed, or a
non-IVM MV -- reads as empty.
+ */
+ public Map<String, MTMVPartitionState> getPartitionStates() {
+ readMvLock();
+ try {
+ if (partitionStates == null) {
+ return Collections.emptyMap();
+ }
+ return
Collections.unmodifiableMap(MTMVPartitionState.copyOf(partitionStates));
+ } finally {
+ readMvUnlock();
+ }
+ }
+
+ // ALTER_PARTITION_STATES replay applies a detached snapshot here,
mirroring alterIvmInfo(). Live
+ // invalidation changes submit their journal from the mutating method
instead.
+ //
+ // A payload without the member carries no state at all, which is not the
same as an empty map that
+ // says the states are now empty: leaving them alone is the only answer
that cannot lose state.
+ public void alterPartitionStates(Map<String, MTMVPartitionState>
partitionStates) {
+ if (partitionStates == null) {
+ return;
+ }
+ writeMvLock();
+ try {
+ this.partitionStates = MTMVPartitionState.copyOf(partitionStates);
+ } finally {
+ writeMvUnlock();
+ }
+ }
+
public void invalidateIvmBaseline() {
EditLogItem editLogItem;
writeMvLock();
@@ -997,6 +1063,11 @@ public class MTMV extends OlapTable {
if (ivmInfo == null) {
ivmInfo = new IvmInfo();
}
+ if (partitionStates == null) {
+ // An image written before the field existed deserializes it as
null, and so does a non-IVM MV.
+ // Both mean "no state", so an empty map is the whole answer.
+ partitionStates = Maps.newLinkedHashMap();
+ }
if (refreshInfo != null && refreshInfo.getRefreshMethod() == null) {
LOG.warn("MTMV {} has unknown refresh method, marking as schema
change", name);
status.setState(MTMVState.SCHEMA_CHANGE);
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVAlterOpType.java
b/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVAlterOpType.java
index a8e81446dea..777dca2aece 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVAlterOpType.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVAlterOpType.java
@@ -22,5 +22,6 @@ public enum MTMVAlterOpType {
ALTER_STATUS,
ALTER_PROPERTY,
ADD_TASK,
- ALTER_IVM_INFO;
+ ALTER_IVM_INFO,
+ ALTER_PARTITION_STATES;
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVPartitionState.java
b/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVPartitionState.java
new file mode 100644
index 00000000000..edb30f84177
--- /dev/null
+++ b/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVPartitionState.java
@@ -0,0 +1,98 @@
+// 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.doris.mtmv;
+
+import com.google.gson.annotations.SerializedName;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Map.Entry;
+
+/**
+ * The per-partition refresh state of one MV partition.
+ *
+ * <p>{@code refreshEpoch} says which generation of data the partition
currently holds, {@code
+ * latestEpoch} says which generation it must hold. A partition whose {@code
latestEpoch} is ahead of
+ * its {@code refreshEpoch} is dirty: it holds rows read before a
metadata-only change of a base table
+ * (a dropped / truncated / replaced / recovered partition emits no row
binlog), so those rows can no
+ * longer be removed incrementally and the partition has to be rebuilt.
+ *
+ * <p>Keyed by MV partition name in {@code MTMV.partitionStates}. The name is
deliberately the only
+ * identity: an MV partition is rewritten by {@code INSERT OVERWRITE} on every
refresh and gets a new
+ * partition id each time, so an id would stop matching as soon as the
partition is refreshed.
+ *
+ * <p>The two values are plain {@code long}s rather than atomics because this
is a persisted DTO: it is
+ * serialized into the alter journal, so it has to stay a plain bean.
+ */
+public class MTMVPartitionState {
+ /** The generation of the data this MV partition currently holds; 0 means
it was never refreshed. */
+ @SerializedName("re")
+ private long refreshEpoch;
+
+ /** The generation the data must reach; starts at 1 and grows on every
invalidation. */
+ @SerializedName("le")
+ private long latestEpoch;
+
+ public MTMVPartitionState() {
+ }
+
+ public MTMVPartitionState(long refreshEpoch, long latestEpoch) {
+ this.refreshEpoch = refreshEpoch;
+ this.latestEpoch = latestEpoch;
+ }
+
+ public MTMVPartitionState(MTMVPartitionState other) {
+ this.refreshEpoch = other.refreshEpoch;
+ this.latestEpoch = other.latestEpoch;
+ }
+
+ /**
+ * Deep-copies a state map, or returns null for null.
+ *
+ * <p>The journal needs this on both sides. A payload is serialized by the
journal thread, which
+ * runs after the submitting thread released the MV lock, so a payload
that shared state with the
+ * live map could be written out half-mutated. The replay path goes
through the same helper so that
+ * both sides of the journal follow one rule instead of two.
+ */
+ public static Map<String, MTMVPartitionState> copyOf(Map<String,
MTMVPartitionState> states) {
+ if (states == null) {
+ return null;
+ }
+ Map<String, MTMVPartitionState> copy = new LinkedHashMap<>();
+ for (Entry<String, MTMVPartitionState> entry : states.entrySet()) {
+ copy.put(entry.getKey(), new MTMVPartitionState(entry.getValue()));
+ }
+ return copy;
+ }
+
+ public long getRefreshEpoch() {
+ return refreshEpoch;
+ }
+
+ public void setRefreshEpoch(long refreshEpoch) {
+ this.refreshEpoch = refreshEpoch;
+ }
+
+ public long getLatestEpoch() {
+ return latestEpoch;
+ }
+
+ public void setLatestEpoch(long latestEpoch) {
+ this.latestEpoch = latestEpoch;
+ }
+}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/persist/AlterMTMV.java
b/fe/fe-core/src/main/java/org/apache/doris/persist/AlterMTMV.java
index 83c81f6ac77..208417698b8 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/persist/AlterMTMV.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/persist/AlterMTMV.java
@@ -22,6 +22,7 @@ import org.apache.doris.common.io.Text;
import org.apache.doris.common.io.Writable;
import org.apache.doris.job.extensions.mtmv.MTMVTask;
import org.apache.doris.mtmv.MTMVAlterOpType;
+import org.apache.doris.mtmv.MTMVPartitionState;
import org.apache.doris.mtmv.MTMVRefreshInfo;
import org.apache.doris.mtmv.MTMVRefreshPartitionSnapshot;
import org.apache.doris.mtmv.MTMVRelation;
@@ -58,6 +59,8 @@ public class AlterMTMV implements Writable {
private Map<String, MTMVRefreshPartitionSnapshot> partitionSnapshots;
@SerializedName("ii")
private IvmInfo ivmInfo;
+ @SerializedName("pst")
+ private Map<String, MTMVPartitionState> partitionStates;
public AlterMTMV(TableNameInfo mvName, MTMVRefreshInfo refreshInfo,
MTMVAlterOpType opType) {
this.mvName = Objects.requireNonNull(mvName, "require mvName object");
@@ -148,6 +151,14 @@ public class AlterMTMV implements Writable {
this.ivmInfo = ivmInfo == null ? null : new IvmInfo(ivmInfo);
}
+ public Map<String, MTMVPartitionState> getPartitionStates() {
+ return partitionStates;
+ }
+
+ public void setPartitionStates(Map<String, MTMVPartitionState>
partitionStates) {
+ this.partitionStates = MTMVPartitionState.copyOf(partitionStates);
+ }
+
@Override
public String toString() {
return "AlterMTMV{"
diff --git a/fe/fe-core/src/test/java/org/apache/doris/mtmv/AlterMTMVTest.java
b/fe/fe-core/src/test/java/org/apache/doris/mtmv/AlterMTMVTest.java
index deeed90db1d..8daa44e63d5 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/mtmv/AlterMTMVTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/mtmv/AlterMTMVTest.java
@@ -32,11 +32,22 @@ import org.apache.doris.mtmv.ivm.IvmUtil;
import org.apache.doris.nereids.exceptions.AnalysisException;
import org.apache.doris.persist.AlterMTMV;
import org.apache.doris.persist.ReplaceTableOperationLog;
+import org.apache.doris.persist.gson.GsonUtils;
import org.apache.doris.utframe.TestWithFeService;
+import com.google.gson.JsonObject;
+import com.google.gson.JsonParser;
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.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.LinkedHashMap;
+import java.util.Map;
import java.util.Set;
@@ -403,6 +414,84 @@ public class AlterMTMVTest extends TestWithFeService {
Assertions.assertEquals(schemaChangeVersion,
mtmv.getSchemaChangeVersion());
}
+ @Test
+ public void testReplayAlterPartitionStates() throws Exception {
+ Config.enable_table_stream = true;
+ createDatabaseAndUse("alter_partition_states_test");
+ createTable("CREATE TABLE alter_partition_states_test.states_base (k1
int, v1 int)\n"
+ + "DUPLICATE KEY(k1)\n"
+ + "DISTRIBUTED BY HASH(k1) BUCKETS 1\n"
+ + "PROPERTIES ('replication_num' = '1', 'binlog.enable' =
'true', 'binlog.format' = 'ROW')");
+ createMvByNereids("CREATE MATERIALIZED VIEW states_mv\n"
+ + " BUILD DEFERRED REFRESH INCREMENTAL ON MANUAL\n"
+ + " DISTRIBUTED BY RANDOM BUCKETS 2\n"
+ + " PROPERTIES ('replication_num' = '1')\n"
+ + " AS SELECT k1, v1 FROM states_base");
+
+ MTMV mtmv = (MTMV) Env.getCurrentInternalCatalog()
+ .getDb("alter_partition_states_test").get()
+ .getTableOrMetaException("states_mv");
+ String partitionName = mtmv.getPartitionNames().iterator().next();
+ TableNameInfo tableName = new TableNameInfo(mtmv.getQualifiedDbName(),
mtmv.getName());
+
+ // A payload carrying state applies it. The live map keeps moving
after the payload was taken,
+ // and for a restart only the bytes in the journal matter, so both are
driven here.
+ MTMVPartitionState state = new MTMVPartitionState(0, 1);
+ Map<String, MTMVPartitionState> states = new LinkedHashMap<>();
+ states.put(partitionName, state);
+ AlterMTMV withState = new AlterMTMV(tableName,
MTMVAlterOpType.ALTER_PARTITION_STATES);
+ withState.setPartitionStates(states);
+ state.setLatestEpoch(7);
+ // The MV starts without any state, so only the replayed payload can
put it there.
+ mtmv.alterPartitionStates(Map.of());
+
+ replayFromJournal(withState);
+
+ Map<String, MTMVPartitionState> applied = mtmv.getPartitionStates();
+ Assertions.assertEquals(Set.of(partitionName), applied.keySet());
+ Assertions.assertEquals(0,
applied.get(partitionName).getRefreshEpoch());
+ Assertions.assertEquals(1,
applied.get(partitionName).getLatestEpoch());
+
+ // An explicit empty map empties the states.
+ AlterMTMV empty = new AlterMTMV(tableName,
MTMVAlterOpType.ALTER_PARTITION_STATES);
+ empty.setPartitionStates(Map.of());
+ Assertions.assertTrue(new String(journalBytes(empty),
StandardCharsets.UTF_8).contains("\"pst\""),
+ "the state member should be written under its serialized
name");
+
+ replayFromJournal(empty);
+
+ Assertions.assertTrue(mtmv.getPartitionStates().isEmpty());
+
+ // A payload written before the member existed leaves the states alone
instead of clearing them.
+ mtmv.alterPartitionStates(Map.of(partitionName, new
MTMVPartitionState(4, 6)));
+ JsonObject legacy =
JsonParser.parseString(GsonUtils.GSON.toJson(withState)).getAsJsonObject();
+ Assertions.assertNotNull(legacy.remove("pst"));
+
+ replayFromJournal(GsonUtils.GSON.fromJson(legacy.toString(),
AlterMTMV.class));
+
+ MTMVPartitionState kept = mtmv.getPartitionStates().get(partitionName);
+ Assertions.assertEquals(4, kept.getRefreshEpoch());
+ Assertions.assertEquals(6, kept.getLatestEpoch());
+ }
+
+ /** The bytes the edit log writes for an alter record. */
+ private static byte[] journalBytes(AlterMTMV alter) throws IOException {
+ ByteArrayOutputStream bytes = new ByteArrayOutputStream();
+ try (DataOutputStream out = new DataOutputStream(bytes)) {
+ alter.write(out);
+ }
+ return bytes.toByteArray();
+ }
+
+ /** Replays an alter record the way a restart does: from what the journal
wrote, not from memory. */
+ private static void replayFromJournal(AlterMTMV alter) throws Exception {
+ AlterMTMV replayed;
+ try (DataInputStream in = new DataInputStream(new
ByteArrayInputStream(journalBytes(alter)))) {
+ replayed = AlterMTMV.read(in);
+ }
+ Env.getCurrentEnv().getAlterInstance().processAlterMTMV(replayed,
true);
+ }
+
@Test
public void testCreateIncrementalMtmvAutoCreatesStream() throws Exception {
createDatabaseAndUse("stream_test");
diff --git a/fe/fe-core/src/test/java/org/apache/doris/mtmv/MTMVTest.java
b/fe/fe-core/src/test/java/org/apache/doris/mtmv/MTMVTest.java
index b3699aedee0..6877004f2c2 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/mtmv/MTMVTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/mtmv/MTMVTest.java
@@ -54,6 +54,8 @@ import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Range;
import com.google.common.collect.Sets;
+import com.google.gson.JsonObject;
+import com.google.gson.JsonParser;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import org.mockito.MockedStatic;
@@ -515,4 +517,169 @@ public class MTMVTest {
Column.IVM_HIDDEN_COLUMN_PREFIX + "SNAPSHOT_COL__",
"k1"), insertedColumnNames);
}
+
+ @Test
+ public void testPartitionStatesSurviveImageRoundTrip() {
+ MTMV mtmv = buildSerializableMTMV();
+ mtmv.alterPartitionStates(Map.of("p202601", new MTMVPartitionState(3,
5)));
+
+ MTMV restored = GsonUtils.GSON.fromJson(GsonUtils.GSON.toJson(mtmv),
MTMV.class);
+
+ Map<String, MTMVPartitionState> states = restored.getPartitionStates();
+ Assertions.assertEquals(Sets.newHashSet("p202601"), states.keySet());
+ Assertions.assertEquals(3, states.get("p202601").getRefreshEpoch());
+ Assertions.assertEquals(5, states.get("p202601").getLatestEpoch());
+ }
+
+ @Test
+ public void testPartitionStatesEmptyOnImageWrittenBeforeTheFieldExisted() {
+ MTMV mtmv = buildSerializableMTMV();
+ mtmv.alterPartitionStates(Map.of("p202601", new MTMVPartitionState(3,
5)));
+ JsonObject image =
JsonParser.parseString(GsonUtils.GSON.toJson(mtmv)).getAsJsonObject();
+ Assertions.assertNotNull(image.remove("pst"));
+
+ // The field is gone from the image, so gsonPostProcess() is the only
thing that can make it a map.
+ MTMV restored = GsonUtils.GSON.fromJson(image.toString(), MTMV.class);
+
+ // Read the field itself: the getter lazily creates the map, so it
would hide a missing init.
+ Assertions.assertNotNull(Deencapsulation.getField(restored,
"partitionStates"));
+ Assertions.assertTrue(restored.getPartitionStates().isEmpty());
+ }
+
+ @Test
+ public void testPartitionStatesGetterIsNeverNull() {
+ MTMV mtmv = new MTMV();
+ // Never loaded from an image and never populated: still a map, not a
null.
+ Assertions.assertTrue(mtmv.getPartitionStates().isEmpty());
+ mtmv.alterPartitionStates(null);
+ Assertions.assertTrue(mtmv.getPartitionStates().isEmpty());
+ }
+
+ @Test
+ public void testPartitionStatesGetterReturnsAnUnmodifiableSnapshot() {
+ MTMV mtmv = buildSerializableMTMV();
+ mtmv.alterPartitionStates(Map.of("p202601", new MTMVPartitionState(3,
5)));
+
+ Map<String, MTMVPartitionState> states = mtmv.getPartitionStates();
+ Assertions.assertThrows(UnsupportedOperationException.class,
+ () -> states.put("p202602", new MTMVPartitionState(0, 1)));
+
+ // The values are copies too: changing one may not reach the state the
MV owns.
+ states.get("p202601").setLatestEpoch(9);
+ Assertions.assertEquals(5,
mtmv.getPartitionStates().get("p202601").getLatestEpoch());
+ }
+
+ @Test
+ public void testAlterPartitionStatesTakesADetachedSnapshot() {
+ MTMVPartitionState live = new MTMVPartitionState(0, 1);
+ Map<String, MTMVPartitionState> liveStates = Maps.newLinkedHashMap();
+ liveStates.put("p202601", live);
+
+ AlterMTMV alterMTMV = new AlterMTMV(
+ new TableNameInfo("db1", "mv1"),
MTMVAlterOpType.ALTER_PARTITION_STATES);
+ alterMTMV.setPartitionStates(liveStates);
+ // A batched edit log serializes the payload after the MV lock was
released, so the payload must
+ // not follow the live map any further.
+ live.setLatestEpoch(2);
+ liveStates.remove("p202601");
+
+ Assertions.assertEquals(1,
alterMTMV.getPartitionStates().get("p202601").getLatestEpoch());
+ }
+
+ @Test
+ public void
testAddTaskResultReplayKeepsPartitionStatesWhenTheJournalHasNoField() {
+ MTMV mtmv = buildSerializableMTMV();
+ mtmv.getIvmInfo().setEnableIvm(true);
+ mtmv.alterPartitionStates(Map.of("p202601", new MTMVPartitionState(3,
5)));
+
+ // A journal written before the field existed carries no state at all:
it must not clear what is
+ // already there.
+ runAddTaskResult(mtmv, null, true);
+
+ Map<String, MTMVPartitionState> states = mtmv.getPartitionStates();
+ Assertions.assertEquals(Sets.newHashSet("p202601"), states.keySet());
+ Assertions.assertEquals(3, states.get("p202601").getRefreshEpoch());
+ Assertions.assertEquals(5, states.get("p202601").getLatestEpoch());
+ }
+
+ @Test
+ public void testAddTaskResultReplayAppliesPartitionStates() {
+ MTMV mtmv = buildSerializableMTMV();
+ mtmv.getIvmInfo().setEnableIvm(true);
+ mtmv.alterPartitionStates(Map.of("p202601", new MTMVPartitionState(0,
1)));
+
+ List<AlterMTMV> journaled = runAddTaskResult(mtmv, Map.of("p202601",
new MTMVPartitionState(3, 5)), true);
+
+ // Replay never writes a journal of its own.
+ Assertions.assertTrue(journaled.isEmpty());
+ MTMVPartitionState state = mtmv.getPartitionStates().get("p202601");
+ Assertions.assertEquals(3, state.getRefreshEpoch());
+ Assertions.assertEquals(5, state.getLatestEpoch());
+ }
+
+ @Test
+ public void testIvmTaskResultJournalsPartitionStates() {
+ MTMV mtmv = buildSerializableMTMV();
+ mtmv.getIvmInfo().setEnableIvm(true);
+ mtmv.alterPartitionStates(Map.of("p202601", new MTMVPartitionState(3,
5)));
+
+ List<AlterMTMV> journaled = runAddTaskResult(mtmv, null, false);
+
+ Assertions.assertEquals(1, journaled.size());
+ MTMVPartitionState journaledState =
journaled.get(0).getPartitionStates().get("p202601");
+ Assertions.assertEquals(3, journaledState.getRefreshEpoch());
+ Assertions.assertEquals(5, journaledState.getLatestEpoch());
+
+ // The payload reaches the journal as JSON, so it has to survive that
trip to be replayable.
+ AlterMTMV readBack = GsonUtils.GSON.fromJson(
+ GsonUtils.GSON.toJson(journaled.get(0)), AlterMTMV.class);
+ Assertions.assertEquals(3,
readBack.getPartitionStates().get("p202601").getRefreshEpoch());
+ Assertions.assertEquals(5,
readBack.getPartitionStates().get("p202601").getLatestEpoch());
+ }
+
+ @Test
+ public void testNonIvmTaskResultDoesNotJournalPartitionStates() {
+ MTMV mtmv = buildSerializableMTMV();
+ Assertions.assertFalse(mtmv.getIvmInfo().isEnableIvm());
+
+ List<AlterMTMV> journaled = runAddTaskResult(mtmv, null, false);
+
+ // The payload of a non-IVM MV has to stay byte-for-byte what it was
before the field existed.
+ Assertions.assertEquals(1, journaled.size());
+ Assertions.assertNull(journaled.get(0).getPartitionStates());
+ }
+
+ /**
+ * Runs one ADD_TASK result through {@link MTMV#addTaskResult}, optionally
carrying {@code
+ * journaledStates} in its payload the way a real journal would, and
returns the payloads that
+ * reached the edit log -- which stays empty on the replay path.
+ */
+ private List<AlterMTMV> runAddTaskResult(MTMV mtmv, Map<String,
MTMVPartitionState> journaledStates,
+ boolean isReplay) {
+ Env env = Mockito.mock(Env.class);
+ EditLog editLog = Mockito.mock(EditLog.class);
+ EditLogItem editLogItem = Mockito.mock(EditLogItem.class);
+ List<AlterMTMV> journaled = Lists.newArrayList();
+ Mockito.when(env.getEditLog()).thenReturn(editLog);
+
Mockito.when(env.getMtmvService()).thenReturn(Mockito.mock(MTMVService.class));
+
Mockito.when(editLog.submitEdit(Mockito.eq(OperationType.OP_ALTER_MTMV),
Mockito.any(AlterMTMV.class)))
+ .thenAnswer(invocation -> {
+ journaled.add(invocation.getArgument(1));
+ return editLogItem;
+ });
+
+ MTMVTask task = new MTMVTask(mtmv, mtmv.getRelation(), null);
+ task.setStatus(TaskStatus.FAILED);
+ AlterMTMV alterMTMV = new AlterMTMV(new TableNameInfo("db1", "mv1"),
MTMVAlterOpType.ADD_TASK);
+ alterMTMV.setTask(task);
+ alterMTMV.setRelation(mtmv.getRelation());
+ alterMTMV.setPartitionSnapshots(Map.of());
+ alterMTMV.setPartitionStates(journaledStates);
+
+ try (MockedStatic<Env> mockedEnv = Mockito.mockStatic(Env.class)) {
+ mockedEnv.when(Env::getCurrentEnv).thenReturn(env);
+ Assertions.assertTrue(mtmv.addTaskResult(alterMTMV, isReplay));
+ }
+ return journaled;
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
