yujun777 commented on code in PR #62606:
URL: https://github.com/apache/doris/pull/62606#discussion_r3914001551
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/CreateMTMVCommand.java:
##########
@@ -50,6 +73,148 @@ public CreateMTMVCommand(CreateMTMVInfo createMTMVInfo) {
public void run(ConnectContext ctx, StmtExecutor executor) throws
Exception {
createMTMVInfo.analyze(ctx);
Env.getCurrentEnv().createTable(this.createMTMVInfo);
Review Comment:
Fixed in 44cb471996a. `CreateMTMVCommand.run` now returns immediately when
`Env.createTable` reports an already-existing object (it returns true only for
the IF NOT EXISTS skip path), so the rollback never force-drops a pre-existing
ordinary table and never recreates an existing MTMV's streams. Covered by the
new regression `test_ivm_create_if_not_exists`: scenario 1 (name belongs to an
ordinary table -- table and data survive, no stream created) and scenario 2
(name belongs to an existing IVM MTMV -- stream id unchanged, incremental
refresh still works).
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/CreateMTMVCommand.java:
##########
@@ -50,6 +73,148 @@ public CreateMTMVCommand(CreateMTMVInfo createMTMVInfo) {
public void run(ConnectContext ctx, StmtExecutor executor) throws
Exception {
createMTMVInfo.analyze(ctx);
Env.getCurrentEnv().createTable(this.createMTMVInfo);
+ List<String> createdStreamNames = new ArrayList<>();
+ try {
+ createIvmStreams(ctx, createdStreamNames);
Review Comment:
Fixed in 44cb471996a. `createTableStream` now provisions streams under the
mv db write lock and skips recreation when an owned, usable stream already
exists, so the CREATE loop and the async IMMEDIATE task's `reconcileIvmStreams`
(which shares the same helper) can no longer drop each other's freshly created
stream or fail on a table-exists race; a stale/disabled stream still falls
through to drop+recreate. Scenario 3 of `test_ivm_create_if_not_exists`
exercises repeated create/refresh/drop cycles.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/RefreshMTMVCommand.java:
##########
@@ -17,31 +17,217 @@
package org.apache.doris.nereids.trees.plans.commands;
+import org.apache.doris.analysis.RedirectStatus;
import org.apache.doris.analysis.StmtType;
+import org.apache.doris.catalog.Database;
import org.apache.doris.catalog.Env;
+import org.apache.doris.catalog.MTMV;
+import org.apache.doris.catalog.TableIf;
+import org.apache.doris.catalog.TableIf.TableType;
+import org.apache.doris.common.AnalysisException;
+import org.apache.doris.common.MetaNotFoundException;
+import org.apache.doris.mtmv.BaseColInfo;
+import org.apache.doris.mtmv.MTMVPartitionInfo.MTMVPartitionType;
+import org.apache.doris.mtmv.MTMVPlanUtil;
+import org.apache.doris.mtmv.MTMVUtil;
+import org.apache.doris.mtmv.ivm.IvmDryRunLimit;
+import org.apache.doris.mtmv.ivm.IvmIncrRefreshManager;
+import org.apache.doris.mtmv.ivm.IvmRewriteContext;
+import org.apache.doris.nereids.NereidsPlanner;
+import org.apache.doris.nereids.StatementContext;
+import org.apache.doris.nereids.glue.LogicalPlanAdapter;
+import org.apache.doris.nereids.trees.plans.Explainable;
+import org.apache.doris.nereids.trees.plans.Plan;
import org.apache.doris.nereids.trees.plans.PlanType;
import org.apache.doris.nereids.trees.plans.commands.info.RefreshMTMVInfo;
+import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
import org.apache.doris.qe.ConnectContext;
+import org.apache.doris.qe.OriginStatement;
import org.apache.doris.qe.StmtExecutor;
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
+
+import java.util.Map;
import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
/**
* refresh mtmv
*/
-public class RefreshMTMVCommand extends Command implements ForwardWithSync {
+public class RefreshMTMVCommand extends Command implements Forward,
Explainable {
private final RefreshMTMVInfo refreshMTMVInfo;
+ // Whether EXPLAIN REFRESH should include up-to-date streams.
+ private final boolean includeExhaustedStreams;
+ // Dry run computes the delta query and streams rows back without writing
anything.
+ private final boolean dryRun;
+ // Only used when dryRun is true: optional offset/count cap for the
returned delta rows.
+ private final Optional<IvmDryRunLimit> dryRunLimit;
+ private Plan explainPlan;
+ private Optional<NereidsPlanner> explainPlanner = Optional.empty();
public RefreshMTMVCommand(RefreshMTMVInfo refreshMTMVInfo) {
+ this(refreshMTMVInfo, false, false, Optional.empty());
+ }
+
+ public RefreshMTMVCommand(RefreshMTMVInfo refreshMTMVInfo, boolean
includeExhaustedStreams) {
+ this(refreshMTMVInfo, includeExhaustedStreams, false,
Optional.empty());
+ }
+
+ public RefreshMTMVCommand(RefreshMTMVInfo refreshMTMVInfo, boolean
includeExhaustedStreams,
+ boolean dryRun, Optional<IvmDryRunLimit> dryRunLimit) {
super(PlanType.REFRESH_MTMV_COMMAND);
this.refreshMTMVInfo = Objects.requireNonNull(refreshMTMVInfo,
"require refreshMTMVInfo object");
+ this.includeExhaustedStreams = includeExhaustedStreams;
+ this.dryRun = dryRun;
+ this.dryRunLimit = Objects.requireNonNull(dryRunLimit, "require
dryRunLimit object");
}
@Override
public void run(ConnectContext ctx, StmtExecutor executor) throws
Exception {
refreshMTMVInfo.analyze(ctx);
- Env.getCurrentEnv().getMtmvService().refreshMTMV(refreshMTMVInfo);
+ if (dryRun) {
+ dryRunRefresh(ctx, executor);
+ } else {
+ Env.getCurrentEnv().getMtmvService().refreshMTMV(refreshMTMVInfo);
+ }
+ }
+
+ // Real refresh forwards to master with sync; dry run is read-only and
must run locally,
+ // streaming rows to the client instead of materializing all delta rows
into one RPC frame.
+ @Override
+ public RedirectStatus toRedirectStatus() {
+ return dryRun ? RedirectStatus.NO_FORWARD :
RedirectStatus.FORWARD_WITH_SYNC;
+ }
+
+ private void dryRunRefresh(ConnectContext ctx, StmtExecutor executor)
throws Exception {
+ MTMV mtmv = getMtmv();
+ if (!mtmv.isIvm()) {
+ throw new org.apache.doris.nereids.exceptions.AnalysisException(
+ "REFRESH MATERIALIZED VIEW ... INCREMENTAL WITH DRY RUN "
+ + "only supports IVM materialized views");
+ }
+
+ ConnectContext internalCtx = MTMVPlanUtil.createMTMVContext(
Review Comment:
Fixed in 44cb471996a. `dryRunRefresh` now restores the caller's thread-local
ConnectContext in a finally block (same save/restore pattern as
ExplainCommand), so a later statement in the same multi-statement request no
longer observes the internal ADMIN context.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/RefreshMTMVCommand.java:
##########
@@ -17,31 +17,217 @@
package org.apache.doris.nereids.trees.plans.commands;
+import org.apache.doris.analysis.RedirectStatus;
import org.apache.doris.analysis.StmtType;
+import org.apache.doris.catalog.Database;
import org.apache.doris.catalog.Env;
+import org.apache.doris.catalog.MTMV;
+import org.apache.doris.catalog.TableIf;
+import org.apache.doris.catalog.TableIf.TableType;
+import org.apache.doris.common.AnalysisException;
+import org.apache.doris.common.MetaNotFoundException;
+import org.apache.doris.mtmv.BaseColInfo;
+import org.apache.doris.mtmv.MTMVPartitionInfo.MTMVPartitionType;
+import org.apache.doris.mtmv.MTMVPlanUtil;
+import org.apache.doris.mtmv.MTMVUtil;
+import org.apache.doris.mtmv.ivm.IvmDryRunLimit;
+import org.apache.doris.mtmv.ivm.IvmIncrRefreshManager;
+import org.apache.doris.mtmv.ivm.IvmRewriteContext;
+import org.apache.doris.nereids.NereidsPlanner;
+import org.apache.doris.nereids.StatementContext;
+import org.apache.doris.nereids.glue.LogicalPlanAdapter;
+import org.apache.doris.nereids.trees.plans.Explainable;
+import org.apache.doris.nereids.trees.plans.Plan;
import org.apache.doris.nereids.trees.plans.PlanType;
import org.apache.doris.nereids.trees.plans.commands.info.RefreshMTMVInfo;
+import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
import org.apache.doris.qe.ConnectContext;
+import org.apache.doris.qe.OriginStatement;
import org.apache.doris.qe.StmtExecutor;
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
+
+import java.util.Map;
import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
/**
* refresh mtmv
*/
-public class RefreshMTMVCommand extends Command implements ForwardWithSync {
+public class RefreshMTMVCommand extends Command implements Forward,
Explainable {
private final RefreshMTMVInfo refreshMTMVInfo;
+ // Whether EXPLAIN REFRESH should include up-to-date streams.
+ private final boolean includeExhaustedStreams;
+ // Dry run computes the delta query and streams rows back without writing
anything.
+ private final boolean dryRun;
+ // Only used when dryRun is true: optional offset/count cap for the
returned delta rows.
+ private final Optional<IvmDryRunLimit> dryRunLimit;
+ private Plan explainPlan;
+ private Optional<NereidsPlanner> explainPlanner = Optional.empty();
public RefreshMTMVCommand(RefreshMTMVInfo refreshMTMVInfo) {
+ this(refreshMTMVInfo, false, false, Optional.empty());
+ }
+
+ public RefreshMTMVCommand(RefreshMTMVInfo refreshMTMVInfo, boolean
includeExhaustedStreams) {
+ this(refreshMTMVInfo, includeExhaustedStreams, false,
Optional.empty());
+ }
+
+ public RefreshMTMVCommand(RefreshMTMVInfo refreshMTMVInfo, boolean
includeExhaustedStreams,
+ boolean dryRun, Optional<IvmDryRunLimit> dryRunLimit) {
super(PlanType.REFRESH_MTMV_COMMAND);
this.refreshMTMVInfo = Objects.requireNonNull(refreshMTMVInfo,
"require refreshMTMVInfo object");
+ this.includeExhaustedStreams = includeExhaustedStreams;
+ this.dryRun = dryRun;
+ this.dryRunLimit = Objects.requireNonNull(dryRunLimit, "require
dryRunLimit object");
}
@Override
public void run(ConnectContext ctx, StmtExecutor executor) throws
Exception {
refreshMTMVInfo.analyze(ctx);
- Env.getCurrentEnv().getMtmvService().refreshMTMV(refreshMTMVInfo);
+ if (dryRun) {
+ dryRunRefresh(ctx, executor);
+ } else {
+ Env.getCurrentEnv().getMtmvService().refreshMTMV(refreshMTMVInfo);
+ }
+ }
+
+ // Real refresh forwards to master with sync; dry run is read-only and
must run locally,
+ // streaming rows to the client instead of materializing all delta rows
into one RPC frame.
+ @Override
+ public RedirectStatus toRedirectStatus() {
+ return dryRun ? RedirectStatus.NO_FORWARD :
RedirectStatus.FORWARD_WITH_SYNC;
+ }
+
+ private void dryRunRefresh(ConnectContext ctx, StmtExecutor executor)
throws Exception {
+ MTMV mtmv = getMtmv();
+ if (!mtmv.isIvm()) {
+ throw new org.apache.doris.nereids.exceptions.AnalysisException(
+ "REFRESH MATERIALIZED VIEW ... INCREMENTAL WITH DRY RUN "
+ + "only supports IVM materialized views");
+ }
+
+ ConnectContext internalCtx = MTMVPlanUtil.createMTMVContext(
+ mtmv, MTMVPlanUtil.DISABLE_RULES_WHEN_RUN_MTMV_TASK);
+ StatementContext stmtCtx = createDryRunStatementContext(mtmv,
internalCtx);
+
+ LogicalPlan queryPlan = new
IvmIncrRefreshManager().buildQueryPlan(mtmv);
+ LogicalPlanAdapter adapter = new LogicalPlanAdapter(queryPlan,
stmtCtx);
+ adapter.setOrigStmt(new OriginStatement(mtmv.getQuerySql(), 0));
+
+ // Execute on a dedicated internal executor (admin identity, MV
session variables) and
+ // stream each batch to the client's real mysql channel (see
executeAndSendResult()).
+ StmtExecutor internalExecutor = new StmtExecutor(internalCtx, adapter);
Review Comment:
Fixed in 44cb471996a. The dry-run delta executor is now registered as a
cancel delegate on the outer StmtExecutor via a new
`setCancelDelegate`/`clearCancelDelegate` hook (cleared in a finally), so
Ctrl+C / KILL QUERY / statement timeout on the outer statement are forwarded to
the internal executor's coordinator. Unit-tested by the new
`testCancelForwardsToCancelDelegate` in StmtExecutorTest.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
