This is an automated email from the ASF dual-hosted git repository.
morrySnow 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 e056f1cddc5 [fix](mtmv) Preserve origin statement for internal refresh
(#68269)
e056f1cddc5 is described below
commit e056f1cddc5531526380a491b31e0a91a5901173
Author: morrySnow <[email protected]>
AuthorDate: Thu Sep 24 15:08:39 2026 +0800
[fix](mtmv) Preserve origin statement for internal refresh (#68269)
### What problem does this PR solve?
Problem Summary:
Internally constructed MTMV refresh commands could lose their origin
statement. Complete refreshes created a `StatementContext` without the
MV query, while `StmtExecutor` copied a null `LogicalPlanAdapter` origin
back into contexts that already had one. Complete and partition
refreshes now use the MV query as their origin and pass the same origin
to the adapter.
Incremental refresh and incremental dry run had a related parsing order
bug: an MV definition with a `SET_VAR` hint was parsed before its new
`StatementContext` was installed on the internal `ConnectContext`. Hint
parsing reads that context and could fail with a null dereference. Both
paths now install the origin-bearing context before parsing the MV
query. The existing dry-run regression case uses a hinted MV definition
to cover both refresh paths.
### Release note
Fix internal materialized view refresh planning to retain its
originating SQL statement, and allow incremental refresh and dry run of
MV queries containing `SET_VAR` hints.
---
.../apache/doris/job/extensions/mtmv/MTMVTask.java | 8 ++++---
.../java/org/apache/doris/mtmv/MTMVPlanUtil.java | 7 +++++-
.../doris/mtmv/ivm/IvmIncrRefreshManager.java | 2 ++
.../trees/plans/commands/RefreshMTMVCommand.java | 3 +++
.../java/org/apache/doris/mtmv/MTMVTaskTest.java | 5 ++++
.../doris/mtmv/ivm/IvmIncrRefreshManagerTest.java | 27 ++++++++++++++++++++++
.../plans/commands/RefreshMTMVCommandTest.java | 18 +++++++++++++++
.../commands/UpdateMvByPartitionCommandTest.java | 9 +++++++-
.../mtmv_p0/ivm/test_ivm_refresh_dry_run.groovy | 2 +-
9 files changed, 75 insertions(+), 6 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 f0445240602..0bf826434b4 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
@@ -78,6 +78,7 @@ import
org.apache.doris.nereids.trees.plans.commands.CreateMTMVCommand;
import
org.apache.doris.nereids.trees.plans.commands.UpdateMvByPartitionCommand;
import
org.apache.doris.nereids.trees.plans.commands.info.RefreshMTMVInfo.RefreshMode;
import org.apache.doris.qe.ConnectContext;
+import org.apache.doris.qe.OriginStatement;
import org.apache.doris.qe.QeProcessorImpl;
import org.apache.doris.qe.StmtExecutor;
import org.apache.doris.rpc.RpcException;
@@ -1157,10 +1158,11 @@ public class MTMVTask extends AbstractTask {
Map<TableIf, String> tableWithPartKey,
Optional<IvmRewriteContext> rewriteContext, RefreshMode
refreshMode)
throws Exception {
- // Create MTMV context first so that new StatementContext() captures
the
- // correct thread-local ConnectContext (with MTMV disabled rules,
etc.).
+ // Create the MTMV context before parsing the MV definition SQL so
SET_VAR hints
+ // resolve against the internal session (with MTMV disabled rules,
etc.).
ConnectContext mtmvCtx = MTMVPlanUtil.createMTMVContext(mtmv,
MTMVPlanUtil.DISABLE_RULES_WHEN_RUN_MTMV_TASK);
- StatementContext statementContext = new StatementContext();
+ StatementContext statementContext = new StatementContext(
+ mtmvCtx, new OriginStatement(mtmv.getQuerySql(), 0));
// Install the StatementContext on the ConnectContext before parsing
// the MV definition SQL. UpdateMvByPartitionCommand.from() calls
// NereidsParser.parseSingle() which, for SQL containing SET_VAR hints,
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 b937c93065a..e9ee4595a0f 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
@@ -172,6 +172,8 @@ public class MTMVPlanUtil {
* executing {@link StmtExecutor} through {@code executorConsumer} before
the command
* runs and clearing it (with {@code null}) after the command finishes, so
task
* cancellation can interrupt the running statement.
+ *
+ * <p>The supplied statement context must contain the originating SQL
statement.
*/
public static void executeCommand(ConnectContext ctx, Command command,
StatementContext stmtCtx, @Nullable String auditStmt,
@@ -180,7 +182,10 @@ public class MTMVPlanUtil {
ctx.getState().setNereids(true);
ctx.getSessionVariable().setEnableMaterializedViewRewrite(false);
ctx.getSessionVariable().setEnableDmlMaterializedViewRewrite(false);
- StmtExecutor executor = new StmtExecutor(ctx, new
LogicalPlanAdapter(command, stmtCtx));
+ LogicalPlanAdapter adapter = new LogicalPlanAdapter(command, stmtCtx);
+
adapter.setOrigStmt(Preconditions.checkNotNull(stmtCtx.getOriginStatement(),
+ "MTMV command origin statement must not be null"));
+ StmtExecutor executor = new StmtExecutor(ctx, adapter);
ctx.setExecutor(executor);
ctx.setQueryId(AbstractTask.generateQueryId());
if (executorConsumer != null) {
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/IvmIncrRefreshManager.java
b/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/IvmIncrRefreshManager.java
index 4559fd54937..cfbdff60d18 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/IvmIncrRefreshManager.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/IvmIncrRefreshManager.java
@@ -97,6 +97,8 @@ public class IvmIncrRefreshManager {
MTMV mtmv = context.getMtmv();
StatementContext statementContext = new StatementContext(
context.getConnectContext(), new
OriginStatement(mtmv.getQuerySql(), 0));
+ // SET_VAR hints are applied while parsing the MV query, before
executeCommand runs.
+ context.getConnectContext().setStatementContext(statementContext);
// The delta may only read the base partitions the MV's partition
definition keeps. A base
// partition outside that set, expired by partition_sync_limit, would
otherwise still be
// read through the delta and the join-opposite snapshot, and its rows
would have no MV
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 55f0e42e735..16507e80bd3 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
@@ -152,6 +152,9 @@ public class RefreshMTMVCommand extends Command implements
Forward, Explainable
stmtCtx.setIvmRewriteContext(Optional.of(IvmRewriteContext.incrementalDryRun(mtmv,
dryRunLimit)));
// Excluded trigger tables must not be validated for binlog / key-type
support.
stmtCtx.setExcludedTriggerTables(mtmv.getExcludedTriggerTables());
+ // The MV query is parsed before the internal executor is created.
SET_VAR hints
+ // need this context already installed on the internal session during
parsing.
+ internalCtx.setStatementContext(stmtCtx);
return stmtCtx;
}
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 47a0d2403e4..5a7aecfffb6 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
@@ -558,6 +558,7 @@ public class MTMVTaskTest {
Mockito.when(mtmv.getExcludedTriggerTables()).thenReturn(excludedTriggerTables);
Mockito.when(mtmv.isIvm()).thenReturn(true);
Mockito.when(mtmv.getName()).thenReturn("test_mv");
+ Mockito.when(mtmv.getQuerySql()).thenReturn("select k1 from
test_db.base_table");
Mockito.when(mtmv.getDatabase()).thenReturn(null);
Mockito.when(mtmvPartitionInfo.getPartitionType()).thenReturn(MTMVPartitionType.FOLLOW_BASE_TABLE);
@@ -583,6 +584,8 @@ public class MTMVTaskTest {
public UpdateMvByPartitionCommand
answer(InvocationOnMock invocation) {
StatementContext statementContext =
invocation.getArgument(3);
Assertions.assertEquals(excludedTriggerTables,
statementContext.getExcludedTriggerTables());
+ Assertions.assertEquals("select k1 from
test_db.base_table",
+
statementContext.getOriginStatement().originStmt);
return command;
}
});
@@ -593,6 +596,8 @@ public class MTMVTaskTest {
public Void answer(InvocationOnMock invocation) {
StatementContext statementContext =
invocation.getArgument(2);
Assertions.assertEquals(excludedTriggerTables,
statementContext.getExcludedTriggerTables());
+ Assertions.assertEquals("select k1 from
test_db.base_table",
+
statementContext.getOriginStatement().originStmt);
return null;
}
});
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/mtmv/ivm/IvmIncrRefreshManagerTest.java
b/fe/fe-core/src/test/java/org/apache/doris/mtmv/ivm/IvmIncrRefreshManagerTest.java
index 3a4a4519e12..59a57d660a0 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/mtmv/ivm/IvmIncrRefreshManagerTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/mtmv/ivm/IvmIncrRefreshManagerTest.java
@@ -90,6 +90,33 @@ public class IvmIncrRefreshManagerTest {
Assertions.assertEquals(excluded,
captured.get().getExcludedTriggerTables());
}
+ @Test
+ public void testIncrementalRefreshParsesSetVarWithItsStatementContext()
throws Exception {
+ MTMV mtmv = mockMtmv();
+ Mockito.when(mtmv.getQuerySql()).thenReturn("SELECT /*+
SET_VAR(query_timeout=10) */ 1 AS k1");
+ Mockito.when(mtmv.getInsertedColumnNames()).thenReturn(List.of("k1"));
+ ConnectContext connectContext = new ConnectContext();
+ IvmIncrRefreshContext context = new IvmIncrRefreshContext(mtmv,
connectContext, "audit",
+ queryId -> { }, null);
+ connectContext.setThreadLocalInfo();
+ try (MockedStatic<MTMVPlanUtil> mockedUtil =
Mockito.mockStatic(MTMVPlanUtil.class)) {
+ mockedUtil.when(() -> MTMVPlanUtil.executeCommand(
+ Mockito.<ConnectContext>any(), Mockito.any(),
Mockito.any(), Mockito.any(), Mockito.any()))
+ .thenAnswer(inv -> {
+ StatementContext stmtCtx = inv.getArgument(2);
+ Assertions.assertSame(stmtCtx,
connectContext.getStatementContext());
+ Assertions.assertEquals(mtmv.getQuerySql(),
stmtCtx.getOriginStatement().originStmt);
+ Assertions.assertEquals(10,
connectContext.getSessionVariable().getQueryTimeoutS());
+ return null;
+ });
+ new IvmIncrRefreshManager().executeInternalRefresh(context);
+ mockedUtil.verify(() -> MTMVPlanUtil.executeCommand(
+ Mockito.eq(connectContext), Mockito.any(), Mockito.any(),
Mockito.any(), Mockito.any()));
+ } finally {
+ ConnectContext.remove();
+ }
+ }
+
@Test
public void testManagerReturnsSuccessForEmptyBundles() throws Exception {
MTMV mtmv = mockMtmv();
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 689d7fb0dad..81becc8b5a9 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
@@ -172,6 +172,24 @@ public class RefreshMTMVCommandTest {
Assertions.assertEquals(excluded, stmtCtx.getExcludedTriggerTables());
}
+ @Test
+ public void testIncrementalDryRunParsesSetVarWithItsStatementContext()
throws Exception {
+ RefreshMTMVInfo info = extractRefreshInfo("REFRESH MATERIALIZED VIEW
db1.mv1 INCREMENTAL");
+ TestRefreshMTMVCommand command = new TestRefreshMTMVCommand(info,
true);
+ MTMV mtmv = Mockito.mock(MTMV.class);
+ Mockito.when(mtmv.getQuerySql()).thenReturn("SELECT /*+
SET_VAR(query_timeout=10) */ 1 AS k1");
+ ConnectContext internalCtx = new ConnectContext();
+ internalCtx.setThreadLocalInfo();
+ try {
+ StatementContext stmtCtx =
command.createDryRunStatementContext(mtmv, internalCtx);
+ new IvmIncrRefreshManager().buildQueryPlan(mtmv);
+ Assertions.assertSame(stmtCtx, internalCtx.getStatementContext());
+ Assertions.assertEquals(10,
internalCtx.getSessionVariable().getQueryTimeoutS());
+ } finally {
+ ConnectContext.remove();
+ }
+ }
+
@Test
public void testIncrementalExplainCarriesExcludedTriggerTables() throws
Exception {
RefreshMTMVInfo info = extractRefreshInfo("REFRESH MATERIALIZED VIEW
db1.mv1 INCREMENTAL");
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/UpdateMvByPartitionCommandTest.java
b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/UpdateMvByPartitionCommandTest.java
index 0eb933b45a4..4228865cfa4 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/UpdateMvByPartitionCommandTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/UpdateMvByPartitionCommandTest.java
@@ -46,6 +46,7 @@ import org.apache.doris.nereids.util.PlanChecker;
import org.apache.doris.planner.ExchangeNode;
import org.apache.doris.planner.OlapTableSink;
import org.apache.doris.planner.PlanFragment;
+import org.apache.doris.qe.OriginStatement;
import org.apache.doris.qe.StmtExecutor;
import org.apache.doris.thrift.TPartitionType;
import org.apache.doris.utframe.TestWithFeService;
@@ -241,6 +242,7 @@ class UpdateMvByPartitionCommandTest extends
TestWithFeService {
void testRunRefreshCommandExecutesIncrementalMtmv() throws Exception {
MTMV mtmv = getMtmv("ivm_mv");
StatementContext statementContext = createStatementCtx("refresh
materialized view test.ivm_mv");
+ OriginStatement originStatement =
statementContext.getOriginStatement();
statementContext.setIvmRewriteContext(Optional.of(IvmRewriteContext.full(mtmv)));
UpdateMvByPartitionCommand command = newRefreshCommand(mtmv);
AtomicReference<StmtExecutor> executorRef = new AtomicReference<>();
@@ -260,12 +262,15 @@ class UpdateMvByPartitionCommandTest extends
TestWithFeService {
executor.getContext().getStatementContext().getIvmRewriteContext().orElseThrow().getMode());
Assertions.assertSame(executor.getContext(),
statementContext.getConnectContext());
Assertions.assertSame(statementContext,
executor.getContext().getStatementContext());
+ Assertions.assertSame(originStatement,
statementContext.getOriginStatement());
+ Assertions.assertSame(originStatement,
executor.getParsedStmt().getOrigStmt());
}
@Test
void testExecuteCommandRebindsTaskStatementContextToExecutionContext()
throws Exception {
MTMV mtmv = getMtmv("ivm_mv");
- StatementContext statementContext = new StatementContext();
+ StatementContext statementContext = createStatementCtx("refresh
materialized view test.ivm_mv");
+ OriginStatement originStatement =
statementContext.getOriginStatement();
statementContext.setIvmRewriteContext(Optional.of(IvmRewriteContext.full(mtmv)));
UpdateMvByPartitionCommand command = UpdateMvByPartitionCommand.from(
mtmv, Sets.newHashSet(), ImmutableMap.of(), statementContext);
@@ -281,6 +286,8 @@ class UpdateMvByPartitionCommandTest extends
TestWithFeService {
Assertions.assertSame(executor.getContext(),
statementContext.getConnectContext());
Assertions.assertSame(statementContext,
executor.getContext().getStatementContext());
+ Assertions.assertSame(originStatement,
statementContext.getOriginStatement());
+ Assertions.assertSame(originStatement,
executor.getParsedStmt().getOrigStmt());
}
@Test
diff --git a/regression-test/suites/mtmv_p0/ivm/test_ivm_refresh_dry_run.groovy
b/regression-test/suites/mtmv_p0/ivm/test_ivm_refresh_dry_run.groovy
index f2583e14e8d..073d400a1b5 100644
--- a/regression-test/suites/mtmv_p0/ivm/test_ivm_refresh_dry_run.groovy
+++ b/regression-test/suites/mtmv_p0/ivm/test_ivm_refresh_dry_run.groovy
@@ -48,7 +48,7 @@ suite("test_ivm_refresh_dry_run") {
BUILD DEFERRED REFRESH INCREMENTAL ON MANUAL
DISTRIBUTED BY RANDOM BUCKETS 2
PROPERTIES ('replication_num' = '1')
- AS SELECT k1, COUNT(*) AS cnt, SUM(v1) AS sum_v1
+ AS SELECT /*+ SET_VAR(query_timeout=180) */ k1, COUNT(*) AS cnt,
SUM(v1) AS sum_v1
FROM test_ivm_refresh_dry_run_base
GROUP BY k1
"""
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
