This is an automated email from the ASF dual-hosted git repository.
yujun777 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 55c6d5f59c4 [refactor](ivm) Route changelog and snapshot handling
through SupportChangelog (#69435)
55c6d5f59c4 is described below
commit 55c6d5f59c4a86dbe50fc77ae9dea3d9e64943f6
Author: yujun <[email protected]>
AuthorDate: Thu Oct 22 21:50:12 2026 +0800
[refactor](ivm) Route changelog and snapshot handling through
SupportChangelog (#69435)
IVM assumed every base scan was an OLAP scan: it read `hasChangelog()`
and the key model directly, and built the pre-/post-refresh snapshots
through OLAP-only methods. That blocks external-table scans
(Iceberg/Hive) from participating, and keeps the snapshot entry split
between `withPreSnapshot` and `withPostSnapshot`.
This change introduces `SupportChangelog`, a scan-level capability that
answers:
- whether the changelog is enabled for this scan (`isChangelogEnabled`),
- whether the changelog contains only appends (`isChangelogAppendOnly`),
- how to build a snapshot (`withSnapshot(boolean preSnapshot,
BaseTableStream stream)`).
`LogicalOlapScan` implements it. `LogicalOlapTableStreamScan` rejects
every method, because IVM never uses a stream scan as a base scan.
`IvmUtil.validateChangelogEnabled`, `IvmNormalizeDupKeysMTMV` and the
CREATE-time key-type judgment in `IvmNormalizeMTMV#judgeKeysType` now go
through the interface. `rewriteSnapshot` moved from `IvmDeltaRewriter`
to `IvmDeltaRewriteHelper`, and the two snapshot methods collapsed into
the single `withSnapshot` entry.
---
.../doris/mtmv/ivm/IvmDeltaRewriteHelper.java | 29 ++++++++++++++
.../apache/doris/mtmv/ivm/IvmDeltaRewriter.java | 32 ---------------
.../apache/doris/mtmv/ivm/IvmDupDeltaRewriter.java | 4 +-
.../apache/doris/mtmv/ivm/IvmJoinDeltaHandler.java | 12 +++---
.../doris/mtmv/ivm/IvmNormalizeDupKeysMTMV.java | 10 ++---
.../java/org/apache/doris/mtmv/ivm/IvmUtil.java | 2 +-
.../nereids/rules/analysis/IvmNormalizeMTMV.java | 8 ++--
.../trees/plans/logical/LogicalOlapScan.java | 31 ++++++++++-----
.../plans/logical/LogicalOlapTableStreamScan.java | 35 +++++++---------
.../trees/plans/logical/SupportChangelog.java | 46 ++++++++++++++++++++++
.../doris/mtmv/ivm/IvmDeltaRewriteHelperTest.java | 17 ++++----
.../apache/doris/mtmv/ivm/IvmDeltaTestBase.java | 21 ++++++++++
.../trees/plans/logical/LogicalOlapScanTest.java | 7 ++--
.../ivm/test_ivm_append_only_dup_mv_3.groovy | 2 +-
14 files changed, 163 insertions(+), 93 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/IvmDeltaRewriteHelper.java
b/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/IvmDeltaRewriteHelper.java
index 9aae4a5b13b..2a7d09c90a0 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/IvmDeltaRewriteHelper.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/IvmDeltaRewriteHelper.java
@@ -44,6 +44,7 @@ import
org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
import org.apache.doris.nereids.trees.plans.logical.LogicalSubQueryAlias;
import org.apache.doris.nereids.trees.plans.logical.LogicalUnion;
+import org.apache.doris.nereids.trees.plans.logical.SupportChangelog;
import org.apache.doris.nereids.types.DataType;
import com.google.common.collect.ImmutableList;
@@ -421,6 +422,34 @@ public class IvmDeltaRewriteHelper {
return new LogicalProject<>(projects.build(), union);
}
+ Pair<Plan, Map<Slot, Slot>> preSnapshot(Plan plan, IvmDeltaRewriteState
rewriteState) {
+ return rewriteSnapshot(plan, rewriteState, true);
+ }
+
+ Pair<Plan, Map<Slot, Slot>> postSnapshot(Plan plan, IvmDeltaRewriteState
rewriteState) {
+ return rewriteSnapshot(plan, rewriteState, false);
+ }
+
+ private Pair<Plan, Map<Slot, Slot>> rewriteSnapshot(Plan plan,
IvmDeltaRewriteState rewriteState,
+ boolean preSnapshot) {
+ Plan rewritten = plan.rewriteDownShortCircuit(node -> {
+ // IVM reads only changelog-capable scans as base scans. A stream
scan is such a scan only
+ // through its supertype, and an IVM query never contains one, so
leave it untouched.
+ if (!(node instanceof SupportChangelog) || node instanceof
LogicalOlapTableStreamScan) {
+ return node;
+ }
+ LogicalOlapScan scan = (LogicalOlapScan) node;
+ if (rewriteState.isExcluded(scan)) {
+ // Excluded tables produce no delta, but their snapshot side
still joins
+ // the delta side; apply the window so the property also saves
their scan.
+ return rewriteState.restrictWindow(scan);
+ }
+ scan = rewriteState.restrictWindow(scan);
+ return remapOlapScanToPlan(scan, scan.withSnapshot(preSnapshot,
rewriteState.getStream(scan)));
+ });
+ return freshPlan(rewritten);
+ }
+
/**
* Deep copy a plan before reusing it in another branch, and return the
copied output mapping.
*/
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/IvmDeltaRewriter.java
b/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/IvmDeltaRewriter.java
index 092b4635b9f..cef49bbd1ce 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/IvmDeltaRewriter.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/IvmDeltaRewriter.java
@@ -25,15 +25,12 @@ import org.apache.doris.common.Pair;
import org.apache.doris.nereids.trees.expressions.Slot;
import org.apache.doris.nereids.trees.plans.Plan;
import org.apache.doris.nereids.trees.plans.logical.LogicalEmptyRelation;
-import org.apache.doris.nereids.trees.plans.logical.LogicalOlapScan;
-import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
import org.apache.doris.qe.ConnectContext;
import com.google.common.base.Preconditions;
import java.util.List;
-import java.util.Map;
import java.util.Optional;
/**
@@ -138,33 +135,4 @@ public class IvmDeltaRewriter {
return false;
}
- static Pair<Plan, Map<Slot, Slot>> preSnapshot(Plan plan,
IvmDeltaRewriteState rewriteState) {
- return rewriteSnapshot(plan, rewriteState, true);
- }
-
- static Pair<Plan, Map<Slot, Slot>> postSnapshot(Plan plan,
IvmDeltaRewriteState rewriteState) {
- return rewriteSnapshot(plan, rewriteState, false);
- }
-
- private static Pair<Plan, Map<Slot, Slot>> rewriteSnapshot(Plan plan,
IvmDeltaRewriteState rewriteState,
- boolean preSnapshot) {
- Plan rewritten = plan.rewriteDownShortCircuit(node -> {
- if (!(node instanceof LogicalOlapScan)) {
- return node;
- }
- LogicalOlapScan scan = (LogicalOlapScan) node;
- if (rewriteState.isExcluded(scan)) {
- // Excluded tables produce no delta, but their snapshot side
still joins
- // the delta side; apply the window so the property also saves
their scan.
- return rewriteState.restrictWindow(scan);
- }
- scan = rewriteState.restrictWindow(scan);
- LogicalPlan snapshotScan = preSnapshot
- ?
scan.withPreSnapshot(Optional.of(rewriteState.getStream(scan)))
- : scan.withPostSnapshot();
- return IvmDeltaRewriteHelper.INSTANCE.remapOlapScanToPlan(scan,
snapshotScan);
- });
- return IvmDeltaRewriteHelper.INSTANCE.freshPlan(rewritten);
- }
-
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/IvmDupDeltaRewriter.java
b/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/IvmDupDeltaRewriter.java
index a3f10826086..f06eea226ef 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/IvmDupDeltaRewriter.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/IvmDupDeltaRewriter.java
@@ -165,14 +165,14 @@ class IvmDupDeltaRewriter extends
PlanVisitor<Optional<Plan>, Void> {
// The left delta joins the right as it was before the window, so
it matches the rows both
// sides already agreed on.
contributions.add(rewriteInnerJoinContribution(join, true,
leftDelta.get(),
- IvmDeltaRewriter.preSnapshot(join.right(), rewriteState)));
+ helper.preSnapshot(join.right(), rewriteState)));
}
Optional<Plan> rightDelta = join.right().accept(this, context);
if (rightDelta.isPresent()) {
// The right delta joins the left as it is after the window, which
is the side carrying the
// rows the left appended in the same window.
contributions.add(rewriteInnerJoinContribution(join, false,
rightDelta.get(),
- IvmDeltaRewriter.postSnapshot(join.left(), rewriteState)));
+ helper.postSnapshot(join.left(), rewriteState)));
}
return helper.combineDeltaPlans(contributions, join.getOutput());
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/IvmJoinDeltaHandler.java
b/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/IvmJoinDeltaHandler.java
index 3b6219f0c4a..e32ca6da718 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/IvmJoinDeltaHandler.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/IvmJoinDeltaHandler.java
@@ -118,18 +118,18 @@ class IvmJoinDeltaHandler {
if (join.getJoinType().isOuterJoin()) {
SideInput leftInput = deltaOnLeft
? sideInput(join.left(), delta.plan)
- : sideInput(join.left(),
IvmDeltaRewriter.postSnapshot(join.left(), rewriteState));
+ : sideInput(join.left(), helper.postSnapshot(join.left(),
rewriteState));
SideInput rightInput = deltaOnLeft
- ? sideInput(join.right(),
IvmDeltaRewriter.preSnapshot(join.right(), rewriteState))
+ ? sideInput(join.right(), helper.preSnapshot(join.right(),
rewriteState))
: sideInput(join.right(), delta.plan);
return rewriteOuterContribution(join, deltaOnLeft, delta,
leftInput, rightInput, rewriteState);
} else if (deltaOnLeft) {
- Pair<Plan, Map<Slot, Slot>> rightPreSnapshot =
IvmDeltaRewriter.preSnapshot(join.right(), rewriteState);
+ Pair<Plan, Map<Slot, Slot>> rightPreSnapshot =
helper.preSnapshot(join.right(), rewriteState);
return rewriteInnerJoinContribution(join, delta.plan,
helper.mapBusinessOutputs(join.left().getOutput(),
delta.plan.getOutput()),
rightPreSnapshot.first, rightPreSnapshot.second, delta);
} else {
- Pair<Plan, Map<Slot, Slot>> leftPostSnapshot =
IvmDeltaRewriter.postSnapshot(join.left(), rewriteState);
+ Pair<Plan, Map<Slot, Slot>> leftPostSnapshot =
helper.postSnapshot(join.left(), rewriteState);
return rewriteInnerJoinContribution(join, leftPostSnapshot.first,
helper.mapBusinessOutputs(join.left().getOutput(),
leftPostSnapshot.first.getOutput()), delta.plan,
helper.mapBusinessOutputs(join.right().getOutput(),
delta.plan.getOutput()), delta);
@@ -146,8 +146,8 @@ class IvmJoinDeltaHandler {
}
Plan deltaSideChild = deltaOnLeft ? join.left() : join.right();
return rewriteNullSideDelta(new NullSideDeltaContext(join,
deltaOnLeft, delta, leftInput, rightInput,
- IvmDeltaRewriter.preSnapshot(deltaSideChild, rewriteState),
- IvmDeltaRewriter.postSnapshot(deltaSideChild, rewriteState),
+ helper.preSnapshot(deltaSideChild, rewriteState),
+ helper.postSnapshot(deltaSideChild, rewriteState),
rewriteState, delta.maxDeltaIndex));
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/IvmNormalizeDupKeysMTMV.java
b/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/IvmNormalizeDupKeysMTMV.java
index 781880a07b8..f16e8fe1a34 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/IvmNormalizeDupKeysMTMV.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/IvmNormalizeDupKeysMTMV.java
@@ -17,8 +17,6 @@
package org.apache.doris.mtmv.ivm;
-import org.apache.doris.catalog.KeysType;
-import org.apache.doris.catalog.OlapTable;
import org.apache.doris.catalog.info.TableNameInfo;
import org.apache.doris.nereids.exceptions.AnalysisException;
import org.apache.doris.nereids.trees.plans.Plan;
@@ -77,15 +75,15 @@ public class IvmNormalizeDupKeysMTMV extends
DefaultPlanRewriter<Set<TableNameIn
@Override
public Plan visitLogicalOlapScan(LogicalOlapScan scan, Set<TableNameInfo>
excludedTriggerTables) {
- OlapTable table = scan.getTable();
- if (table.getKeysType() != KeysType.DUP_KEYS) {
+ if (!scan.isChangelogAppendOnly()) {
// An AnalysisException rather than an IvmException: a refresh
turns every IvmException into a
// fallback, which would express "the premise of this MV is
broken" as "try a complete refresh",
// even though that refresh runs this same normalize and fails the
same way. The premise being
// broken is not a retry.
throw new AnalysisException(
- "IVM detail MV error: table " + table.getName() + " is " +
table.getKeysType()
- + ", but a detail MV can only be built over
duplicate-key detail tables");
+ "IVM detail MV error: table " + scan.getTable().getName()
+ + " does not provide an append-only changelog, but
a detail MV can only be "
+ + "built over scans whose changelog contains only
appends");
}
IvmUtil.validateChangelogEnabled(excludedTriggerTables, scan);
return scan;
diff --git a/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/IvmUtil.java
b/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/IvmUtil.java
index 5ff1d9357b5..575fd2cd0a4 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/IvmUtil.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/IvmUtil.java
@@ -382,7 +382,7 @@ public class IvmUtil {
return;
}
OlapTable table = scan.getTable();
- if (!table.hasChangelog()) {
+ if (!scan.isChangelogEnabled()) {
throw new IvmException(IvmFailureReason.CHANGELOG_NOT_ENABLED,
"SQL can be incrementally refreshed, but changelog is not
enabled for table: "
+ table.getName()
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/IvmNormalizeMTMV.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/IvmNormalizeMTMV.java
index 8c0aac09430..fe194e0b89c 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/IvmNormalizeMTMV.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/IvmNormalizeMTMV.java
@@ -29,8 +29,8 @@ import org.apache.doris.nereids.jobs.JobContext;
import org.apache.doris.nereids.trees.plans.Plan;
import org.apache.doris.nereids.trees.plans.logical.LogicalAggregate;
import org.apache.doris.nereids.trees.plans.logical.LogicalJoin;
-import org.apache.doris.nereids.trees.plans.logical.LogicalOlapScan;
import org.apache.doris.nereids.trees.plans.logical.LogicalRepeat;
+import org.apache.doris.nereids.trees.plans.logical.SupportChangelog;
import org.apache.doris.nereids.trees.plans.visitor.CustomRewriter;
import com.google.common.base.Preconditions;
@@ -151,9 +151,9 @@ public class IvmNormalizeMTMV implements CustomRewriter {
if (needsRowIdMaintenance) {
return KeysType.UNIQUE_KEYS;
}
- List<LogicalOlapScan> scans =
plan.collectToList(LogicalOlapScan.class::isInstance);
- for (LogicalOlapScan scan : scans) {
- if (scan.getTable().getKeysType() != KeysType.DUP_KEYS) {
+ List<SupportChangelog> scans =
plan.collectToList(SupportChangelog.class::isInstance);
+ for (SupportChangelog scan : scans) {
+ if (!scan.isChangelogAppendOnly()) {
return KeysType.UNIQUE_KEYS;
}
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalOlapScan.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalOlapScan.java
index 3008fccdc61..97f446d0519 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalOlapScan.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalOlapScan.java
@@ -23,6 +23,7 @@ import org.apache.doris.catalog.KeysType;
import org.apache.doris.catalog.MTMV;
import org.apache.doris.catalog.OlapTable;
import org.apache.doris.catalog.Table;
+import org.apache.doris.catalog.stream.BaseTableStream;
import org.apache.doris.catalog.stream.OlapTableStream;
import org.apache.doris.catalog.stream.OlapTableStreamWrapper;
import org.apache.doris.catalog.stream.StreamReadMode;
@@ -84,7 +85,8 @@ import java.util.stream.Collectors;
/**
* Logical OlapScan.
*/
-public class LogicalOlapScan extends LogicalCatalogRelation implements
OlapScan, SupportPruneNestedColumn {
+public class LogicalOlapScan extends LogicalCatalogRelation
+ implements OlapScan, SupportPruneNestedColumn, SupportChangelog {
private static final Logger LOG =
LogManager.getLogger(LogicalOlapScan.class);
@@ -961,10 +963,25 @@ public class LogicalOlapScan extends
LogicalCatalogRelation implements OlapScan,
return scoreRangeInfo;
}
- /** Build a pre-refresh snapshot as a stream scan in snapshot mode. */
- public LogicalPlan withPreSnapshot(Optional<OlapTableStream> stream) {
+ @Override
+ public boolean isChangelogEnabled() {
+ return getTable().hasChangelog();
+ }
+
+ @Override
+ public boolean isChangelogAppendOnly() {
+ return getTable().getKeysType() == KeysType.DUP_KEYS;
+ }
+
+ @Override
+ public LogicalPlan withSnapshot(boolean preSnapshot, BaseTableStream
stream) {
+ if (!preSnapshot) {
+ return LogicalPlanDeepCopier.INSTANCE.deepCopy(
+ (LogicalPlan) this, new DeepCopierContext());
+ }
+ OlapTableStream olapStream = (OlapTableStream) stream;
OlapTableStreamWrapper streamWrapper = new OlapTableStreamWrapper(
- stream.get(), getTable(), selectedPartitionIds);
+ olapStream, getTable(), selectedPartitionIds);
return new LogicalOlapTableStreamScan(
StatementScopeIdGenerator.newRelationId(),
streamWrapper,
@@ -977,12 +994,6 @@ public class LogicalOlapScan extends
LogicalCatalogRelation implements OlapScan,
).withReadMode(StreamReadMode.SNAPSHOT);
}
- /** Build a post-refresh snapshot as a regular olap scan. */
- public LogicalPlan withPostSnapshot() {
- return LogicalPlanDeepCopier.INSTANCE.deepCopy(
- (LogicalPlan) this, new DeepCopierContext());
- }
-
protected List<SlotReference> createSlotsVectorized(List<Column> columns) {
List<String> qualified = qualified();
SlotReference[] slots = new SlotReference[columns.size()];
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalOlapTableStreamScan.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalOlapTableStreamScan.java
index 6ab3587740c..b0535ea2020 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalOlapTableStreamScan.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalOlapTableStreamScan.java
@@ -23,14 +23,14 @@ import org.apache.doris.catalog.KeysType;
import org.apache.doris.catalog.OlapTable;
import org.apache.doris.catalog.Table;
import org.apache.doris.catalog.constraint.TableIdentifier;
-import org.apache.doris.catalog.stream.OlapTableStream;
+import org.apache.doris.catalog.stream.BaseTableStream;
import org.apache.doris.catalog.stream.OlapTableStreamWrapper;
import org.apache.doris.catalog.stream.StreamReadMode;
import org.apache.doris.common.IdGenerator;
+import org.apache.doris.nereids.exceptions.AnalysisException;
import org.apache.doris.nereids.memo.GroupExpression;
import org.apache.doris.nereids.properties.LogicalProperties;
import org.apache.doris.nereids.properties.OrderKey;
-import org.apache.doris.nereids.rules.analysis.BindRelation;
import org.apache.doris.nereids.trees.TableSample;
import org.apache.doris.nereids.trees.expressions.ExprId;
import org.apache.doris.nereids.trees.expressions.NamedExpression;
@@ -547,27 +547,22 @@ public class LogicalOlapTableStreamScan extends
LogicalOlapScan {
}
@Override
- public LogicalPlan withPreSnapshot(Optional<OlapTableStream> stream) {
- return withReadMode(StreamReadMode.SNAPSHOT);
+ public boolean isChangelogEnabled() {
+ throw unsupportedChangelogScan();
}
@Override
- public LogicalPlan withPostSnapshot() {
- OlapTable baseTable = getTable().getBaseTable();
- LogicalOlapScan scan = new LogicalOlapScan(
- StatementScopeIdGenerator.newRelationId(),
- baseTable,
- qualifier,
- ImmutableList.of(),
- baseTable.getPartitionIds(),
- baseTable.getBaseIndexId(),
- PreAggStatus.unset(),
- ImmutableList.of(),
- ImmutableList.of(),
- Optional.empty(),
- ImmutableList.of());
- return BindRelation.checkAndAddDeleteSignFilter(
- scan, ConnectContext.get(), baseTable);
+ public boolean isChangelogAppendOnly() {
+ throw unsupportedChangelogScan();
+ }
+
+ @Override
+ public LogicalPlan withSnapshot(boolean preSnapshot, BaseTableStream
stream) {
+ throw unsupportedChangelogScan();
+ }
+
+ private static AnalysisException unsupportedChangelogScan() {
+ return new AnalysisException("IVM does not support
LogicalOlapTableStreamScan as a base scan");
}
@Override
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/SupportChangelog.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/SupportChangelog.java
new file mode 100644
index 00000000000..f048687348b
--- /dev/null
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/SupportChangelog.java
@@ -0,0 +1,46 @@
+// 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.nereids.trees.plans.logical;
+
+import org.apache.doris.catalog.stream.BaseTableStream;
+
+/**
+ * A scan that can expose a changelog and answer its shape.
+ *
+ * <p>Implementations answer whether the changelog is enabled and whether the
changelog this scan
+ * reads contains only appends, and build the pre- and post-refresh snapshots
an incremental
+ * refresh reads. It says nothing about the plan a refresh runs: an {@link
LogicalOlapTableStreamScan}
+ * inherits this interface only through {@link LogicalOlapScan}, and it
rejects every method because
+ * a stream scan is never a base scan of an IVM query.
+ */
+public interface SupportChangelog {
+ /** Whether the changelog this scan reads is enabled on its source. */
+ boolean isChangelogEnabled();
+
+ /**
+ * True when the changelog this scan reads contains only appends: no
DELETE or UPDATE_BEFORE,
+ * so every change carries dml_factor = +1.
+ */
+ boolean isChangelogAppendOnly();
+
+ /**
+ * Build a snapshot of this scan from {@code stream}: the pre-refresh
snapshot when
+ * {@code preSnapshot} is true, the post-refresh snapshot otherwise.
+ */
+ LogicalPlan withSnapshot(boolean preSnapshot, BaseTableStream stream);
+}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/mtmv/ivm/IvmDeltaRewriteHelperTest.java
b/fe/fe-core/src/test/java/org/apache/doris/mtmv/ivm/IvmDeltaRewriteHelperTest.java
index 876a451b4ed..c661dfbb5d6 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/mtmv/ivm/IvmDeltaRewriteHelperTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/mtmv/ivm/IvmDeltaRewriteHelperTest.java
@@ -29,6 +29,7 @@ import org.apache.doris.catalog.ScalarType;
import org.apache.doris.catalog.Type;
import org.apache.doris.catalog.stream.OlapTableStream;
import org.apache.doris.catalog.stream.OlapTableStreamWrapper;
+import org.apache.doris.catalog.stream.StreamReadMode;
import org.apache.doris.nereids.exceptions.AnalysisException;
import org.apache.doris.nereids.trees.expressions.Alias;
import org.apache.doris.nereids.trees.expressions.Expression;
@@ -70,7 +71,7 @@ class IvmDeltaRewriteHelperTest extends IvmDeltaTestBase {
@Test
void testIsIncrementalDeltaScan_streamScanNotIncremental_returnsFalse() {
- LogicalOlapTableStreamScan scan = (LogicalOlapTableStreamScan)
buildDeltaScan().withPreSnapshot(Optional.empty());
+ LogicalOlapTableStreamScan scan =
buildDeltaScan().withReadMode(StreamReadMode.SNAPSHOT);
Assertions.assertFalse(helper.isIncrementalDeltaScan(scan));
}
@@ -161,7 +162,7 @@ class IvmDeltaRewriteHelperTest extends IvmDeltaTestBase {
.getDbOrAnalysisException("test_db")
.getTableOrAnalysisException(IvmUtil.streamName(0L,
scan.getTable().getFullQualifiers()));
- LogicalPlan preSnapshot = (LogicalPlan)
scan.withPreSnapshot(Optional.of(stream));
+ LogicalPlan preSnapshot = scan.withSnapshot(true, stream);
LogicalPlan remapped = helper.remapOlapScanToPlan(scan,
preSnapshot);
Assertions.assertInstanceOf(LogicalProject.class, remapped);
@@ -183,7 +184,7 @@ class IvmDeltaRewriteHelperTest extends IvmDeltaTestBase {
void testRemapScanOutputForPostSnapshotPreservesExprId() {
LogicalOlapScan scan = buildScanForTable(2, "t_post");
- LogicalPlan postSnapshot = (LogicalPlan) scan.withPostSnapshot();
+ LogicalPlan postSnapshot = scan.withSnapshot(false,
getRegisteredStream(scan.getTable(), 0L));
LogicalPlan remapped = helper.remapOlapScanToPlan(scan, postSnapshot);
Assertions.assertInstanceOf(LogicalProject.class, remapped);
@@ -211,7 +212,7 @@ class IvmDeltaRewriteHelperTest extends IvmDeltaTestBase {
LogicalProject<LogicalOlapTableStreamScan> oldProject = new
LogicalProject<>(
ImmutableList.of(aliasId, passthroughName), deltaScan);
- LogicalPlan newChild = (LogicalPlan)
deltaScan.withPreSnapshot(Optional.empty());
+ LogicalPlan newChild = deltaScan.withReadMode(StreamReadMode.SNAPSHOT);
LogicalProject<?> remapped = helper.remapStreamScanToPlan(oldProject,
newChild);
Assertions.assertInstanceOf(LogicalOlapTableStreamScan.class,
remapped.child());
@@ -229,7 +230,7 @@ class IvmDeltaRewriteHelperTest extends IvmDeltaTestBase {
LogicalProject<LogicalOlapTableStreamScan> oldProject = new
LogicalProject<>(
ImmutableList.of(aliasId, passthroughName), deltaScan);
- LogicalPlan newChild = (LogicalPlan) deltaScan.withPostSnapshot();
+ LogicalPlan newChild = buildPostSnapshotScan(deltaScan);
LogicalProject<?> remapped = helper.remapStreamScanToPlan(oldProject,
newChild);
Assertions.assertInstanceOf(LogicalOlapScan.class, remapped.child());
@@ -245,7 +246,7 @@ class IvmDeltaRewriteHelperTest extends IvmDeltaTestBase {
LogicalProject<LogicalOlapTableStreamScan> oldProject = new
LogicalProject<>(
ImmutableList.of(missingStreamColumn), deltaScan);
- LogicalPlan newChild = (LogicalPlan) deltaScan.withPostSnapshot();
+ LogicalPlan newChild = buildPostSnapshotScan(deltaScan);
LogicalProject<?> remapped = helper.remapStreamScanToPlan(oldProject,
newChild);
Assertions.assertInstanceOf(Alias.class,
remapped.getProjects().get(0));
Assertions.assertInstanceOf(NullLiteral.class,
@@ -259,7 +260,7 @@ class IvmDeltaRewriteHelperTest extends IvmDeltaTestBase {
LogicalProject<LogicalOlapTableStreamScan> oldProject = new
LogicalProject<>(
ImmutableList.of(hiddenAlias), deltaScan);
- LogicalPlan newChild = (LogicalPlan) deltaScan.withPostSnapshot();
+ LogicalPlan newChild = buildPostSnapshotScan(deltaScan);
LogicalProject<?> remapped = helper.remapStreamScanToPlan(oldProject,
newChild);
Assertions.assertInstanceOf(Alias.class,
remapped.getProjects().get(0));
Assertions.assertInstanceOf(NullLiteral.class, ((Alias)
remapped.getProjects().get(0)).child());
@@ -286,7 +287,7 @@ class IvmDeltaRewriteHelperTest extends IvmDeltaTestBase {
Assertions.assertEquals(new BigIntLiteral(0L),
findAliasChild(streamProject, Column.SEQUENCE_COL));
- LogicalPlan snapshotPlan = (LogicalPlan)
streamScan.withPostSnapshot();
+ LogicalPlan snapshotPlan = buildPostSnapshotScan(streamScan);
LogicalProject<?> remapped =
helper.remapStreamScanToPlan(streamProject, snapshotPlan);
Assertions.assertEquals(helper.findSlotByName(snapshotPlan.getOutput(),
Column.DELETE_SIGN),
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/mtmv/ivm/IvmDeltaTestBase.java
b/fe/fe-core/src/test/java/org/apache/doris/mtmv/ivm/IvmDeltaTestBase.java
index e0c7b676c25..7ac4e359b7a 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/mtmv/ivm/IvmDeltaTestBase.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/mtmv/ivm/IvmDeltaTestBase.java
@@ -53,6 +53,7 @@ import org.apache.doris.nereids.StatementContext;
import org.apache.doris.nereids.analyzer.UnboundTableSink;
import org.apache.doris.nereids.jobs.JobContext;
import org.apache.doris.nereids.properties.PhysicalProperties;
+import org.apache.doris.nereids.rules.analysis.BindRelation;
import org.apache.doris.nereids.trees.expressions.Add;
import org.apache.doris.nereids.trees.expressions.Alias;
import org.apache.doris.nereids.trees.expressions.Expression;
@@ -65,11 +66,13 @@ import
org.apache.doris.nereids.trees.expressions.functions.agg.Max;
import org.apache.doris.nereids.trees.expressions.functions.agg.Min;
import org.apache.doris.nereids.trees.expressions.functions.agg.Sum;
import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.PreAggStatus;
import
org.apache.doris.nereids.trees.plans.commands.insert.InsertIntoTableCommand;
import org.apache.doris.nereids.trees.plans.logical.LogicalAggregate;
import org.apache.doris.nereids.trees.plans.logical.LogicalFilter;
import org.apache.doris.nereids.trees.plans.logical.LogicalOlapScan;
import org.apache.doris.nereids.trees.plans.logical.LogicalOlapTableStreamScan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
import org.apache.doris.nereids.trees.plans.logical.LogicalResultSink;
import org.apache.doris.nereids.util.MemoTestUtils;
@@ -145,6 +148,24 @@ abstract class IvmDeltaTestBase {
ImmutableList.of("test_db"));
}
+ /** Builds the regular post-refresh snapshot scan of a delta scan's base
table. */
+ protected LogicalPlan buildPostSnapshotScan(LogicalOlapTableStreamScan
streamScan) {
+ OlapTable baseTable = streamScan.getTable().getBaseTable();
+ LogicalOlapScan scan = new LogicalOlapScan(
+ PlanConstructor.getNextRelationId(),
+ baseTable,
+ ImmutableList.of("test_db"),
+ ImmutableList.of(),
+ baseTable.getPartitionIds(),
+ baseTable.getBaseIndexId(),
+ PreAggStatus.unset(),
+ ImmutableList.of(),
+ ImmutableList.of(),
+ Optional.empty(),
+ ImmutableList.of());
+ return BindRelation.checkAndAddDeleteSignFilter(scan,
ConnectContext.get(), baseTable);
+ }
+
private OlapTableStreamWrapper buildStreamWrapper(OlapTable baseTable) {
OlapTableStream stream = registerTestStream(baseTable, 1L);
return new OlapTableStreamWrapper(stream, baseTable,
ImmutableList.of());
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/logical/LogicalOlapScanTest.java
b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/logical/LogicalOlapScanTest.java
index 0195f136968..9959dc0dd55 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/logical/LogicalOlapScanTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/logical/LogicalOlapScanTest.java
@@ -226,7 +226,7 @@ public class LogicalOlapScanTest {
LogicalOlapScan scan = createMockScan(ImmutableList.of());
OlapTableStream stream = Mockito.mock(OlapTableStream.class);
- LogicalPlan snapshot = scan.withPreSnapshot(Optional.of(stream));
+ LogicalPlan snapshot = scan.withSnapshot(true, stream);
Assertions.assertInstanceOf(LogicalOlapTableStreamScan.class,
snapshot);
LogicalOlapTableStreamScan streamScan = (LogicalOlapTableStreamScan)
snapshot;
Assertions.assertTrue(streamScan.isSnapshot());
@@ -237,7 +237,8 @@ public class LogicalOlapScanTest {
@Test
public void testWithPostSnapshotReturnsRegularScan() {
LogicalOlapScan scan1 = createMockScan(ImmutableList.of());
- LogicalPlan snapshot = scan1.withPostSnapshot();
+ OlapTableStream stream = Mockito.mock(OlapTableStream.class);
+ LogicalPlan snapshot = scan1.withSnapshot(false, stream);
Assertions.assertInstanceOf(LogicalOlapScan.class, snapshot);
Assertions.assertFalse(snapshot instanceof LogicalOlapTableStreamScan);
@@ -255,7 +256,7 @@ public class LogicalOlapScanTest {
OlapTableStream stream = Mockito.mock(OlapTableStream.class);
LogicalOlapTableStreamScan streamScan = (LogicalOlapTableStreamScan)
scan
- .withPreSnapshot(Optional.of(stream));
+ .withSnapshot(true, stream);
streamScan = streamScan.withSelectedPartitionIds(ImmutableList.of(),
true);
LogicalOlapTableStreamScan copiedStreamScan =
streamScan.withSelectedPartitionIds(ImmutableList.of());
diff --git
a/regression-test/suites/mtmv_p0/ivm/test_ivm_append_only_dup_mv_3.groovy
b/regression-test/suites/mtmv_p0/ivm/test_ivm_append_only_dup_mv_3.groovy
index 88c31ad7651..cd94249647e 100644
--- a/regression-test/suites/mtmv_p0/ivm/test_ivm_append_only_dup_mv_3.groovy
+++ b/regression-test/suites/mtmv_p0/ivm/test_ivm_append_only_dup_mv_3.groovy
@@ -311,7 +311,7 @@ suite("test_ivm_append_only_dup_mv_3") {
test {
sql """REFRESH MATERIALIZED VIEW dup_mv_premise INCREMENTAL WITH DRY
RUN"""
- exception "can only be built over duplicate-key detail tables"
+ exception "does not provide an append-only changelog"
}
// A rebuild would normalize the same way and fail the same way, so the MV
is left in the state that
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
