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 8c30d35a733426..28613904211b23 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 void processAlterMTMV(AlterMTMV alterMTMV, boolean isReplay) { // 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 b162a2b2d511b7..eab4fb5f2f6a99 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.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; @@ -103,6 +104,19 @@ 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. + * + *

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. + * + *

Null means the same thing and has the same two causes: an image written before the field + * existed, and a non-IVM MV. {@link #gsonPostProcess()} turns it into an empty map on load. + */ + @SerializedName("pst") + private Map 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 +304,11 @@ public boolean addTaskResult(AlterMTMV alterMTMV, boolean isReplay) { // 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 +342,10 @@ public boolean addTaskResult(AlterMTMV alterMTMV, boolean isReplay) { } 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 +621,33 @@ public void alterIvmInfo(IvmInfo ivmInfo) { } } + /** + * Read under the MV lock, like {@link #getIvmInfo()}: the map may be null before + * {@link #gsonPostProcess()} has run, and a reader must never see a half-applied replay payload. + */ + public Map getPartitionStates() { + writeMvLock(); + try { + if (partitionStates == null) { + partitionStates = Maps.newLinkedHashMap(); + } + return partitionStates; + } finally { + writeMvUnlock(); + } + } + + // ALTER_PARTITION_STATES replay applies a detached snapshot here, mirroring alterIvmInfo(). Live + // invalidation changes submit their journal from the mutating method instead. + public void alterPartitionStates(Map partitionStates) { + writeMvLock(); + try { + this.partitionStates = MTMVPartitionState.copyOf(partitionStates); + } finally { + writeMvUnlock(); + } + } + public void invalidateIvmBaseline() { EditLogItem editLogItem; writeMvLock(); @@ -997,6 +1047,11 @@ public void gsonPostProcess() throws IOException { 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 a8e81446deaf0f..777dca2aece5ff 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 00000000000000..edb30f84177477 --- /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. + * + *

{@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. + * + *

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. + * + *

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. + * + *

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 copyOf(Map states) { + if (states == null) { + return null; + } + Map copy = new LinkedHashMap<>(); + for (Entry 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 83c81f6ac77426..208417698b8cca 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.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 partitionSnapshots; @SerializedName("ii") private IvmInfo ivmInfo; + @SerializedName("pst") + private Map partitionStates; public AlterMTMV(TableNameInfo mvName, MTMVRefreshInfo refreshInfo, MTMVAlterOpType opType) { this.mvName = Objects.requireNonNull(mvName, "require mvName object"); @@ -148,6 +151,14 @@ public void setIvmInfo(IvmInfo ivmInfo) { this.ivmInfo = ivmInfo == null ? null : new IvmInfo(ivmInfo); } + public Map getPartitionStates() { + return partitionStates; + } + + public void setPartitionStates(Map 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 deeed90db1d0e0..65d56d8959c8b5 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 @@ -37,6 +37,8 @@ import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; +import java.util.LinkedHashMap; +import java.util.Map; import java.util.Set; @@ -403,6 +405,44 @@ public void testAlterIvmInfoPersistence() throws Exception { 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(); + + MTMVPartitionState state = new MTMVPartitionState(0, 1); + Map states = new LinkedHashMap<>(); + states.put(partitionName, state); + TableNameInfo tableName = new TableNameInfo(mtmv.getQualifiedDbName(), mtmv.getName()); + AlterMTMV replayAlter = new AlterMTMV(tableName, MTMVAlterOpType.ALTER_PARTITION_STATES); + replayAlter.setPartitionStates(states); + // The live map keeps moving after the payload was taken; the payload must not follow it. + state.setLatestEpoch(7); + // The MV starts without any state, so only the replayed payload can put it there. + mtmv.alterPartitionStates(Map.of()); + + Env.getCurrentEnv().getAlterInstance().processAlterMTMV(replayAlter, true); + + Map replayed = mtmv.getPartitionStates(); + Assertions.assertEquals(Set.of(partitionName), replayed.keySet()); + Assertions.assertEquals(0, replayed.get(partitionName).getRefreshEpoch()); + Assertions.assertEquals(1, replayed.get(partitionName).getLatestEpoch()); + } + @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 b3699aedee08ad..17119e4b4cd623 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.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,149 @@ public void testGetInsertedColumnNamesIncludesAllIvmHiddenColumns() { Column.IVM_HIDDEN_COLUMN_PREFIX + "SNAPSHOT_COL__", "k1"), insertedColumnNames); } + + @Test + public void testPartitionStatesSurviveImageRoundTrip() { + MTMV mtmv = buildSerializableMTMV(); + mtmv.getPartitionStates().put("p202601", new MTMVPartitionState(3, 5)); + + MTMV restored = GsonUtils.GSON.fromJson(GsonUtils.GSON.toJson(mtmv), MTMV.class); + + Map 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.getPartitionStates().put("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 testAlterPartitionStatesTakesADetachedSnapshot() { + MTMVPartitionState live = new MTMVPartitionState(0, 1); + Map 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 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 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 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()); + } + + @Test + public void testNonIvmTaskResultDoesNotJournalPartitionStates() { + MTMV mtmv = buildSerializableMTMV(); + Assertions.assertFalse(mtmv.getIvmInfo().isEnableIvm()); + + List 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 runAddTaskResult(MTMV mtmv, Map journaledStates, + boolean isReplay) { + Env env = Mockito.mock(Env.class); + EditLog editLog = Mockito.mock(EditLog.class); + EditLogItem editLogItem = Mockito.mock(EditLogItem.class); + List 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 mockedEnv = Mockito.mockStatic(Env.class)) { + mockedEnv.when(Env::getCurrentEnv).thenReturn(env); + Assertions.assertTrue(mtmv.addTaskResult(alterMTMV, isReplay)); + } + return journaled; + } }