branch-4.1:[feature](fe) Show compute group for MTMV refresh task (#63206) (#63933)

seawinde · web-flow · commit 5491bbdf3d9c · 2026-06-01T18:32:52.000+08:00
pr: #63206 commitId: 66dbb85
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
@@ -24,6 +24,7 @@
 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.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 enum MTMVTaskRefreshMode {
     MTMVTaskRefreshMode refreshMode;
     @SerializedName("lastQueryId")
     String lastQueryId;
+    @SerializedName("cg")
+    private String computeGroup;
 
     private MTMV mtmv;
     private MTMVRelation relation;
@@ -322,6 +327,8 @@ private void exec(Set<String> refreshPartitionNames,
             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 @@ private void exec(Set<String> refreshPartitionNames,
         }
     }
 
+    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 TRow getTvfInfo(String jobName) {
                 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
@@ -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 boolean isComplete() {
         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
@@ -23,6 +23,8 @@
 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.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 void postCreateMTMV(MTMV mtmv) {
         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 void alterJob(MTMV mtmv, boolean isReplay) {
     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 void onCommit(MTMV mtmv) throws DdlException, JobException {
             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
@@ -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
