From 8ceff3bd443985d6c72625484e768090c4c31362 Mon Sep 17 00:00:00 2001 From: seawinde Date: Fri, 15 May 2026 18:17:30 +0800 Subject: [PATCH] [feature](fe) Show compute group for MTMV refresh task (#63206) 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`. Expose ComputeGroup in tasks("type" = "mv") for asynchronous materialized view refresh tasks. --- .../doris/job/extensions/mtmv/MTMVTask.java | 31 ++++++- .../job/extensions/mtmv/MTMVTaskContext.java | 12 ++- .../org/apache/doris/mtmv/MTMVJobManager.java | 25 +++++- .../apache/doris/mtmv/MTMVJobManagerTest.java | 88 +++++++++++++++++++ .../org/apache/doris/mtmv/MTMVTaskTest.java | 88 ++++++++++++++++++- 5 files changed, 237 insertions(+), 7 deletions(-) create mode 100644 fe/fe-core/src/test/java/org/apache/doris/mtmv/MTMVJobManagerTest.java 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 40f65f0b3b7460..f015ec3d0cba8a 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.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 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 refreshPartitionNames, Map 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 entry : snapshots.entrySet()) { statementContext.setSnapshot(entry.getKey(), entry.getValue()); @@ -354,6 +361,26 @@ private void exec(Set 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 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 index 1c53f5dd301ddb..1d81a5d6a44683 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 partitions, boolean isComplete) { + public MTMVTaskContext(MTMVTaskTriggerMode triggerMode, List partitions, boolean isComplete, + String computeGroup) { this.triggerMode = triggerMode; this.partitions = partitions; this.isComplete = isComplete; + this.computeGroup = computeGroup; } public List 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 index 1edf590f310341..6804af4330fe4b 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.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.RefreshMTMVInfo; import org.apache.doris.nereids.trees.plans.commands.info.ResumeMTMVInfo; import org.apache.doris.nereids.trees.plans.commands.info.TableNameInfo; +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 new file mode 100644 index 00000000000000..0a9db00ea7dea1 --- /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.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.nereids.trees.plans.commands.info.TableNameInfo; +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 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 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 1d682c47d05c67..44b88cc506f018 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 void setUp() @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 result = task.calculateNeedRefreshPartitions(null); Assert.assertEquals(allPartitionNames, result); @@ -109,7 +117,8 @@ public void testCalculateNeedRefreshPartitionsManualComplete() throws AnalysisEx @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 result = task.calculateNeedRefreshPartitions(null); Assert.assertEquals(Lists.newArrayList(poneName), result); @@ -177,4 +186,79 @@ public void testCalculateNeedRefreshPartitionsSystemNotSyncAuto() throws Analysi List 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()); + } }