This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.2 by this push:
new 5b34bb36351 branch-4.2: [improvement](mtmv) Journal MV alter ops under
the MV lock and read IVM info lock-free #69256 (#70097)
5b34bb36351 is described below
commit 5b34bb3635173fbc33084eb97377ab9ceafbc82e
Author: morrySnow <[email protected]>
AuthorDate: Wed Nov 4 08:28:43 2026 +0800
branch-4.2: [improvement](mtmv) Journal MV alter ops under the MV lock and
read IVM info lock-free #69256 (#70097)
picked from master #69256
Co-authored-by: yujun <[email protected]>
---
.../main/java/org/apache/doris/alter/Alter.java | 21 ++-
.../main/java/org/apache/doris/catalog/MTMV.java | 60 +++++++-
.../java/org/apache/doris/persist/EditLog.java | 68 ++++++++-
.../apache/doris/mtmv/MTMVAlterJournalTest.java | 152 +++++++++++++++++++++
.../apache/doris/persist/EditLogSubmitTest.java | 86 ++++++++++++
5 files changed, 372 insertions(+), 15 deletions(-)
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 6997f9be3e0..181f84a8cdb 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
@@ -90,6 +90,7 @@ import
org.apache.doris.nereids.trees.plans.commands.DropMaterializedViewCommand
import org.apache.doris.persist.AlterMTMV;
import org.apache.doris.persist.AlterViewInfo;
import org.apache.doris.persist.BatchModifyPartitionsInfo;
+import org.apache.doris.persist.EditLog.EditLogItem;
import org.apache.doris.persist.ModifyCommentOperationLog;
import org.apache.doris.persist.ModifyPartitionInfo;
import org.apache.doris.persist.ModifyTableEngineOperationLog;
@@ -1299,24 +1300,22 @@ public class Alter {
public void processAlterMTMV(AlterMTMV alterMTMV, boolean isReplay) {
TableNameInfo tbl = alterMTMV.getMvName();
MTMV mtmv = null;
- boolean alterSuccess = true;
+ EditLogItem editLogItem = null;
try {
Database db =
Env.getCurrentInternalCatalog().getDbOrDdlException(tbl.getDb());
mtmv = (MTMV) db.getTableOrMetaException(tbl.getTbl(),
TableType.MATERIALIZED_VIEW);
switch (alterMTMV.getOpType()) {
case ALTER_REFRESH_INFO:
- mtmv.alterRefreshInfo(alterMTMV.getRefreshInfo());
+ editLogItem = mtmv.alterRefreshInfo(alterMTMV, isReplay);
break;
case ALTER_STATUS:
- mtmv.alterStatus(alterMTMV.getStatus());
+ editLogItem = mtmv.alterStatus(alterMTMV, isReplay);
break;
case ALTER_PROPERTY:
- mtmv.alterMvProperties(alterMTMV.getMvProperties());
+ editLogItem = mtmv.alterMvProperties(alterMTMV, isReplay);
break;
case ADD_TASK:
- alterSuccess = mtmv.addTaskResult(alterMTMV.getTask(),
alterMTMV.getRelation(),
- alterMTMV.getPartitionSnapshots(),
- isReplay);
+ mtmv.addTaskResult(alterMTMV, isReplay);
// If it is not a replay thread, it means that the current
service is already a new version
// and does not require compatibility
if (isReplay) {
@@ -1326,13 +1325,13 @@ public class Alter {
default:
throw new RuntimeException("Unknown type value: " +
alterMTMV.getOpType());
}
+ // Await before rebuilding jobs, which can fail after the metadata
has changed.
+ if (editLogItem != null) {
+ editLogItem.await();
+ }
if (alterMTMV.isNeedRebuildJob()) {
Env.getCurrentEnv().getMtmvService().alterJob(mtmv, isReplay);
}
- // 4. log it and replay it in the follower
- if (!isReplay && alterSuccess) {
- Env.getCurrentEnv().getEditLog().logAlterMTMV(alterMTMV);
- }
} catch (UserException e) {
// if MTMV has been dropped, ignore this exception
LOG.warn(e);
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 24fe0adb30f..fe07e66bcab 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
@@ -45,6 +45,9 @@ import org.apache.doris.mtmv.MTMVRelation;
import org.apache.doris.mtmv.MTMVSnapshotIf;
import org.apache.doris.mtmv.MTMVStatus;
import org.apache.doris.nereids.rules.analysis.SessionVarGuardRewriter;
+import org.apache.doris.persist.AlterMTMV;
+import org.apache.doris.persist.EditLog.EditLogItem;
+import org.apache.doris.persist.OperationType;
import org.apache.doris.qe.ConnectContext;
import com.google.common.collect.Maps;
@@ -176,6 +179,17 @@ public class MTMV extends OlapTable {
}
}
+ /** Apply and enqueue under one MV lock; the caller awaits after releasing
the lock. */
+ public EditLogItem alterRefreshInfo(AlterMTMV alterMTMV, boolean isReplay)
{
+ writeMvLock();
+ try {
+ alterRefreshInfo(alterMTMV.getRefreshInfo());
+ return isReplay ? null : submitAlterLog(alterMTMV);
+ } finally {
+ writeMvUnlock();
+ }
+ }
+
public MTMVRefreshInfo alterRefreshInfo(MTMVRefreshInfo newRefreshInfo) {
writeMvLock();
try {
@@ -185,6 +199,17 @@ public class MTMV extends OlapTable {
}
}
+ /** Apply and enqueue under one MV lock; the caller awaits after releasing
the lock. */
+ public EditLogItem alterStatus(AlterMTMV alterMTMV, boolean isReplay) {
+ writeMvLock();
+ try {
+ alterStatus(alterMTMV.getStatus());
+ return isReplay ? null : submitAlterLog(alterMTMV);
+ } finally {
+ writeMvUnlock();
+ }
+ }
+
public MTMVStatus alterStatus(MTMVStatus newStatus) {
writeMvLock();
try {
@@ -204,6 +229,17 @@ public class MTMV extends OlapTable {
public boolean addTaskResult(MTMVTask task, MTMVRelation relation,
Map<String, MTMVRefreshPartitionSnapshot> partitionSnapshots,
boolean isReplay) {
+ return addTaskResult(task, relation, partitionSnapshots, isReplay,
null);
+ }
+
+ public boolean addTaskResult(AlterMTMV alterMTMV, boolean isReplay) {
+ return addTaskResult(alterMTMV.getTask(), alterMTMV.getRelation(),
alterMTMV.getPartitionSnapshots(),
+ isReplay, alterMTMV);
+ }
+
+ private boolean addTaskResult(MTMVTask task, MTMVRelation relation,
+ Map<String, MTMVRefreshPartitionSnapshot> partitionSnapshots,
boolean isReplay, AlterMTMV alterMTMV) {
+ EditLogItem editLogItem = null;
MTMVCache mtmvCacheWithGuard = null;
MTMVCache mtmvCacheWithoutGuard = null;
boolean needUpdateCache = false;
@@ -270,7 +306,29 @@ public class MTMV extends OlapTable {
}
Env.getCurrentEnv().getMtmvService()
.refreshComplete(this, relation, task);
- return true;
+ if (!isReplay && alterMTMV != null) {
+ editLogItem = submitAlterLog(alterMTMV);
+ }
+ } finally {
+ writeMvUnlock();
+ }
+ if (editLogItem != null) {
+ editLogItem.await();
+ }
+ return true;
+ }
+
+ private EditLogItem submitAlterLog(AlterMTMV alterMTMV) {
+ // All journaled MTMV operations use this lock order, including
properties and task results.
+ return
Env.getCurrentEnv().getEditLog().submitEdit(OperationType.OP_ALTER_MTMV,
alterMTMV);
+ }
+
+ /** Apply and enqueue under one MV lock; the caller awaits after releasing
the lock. */
+ public EditLogItem alterMvProperties(AlterMTMV alterMTMV, boolean
isReplay) {
+ writeMvLock();
+ try {
+ alterMvProperties(alterMTMV.getMvProperties());
+ return isReplay ? null : submitAlterLog(alterMTMV);
} finally {
writeMvUnlock();
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/persist/EditLog.java
b/fe/fe-core/src/main/java/org/apache/doris/persist/EditLog.java
index c5450abd6ba..5ad8af21e38 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/persist/EditLog.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/persist/EditLog.java
@@ -130,13 +130,14 @@ import java.util.concurrent.atomic.AtomicLong;
public class EditLog {
public static final Logger LOG = LogManager.getLogger(EditLog.class);
- // Helper class to hold log edit requests
- private static class EditLogItem {
+ // Helper class to hold log edit requests.
+ // Public so that callers can enqueue inside a lock and await outside it.
+ public static class EditLogItem {
static AtomicLong nextUid = new AtomicLong(0);
final short op;
final Writable writable;
final Object lock = new Object();
- boolean finished = false;
+ volatile boolean finished = false;
long logId = -1;
long uid = -1;
@@ -145,6 +146,24 @@ public class EditLog {
this.writable = writable;
uid = nextUid.getAndIncrement();
}
+
+ /**
+ * Wait for this edit log entry to be flushed to persistent storage.
+ * Returns the assigned log ID.
+ */
+ public long await() {
+ synchronized (lock) {
+ while (!finished) {
+ try {
+ lock.wait();
+ } catch (InterruptedException e) {
+ LOG.error("Fatal Error : write stream Exception");
+ System.exit(-1);
+ }
+ }
+ }
+ return logId;
+ }
}
private final BlockingQueue<EditLogItem> logEditQueue = new
LinkedBlockingQueue<>();
@@ -1589,6 +1608,49 @@ public class EditLog {
return req.logId;
}
+ /**
+ * Submit an edit log entry to the batch queue without waiting for it to
be flushed.
+ * The entry is enqueued in FIFO order, so calling this inside a write
lock guarantees
+ * that edit log entries are ordered by lock acquisition order.
+ *
+ * <p>The caller MUST call {@link EditLogItem#await()} after releasing the
lock to ensure
+ * the entry is persisted before proceeding.
+ *
+ * <p>If batch edit log is disabled, this falls back to a synchronous
direct write
+ * and the returned item is already completed.
+ *
+ * @return an {@link EditLogItem} handle to await completion
+ */
+ public EditLogItem submitEdit(short op, Writable writable) {
+ if (this.getNumEditStreams() == 0) {
+ LOG.error("Fatal Error : no editLog stream", new Exception());
+ throw new Error("Fatal Error : no editLog stream");
+ }
+
+ EditLogItem req = new EditLogItem(op, writable);
+ if (Config.enable_batch_editlog && op != OperationType.OP_TIMESTAMP) {
+ while (true) {
+ try {
+ logEditQueue.put(req);
+ break;
+ } catch (InterruptedException e) {
+ LOG.warn("Interrupted during put, will sleep and retry.");
+ try {
+ Thread.sleep(100);
+ } catch (InterruptedException ex) {
+ LOG.warn("interrupted during sleep, will retry.", ex);
+ }
+ }
+ }
+ } else {
+ // Non-batch mode: write directly (synchronous)
+ long logId = logEditDirectly(op, writable);
+ req.logId = logId;
+ req.finished = true;
+ }
+ return req;
+ }
+
private synchronized long logEditDirectly(short op, Writable writable) {
long logId = -1;
try {
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/mtmv/MTMVAlterJournalTest.java
b/fe/fe-core/src/test/java/org/apache/doris/mtmv/MTMVAlterJournalTest.java
new file mode 100644
index 00000000000..d6616f95b8e
--- /dev/null
+++ b/fe/fe-core/src/test/java/org/apache/doris/mtmv/MTMVAlterJournalTest.java
@@ -0,0 +1,152 @@
+// 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 org.apache.doris.catalog.Env;
+import org.apache.doris.catalog.MTMV;
+import org.apache.doris.common.jmockit.Deencapsulation;
+import org.apache.doris.info.TableNameInfo;
+import org.apache.doris.job.common.TaskStatus;
+import org.apache.doris.job.extensions.mtmv.MTMVTask;
+import org.apache.doris.mtmv.MTMVRefreshEnum.MTMVState;
+import org.apache.doris.persist.AlterMTMV;
+import org.apache.doris.persist.EditLog;
+import org.apache.doris.persist.EditLog.EditLogItem;
+import org.apache.doris.persist.OperationType;
+
+import org.junit.Assert;
+import org.junit.Test;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+public class MTMVAlterJournalTest {
+ private MTMV createMv() {
+ MTMV mv = Mockito.spy(new MTMV());
+ mv.setStatus(new MTMVStatus());
+ mv.setRefreshInfo(new MTMVRefreshInfo());
+ mv.setMvProperties(new HashMap<>());
+ mv.setJobInfo(new MTMVJobInfo("journal_test"));
+ mv.setMvPartitionInfo(new MTMVPartitionInfo());
+ mv.setRefreshSnapshot(new MTMVRefreshSnapshot());
+ Mockito.doReturn(Collections.emptySet()).when(mv).getPartitionNames();
+ return mv;
+ }
+
+ private AlterMTMV operation(MTMV mv, MTMVAlterOpType type) {
+ AlterMTMV alter = new AlterMTMV(new TableNameInfo("db", "mv"), type);
+ alter.setRefreshInfo(new MTMVRefreshInfo());
+ alter.setStatus(new MTMVStatus(MTMVState.SCHEMA_CHANGE, "changed"));
+ alter.setMvProperties(Collections.singletonMap("grace_period", "12"));
+ MTMVTask task = new MTMVTask(mv, null, null);
+ task.setTaskId(1L);
+ task.setStatus(TaskStatus.FAILED);
+ alter.setTask(task);
+ alter.setPartitionSnapshots(Collections.emptyMap());
+ return alter;
+ }
+
+ private void verifyJournalBoundary(MTMVAlterOpType type) {
+ MTMV mv = createMv();
+ AlterMTMV alter = operation(mv, type);
+ ReentrantReadWriteLock lock = Deencapsulation.getField(mv, "mvRwLock");
+ Env env = Mockito.mock(Env.class);
+ EditLog log = Mockito.mock(EditLog.class);
+ EditLogItem item = Mockito.mock(EditLogItem.class);
+ Mockito.when(env.getEditLog()).thenReturn(log);
+
Mockito.when(env.getMtmvService()).thenReturn(Mockito.mock(MTMVService.class));
+ Mockito.when(log.submitEdit(OperationType.OP_ALTER_MTMV,
alter)).thenAnswer(invocation -> {
+ Assert.assertTrue("journal enqueue must hold the mutation lock",
lock.isWriteLockedByCurrentThread());
+ if (type == MTMVAlterOpType.ALTER_STATUS) {
+ Assert.assertEquals(MTMVState.SCHEMA_CHANGE,
mv.getStatus().getState());
+ }
+ return item;
+ });
+ Mockito.when(item.await()).thenAnswer(invocation -> {
+ Assert.assertFalse("journal flush must not hold the MV lock",
lock.isWriteLockedByCurrentThread());
+ return 1L;
+ });
+ try (MockedStatic<Env> envStatic = Mockito.mockStatic(Env.class)) {
+ envStatic.when(Env::getCurrentEnv).thenReturn(env);
+ EditLogItem pending = null;
+ switch (type) {
+ case ALTER_REFRESH_INFO:
+ pending = mv.alterRefreshInfo(alter, false);
+ break;
+ case ALTER_STATUS:
+ pending = mv.alterStatus(alter, false);
+ break;
+ case ALTER_PROPERTY:
+ pending = mv.alterMvProperties(alter, false);
+ break;
+ case ADD_TASK:
+ Assert.assertTrue(mv.addTaskResult(alter, false));
+ break;
+ default:
+ throw new AssertionError(type);
+ }
+ if (pending != null) {
+ pending.await();
+ }
+ Mockito.verify(log).submitEdit(OperationType.OP_ALTER_MTMV, alter);
+ Mockito.verify(item).await();
+ }
+ }
+
+ @Test
+ public void testRefreshJournalBoundary() {
+ verifyJournalBoundary(MTMVAlterOpType.ALTER_REFRESH_INFO);
+ }
+
+ @Test
+ public void testStatusJournalBoundary() {
+ verifyJournalBoundary(MTMVAlterOpType.ALTER_STATUS);
+ }
+
+ @Test
+ public void testPropertiesJournalBoundary() {
+ verifyJournalBoundary(MTMVAlterOpType.ALTER_PROPERTY);
+ }
+
+ @Test
+ public void testTaskJournalBoundary() {
+ verifyJournalBoundary(MTMVAlterOpType.ADD_TASK);
+ }
+
+ @Test
+ public void testReplayAndStaleTaskDoNotJournal() {
+ MTMV mv = createMv();
+ AlterMTMV task = operation(mv, MTMVAlterOpType.ADD_TASK);
+ Env env = Mockito.mock(Env.class);
+ EditLog log = Mockito.mock(EditLog.class);
+ Mockito.when(env.getEditLog()).thenReturn(log);
+
Mockito.when(env.getMtmvService()).thenReturn(Mockito.mock(MTMVService.class));
+ try (MockedStatic<Env> envStatic = Mockito.mockStatic(Env.class)) {
+ envStatic.when(Env::getCurrentEnv).thenReturn(env);
+ Assert.assertNull(mv.alterRefreshInfo(operation(mv,
MTMVAlterOpType.ALTER_REFRESH_INFO), true));
+ Assert.assertNull(mv.alterStatus(operation(mv,
MTMVAlterOpType.ALTER_STATUS), true));
+ Assert.assertNull(mv.alterMvProperties(operation(mv,
MTMVAlterOpType.ALTER_PROPERTY), true));
+ Assert.assertFalse(mv.addTaskResult(task, false));
+ Assert.assertTrue(mv.addTaskResult(task, true));
+ Mockito.verifyNoInteractions(log);
+ }
+ }
+}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/persist/EditLogSubmitTest.java
b/fe/fe-core/src/test/java/org/apache/doris/persist/EditLogSubmitTest.java
new file mode 100644
index 00000000000..ef3bb6b0355
--- /dev/null
+++ b/fe/fe-core/src/test/java/org/apache/doris/persist/EditLogSubmitTest.java
@@ -0,0 +1,86 @@
+// 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.persist;
+
+import org.apache.doris.common.Config;
+import org.apache.doris.common.io.Text;
+import org.apache.doris.common.jmockit.Deencapsulation;
+import org.apache.doris.journal.Journal;
+import org.apache.doris.persist.EditLog.EditLogItem;
+
+import org.junit.Assert;
+import org.junit.Test;
+import org.mockito.Mockito;
+
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+
+public class EditLogSubmitTest {
+ @Test
+ public void testQueueOrderAndAwaitCompletion() throws Exception {
+ boolean batch = Config.enable_batch_editlog;
+ Config.enable_batch_editlog = true;
+ try {
+ EditLog log = Mockito.mock(EditLog.class,
Mockito.CALLS_REAL_METHODS);
+ BlockingQueue<EditLogItem> queue = new
java.util.concurrent.LinkedBlockingQueue<>();
+ Deencapsulation.setField(log, "logEditQueue", queue);
+ Mockito.when(log.getNumEditStreams()).thenReturn(1);
+ EditLogItem first = log.submitEdit(OperationType.OP_ALTER_MTMV,
new Text("first"));
+ EditLogItem second = log.submitEdit(OperationType.OP_ALTER_MTMV,
new Text("second"));
+ Assert.assertSame(first, queue.remove());
+ Assert.assertSame(second, queue.remove());
+ Assert.assertTrue(queue.isEmpty());
+ CompletableFuture<Long> flushed =
CompletableFuture.supplyAsync(first::await);
+ Assert.assertFalse(flushed.isDone());
+ synchronized (first.lock) {
+ first.logId = 42L;
+ first.finished = true;
+ first.lock.notifyAll();
+ }
+ Assert.assertEquals(Long.valueOf(42), flushed.get(10,
TimeUnit.SECONDS));
+ Assert.assertEquals(42L, first.await());
+ } finally {
+ Config.enable_batch_editlog = batch;
+ }
+ }
+
+ @Test
+ public void testDirectSubmissionIsAlreadyDurable() throws Exception {
+ boolean batch = Config.enable_batch_editlog;
+ int rollNum = Config.edit_log_roll_num;
+ Config.enable_batch_editlog = false;
+ Config.edit_log_roll_num = Integer.MAX_VALUE;
+ try {
+ EditLog log = Mockito.mock(EditLog.class,
Mockito.CALLS_REAL_METHODS);
+ Journal journal = Mockito.mock(Journal.class);
+ Deencapsulation.setField(log, "journal", journal);
+ Deencapsulation.setField(log, "lastEditLogRollTimeMs",
System.currentTimeMillis());
+ Text record = new Text("delta");
+ Mockito.when(journal.write(OperationType.OP_ALTER_MTMV,
record)).thenReturn(7L);
+ EditLogItem item = log.submitEdit(OperationType.OP_ALTER_MTMV,
record);
+ Assert.assertTrue(item.finished);
+ Assert.assertEquals(7L, item.await());
+ Mockito.verify(journal).write(OperationType.OP_ALTER_MTMV, record);
+ } finally {
+ Config.enable_batch_editlog = batch;
+ Config.edit_log_roll_num = rollNum;
+ }
+ }
+
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
