This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new 5491bbdf3d9 branch-4.1:[feature](fe) Show compute group for MTMV
refresh task (#63206) (#63933)
5491bbdf3d9 is described below
commit 5491bbdf3d9c34d5460b5e86fe3b5f16db407a9c
Author: seawinde <[email protected]>
AuthorDate: Mon Jun 1 18:32:52 2026 +0800
branch-4.1:[feature](fe) Show compute group for MTMV refresh task (#63206)
(#63933)
pr: #63206
commitId: 66dbb85
---
.../apache/doris/job/extensions/mtmv/MTMVTask.java | 31 +++++++-
.../doris/job/extensions/mtmv/MTMVTaskContext.java | 12 ++-
.../java/org/apache/doris/mtmv/MTMVJobManager.java | 25 +++++-
.../org/apache/doris/mtmv/MTMVJobManagerTest.java | 88 ++++++++++++++++++++++
.../java/org/apache/doris/mtmv/MTMVTaskTest.java | 88 +++++++++++++++++++++-
5 files changed, 237 insertions(+), 7 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/mtmv/MTMVTask.java
b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/mtmv/MTMVTask.java
index ccf1e9be899..5d9b9f6c50e 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/mtmv/MTMVTask.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/mtmv/MTMVTask.java
@@ -24,6 +24,7 @@ import org.apache.doris.catalog.Env;
import org.apache.doris.catalog.MTMV;
import org.apache.doris.catalog.ScalarType;
import org.apache.doris.catalog.TableIf;
+import org.apache.doris.cloud.qe.ComputeGroupException;
import org.apache.doris.common.AnalysisException;
import org.apache.doris.common.Config;
import org.apache.doris.common.DdlException;
@@ -70,6 +71,7 @@ import org.apache.doris.thrift.TRow;
import org.apache.doris.thrift.TStatusCode;
import org.apache.doris.thrift.TUniqueId;
+import com.google.common.base.Strings;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Lists;
@@ -114,7 +116,8 @@ public class MTMVTask extends AbstractTask {
new Column("NeedRefreshPartitions", ScalarType.createStringType()),
new Column("CompletedPartitions", ScalarType.createStringType()),
new Column("Progress", ScalarType.createStringType()),
- new Column("LastQueryId", ScalarType.createStringType()));
+ new Column("LastQueryId", ScalarType.createStringType()),
+ new Column("ComputeGroup", ScalarType.createStringType()));
public static final ImmutableMap<String, Integer> COLUMN_TO_INDEX;
@@ -152,6 +155,8 @@ public class MTMVTask extends AbstractTask {
MTMVTaskRefreshMode refreshMode;
@SerializedName("lastQueryId")
String lastQueryId;
+ @SerializedName("cg")
+ private String computeGroup;
private MTMV mtmv;
private MTMVRelation relation;
@@ -322,6 +327,8 @@ public class MTMVTask extends AbstractTask {
Map<TableIf, String> tableWithPartKey)
throws Exception {
ConnectContext ctx = MTMVPlanUtil.createMTMVContext(mtmv,
MTMVPlanUtil.DISABLE_RULES_WHEN_RUN_MTMV_TASK);
+ setComputeGroup(ctx);
+ recordComputeGroup(ctx);
StatementContext statementContext = new StatementContext();
for (Entry<MvccTableInfo, MvccSnapshot> entry : snapshots.entrySet()) {
statementContext.setSnapshot(entry.getKey(), entry.getValue());
@@ -354,6 +361,26 @@ public class MTMVTask extends AbstractTask {
}
}
+ private void setComputeGroup(ConnectContext ctx) {
+ String taskComputeGroup = taskContext.getComputeGroup();
+ if (Config.isCloudMode() && !Strings.isNullOrEmpty(taskComputeGroup)) {
+ ctx.setCloudCluster(taskComputeGroup);
+ }
+ }
+
+ private void recordComputeGroup(ConnectContext ctx) {
+ if (!Config.isCloudMode()) {
+ computeGroup = FeConstants.null_string;
+ return;
+ }
+ try {
+ computeGroup = ctx.getCloudCluster(false);
+ } catch (ComputeGroupException e) {
+ computeGroup = FeConstants.null_string;
+ LOG.warn("failed to resolve compute group for mtmv task, taskId:
{}", getTaskId(), e);
+ }
+ }
+
private String getDummyStmt(Set<String> refreshPartitionNames) {
String mvName = mtmv.getName();
DatabaseIf database = mtmv.getDatabase();
@@ -532,6 +559,8 @@ public class MTMVTask extends AbstractTask {
new TCell().setStringVal(getProgress()));
trow.addToColumnValue(
new TCell().setStringVal(lastQueryId));
+ trow.addToColumnValue(new TCell().setStringVal(
+ computeGroup == null || computeGroup.isEmpty() ?
FeConstants.null_string : computeGroup));
return trow;
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/mtmv/MTMVTaskContext.java
b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/mtmv/MTMVTaskContext.java
index 1c53f5dd301..1d81a5d6a44 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/mtmv/MTMVTaskContext.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/mtmv/MTMVTaskContext.java
@@ -34,14 +34,19 @@ public class MTMVTaskContext {
@SerializedName(value = "isComplete")
private boolean isComplete;
+ @SerializedName(value = "computeGroup")
+ private String computeGroup;
+
public MTMVTaskContext(MTMVTaskTriggerMode triggerMode) {
this.triggerMode = triggerMode;
}
- public MTMVTaskContext(MTMVTaskTriggerMode triggerMode, List<String>
partitions, boolean isComplete) {
+ public MTMVTaskContext(MTMVTaskTriggerMode triggerMode, List<String>
partitions, boolean isComplete,
+ String computeGroup) {
this.triggerMode = triggerMode;
this.partitions = partitions;
this.isComplete = isComplete;
+ this.computeGroup = computeGroup;
}
public List<String> getPartitions() {
@@ -56,12 +61,17 @@ public class MTMVTaskContext {
return isComplete;
}
+ public String getComputeGroup() {
+ return computeGroup;
+ }
+
@Override
public String toString() {
return "MTMVTaskContext{"
+ "triggerMode=" + triggerMode
+ ", partitions=" + partitions
+ ", isComplete=" + isComplete
+ + ", computeGroup=" + computeGroup
+ '}';
}
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVJobManager.java
b/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVJobManager.java
index 8792868d7db..62d74ea6932 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVJobManager.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVJobManager.java
@@ -23,6 +23,8 @@ import org.apache.doris.catalog.Env;
import org.apache.doris.catalog.MTMV;
import org.apache.doris.catalog.Table;
import org.apache.doris.catalog.TableIf.TableType;
+import org.apache.doris.cloud.qe.ComputeGroupException;
+import org.apache.doris.common.Config;
import org.apache.doris.common.DdlException;
import org.apache.doris.common.MetaNotFoundException;
import org.apache.doris.common.util.TimeUtils;
@@ -42,6 +44,7 @@ import
org.apache.doris.nereids.trees.plans.commands.info.CancelMTMVTaskInfo;
import org.apache.doris.nereids.trees.plans.commands.info.PauseMTMVInfo;
import org.apache.doris.nereids.trees.plans.commands.info.RefreshMTMVInfo;
import org.apache.doris.nereids.trees.plans.commands.info.ResumeMTMVInfo;
+import org.apache.doris.qe.ConnectContext;
import com.google.common.collect.Lists;
import org.apache.commons.lang3.StringUtils;
@@ -64,7 +67,7 @@ public class MTMVJobManager implements MTMVHookService {
if (!mtmv.getRefreshInfo().getBuildMode().equals(BuildMode.IMMEDIATE))
{
return;
}
- MTMVTaskContext mtmvTaskContext = new
MTMVTaskContext(MTMVTaskTriggerMode.SYSTEM, null, true);
+ MTMVTaskContext mtmvTaskContext = new
MTMVTaskContext(MTMVTaskTriggerMode.SYSTEM, null, true, null);
try {
Env.getCurrentEnv().getJobManager().triggerJob(mtmv.getId(),
mtmvTaskContext);
} catch (JobException e) {
@@ -155,10 +158,26 @@ public class MTMVJobManager implements MTMVHookService {
public void refreshMTMV(RefreshMTMVInfo info) throws DdlException,
MetaNotFoundException, JobException {
MTMVJob job = getJobByTableNameInfo(info.getMvName());
MTMVTaskContext mtmvTaskContext = new
MTMVTaskContext(MTMVTaskTriggerMode.MANUAL, info.getPartitions(),
- info.isComplete());
+ info.isComplete(), getCurrentComputeGroup());
Env.getCurrentEnv().getJobManager().triggerJob(job.getJobId(),
mtmvTaskContext);
}
+ private String getCurrentComputeGroup() {
+ if (!Config.isCloudMode()) {
+ return null;
+ }
+ ConnectContext ctx = ConnectContext.get();
+ if (ctx == null) {
+ return null;
+ }
+ try {
+ return ctx.getCloudCluster(false);
+ } catch (ComputeGroupException e) {
+ LOG.warn("failed to resolve compute group for refresh mtmv", e);
+ return null;
+ }
+ }
+
@Override
public void refreshComplete(MTMV mtmv, MTMVRelation relation, MTMVTask
task) {
@@ -202,7 +221,7 @@ public class MTMVJobManager implements MTMVHookService {
return;
}
MTMVTaskContext mtmvTaskContext = new
MTMVTaskContext(MTMVTaskTriggerMode.COMMIT, Lists.newArrayList(),
- false);
+ false, null);
Env.getCurrentEnv().getJobManager().triggerJob(job.getJobId(),
mtmvTaskContext);
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/mtmv/MTMVJobManagerTest.java
b/fe/fe-core/src/test/java/org/apache/doris/mtmv/MTMVJobManagerTest.java
new file mode 100644
index 00000000000..ef5486d7512
--- /dev/null
+++ b/fe/fe-core/src/test/java/org/apache/doris/mtmv/MTMVJobManagerTest.java
@@ -0,0 +1,88 @@
+// 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.Database;
+import org.apache.doris.catalog.Env;
+import org.apache.doris.catalog.MTMV;
+import org.apache.doris.catalog.TableIf.TableType;
+import org.apache.doris.common.Config;
+import org.apache.doris.datasource.InternalCatalog;
+import org.apache.doris.info.TableNameInfo;
+import org.apache.doris.job.extensions.mtmv.MTMVJob;
+import org.apache.doris.job.extensions.mtmv.MTMVTask.MTMVTaskTriggerMode;
+import org.apache.doris.job.extensions.mtmv.MTMVTaskContext;
+import org.apache.doris.job.manager.JobManager;
+import org.apache.doris.nereids.trees.plans.commands.info.RefreshMTMVInfo;
+import org.apache.doris.qe.ConnectContext;
+
+import com.google.common.collect.Lists;
+import org.junit.Assert;
+import org.junit.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
+
+public class MTMVJobManagerTest {
+
+ @Test
+ public void testRefreshMTMVPassesCurrentComputeGroupToTaskContext() throws
Exception {
+ String originCloudUniqueId = Config.cloud_unique_id;
+ ConnectContext previousContext = ConnectContext.get();
+ try (MockedStatic<Env> mockedEnv = Mockito.mockStatic(Env.class)) {
+ Config.cloud_unique_id = "test_cloud";
+ Env env = Mockito.mock(Env.class);
+ InternalCatalog internalCatalog =
Mockito.mock(InternalCatalog.class);
+ Database db = Mockito.mock(Database.class);
+ MTMV mtmv = Mockito.mock(MTMV.class);
+ MTMVJob job = Mockito.mock(MTMVJob.class);
+ JobManager jobManager = Mockito.mock(JobManager.class);
+
+ mockedEnv.when(Env::getCurrentEnv).thenReturn(env);
+
mockedEnv.when(Env::getCurrentInternalCatalog).thenReturn(internalCatalog);
+
Mockito.when(internalCatalog.getDbOrDdlException("db1")).thenReturn(db);
+ Mockito.when(db.getTableOrMetaException(Mockito.eq("mv1"),
Mockito.eq(TableType.MATERIALIZED_VIEW)))
+ .thenReturn(mtmv);
+ Mockito.when(env.getJobManager()).thenReturn(jobManager);
+ Mockito.when(jobManager.getJob(mtmv.getId())).thenReturn(job);
+ Mockito.when(job.getJobId()).thenReturn(100L);
+
+ ConnectContext ctx = new ConnectContext();
+ ctx.setCloudCluster("cg1");
+ ctx.setThreadLocalInfo();
+
+ RefreshMTMVInfo info = new RefreshMTMVInfo(new
TableNameInfo("db1", "mv1"),
+ Lists.newArrayList("p1"), false);
+ new MTMVJobManager().refreshMTMV(info);
+
+ ArgumentCaptor<MTMVTaskContext> captor =
ArgumentCaptor.forClass(MTMVTaskContext.class);
+ Mockito.verify(jobManager).triggerJob(Mockito.eq(100L),
captor.capture());
+ MTMVTaskContext taskContext = captor.getValue();
+ Assert.assertEquals(MTMVTaskTriggerMode.MANUAL,
taskContext.getTriggerMode());
+ Assert.assertEquals(Lists.newArrayList("p1"),
taskContext.getPartitions());
+ Assert.assertFalse(taskContext.isComplete());
+ Assert.assertEquals("cg1", taskContext.getComputeGroup());
+ } finally {
+ Config.cloud_unique_id = originCloudUniqueId;
+ ConnectContext.remove();
+ if (previousContext != null) {
+ previousContext.setThreadLocalInfo();
+ }
+ }
+ }
+}
diff --git a/fe/fe-core/src/test/java/org/apache/doris/mtmv/MTMVTaskTest.java
b/fe/fe-core/src/test/java/org/apache/doris/mtmv/MTMVTaskTest.java
index 1d682c47d05..44b88cc506f 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/mtmv/MTMVTaskTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/mtmv/MTMVTaskTest.java
@@ -18,15 +18,22 @@
package org.apache.doris.mtmv;
import org.apache.doris.analysis.TableName;
+import org.apache.doris.catalog.Column;
import org.apache.doris.catalog.MTMV;
import org.apache.doris.common.AnalysisException;
+import org.apache.doris.common.Config;
import org.apache.doris.common.DdlException;
+import org.apache.doris.common.FeConstants;
import org.apache.doris.common.MetaNotFoundException;
+import org.apache.doris.common.jmockit.Deencapsulation;
import org.apache.doris.job.extensions.mtmv.MTMVTask;
import org.apache.doris.job.extensions.mtmv.MTMVTask.MTMVTaskTriggerMode;
import org.apache.doris.job.extensions.mtmv.MTMVTaskContext;
import org.apache.doris.mtmv.MTMVPartitionInfo.MTMVPartitionType;
import org.apache.doris.mtmv.MTMVRefreshEnum.RefreshMethod;
+import org.apache.doris.persist.gson.GsonUtils;
+import org.apache.doris.qe.ConnectContext;
+import org.apache.doris.thrift.TRow;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
@@ -57,6 +64,7 @@ public class MTMVTaskTest {
private MTMVPartitionInfo mtmvPartitionInfo;
@Mocked
private MTMVRefreshInfo mtmvRefreshInfo;
+ private static final String COMPUTE_GROUP = "ComputeGroup";
@Before
public void setUp()
@@ -101,7 +109,7 @@ public class MTMVTaskTest {
@Test
public void testCalculateNeedRefreshPartitionsManualComplete() throws
AnalysisException {
- MTMVTaskContext context = new
MTMVTaskContext(MTMVTaskTriggerMode.MANUAL, null, true);
+ MTMVTaskContext context = new
MTMVTaskContext(MTMVTaskTriggerMode.MANUAL, null, true, null);
MTMVTask task = new MTMVTask(mtmv, relation, context);
List<String> result = task.calculateNeedRefreshPartitions(null);
Assert.assertEquals(allPartitionNames, result);
@@ -109,7 +117,8 @@ public class MTMVTaskTest {
@Test
public void testCalculateNeedRefreshPartitionsManualPartitions() throws
AnalysisException {
- MTMVTaskContext context = new
MTMVTaskContext(MTMVTaskTriggerMode.MANUAL, Lists.newArrayList(poneName),
false);
+ MTMVTaskContext context = new
MTMVTaskContext(MTMVTaskTriggerMode.MANUAL, Lists.newArrayList(poneName),
+ false, null);
MTMVTask task = new MTMVTask(mtmv, relation, context);
List<String> result = task.calculateNeedRefreshPartitions(null);
Assert.assertEquals(Lists.newArrayList(poneName), result);
@@ -177,4 +186,79 @@ public class MTMVTaskTest {
List<String> result = task.calculateNeedRefreshPartitions(null);
Assert.assertEquals(Lists.newArrayList(ptwoName), result);
}
+
+ @Test
+ public void testTaskSchemaContainsComputeGroup() {
+ Column lastColumn = MTMVTask.SCHEMA.get(MTMVTask.SCHEMA.size() - 1);
+ Assert.assertEquals(COMPUTE_GROUP, lastColumn.getName());
+ Assert.assertEquals(MTMVTask.SCHEMA.size() - 1,
+
MTMVTask.COLUMN_TO_INDEX.get(COMPUTE_GROUP.toLowerCase()).intValue());
+ }
+
+ @Test
+ public void testGetTvfInfoReturnsComputeGroup() {
+ MTMVTask task = new MTMVTask(mtmv, relation, new
MTMVTaskContext(MTMVTaskTriggerMode.MANUAL));
+ Deencapsulation.setField(task, "computeGroup", "cg1");
+
+ TRow row = task.getTvfInfo("job1");
+
+ Assert.assertEquals("cg1", row.getColumnValue()
+
.get(MTMVTask.COLUMN_TO_INDEX.get(COMPUTE_GROUP.toLowerCase())).getStringVal());
+ }
+
+ @Test
+ public void testRecordComputeGroupFromContext() {
+ String originCloudUniqueId = Config.cloud_unique_id;
+ try {
+ Config.cloud_unique_id = "test_cloud";
+ ConnectContext ctx = new ConnectContext();
+ ctx.setCloudCluster("cg1");
+ MTMVTask task = new MTMVTask(mtmv, relation, new
MTMVTaskContext(MTMVTaskTriggerMode.MANUAL));
+
+ Deencapsulation.invoke(task, "recordComputeGroup", ctx);
+ TRow row = task.getTvfInfo("job1");
+
+ Assert.assertEquals("cg1", row.getColumnValue()
+
.get(MTMVTask.COLUMN_TO_INDEX.get(COMPUTE_GROUP.toLowerCase())).getStringVal());
+ } finally {
+ Config.cloud_unique_id = originCloudUniqueId;
+ }
+ }
+
+ @Test
+ public void testSetComputeGroupFromTaskContext() {
+ String originCloudUniqueId = Config.cloud_unique_id;
+ try {
+ Config.cloud_unique_id = "test_cloud";
+ ConnectContext ctx = new ConnectContext();
+ MTMVTaskContext context = new
MTMVTaskContext(MTMVTaskTriggerMode.MANUAL, null, true, "cg1");
+ MTMVTask task = new MTMVTask(mtmv, relation, context);
+
+ Deencapsulation.invoke(task, "setComputeGroup", ctx);
+
+ Assert.assertEquals("cg1",
ctx.getSessionVariable().getCloudCluster());
+ } finally {
+ Config.cloud_unique_id = originCloudUniqueId;
+ }
+ }
+
+ @Test
+ public void testGetTvfInfoReturnsNullStringForMissingComputeGroup() {
+ MTMVTask task = new MTMVTask(mtmv, relation, new
MTMVTaskContext(MTMVTaskTriggerMode.MANUAL));
+
+ TRow row = task.getTvfInfo("job1");
+
+ Assert.assertEquals(FeConstants.null_string, row.getColumnValue()
+
.get(MTMVTask.COLUMN_TO_INDEX.get(COMPUTE_GROUP.toLowerCase())).getStringVal());
+ }
+
+ @Test
+ public void testDeserializeOldTaskWithoutComputeGroup() {
+ MTMVTask task = GsonUtils.GSON.fromJson("{\"di\":1,\"mi\":2}",
MTMVTask.class);
+
+ TRow row = task.getTvfInfo("job1");
+
+ Assert.assertEquals(FeConstants.null_string, row.getColumnValue()
+
.get(MTMVTask.COLUMN_TO_INDEX.get(COMPUTE_GROUP.toLowerCase())).getStringVal());
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
