Skip to content

[fix](mtmv) Clear stale partition state for prepared statements #66370

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
seawinde wants to merge 3 commits into
base: master
Choose a base branch
from
+144 −3
Open

[fix](mtmv) Clear stale partition state for prepared statements #66370

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
e737a7c
[fix](fe) Clear stale partition state for prepared statements
seawinde Aug 3, 2026
5b4221b
[test](fe) Cover prepared statement partition state reset
seawinde Aug 3, 2026
b959429
[fix](fe) Reset MV state for prepared executions
seawinde Aug 4, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 16 additions & 0 deletions fe/fe-core/src/main/java/org/apache/doris/nereids/StatementContext.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1443,6 +1443,22 @@ public Map<BaseTableInfo, Collection<Partition>> getMvCanRewritePartitionsMap()
return mvCanRewritePartitionsMap;
}

/** Clear materialized-view planning state retained by a prepared statement between executions. */
public void resetMaterializedViewStateForPreparedExecution() {
tableUsedPartitionNameMap.clear();
commonTableIdToRelationIdToMap.clear();
mvCanRewritePartitionsMap.clear();
materializedViewRewriteDuration = 0;
hints.removeIf(UseMvHint.class::isInstance);

@github-actions github-actions Bot Aug 4, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P2] Remove retained MV hooks when rewrite is disabled

An enabled execution installs InitMaterializationContextHook.INSTANCE in this reused context. If the session then disables MV rewrite, AddInitMaterializationHook merely declines to add another hook; it does not remove the old one. containMaterializedViewHook therefore remains true, so CollectRelation still discovers MVs and lock() still acquires all retained MV-related table locks before initMaterializationContext finally rechecks the disabled flag. A writer holding any such lock can make a base-table EXECUTE wait for the one-minute planner timeout or fail even though MV rewrite is off. Remove only materialization hook instances here (preserving unrelated hooks) and clear their candidate/related-table state so the current collect pass can re-add them only when enabled.

tmpPlanForMvRewrite.clear();
rewrittenPlansByMv.clear();
needPreMvRewriteRuleMasks.clear();
needPreMvRewrite = false;
preMvRewritten = false;
materializationRewrittenSuccessSet.clear();
relationIdToStatisticsMap.clear();

@github-actions github-actions Bot Aug 4, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P1] Clear retained MV candidates before rebuilding them

A dropped MTMV can still be selected on the next execution:

EXECUTE 1: ResultSink -> Scan(base)   candidateMTMVs={mv_old}
DROP MATERIALIZED VIEW mv_old
EXECUTE 2: ResultSink -> Scan(mv_old [dropped, old table id])

DROP marks the old object isDropped and removes it from the live relation manager, but this reset leaves candidateMTMVs/mtmvRelatedTables alive. getAvailableMTMVs then consumes the retained set directly and checks rewrite status/partitions, not isDropped or the current catalog generation; AsyncMaterializationContext can consequently generate a scan from that old object and partition IDs. Clear candidateMTMVs, candidateMVs, and mtmvRelatedTables here so collection rebuilds them, and cover DROP or same-name replacement between two real EXECUTEs.

}

public void setPrepareStage(boolean isPrepare) {
this.prepareStage = isPrepare;
}
Expand Down
4 changes: 1 addition & 3 deletions ...ore/src/main/java/org/apache/doris/nereids/rules/exploration/mv/PartitionCompensator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -298,10 +298,8 @@ public static Map<List<String>, Set<String>> getQueryUsedPartitions(StatementCon
continue tableLoop;
}
// If currentUsedRelationIdSet is not empty, need check relation id to get concrete used partitions
BitSet usedPartitionRelation = new BitSet();
usedPartitionRelation.set(tableUsedPartitionPair.key().asInt());
if (!currentUsedRelationIdSet.isEmpty()
&& !currentUsedRelationIdSet.intersects(usedPartitionRelation)) {
&& !currentUsedRelationIdSet.get(tableUsedPartitionPair.key().asInt())) {
continue;
}
usedPartitionSet.addAll(tableUsedPartitionPair.value());
Expand Down
3 changes: 3 additions & 0 deletions fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/ExecuteCommand.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -91,6 +91,9 @@ public void run(ConnectContext ctx, StmtExecutor executor) throws Exception {
}
PrepareCommand prepareCommand = preparedStmtCtx.command;
StatementContext statementContext = preparedStmtCtx.getStatementContext();
// Prepared statements reuse StatementContext across executions. Discard partition and MV
// planning results collected by the previous execution before planning the current one.
statementContext.resetMaterializedViewStateForPreparedExecution();
statementContext.setPrepareStage(false);
statementContext.setIsInsert(false);
// A prepared EXECUTE reuses this one StatementContext across executions; drop the connector
Expand Down
124 changes: 124 additions & 0 deletions ...-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/ExecuteCommandTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,28 +19,40 @@

import org.apache.doris.analysis.TableScanParams;
import org.apache.doris.catalog.DatabaseIf;
import org.apache.doris.catalog.Partition;
import org.apache.doris.catalog.TableIf;
import org.apache.doris.common.Pair;
import org.apache.doris.datasource.CatalogIf;
import org.apache.doris.datasource.mvcc.MvccSnapshot;
import org.apache.doris.datasource.mvcc.MvccTable;
import org.apache.doris.mtmv.BaseTableInfo;
import org.apache.doris.nereids.NereidsPlanner;
import org.apache.doris.nereids.StatementContext;
import org.apache.doris.nereids.analyzer.UnboundRelation;
import org.apache.doris.nereids.glue.LogicalPlanAdapter;
import org.apache.doris.nereids.hint.Hint;
import org.apache.doris.nereids.hint.UseMvHint;
import org.apache.doris.nereids.parser.NereidsParser;
import org.apache.doris.nereids.rules.RuleType;
import org.apache.doris.nereids.trees.expressions.SubqueryExpr;
import org.apache.doris.nereids.trees.plans.RelationId;
import org.apache.doris.nereids.trees.plans.commands.merge.MergeIntoCommand;
import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
import org.apache.doris.nereids.util.MemoTestUtils;
import org.apache.doris.qe.ConnectContext;
import org.apache.doris.qe.OriginStatement;
import org.apache.doris.qe.PreparedStatementContext;
import org.apache.doris.qe.SessionVariable;
import org.apache.doris.qe.StmtExecutor;
import org.apache.doris.statistics.Statistics;

import com.google.common.collect.ImmutableMap;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import org.mockito.Mockito;

import java.util.Collections;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.atomic.AtomicInteger;

Expand Down Expand Up @@ -77,6 +89,101 @@ public void testResolvedScanOptionsAreResetForEveryExecute() throws Exception {
Mockito.verify(executor, Mockito.times(2)).execute();
}

@Test
public void testPartitionStateIsResetForEveryExecute() throws Exception {
String sql = "select 1";
LogicalPlan logicalPlan = new NereidsParser().parseSingle(sql);
ConnectContext connectContext = Mockito.mock(ConnectContext.class);
StatementContext statementContext = new StatementContext();
PrepareCommand prepareCommand = new PrepareCommand(
"stmt", logicalPlan, Collections.emptyList(), new OriginStatement(sql, 0));
PreparedStatementContext preparedStatement = new PreparedStatementContext(
prepareCommand, connectContext, statementContext, "stmt");
StmtExecutor executor = Mockito.mock(StmtExecutor.class);
Mockito.when(connectContext.getPreparedStementContext("stmt")).thenReturn(preparedStatement);
Mockito.when(connectContext.getSessionVariable()).thenReturn(new SessionVariable());
Mockito.when(connectContext.getStatementContext()).thenReturn(statementContext);
Mockito.when(executor.getContext()).thenReturn(connectContext);

List<String> tableQualifier = Collections.singletonList("table");
AtomicInteger relationId = new AtomicInteger();
Mockito.doAnswer(invocation -> {
int currentRelationId = relationId.getAndIncrement();
statementContext.getTableUsedPartitionNameMap().put(tableQualifier,
Pair.of(new RelationId(currentRelationId), Collections.singleton("p")));
statementContext.getCommonTableIdToRelationIdMap().put(0, currentRelationId);
return null;
}).when(executor).execute();

statementContext.getTableUsedPartitionNameMap().put(
tableQualifier, Pair.of(new RelationId(100), Collections.singleton("old")));
statementContext.getCommonTableIdToRelationIdMap().put(0, 100);

new ExecuteCommand("stmt", prepareCommand, statementContext).run(connectContext, executor);
Assertions.assertEquals(1, statementContext.getTableUsedPartitionNameMap().size());
Assertions.assertEquals(0, statementContext.getTableUsedPartitionNameMap()
.get(tableQualifier).iterator().next().key().asInt());
Assertions.assertEquals(Collections.singleton(0),
statementContext.getCommonTableIdToRelationIdMap().get(0));

new ExecuteCommand("stmt", prepareCommand, statementContext).run(connectContext, executor);
Assertions.assertEquals(1, statementContext.getTableUsedPartitionNameMap().size());
Assertions.assertEquals(1, statementContext.getTableUsedPartitionNameMap()
.get(tableQualifier).iterator().next().key().asInt());
Assertions.assertEquals(Collections.singleton(1),
statementContext.getCommonTableIdToRelationIdMap().get(0));
Mockito.verify(executor, Mockito.times(2)).execute();
}

@Test
public void testMaterializedViewStateIsResetForEveryExecute() throws Exception {
String sql = "select 1";
LogicalPlan logicalPlan = new NereidsParser().parseSingle(sql);
ConnectContext connectContext = MemoTestUtils.createConnectContext();
StatementContext statementContext = new StatementContext(
connectContext, new OriginStatement(sql, 0));
connectContext.setStatementContext(statementContext);
PrepareCommand prepareCommand = new PrepareCommand(
"stmt", logicalPlan, Collections.emptyList(), new OriginStatement(sql, 0));
PreparedStatementContext preparedStatement = new PreparedStatementContext(
prepareCommand, connectContext, statementContext, "stmt");
StmtExecutor executor = Mockito.mock(StmtExecutor.class);
connectContext.addPreparedStatementContext("stmt", preparedStatement);
Mockito.when(executor.getContext()).thenReturn(connectContext);

Hint retainedHint = new Hint("Distribute");
statementContext.addHint(retainedHint);
statementContext.setForceRecordTmpPlan(true);
AtomicInteger executionCount = new AtomicInteger();
Mockito.doAnswer(invocation -> {
Assertions.assertTrue(statementContext.getTableUsedPartitionNameMap().isEmpty());
Assertions.assertTrue(statementContext.getCommonTableIdToRelationIdMap().isEmpty());
Assertions.assertTrue(statementContext.getMvCanRewritePartitionsMap().isEmpty());
Assertions.assertEquals(0, statementContext.getMaterializedViewRewriteDuration());
Assertions.assertEquals(Collections.singletonList(retainedHint), statementContext.getHints());
Assertions.assertTrue(statementContext.getTmpPlanForMvRewrite().isEmpty());
Assertions.assertTrue(statementContext.getRewrittenPlansByMv().isEmpty());
Assertions.assertTrue(statementContext.getNeedPreMvRewriteRuleMasks().isEmpty());
Assertions.assertFalse(statementContext.isNeedPreMvRewrite());
Assertions.assertFalse(statementContext.isPreMvRewritten());
Assertions.assertTrue(statementContext.getMaterializationRewrittenSuccessSet().isEmpty());
Assertions.assertTrue(statementContext.getRelationIdToStatisticsMap().isEmpty());
Assertions.assertTrue(statementContext.isForceRecordTmpPlan());
NereidsPlanner planner = new NereidsPlanner(statementContext);
planner.plan(new LogicalPlanAdapter(logicalPlan, statementContext));
Assertions.assertNotNull(planner.getPhysicalPlan());
populateMaterializedViewState(statementContext, logicalPlan);
executionCount.incrementAndGet();
return null;
}).when(executor).execute();

populateMaterializedViewState(statementContext, logicalPlan);
new ExecuteCommand("stmt", prepareCommand, statementContext).run(connectContext, executor);
new ExecuteCommand("stmt", prepareCommand, statementContext).run(connectContext, executor);

Assertions.assertEquals(2, executionCount.get());
}

@Test
public void testResolvedScanOptionsAreResetForPreparedDeleteUsing() throws Exception {
String sql = "delete from target using source@options('scan.mode'='latest') "
Expand Down Expand Up @@ -180,6 +287,23 @@ private String resolveNextSnapshot(TableScanParams scanParams, AtomicInteger sna
.get("scan.snapshot-id");
}

private void populateMaterializedViewState(StatementContext statementContext, LogicalPlan logicalPlan) {
statementContext.getTableUsedPartitionNameMap().put(Collections.singletonList("table"),
Pair.of(new RelationId(1), Collections.singleton("partition")));
statementContext.getCommonTableIdToRelationIdMap().put(1, 1);
statementContext.getMvCanRewritePartitionsMap().put(Mockito.mock(BaseTableInfo.class),
Collections.singleton(Mockito.mock(Partition.class)));
statementContext.addMaterializedViewRewriteDuration(1);
statementContext.addHint(Mockito.mock(UseMvHint.class));
statementContext.addTmpPlanForMvRewrite(logicalPlan);
statementContext.addRewrittenPlanByMv(logicalPlan);
statementContext.ruleSetApplied(RuleType.REORDER_JOIN);
statementContext.setNeedPreMvRewrite(true);
statementContext.setPreMvRewritten(true);
statementContext.addMaterializationRewrittenSuccess(Collections.singletonList("mv"));
statementContext.addStatistics(new RelationId(1), Mockito.mock(Statistics.class));
}

private void assertPreparedCommandResetsScanOptions(
String sql, LogicalPlan command, LogicalPlan relationRoot) throws Exception {
UnboundRelation relation = relationRoot.<UnboundRelation>collectToList(
Expand Down
Loading