This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 35209ae701c [fix](fe) Let MV dry runs and EXPLAIN REFRESH borrow the
session's compute group (#70125)
35209ae701c is described below
commit 35209ae701ccd255d4d22eaeac744919331d27e7
Author: Mingyu Chen (Rayner) <[email protected]>
AuthorDate: Thu Nov 5 07:28:23 2026 +0000
[fix](fe) Let MV dry runs and EXPLAIN REFRESH borrow the session's compute
group (#70125)
### What problem does this PR solve?
Issue Number: None
Related PR: #69749, #62606
Problem Summary:
**In short.** In cloud mode, `REFRESH MATERIALIZED VIEW ... INCREMENTAL
WITH DRY RUN` and `EXPLAIN REFRESH MATERIALIZED VIEW` of an MV that
declares no `compute_group` plan (and the dry run executes) in whatever
compute group the `admin` user resolves to, not in the group of the
session that issues them, where a manual `REFRESH` of the same MV runs.
This PR binds both to the session's group, the way a manual REFRESH
borrows it.
**Background.** Both statements build an internal context with the admin
identity and the MV's session variables
(`MTMVPlanUtil.createMTMVContext`). Its `cloud_cluster` is empty, so it
resolves a group the way admin would: admin's `default_compute_group`,
else the first group admin may use. A manual REFRESH passes the
session's group to its task instead (`MTMVJobManager#refreshMTMV`), and
the task binds it when the MV declares none
(`MTMVTask#applyComputeGroup`). #69749 made both statements honour a
declared group; an MV that declares none, which is every MV created
before #69749 and any MV created without the property, still goes to
admin's group. Found while reviewing #69749 (thread 4401555648).
**The problem, and what it cost.** Measured on master `856194ece0e` with
a unit test in cloud mode: groups `cg_of_admin` and `cg_of_session`, the
session on `USE @cg_of_session`, an MV that declares no group.
| | manual REFRESH | dry run | EXPLAIN REFRESH |
|---|---|---|---|
| compute group | `cg_of_session` | `cg_of_admin` | `cg_of_admin` |
- If the MV's `workload_group` exists only in the session's group (a
workload group belongs to one compute group in cloud mode), the dry run
fails with `Can not find workload group <wg> in compute croup
cg_of_admin`, while the manual REFRESH succeeds.
- Otherwise the dry run's delta query runs on the backends of another
group, which the session's user may have no USAGE on, and starts that
group if it is suspended. EXPLAIN REFRESH assigns backends there and can
start it as well. Results are not affected: the dry run only reads, and
storage is shared.
**How this PR fixes it.**
- The private lookup of the session's group in `MTMVJobManager` moves
unchanged to `MTMVPlanUtil.getSessionComputeGroup()`, so a manual
REFRESH, a dry run and EXPLAIN REFRESH resolve it with the same code.
- `RefreshMTMVCommand` builds both internal contexts through one helper.
It reads the session's group first, while the session is still the
thread-local context, then binds the declared group, else the session's
group, as `MTMVTask#applyComputeGroup` does. With neither, the context
stays unbound and the existing resolution applies; outside cloud mode
nothing changes.
- As in `MTMVTask`, the dry run re-checks only a declared group, since
the MV never chose the session's group. EXPLAIN REFRESH executes nothing
and still checks nothing.
**Results.**
| MV | session | dry run / EXPLAIN REFRESH, before | after |
|---|---|---|---|
| declares no group | `USE @cg_of_session` | `cg_of_admin` |
`cg_of_session`, as a manual REFRESH |
| declares no group | resolves no group | admin's resolution | unchanged
|
| declares `cg_declared` | `USE @cg_of_session` | `cg_declared` |
unchanged |
| any, outside cloud mode | | no compute group | unchanged |
branch-4.1 and branch-4.2 have neither statement.
---
.../java/org/apache/doris/mtmv/MTMVJobManager.java | 21 +-----
.../java/org/apache/doris/mtmv/MTMVPlanUtil.java | 23 +++++++
.../trees/plans/commands/RefreshMTMVCommand.java | 43 ++++++++-----
.../plans/commands/RefreshMTMVCommandTest.java | 75 +++++++++++++++++++++-
4 files changed, 124 insertions(+), 38 deletions(-)
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 9a23ca479a9..9b09638f818 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,8 +24,6 @@ import org.apache.doris.catalog.MTMV;
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;
@@ -45,7 +43,6 @@ 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.RefreshMTMVInfo.RefreshMode;
import org.apache.doris.nereids.trees.plans.commands.info.ResumeMTMVInfo;
-import org.apache.doris.qe.ConnectContext;
import org.apache.commons.lang3.StringUtils;
import org.apache.logging.log4j.LogManager;
@@ -159,26 +156,10 @@ public class MTMVJobManager implements MTMVHookService {
public void refreshMTMV(RefreshMTMVInfo info) throws DdlException,
MetaNotFoundException, JobException {
MTMVJob job = getJobByTableNameInfo(info.getMvName());
MTMVTaskContext mtmvTaskContext =
MTMVTaskContext.of(MTMVTaskTriggerMode.MANUAL, info.getPartitions(),
- info.getRefreshMode(), info.allowFallback(),
getCurrentComputeGroup());
+ info.getRefreshMode(), info.allowFallback(),
MTMVPlanUtil.getSessionComputeGroup());
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) {
diff --git a/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVPlanUtil.java
b/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVPlanUtil.java
index 2cbc778e735..d47ec105c4e 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVPlanUtil.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVPlanUtil.java
@@ -36,6 +36,8 @@ import org.apache.doris.catalog.TableIf;
import org.apache.doris.catalog.TableIf.TableType;
import org.apache.doris.catalog.Type;
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.ErrorCode;
import org.apache.doris.common.ErrorReport;
import org.apache.doris.common.FeConstants;
@@ -153,6 +155,27 @@ public class MTMVPlanUtil {
return ctx;
}
+ /**
+ * The compute group of the current session, which a manual refresh of an
MV that declares none
+ * borrows. Null outside cloud mode, without a session, or when the
session resolves none.
+ */
+ @Nullable
+ public static String getSessionComputeGroup() {
+ 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;
+ }
+ }
+
/**
* Execute a Nereids command in an MTMV context with optional audit
logging and an
* executor registration callback. The callback is invoked right before
the command
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/RefreshMTMVCommand.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/RefreshMTMVCommand.java
index 3a351828293..a728038ca26 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/RefreshMTMVCommand.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/RefreshMTMVCommand.java
@@ -49,6 +49,7 @@ import org.apache.doris.qe.StmtExecutor;
import org.apache.doris.resource.computegroup.ComputeGroupBindingUtil;
import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Strings;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
@@ -147,25 +148,40 @@ public class RefreshMTMVCommand extends Command
implements Forward, Explainable
}
/**
- * The internal context the dry run plans and executes its delta query on.
Like the refresh
- * contexts of MTMVTask, it is bound to the compute group the MV declares,
which is re-checked
- * because it can have been dropped since CREATE / ALTER, so the delta
runs, and its workload
- * group is looked up, in the group every refresh of that MV runs in. An
MV that declares none
- * keeps the existing resolution.
+ * The internal context the dry run plans and executes its delta query on.
It is bound to the
+ * compute group a manual REFRESH of the MV from this session runs in, so
the delta runs, and its
+ * workload group is looked up, in that group: the group the MV declares,
which is re-checked
+ * because it can have been dropped since CREATE / ALTER, or else the
group of this session.
*/
@VisibleForTesting
ConnectContext createDryRunConnectContext(MTMV mtmv) throws UserException {
- ConnectContext internalCtx = MTMVPlanUtil.createMTMVContext(
- mtmv, MTMVPlanUtil.DISABLE_RULES_WHEN_RUN_MTMV_TASK);
// Read once, so the group that is checked is the group that is bound.
Optional<String> declaredComputeGroup = mtmv.getComputeGroup();
- // A no-op outside cloud mode.
- declaredComputeGroup.ifPresent(internalCtx::setCloudCluster);
+ ConnectContext internalCtx = createRefreshContext(mtmv,
declaredComputeGroup);
+ // Like MTMVTask, only a declared group is re-checked: the MV never
chose the session's group.
ComputeGroupBindingUtil.checkComputeGroupBeforeTask(internalCtx.getCurrentUserIdentity(),
declaredComputeGroup.orElse(null));
return internalCtx;
}
+ /**
+ * An internal context (admin identity, MV session variables) bound the
way MTMVTask binds the
+ * contexts of a manual refresh: to the group the MV declares, else to the
session's group, which
+ * MTMVJobManager#refreshMTMV hands such a refresh. Left unbound when
there is neither, so the
+ * implicit resolution applies. The session's group is read first, while
this session is still
+ * the thread-local context that the internal context replaces.
+ */
+ private static ConnectContext createRefreshContext(MTMV mtmv,
Optional<String> declaredComputeGroup) {
+ String sessionComputeGroup = MTMVPlanUtil.getSessionComputeGroup();
+ ConnectContext ctx = MTMVPlanUtil.createMTMVContext(mtmv,
MTMVPlanUtil.DISABLE_RULES_WHEN_RUN_MTMV_TASK);
+ String computeGroup = declaredComputeGroup.orElse(sessionComputeGroup);
+ if (!Strings.isNullOrEmpty(computeGroup)) {
+ // A no-op outside cloud mode.
+ ctx.setCloudCluster(computeGroup);
+ }
+ return ctx;
+ }
+
@VisibleForTesting
StatementContext createDryRunStatementContext(MTMV mtmv, ConnectContext
internalCtx) {
StatementContext stmtCtx = new StatementContext(
@@ -199,12 +215,9 @@ public class RefreshMTMVCommand extends Command implements
Forward, Explainable
}
protected ConnectContext createExplainConnectContext(MTMV mtmv) {
- ConnectContext explainCtx = MTMVPlanUtil.createMTMVContext(
- mtmv, MTMVPlanUtil.DISABLE_RULES_WHEN_RUN_MTMV_TASK);
- // Plan, and pick backends, in the group the MV declares, as MTMVTask
does for its planning
- // context. Nothing is executed here, so the group is not re-checked.
- mtmv.getComputeGroup().ifPresent(explainCtx::setCloudCluster);
- return explainCtx;
+ // Plan, and pick backends, in the group a manual REFRESH from this
session runs in, as MTMVTask
+ // does for its planning context. Nothing is executed here, so the
group is not re-checked.
+ return createRefreshContext(mtmv, mtmv.getComputeGroup());
}
protected LogicalPlan createRefreshCommand(MTMV mtmv, StatementContext
statementContext) throws Exception {
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/RefreshMTMVCommandTest.java
b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/RefreshMTMVCommandTest.java
index e56e67909d3..eab6575a435 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/RefreshMTMVCommandTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/RefreshMTMVCommandTest.java
@@ -60,6 +60,9 @@ import java.util.Set;
public class RefreshMTMVCommandTest {
private static final String DECLARED_CG = "cg_declared";
+ // Listed first in enterCloudMode, so it is the group admin resolves to
when nothing is bound.
+ private static final String ADMIN_CG = "cg_of_admin";
+ private static final String SESSION_CG = "cg_of_session";
private final NereidsParser parser = new NereidsParser();
@@ -267,6 +270,16 @@ public class RefreshMTMVCommandTest {
Deencapsulation.setField(Env.getCurrentEnv(), "accessManager",
accessManager);
}
+ // The session that issues the statement, in this compute group as after
`USE @<group>`; null for a
+ // session that has not chosen one.
+ private void enterSession(String computeGroup) {
+ ConnectContext session = new ConnectContext();
+ if (computeGroup != null) {
+ session.setCloudCluster(computeGroup);
+ }
+ session.setThreadLocalInfo();
+ }
+
private MTMV newMtmvDeclaring(String declaredComputeGroup) {
MTMV mtmv = Mockito.mock(MTMV.class);
Mockito.when(mtmv.getComputeGroup()).thenReturn(Optional.ofNullable(declaredComputeGroup));
@@ -315,15 +328,71 @@ public class RefreshMTMVCommandTest {
Assertions.assertEquals(DECLARED_CG,
explainCtx.getSessionVariable().getCloudCluster());
}
- // An MV that declares nothing keeps the existing resolution: neither
context names a group, and
- // the dry run has nothing to re-check, even with no compute group left at
all.
+ // A manual REFRESH of an MV that declares no compute group runs in the
group of the session that
+ // issues it (MTMVJobManager#refreshMTMV), so its dry run runs there too.
Unbound, the internal
+ // context resolves a group the way admin, its identity, does: the first
group admin may use.
+ @Test
+ public void testDryRunWithoutDeclarationRunsInSessionComputeGroup() throws
Exception {
+ enterCloudMode(Lists.newArrayList(ADMIN_CG, SESSION_CG));
+ enterSession(SESSION_CG);
+
+ ConnectContext internalCtx =
newIncrementalCommand(true).createDryRunConnectContext(
+ newMtmvDeclaring(null));
+
+ Assertions.assertEquals(SESSION_CG,
internalCtx.getCloudCluster(false));
+ }
+
+ // EXPLAIN REFRESH plans, and picks backends, in the group a manual
REFRESH from this session runs in.
+ @Test
+ public void testExplainWithoutDeclarationPlansInSessionComputeGroup()
throws Exception {
+ enterCloudMode(Lists.newArrayList(ADMIN_CG, SESSION_CG));
+ enterSession(SESSION_CG);
+
+ ConnectContext explainCtx =
newIncrementalCommand(false).createExplainConnectContext(
+ newMtmvDeclaring(null));
+
+ Assertions.assertEquals(SESSION_CG, explainCtx.getCloudCluster(false));
+ }
+
+ // Like a refresh (MTMVTask), the dry run re-checks only a declared group:
the group it borrows from
+ // the session is not one the MV chose, so it is bound as it is, even
after it is gone.
+ @Test
+ public void testDryRunRechecksOnlyTheDeclaredComputeGroup() throws
Exception {
+ enterCloudMode(Lists.newArrayList(ADMIN_CG));
+ enterSession("cg_dropped_from_session");
+
+ ConnectContext internalCtx =
newIncrementalCommand(true).createDryRunConnectContext(
+ newMtmvDeclaring(null));
+
+ Assertions.assertEquals("cg_dropped_from_session",
internalCtx.getSessionVariable().getCloudCluster());
+ }
+
+ // As for a manual REFRESH, a declared group wins over the group of the
session.
+ @Test
+ public void testDeclaredComputeGroupWinsOverSessionComputeGroup() throws
Exception {
+ enterCloudMode(Lists.newArrayList(ADMIN_CG, SESSION_CG, DECLARED_CG));
+ MTMV mtmv = newMtmvDeclaring(DECLARED_CG);
+
+ enterSession(SESSION_CG);
+ Assertions.assertEquals(DECLARED_CG, newIncrementalCommand(true)
+
.createDryRunConnectContext(mtmv).getSessionVariable().getCloudCluster());
+ enterSession(SESSION_CG);
+ Assertions.assertEquals(DECLARED_CG, newIncrementalCommand(false)
+
.createExplainConnectContext(mtmv).getSessionVariable().getCloudCluster());
+ }
+
+ // With neither a declared group nor a group the session resolves, the
existing resolution applies:
+ // neither context names a group, and the dry run has nothing to re-check,
even with no compute
+ // group left at all.
@Test
- public void testContextsWithoutDeclarationKeepImplicitComputeGroup()
throws Exception {
+ public void
testContextsWithoutDeclarationOrSessionComputeGroupStayUnbound() throws
Exception {
enterCloudMode(Lists.newArrayList());
MTMV mtmv = newMtmvDeclaring(null);
+ enterSession(null);
Assertions.assertTrue(Strings.isNullOrEmpty(newIncrementalCommand(true)
.createDryRunConnectContext(mtmv).getSessionVariable().getCloudCluster()));
+ enterSession(null);
Assertions.assertTrue(Strings.isNullOrEmpty(newIncrementalCommand(false)
.createExplainConnectContext(mtmv).getSessionVariable().getCloudCluster()));
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
