[feature](fe) Show compute group for MTMV refresh task (#63206)
### What problem does this PR solve?
Problem Summary:
Async materialized view refresh task status does not show which compute
group runs the refresh task. This is needed for cloud deployments where
users inspect a single MTMV refresh task and want to know the compute
group used by that task.
Root cause: In `MTMVJobManager.refreshMTMV()`, manual refresh only
creates `MTMVTaskContext` with trigger mode, partitions, and
complete-refresh flag. The refresh execution later creates a new
background `ConnectContext` in `MTMVTask.exec()`, so the foreground
connection's selected compute group is not explicitly carried into the
task context.
| File | Change Description |
|------|-------------------|
| `MTMVJobManager.java` | Resolves the current connection's compute
group in cloud mode and stores it in `MTMVTaskContext` for manual
refresh. |
| `MTMVTaskContext.java` | Adds a serialized `computeGroup` field. |
| `MTMVTask.java` | Applies the task compute group to the background
refresh `ConnectContext`, records it, and exposes `ComputeGroup` in
`tasks("type" = "mv")`. |
| `MTMVTaskTest.java` | Covers schema, TVF output, context recording,
task context application, and old task JSON compatibility. |
| `MTMVJobManagerTest.java` | Covers passing the current compute group
from manual refresh into `MTMVTaskContext`. |
```mermaid
graph TD
A[User REFRESH MTMV] --> B[MTMVJobManager.refreshMTMV]
B --> C[Resolve current ConnectContext compute group]
C --> D[MTMVTaskContext.computeGroup]
D --> E[MTMVTask.exec creates background ConnectContext]
E --> F[Set cloud cluster on refresh ConnectContext]
F --> G[Execute refresh SQL]
F --> H[Expose ComputeGroup in tasks type mv]
```
Example:
After refreshing an async materialized view from compute group
`cg_analytics`:
```sql
REFRESH MATERIALIZED VIEW mv_sales;
```
Users can inspect the refresh task with `tasks("type" = "mv")` and read
the new `ComputeGroup` column:
```sql
SELECT
TaskId,
MvName,
Status,
RefreshMode,
LastQueryId,
ComputeGroup
FROM tasks("type" = "mv")
WHERE MvName = "mv_sales"
ORDER BY CreateTime DESC
LIMIT 1;
```
Example result:
```text
+--------+----------+---------+-------------+----------------------------------+--------------+
| TaskId | MvName | Status | RefreshMode | LastQueryId | ComputeGroup |
+--------+----------+---------+-------------+----------------------------------+--------------+
| 12003 | mv_sales | SUCCESS | COMPLETE | 7f12c8d0f2a84b2b-9d8f2f4c1a5e6b7c | cg_analytics |
+--------+----------+---------+-------------+----------------------------------+--------------+
```
If the task has no recorded compute group, `ComputeGroup` is shown as
`\N`.
### Release note
Expose ComputeGroup in tasks("type" = "mv") for asynchronous
materialized view refresh tasks.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 56a6ea5..7c9e659 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
@@ -25,6 +25,7 @@
import org.apache.doris.catalog.ScalarType;
import org.apache.doris.catalog.TableIf;
import org.apache.doris.catalog.info.TableNameInfo;
+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 @@
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 @@
MTMVTaskRefreshMode refreshMode;
@SerializedName("lastQueryId")
String lastQueryId;
+ @SerializedName("cg")
+ private String computeGroup;
private MTMV mtmv;
private MTMVRelation relation;
@@ -322,6 +327,8 @@
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 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 @@
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 1c53f5d..1d81a5d 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 @@
@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 @@
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 7bee416..67a5bbe 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
@@ -24,6 +24,8 @@
import org.apache.doris.catalog.Table;
import org.apache.doris.catalog.TableIf.TableType;
import org.apache.doris.catalog.info.TableNameInfo;
+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 @@
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 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 @@
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 0000000..b97dc6b
--- /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.catalog.info.TableNameInfo;
+import org.apache.doris.common.Config;
+import org.apache.doris.datasource.InternalCatalog;
+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 4e4b51a..505f210 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
@@ -17,15 +17,22 @@
package org.apache.doris.mtmv;
+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;
@@ -52,6 +59,7 @@
private MTMVRefreshInfo mtmvRefreshInfo = Mockito.mock(MTMVRefreshInfo.class);
private MockedStatic<MTMVUtil> mtmvUtilStatic;
private MockedStatic<MTMVPartitionUtil> mtmvPartitionUtilStatic;
+ private static final String COMPUTE_GROUP = "ComputeGroup";
@Before
public void setUp()
@@ -87,7 +95,7 @@
@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);
@@ -95,7 +103,8 @@
@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);
@@ -139,4 +148,79 @@
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());
+ }
}
