From bc8c6e5776b32c78f070e92079d2193d3ecae18b Mon Sep 17 00:00:00 2001 From: yujun Date: Wed, 8 Jul 2026 18:57:50 +0800 Subject: [PATCH 1/8] [fix](nereids) Preserve LogicalOlapTableStreamScan type in deep copy and normalize after CTE inline Two fixes for LogicalOlapTableStreamScan handling in Nereids planner: 1. Override withVirtualColumns to return LogicalOlapTableStreamScan instead of LogicalOlapScan, preventing type downgrade during LogicalPlanDeepCopier.deepCopy used by CTEInline. 2. Move NormalizeOlapTableStreamScan execution to after CTE inline so StreamScans exposed by CTE expansion get normalized. Previously it ran inside notTraverseChildrenOf(CTEAnchor) which blocked traversal into CTE producers. This fixes CTE+stream hidden column queries and IVM incremental refresh failures caused by un-normalized stream scans. --- .../doris/nereids/jobs/executor/Rewriter.java | 15 +++++++------- .../logical/LogicalOlapTableStreamScan.java | 20 +++++++++++++++++++ 2 files changed, 27 insertions(+), 8 deletions(-) diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/jobs/executor/Rewriter.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/jobs/executor/Rewriter.java index 4728abaf805181..247245bf86b6ad 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/jobs/executor/Rewriter.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/jobs/executor/Rewriter.java @@ -892,14 +892,6 @@ private static List getWholeTreeRewriteJobs( ImmutableSet.of(LogicalCTEAnchor.class), () -> { List rewriteJobs = Lists.newArrayListWithExpectedSize(300); - if (Config.enable_table_stream) { - rewriteJobs.addAll(jobs( - topic("normalize olap table stream scan", - topDown(new NormalizeOlapTableStreamScan()) - ) - ) - ); - } rewriteJobs.addAll(jobs( topic("cte inline and pull up all cte anchor", custom(RuleType.PULL_UP_CTE_ANCHOR, PullUpCteAnchor::new), @@ -967,6 +959,13 @@ private static List getWholeTreeRewriteJobs( return rewriteJobs; } )); + if (Config.enable_table_stream) { + builder.addAll(jobs( + topic("normalize olap table stream scan after cte inline", + topDown(new NormalizeOlapTableStreamScan()) + ) + )); + } return builder.build(); } 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 f1e20a0dae522a..4608e441524ab7 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 @@ -41,6 +41,7 @@ import org.apache.doris.nereids.util.Utils; import com.google.common.collect.ImmutableList; +import com.google.common.collect.Lists; import com.google.common.collect.Maps; import org.apache.commons.lang3.tuple.Pair; @@ -50,6 +51,7 @@ import java.util.Objects; import java.util.Optional; import java.util.Set; +import java.util.stream.Collectors; /** * Logical OlapTableStreamScan @@ -291,6 +293,24 @@ public LogicalOlapTableStreamScan withPartitionPrunablePredicates( partitionPrunablePredicates, scanParams, isReset, isSnapshot)); } + @Override + public LogicalOlapTableStreamScan withVirtualColumns(List virtualColumns) { + LogicalProperties logicalProperties = getLogicalProperties(); + List output = Lists.newArrayList(logicalProperties.getOutput()); + output.addAll(virtualColumns.stream().map(NamedExpression::toSlot).collect(Collectors.toList())); + LogicalProperties finalLogicalProperties = new LogicalProperties(() -> output, this::computeDataTrait); + return AbstractPlan.copyWithSameId(this, () -> + new LogicalOlapTableStreamScan(relationId, (Table) table, qualifier, + groupExpression, Optional.of(finalLogicalProperties), + selectedPartitionIds, partitionPruned, hasPartitionPredicate, selectedTabletIds, + selectedIndexId, indexSelected, preAggStatus, manuallySpecifiedPartitions, + hints, cacheSlotWithSlotName, cachedOutput, tableSample, directMvScan, + colToSubPathsMap, manuallySpecifiedTabletIds, operativeSlots, virtualColumns, + scoreOrderKeys, scoreLimit, scoreRangeInfo, annOrderKeys, annLimit, tableAlias, + partitionPrunablePredicates, scanParams, + isReset, isSnapshot)); + } + /** withTableScanParams */ @Override public LogicalOlapTableStreamScan withTableScanParams(TableScanParams scanParams) { From 9bdec4c6a5148112f85ea3cc6f7490a5d67d8c6a Mon Sep 17 00:00:00 2001 From: yujun Date: Fri, 10 Jul 2026 18:31:24 +0800 Subject: [PATCH 2/8] [fix](fe) Move stream scan normalization into CTE before-derive pipeline Issue Number: close #65339 Related PR: Problem Summary: Normalize stream scan rules were executed in a late CTE-related rewrite topic, which could be skipped in trees with CTE-related traversal constraints, causing delta/rewrite paths to miss synthetic change_type/version project aliases and fail when buildDmlFactorExpr accessed hidden stream columns. This moves into the early table/physical optimization topic after partition pruning and removes the separate late topic from , and updates regression baseline for to the fixed incremental result behavior. None - Test: Regression test / Unit Test / Manual test / No need to test (with reason) - test_ivm_basic_mtmv (baseline .out updated) - Behavior changed: Yes (CTE stream-scan normalization now runs in the proper rewrite phase) - Does this need documentation: No --- .../apache/doris/nereids/jobs/executor/Rewriter.java | 11 +++-------- 1 file changed, 3 insertions(+), 8 deletions(-) diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/jobs/executor/Rewriter.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/jobs/executor/Rewriter.java index 247245bf86b6ad..ab4da685778584 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/jobs/executor/Rewriter.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/jobs/executor/Rewriter.java @@ -17,7 +17,6 @@ package org.apache.doris.nereids.jobs.executor; -import org.apache.doris.common.Config; import org.apache.doris.nereids.CascadesContext; import org.apache.doris.nereids.jobs.JobContext; import org.apache.doris.nereids.jobs.rewrite.CostBasedRewriteJob; @@ -728,6 +727,9 @@ public class Rewriter extends AbstractBatchJobExecutor { new LogicalResultSinkToShortCircuitPointQuery(), new PruneOlapScanPartition(), new PruneEmptyPartition(), + // Stream lowering needs the pruned partitions and must finish before + // OperativeColumnDerive treats stream virtual columns as scan slots. + new NormalizeOlapTableStreamScan(), new PruneFileScanPartition(), new PushDownFilterIntoSchemaScan(), new PruneOlapScanTablet() @@ -959,13 +961,6 @@ private static List getWholeTreeRewriteJobs( return rewriteJobs; } )); - if (Config.enable_table_stream) { - builder.addAll(jobs( - topic("normalize olap table stream scan after cte inline", - topDown(new NormalizeOlapTableStreamScan()) - ) - )); - } return builder.build(); } From f42fcb3da7dfef912bd993a1147632fa00f74a26 Mon Sep 17 00:00:00 2001 From: yujun Date: Fri, 10 Jul 2026 18:42:13 +0800 Subject: [PATCH 3/8] add is incremental --- .../src/main/java/org/apache/doris/A.java | 436 ++++++++++++++++++ .../logical/LogicalOlapTableStreamScan.java | 6 +- 2 files changed, 441 insertions(+), 1 deletion(-) create mode 100644 fe/fe-core/src/main/java/org/apache/doris/A.java diff --git a/fe/fe-core/src/main/java/org/apache/doris/A.java b/fe/fe-core/src/main/java/org/apache/doris/A.java new file mode 100644 index 00000000000000..1dfd6eccab2265 --- /dev/null +++ b/fe/fe-core/src/main/java/org/apache/doris/A.java @@ -0,0 +1,436 @@ +// 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.analysis.TableScanParams; +import org.apache.doris.catalog.Column; +import org.apache.doris.catalog.OlapTable; +import org.apache.doris.catalog.Table; +import org.apache.doris.catalog.stream.OlapTableStream; +import org.apache.doris.catalog.stream.OlapTableStreamWrapper; +import org.apache.doris.common.IdGenerator; +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.trees.TableSample; +import org.apache.doris.nereids.trees.expressions.ExprId; +import org.apache.doris.nereids.trees.expressions.NamedExpression; +import org.apache.doris.nereids.trees.expressions.Slot; +import org.apache.doris.nereids.trees.expressions.SlotReference; +import org.apache.doris.nereids.trees.expressions.StatementScopeIdGenerator; +import org.apache.doris.nereids.trees.plans.AbstractPlan; +import org.apache.doris.nereids.trees.plans.PartitionPrunablePredicate; +import org.apache.doris.nereids.trees.plans.PreAggStatus; +import org.apache.doris.nereids.trees.plans.RelationId; +import org.apache.doris.nereids.trees.plans.ScoreRangeInfo; +import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor; +import org.apache.doris.nereids.util.Utils; +import org.apache.doris.qe.ConnectContext; + +import com.google.common.collect.ImmutableList; +import com.google.common.collect.Lists; +import com.google.common.collect.Maps; +import org.apache.commons.lang3.tuple.Pair; + +import java.util.Collection; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Optional; +import java.util.Set; +import java.util.stream.Collectors; + +/** + * Logical OlapTableStreamScan + */ +public class LogicalOlapTableStreamScan extends LogicalOlapScan { + private final boolean isReset; + private final boolean isSnapshot; + + /** + * LogicalOlapTableStreamScan construct method + */ + public LogicalOlapTableStreamScan(RelationId id, OlapTable table, List qualifier, List tabletIds, + List hints, Optional tableSample, Collection operativeSlots) { + super(id, table, qualifier, tabletIds, hints, tableSample, operativeSlots); + this.isReset = false; + this.isSnapshot = false; + } + + /** + * LogicalOlapTableStreamScan construct method + */ + public LogicalOlapTableStreamScan(RelationId id, OlapTable table, List qualifier, + List specifiedPartitions, List tabletIds, List hints, + Optional tableSample, List operativeSlots) { + super(id, table, qualifier, specifiedPartitions, tabletIds, hints, tableSample, operativeSlots); + this.isReset = false; + this.isSnapshot = false; + } + + /** + * LogicalOlapTableStreamScan construct method + */ + public LogicalOlapTableStreamScan(RelationId id, Table table, List qualifier, + Optional groupExpression, + Optional logicalProperties, + List selectedPartitionIds, boolean partitionPruned, + boolean hasPartitionPredicate, + List selectedTabletIds, long selectedIndexId, boolean indexSelected, + PreAggStatus preAggStatus, List specifiedPartitions, + List hints, Map, Slot> cacheSlotWithSlotName, + Optional> cachedOutput, Optional tableSample, + boolean directMvScan, + Map>> colToSubPathsMap, List specifiedTabletIds, + Collection operativeSlots, List virtualColumns, + List scoreOrderKeys, Optional scoreLimit, + Optional scoreRangeInfo, + List annOrderKeys, Optional annLimit, String tableAlias, + Optional partitionPrunablePredicates, + Optional scanParams, + boolean isReset, boolean isSnapshot) { + super(id, table, qualifier, groupExpression, logicalProperties, + selectedPartitionIds, partitionPruned, hasPartitionPredicate, selectedTabletIds, selectedIndexId, + indexSelected, preAggStatus, specifiedPartitions, hints, cacheSlotWithSlotName, cachedOutput, + tableSample, directMvScan, colToSubPathsMap, specifiedTabletIds, operativeSlots, virtualColumns, + scoreOrderKeys, scoreLimit, scoreRangeInfo, annOrderKeys, annLimit, tableAlias, + partitionPrunablePredicates, scanParams); + this.isReset = isReset; + this.isSnapshot = isSnapshot; + } + + @Override + public LogicalOlapTableStreamScan withManuallySpecifiedTabletIds(List manuallySpecifiedTabletIds) { + return AbstractPlan.copyWithSameId(this, () -> + new LogicalOlapTableStreamScan(relationId, (Table) table, qualifier, + Optional.empty(), Optional.of(getLogicalProperties()), + selectedPartitionIds, partitionPruned, hasPartitionPredicate, selectedTabletIds, + selectedIndexId, indexSelected, preAggStatus, manuallySpecifiedPartitions, + hints, cacheSlotWithSlotName, cachedOutput, tableSample, directMvScan, + colToSubPathsMap, manuallySpecifiedTabletIds, operativeSlots, virtualColumns, + scoreOrderKeys, scoreLimit, scoreRangeInfo, annOrderKeys, annLimit, tableAlias, + partitionPrunablePredicates, scanParams, + isReset, isSnapshot)); + } + + @Override + public List computeOutput() { + if (cachedOutput.isPresent()) { + return cachedOutput.get(); + } + List baseSchema = table.getBaseSchema(true); + List slotFromColumn = createSlotsVectorized(baseSchema); + + ConnectContext connectContext = ConnectContext.get(); + boolean ivmRewriteEnabled = connectContext != null + && connectContext.getStatementContext() != null + && connectContext.getStatementContext().getIvmRewriteContext().isPresent(); + + ImmutableList.Builder slots = ImmutableList.builder(); + IdGenerator exprIdGenerator = StatementScopeIdGenerator.getExprIdGenerator(); + for (int i = 0; i < baseSchema.size(); i++) { + // skip binlog before column + final int index = i; + Column col = baseSchema.get(i); + if (col.getName().startsWith(Column.BINLOG_BEFORE_PREFIX)) { + continue; + } + // Keep visible columns, and IVM row-id during IVM rewrite. Skip all other hidden columns. + if (!col.isVisible() && !(Column.IVM_ROW_ID_COL.equals(col.getName()) && ivmRewriteEnabled)) { + continue; + } + Pair key = Pair.of(selectedIndexId, col.getName()); + Slot slot = cacheSlotWithSlotName.computeIfAbsent(key, k -> slotFromColumn.get(index)); + slots.add(slot); + if (colToSubPathsMap.containsKey(key.getValue())) { + for (List subPath : colToSubPathsMap.get(key.getValue())) { + if (!subPath.isEmpty()) { + SlotReference slotReference = SlotReference.fromColumn( + exprIdGenerator.getNextId(), table, baseSchema.get(i), qualified() + ).withSubPath(subPath); + slots.add(slotReference); + subPathToSlotMap.computeIfAbsent(slot, k -> Maps.newHashMap()) + .put(subPath, slotReference); + } + } + } + } + if (!isSnapshot && !isReset) { + // add stream exclusive virtual columns. + slots.add(SlotReference.fromColumn( + exprIdGenerator.getNextId(), table, Column.STREAM_SEQ_VIRTUAL_COLUMN, qualified())); + slots.add(SlotReference.fromColumn( + exprIdGenerator.getNextId(), table, Column.STREAM_CHANGE_TYPE_VIRTUAL_COLUMN, qualified())); + } + for (NamedExpression virtualColumn : virtualColumns) { + slots.add(virtualColumn.toSlot()); + } + return slots.build(); + } + + /** + * withSelectedTabletIds + */ + @Override + public LogicalOlapTableStreamScan withSelectedTabletIds(List selectedTabletIds) { + return AbstractPlan.copyWithSameId(this, () -> + new LogicalOlapTableStreamScan(relationId, (Table) table, qualifier, + Optional.empty(), Optional.of(getLogicalProperties()), + selectedPartitionIds, partitionPruned, hasPartitionPredicate, selectedTabletIds, + selectedIndexId, indexSelected, preAggStatus, manuallySpecifiedPartitions, + hints, cacheSlotWithSlotName, cachedOutput, tableSample, directMvScan, colToSubPathsMap, + manuallySpecifiedTabletIds, operativeSlots, virtualColumns, scoreOrderKeys, scoreLimit, + scoreRangeInfo, annOrderKeys, annLimit, tableAlias, partitionPrunablePredicates, + scanParams, isReset, isSnapshot)); + } + + /** withCachedOutput */ + @Override + public LogicalOlapTableStreamScan withCachedOutput(List outputSlots) { + return AbstractPlan.copyWithSameId(this, () -> + new LogicalOlapTableStreamScan(relationId, (Table) table, qualifier, + groupExpression, Optional.empty(), + selectedPartitionIds, partitionPruned, hasPartitionPredicate, selectedTabletIds, + selectedIndexId, indexSelected, preAggStatus, manuallySpecifiedPartitions, hints, + cacheSlotWithSlotName, Optional.of(outputSlots), tableSample, directMvScan, colToSubPathsMap, + manuallySpecifiedTabletIds, operativeSlots, virtualColumns, scoreOrderKeys, scoreLimit, + scoreRangeInfo, annOrderKeys, annLimit, tableAlias, partitionPrunablePredicates, + scanParams, isReset, isSnapshot)); + } + + @Override + public LogicalOlapTableStreamScan withOperativeSlots(Collection operativeSlots) { + return AbstractPlan.copyWithSameId(this, () -> + new LogicalOlapTableStreamScan(relationId, (Table) table, qualifier, + groupExpression, Optional.of(getLogicalProperties()), + selectedPartitionIds, partitionPruned, hasPartitionPredicate, selectedTabletIds, + selectedIndexId, indexSelected, preAggStatus, manuallySpecifiedPartitions, + hints, cacheSlotWithSlotName, cachedOutput, tableSample, directMvScan, colToSubPathsMap, + manuallySpecifiedTabletIds, operativeSlots, virtualColumns, scoreOrderKeys, scoreLimit, + scoreRangeInfo, annOrderKeys, annLimit, tableAlias, partitionPrunablePredicates, + scanParams, isReset, isSnapshot)); + } + + /** + * withPreAggStatus + */ + @Override + public LogicalOlapTableStreamScan withPreAggStatus(PreAggStatus preAggStatus) { + return AbstractPlan.copyWithSameId(this, () -> + new LogicalOlapTableStreamScan(relationId, (Table) table, qualifier, + Optional.empty(), Optional.of(getLogicalProperties()), + selectedPartitionIds, partitionPruned, hasPartitionPredicate, selectedTabletIds, + selectedIndexId, indexSelected, preAggStatus, manuallySpecifiedPartitions, + hints, cacheSlotWithSlotName, cachedOutput, tableSample, directMvScan, + colToSubPathsMap, manuallySpecifiedTabletIds, operativeSlots, virtualColumns, + scoreOrderKeys, scoreLimit, scoreRangeInfo, annOrderKeys, annLimit, tableAlias, + partitionPrunablePredicates, scanParams, isReset, isSnapshot)); + } + + /** + * withGroupExpression + */ + @Override + public LogicalOlapTableStreamScan withGroupExpression(Optional groupExpression) { + return AbstractPlan.copyWithSameId(this, () -> + new LogicalOlapTableStreamScan(relationId, (Table) table, qualifier, + groupExpression, Optional.of(getLogicalProperties()), + selectedPartitionIds, partitionPruned, hasPartitionPredicate, selectedTabletIds, + selectedIndexId, indexSelected, preAggStatus, manuallySpecifiedPartitions, + hints, cacheSlotWithSlotName, cachedOutput, tableSample, directMvScan, + colToSubPathsMap, manuallySpecifiedTabletIds, operativeSlots, virtualColumns, + scoreOrderKeys, scoreLimit, scoreRangeInfo, annOrderKeys, annLimit, tableAlias, + partitionPrunablePredicates, scanParams, isReset, isSnapshot)); + } + + /** + * withSelectedPartitionIds + */ + @Override + public LogicalOlapTableStreamScan withSelectedPartitionIds(List selectedPartitionIdsd) { + return withSelectedPartitionIds(selectedPartitionIdsd, false); + } + + /** + * withSelectedPartitionIds + */ + @Override + public LogicalOlapTableStreamScan withSelectedPartitionIds(List selectedPartitionIds, + boolean isPartitionPruned) { + return AbstractPlan.copyWithSameId(this, () -> + new LogicalOlapTableStreamScan(relationId, (Table) table, qualifier, + groupExpression, Optional.of(getLogicalProperties()), + selectedPartitionIds, isPartitionPruned, hasPartitionPredicate, selectedTabletIds, + selectedIndexId, indexSelected, preAggStatus, manuallySpecifiedPartitions, + hints, cacheSlotWithSlotName, cachedOutput, tableSample, directMvScan, + colToSubPathsMap, manuallySpecifiedTabletIds, operativeSlots, virtualColumns, + scoreOrderKeys, scoreLimit, scoreRangeInfo, annOrderKeys, annLimit, tableAlias, + partitionPrunablePredicates, scanParams, isReset, isSnapshot)); + } + + /** + * Returns a new {@code LogicalOlapScan} carrying the supplied + * {@link PartitionPrunablePredicate}. It is preserved across all other + * {@code with*} builders so partition-derived conjuncts can be removed + * safely after MV rewrite has had a chance to match the plan. + */ + @Override + public LogicalOlapTableStreamScan withPartitionPrunablePredicates( + Optional partitionPrunablePredicates) { + return AbstractPlan.copyWithSameId(this, () -> + new LogicalOlapTableStreamScan(relationId, (Table) table, qualifier, + groupExpression, Optional.of(getLogicalProperties()), + selectedPartitionIds, partitionPruned, hasPartitionPredicate, selectedTabletIds, + selectedIndexId, indexSelected, preAggStatus, manuallySpecifiedPartitions, + hints, cacheSlotWithSlotName, cachedOutput, tableSample, directMvScan, + colToSubPathsMap, manuallySpecifiedTabletIds, operativeSlots, virtualColumns, + scoreOrderKeys, scoreLimit, scoreRangeInfo, annOrderKeys, annLimit, tableAlias, + partitionPrunablePredicates, scanParams, isReset, isSnapshot)); + } + + @Override + public LogicalOlapTableStreamScan withVirtualColumns(List virtualColumns) { + LogicalProperties logicalProperties = getLogicalProperties(); + List output = Lists.newArrayList(logicalProperties.getOutput()); + output.addAll(virtualColumns.stream().map(NamedExpression::toSlot).collect(Collectors.toList())); + LogicalProperties finalLogicalProperties = new LogicalProperties(() -> output, this::computeDataTrait); + return AbstractPlan.copyWithSameId(this, () -> + new LogicalOlapTableStreamScan(relationId, (Table) table, qualifier, + groupExpression, Optional.of(finalLogicalProperties), + selectedPartitionIds, partitionPruned, hasPartitionPredicate, selectedTabletIds, + selectedIndexId, indexSelected, preAggStatus, manuallySpecifiedPartitions, + hints, cacheSlotWithSlotName, cachedOutput, tableSample, directMvScan, + colToSubPathsMap, manuallySpecifiedTabletIds, operativeSlots, virtualColumns, + scoreOrderKeys, scoreLimit, scoreRangeInfo, annOrderKeys, annLimit, tableAlias, + partitionPrunablePredicates, scanParams, + isReset, isSnapshot)); + } + + /** withTableScanParams */ + @Override + public LogicalOlapTableStreamScan withTableScanParams(TableScanParams scanParams) { + return AbstractPlan.copyWithSameId(this, () -> + new LogicalOlapTableStreamScan(relationId, (Table) table, qualifier, + groupExpression, Optional.of(getLogicalProperties()), + selectedPartitionIds, partitionPruned, hasPartitionPredicate, selectedTabletIds, + selectedIndexId, indexSelected, preAggStatus, manuallySpecifiedPartitions, + hints, cacheSlotWithSlotName, cachedOutput, tableSample, directMvScan, colToSubPathsMap, + manuallySpecifiedTabletIds, operativeSlots, virtualColumns, scoreOrderKeys, scoreLimit, + scoreRangeInfo, annOrderKeys, annLimit, tableAlias, partitionPrunablePredicates, + Optional.of(scanParams), isReset, isSnapshot)); + } + + /** + * withRelationId + */ + @Override + public LogicalOlapTableStreamScan withRelationId(RelationId relationId) { + return AbstractPlan.copyWithSameId(this, () -> + new LogicalOlapTableStreamScan(relationId, (Table) table, qualifier, + groupExpression, Optional.empty(), + selectedPartitionIds, partitionPruned, hasPartitionPredicate, selectedTabletIds, + selectedIndexId, indexSelected, preAggStatus, manuallySpecifiedPartitions, + hints, cacheSlotWithSlotName, cachedOutput, tableSample, directMvScan, colToSubPathsMap, + manuallySpecifiedTabletIds, operativeSlots, virtualColumns, scoreOrderKeys, scoreLimit, + scoreRangeInfo, annOrderKeys, annLimit, tableAlias, partitionPrunablePredicates, + scanParams, isReset, isSnapshot)); + } + + /** + * withIsSnapshot + */ + public LogicalOlapTableStreamScan withIsSnapshot(boolean isSnapshot) { + return AbstractPlan.copyWithSameId(this, () -> + new LogicalOlapTableStreamScan(relationId, (Table) table, qualifier, + groupExpression, Optional.empty(), + selectedPartitionIds, partitionPruned, hasPartitionPredicate, selectedTabletIds, + selectedIndexId, indexSelected, preAggStatus, manuallySpecifiedPartitions, + hints, cacheSlotWithSlotName, cachedOutput, tableSample, directMvScan, colToSubPathsMap, + manuallySpecifiedTabletIds, operativeSlots, virtualColumns, scoreOrderKeys, scoreLimit, + scoreRangeInfo, annOrderKeys, annLimit, tableAlias, partitionPrunablePredicates, + scanParams, isReset, isSnapshot)); + } + + /** + * withIsReset + */ + public LogicalOlapTableStreamScan withIsReset(boolean isReset) { + return AbstractPlan.copyWithSameId(this, () -> + new LogicalOlapTableStreamScan(relationId, (Table) table, qualifier, + groupExpression, Optional.empty(), + selectedPartitionIds, partitionPruned, hasPartitionPredicate, selectedTabletIds, + selectedIndexId, indexSelected, preAggStatus, manuallySpecifiedPartitions, + hints, cacheSlotWithSlotName, cachedOutput, tableSample, directMvScan, colToSubPathsMap, + manuallySpecifiedTabletIds, operativeSlots, virtualColumns, scoreOrderKeys, scoreLimit, + scoreRangeInfo, annOrderKeys, annLimit, tableAlias, partitionPrunablePredicates, + scanParams, isReset, isSnapshot)); + } + + @Override + public R accept(PlanVisitor visitor, C context) { + return visitor.visitLogicalOlapTableStreamScan(this, context); + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } + if (!super.equals(o)) { + return false; + } + LogicalOlapTableStreamScan that = (LogicalOlapTableStreamScan) o; + return Objects.equals(isReset, that.isReset) + && Objects.equals(isSnapshot, that.isSnapshot); + } + + @Override + public OlapTableStreamWrapper getTable() { + return (OlapTableStreamWrapper) super.getTable(); + } + + @Override + public String toString() { + return Utils.toSqlStringSkipNull("LogicalOlapTableStreamScan[" + id.asInt() + "]", + "qualified", qualifiedName(), + "alias", tableAlias, + "indexName", getSelectedMaterializedIndexName().orElse(""), + "selectedIndexId", selectedIndexId, + "preAgg", preAggStatus, + "operativeCol", operativeSlots, + "stats", statistics, + "virtualColumns", virtualColumns, + "isSnapshot", isSnapshot, + "isReset", isReset); + } + + public boolean isSnapshot() { + return isSnapshot; + } + + public boolean isReset() { + return isReset; + } + + public boolean isIncremental() { + return !isSnapshot && !isReset; + } +} 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 4608e441524ab7..4d1a52326661f4 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 @@ -160,7 +160,7 @@ public List computeOutput() { } } } - if (!isSnapshot && !isReset) { + if (isIncremental()) { // add stream exclusive virtual columns. slots.add(SlotReference.fromColumn( exprIdGenerator.getNextId(), table, Column.STREAM_SEQ_VIRTUAL_COLUMN, qualified())); @@ -419,4 +419,8 @@ public boolean isSnapshot() { public boolean isReset() { return isReset; } + + public boolean isIncremental() { + return !isSnapshot && !isReset; + } } From 44b0ee46ffa45649efeda3d1c3d432a88330d382 Mon Sep 17 00:00:00 2001 From: yujun Date: Fri, 10 Jul 2026 18:47:15 +0800 Subject: [PATCH 4/8] drop temp file --- .../src/main/java/org/apache/doris/A.java | 436 ------------------ 1 file changed, 436 deletions(-) delete mode 100644 fe/fe-core/src/main/java/org/apache/doris/A.java diff --git a/fe/fe-core/src/main/java/org/apache/doris/A.java b/fe/fe-core/src/main/java/org/apache/doris/A.java deleted file mode 100644 index 1dfd6eccab2265..00000000000000 --- a/fe/fe-core/src/main/java/org/apache/doris/A.java +++ /dev/null @@ -1,436 +0,0 @@ -// 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.analysis.TableScanParams; -import org.apache.doris.catalog.Column; -import org.apache.doris.catalog.OlapTable; -import org.apache.doris.catalog.Table; -import org.apache.doris.catalog.stream.OlapTableStream; -import org.apache.doris.catalog.stream.OlapTableStreamWrapper; -import org.apache.doris.common.IdGenerator; -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.trees.TableSample; -import org.apache.doris.nereids.trees.expressions.ExprId; -import org.apache.doris.nereids.trees.expressions.NamedExpression; -import org.apache.doris.nereids.trees.expressions.Slot; -import org.apache.doris.nereids.trees.expressions.SlotReference; -import org.apache.doris.nereids.trees.expressions.StatementScopeIdGenerator; -import org.apache.doris.nereids.trees.plans.AbstractPlan; -import org.apache.doris.nereids.trees.plans.PartitionPrunablePredicate; -import org.apache.doris.nereids.trees.plans.PreAggStatus; -import org.apache.doris.nereids.trees.plans.RelationId; -import org.apache.doris.nereids.trees.plans.ScoreRangeInfo; -import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor; -import org.apache.doris.nereids.util.Utils; -import org.apache.doris.qe.ConnectContext; - -import com.google.common.collect.ImmutableList; -import com.google.common.collect.Lists; -import com.google.common.collect.Maps; -import org.apache.commons.lang3.tuple.Pair; - -import java.util.Collection; -import java.util.List; -import java.util.Map; -import java.util.Objects; -import java.util.Optional; -import java.util.Set; -import java.util.stream.Collectors; - -/** - * Logical OlapTableStreamScan - */ -public class LogicalOlapTableStreamScan extends LogicalOlapScan { - private final boolean isReset; - private final boolean isSnapshot; - - /** - * LogicalOlapTableStreamScan construct method - */ - public LogicalOlapTableStreamScan(RelationId id, OlapTable table, List qualifier, List tabletIds, - List hints, Optional tableSample, Collection operativeSlots) { - super(id, table, qualifier, tabletIds, hints, tableSample, operativeSlots); - this.isReset = false; - this.isSnapshot = false; - } - - /** - * LogicalOlapTableStreamScan construct method - */ - public LogicalOlapTableStreamScan(RelationId id, OlapTable table, List qualifier, - List specifiedPartitions, List tabletIds, List hints, - Optional tableSample, List operativeSlots) { - super(id, table, qualifier, specifiedPartitions, tabletIds, hints, tableSample, operativeSlots); - this.isReset = false; - this.isSnapshot = false; - } - - /** - * LogicalOlapTableStreamScan construct method - */ - public LogicalOlapTableStreamScan(RelationId id, Table table, List qualifier, - Optional groupExpression, - Optional logicalProperties, - List selectedPartitionIds, boolean partitionPruned, - boolean hasPartitionPredicate, - List selectedTabletIds, long selectedIndexId, boolean indexSelected, - PreAggStatus preAggStatus, List specifiedPartitions, - List hints, Map, Slot> cacheSlotWithSlotName, - Optional> cachedOutput, Optional tableSample, - boolean directMvScan, - Map>> colToSubPathsMap, List specifiedTabletIds, - Collection operativeSlots, List virtualColumns, - List scoreOrderKeys, Optional scoreLimit, - Optional scoreRangeInfo, - List annOrderKeys, Optional annLimit, String tableAlias, - Optional partitionPrunablePredicates, - Optional scanParams, - boolean isReset, boolean isSnapshot) { - super(id, table, qualifier, groupExpression, logicalProperties, - selectedPartitionIds, partitionPruned, hasPartitionPredicate, selectedTabletIds, selectedIndexId, - indexSelected, preAggStatus, specifiedPartitions, hints, cacheSlotWithSlotName, cachedOutput, - tableSample, directMvScan, colToSubPathsMap, specifiedTabletIds, operativeSlots, virtualColumns, - scoreOrderKeys, scoreLimit, scoreRangeInfo, annOrderKeys, annLimit, tableAlias, - partitionPrunablePredicates, scanParams); - this.isReset = isReset; - this.isSnapshot = isSnapshot; - } - - @Override - public LogicalOlapTableStreamScan withManuallySpecifiedTabletIds(List manuallySpecifiedTabletIds) { - return AbstractPlan.copyWithSameId(this, () -> - new LogicalOlapTableStreamScan(relationId, (Table) table, qualifier, - Optional.empty(), Optional.of(getLogicalProperties()), - selectedPartitionIds, partitionPruned, hasPartitionPredicate, selectedTabletIds, - selectedIndexId, indexSelected, preAggStatus, manuallySpecifiedPartitions, - hints, cacheSlotWithSlotName, cachedOutput, tableSample, directMvScan, - colToSubPathsMap, manuallySpecifiedTabletIds, operativeSlots, virtualColumns, - scoreOrderKeys, scoreLimit, scoreRangeInfo, annOrderKeys, annLimit, tableAlias, - partitionPrunablePredicates, scanParams, - isReset, isSnapshot)); - } - - @Override - public List computeOutput() { - if (cachedOutput.isPresent()) { - return cachedOutput.get(); - } - List baseSchema = table.getBaseSchema(true); - List slotFromColumn = createSlotsVectorized(baseSchema); - - ConnectContext connectContext = ConnectContext.get(); - boolean ivmRewriteEnabled = connectContext != null - && connectContext.getStatementContext() != null - && connectContext.getStatementContext().getIvmRewriteContext().isPresent(); - - ImmutableList.Builder slots = ImmutableList.builder(); - IdGenerator exprIdGenerator = StatementScopeIdGenerator.getExprIdGenerator(); - for (int i = 0; i < baseSchema.size(); i++) { - // skip binlog before column - final int index = i; - Column col = baseSchema.get(i); - if (col.getName().startsWith(Column.BINLOG_BEFORE_PREFIX)) { - continue; - } - // Keep visible columns, and IVM row-id during IVM rewrite. Skip all other hidden columns. - if (!col.isVisible() && !(Column.IVM_ROW_ID_COL.equals(col.getName()) && ivmRewriteEnabled)) { - continue; - } - Pair key = Pair.of(selectedIndexId, col.getName()); - Slot slot = cacheSlotWithSlotName.computeIfAbsent(key, k -> slotFromColumn.get(index)); - slots.add(slot); - if (colToSubPathsMap.containsKey(key.getValue())) { - for (List subPath : colToSubPathsMap.get(key.getValue())) { - if (!subPath.isEmpty()) { - SlotReference slotReference = SlotReference.fromColumn( - exprIdGenerator.getNextId(), table, baseSchema.get(i), qualified() - ).withSubPath(subPath); - slots.add(slotReference); - subPathToSlotMap.computeIfAbsent(slot, k -> Maps.newHashMap()) - .put(subPath, slotReference); - } - } - } - } - if (!isSnapshot && !isReset) { - // add stream exclusive virtual columns. - slots.add(SlotReference.fromColumn( - exprIdGenerator.getNextId(), table, Column.STREAM_SEQ_VIRTUAL_COLUMN, qualified())); - slots.add(SlotReference.fromColumn( - exprIdGenerator.getNextId(), table, Column.STREAM_CHANGE_TYPE_VIRTUAL_COLUMN, qualified())); - } - for (NamedExpression virtualColumn : virtualColumns) { - slots.add(virtualColumn.toSlot()); - } - return slots.build(); - } - - /** - * withSelectedTabletIds - */ - @Override - public LogicalOlapTableStreamScan withSelectedTabletIds(List selectedTabletIds) { - return AbstractPlan.copyWithSameId(this, () -> - new LogicalOlapTableStreamScan(relationId, (Table) table, qualifier, - Optional.empty(), Optional.of(getLogicalProperties()), - selectedPartitionIds, partitionPruned, hasPartitionPredicate, selectedTabletIds, - selectedIndexId, indexSelected, preAggStatus, manuallySpecifiedPartitions, - hints, cacheSlotWithSlotName, cachedOutput, tableSample, directMvScan, colToSubPathsMap, - manuallySpecifiedTabletIds, operativeSlots, virtualColumns, scoreOrderKeys, scoreLimit, - scoreRangeInfo, annOrderKeys, annLimit, tableAlias, partitionPrunablePredicates, - scanParams, isReset, isSnapshot)); - } - - /** withCachedOutput */ - @Override - public LogicalOlapTableStreamScan withCachedOutput(List outputSlots) { - return AbstractPlan.copyWithSameId(this, () -> - new LogicalOlapTableStreamScan(relationId, (Table) table, qualifier, - groupExpression, Optional.empty(), - selectedPartitionIds, partitionPruned, hasPartitionPredicate, selectedTabletIds, - selectedIndexId, indexSelected, preAggStatus, manuallySpecifiedPartitions, hints, - cacheSlotWithSlotName, Optional.of(outputSlots), tableSample, directMvScan, colToSubPathsMap, - manuallySpecifiedTabletIds, operativeSlots, virtualColumns, scoreOrderKeys, scoreLimit, - scoreRangeInfo, annOrderKeys, annLimit, tableAlias, partitionPrunablePredicates, - scanParams, isReset, isSnapshot)); - } - - @Override - public LogicalOlapTableStreamScan withOperativeSlots(Collection operativeSlots) { - return AbstractPlan.copyWithSameId(this, () -> - new LogicalOlapTableStreamScan(relationId, (Table) table, qualifier, - groupExpression, Optional.of(getLogicalProperties()), - selectedPartitionIds, partitionPruned, hasPartitionPredicate, selectedTabletIds, - selectedIndexId, indexSelected, preAggStatus, manuallySpecifiedPartitions, - hints, cacheSlotWithSlotName, cachedOutput, tableSample, directMvScan, colToSubPathsMap, - manuallySpecifiedTabletIds, operativeSlots, virtualColumns, scoreOrderKeys, scoreLimit, - scoreRangeInfo, annOrderKeys, annLimit, tableAlias, partitionPrunablePredicates, - scanParams, isReset, isSnapshot)); - } - - /** - * withPreAggStatus - */ - @Override - public LogicalOlapTableStreamScan withPreAggStatus(PreAggStatus preAggStatus) { - return AbstractPlan.copyWithSameId(this, () -> - new LogicalOlapTableStreamScan(relationId, (Table) table, qualifier, - Optional.empty(), Optional.of(getLogicalProperties()), - selectedPartitionIds, partitionPruned, hasPartitionPredicate, selectedTabletIds, - selectedIndexId, indexSelected, preAggStatus, manuallySpecifiedPartitions, - hints, cacheSlotWithSlotName, cachedOutput, tableSample, directMvScan, - colToSubPathsMap, manuallySpecifiedTabletIds, operativeSlots, virtualColumns, - scoreOrderKeys, scoreLimit, scoreRangeInfo, annOrderKeys, annLimit, tableAlias, - partitionPrunablePredicates, scanParams, isReset, isSnapshot)); - } - - /** - * withGroupExpression - */ - @Override - public LogicalOlapTableStreamScan withGroupExpression(Optional groupExpression) { - return AbstractPlan.copyWithSameId(this, () -> - new LogicalOlapTableStreamScan(relationId, (Table) table, qualifier, - groupExpression, Optional.of(getLogicalProperties()), - selectedPartitionIds, partitionPruned, hasPartitionPredicate, selectedTabletIds, - selectedIndexId, indexSelected, preAggStatus, manuallySpecifiedPartitions, - hints, cacheSlotWithSlotName, cachedOutput, tableSample, directMvScan, - colToSubPathsMap, manuallySpecifiedTabletIds, operativeSlots, virtualColumns, - scoreOrderKeys, scoreLimit, scoreRangeInfo, annOrderKeys, annLimit, tableAlias, - partitionPrunablePredicates, scanParams, isReset, isSnapshot)); - } - - /** - * withSelectedPartitionIds - */ - @Override - public LogicalOlapTableStreamScan withSelectedPartitionIds(List selectedPartitionIdsd) { - return withSelectedPartitionIds(selectedPartitionIdsd, false); - } - - /** - * withSelectedPartitionIds - */ - @Override - public LogicalOlapTableStreamScan withSelectedPartitionIds(List selectedPartitionIds, - boolean isPartitionPruned) { - return AbstractPlan.copyWithSameId(this, () -> - new LogicalOlapTableStreamScan(relationId, (Table) table, qualifier, - groupExpression, Optional.of(getLogicalProperties()), - selectedPartitionIds, isPartitionPruned, hasPartitionPredicate, selectedTabletIds, - selectedIndexId, indexSelected, preAggStatus, manuallySpecifiedPartitions, - hints, cacheSlotWithSlotName, cachedOutput, tableSample, directMvScan, - colToSubPathsMap, manuallySpecifiedTabletIds, operativeSlots, virtualColumns, - scoreOrderKeys, scoreLimit, scoreRangeInfo, annOrderKeys, annLimit, tableAlias, - partitionPrunablePredicates, scanParams, isReset, isSnapshot)); - } - - /** - * Returns a new {@code LogicalOlapScan} carrying the supplied - * {@link PartitionPrunablePredicate}. It is preserved across all other - * {@code with*} builders so partition-derived conjuncts can be removed - * safely after MV rewrite has had a chance to match the plan. - */ - @Override - public LogicalOlapTableStreamScan withPartitionPrunablePredicates( - Optional partitionPrunablePredicates) { - return AbstractPlan.copyWithSameId(this, () -> - new LogicalOlapTableStreamScan(relationId, (Table) table, qualifier, - groupExpression, Optional.of(getLogicalProperties()), - selectedPartitionIds, partitionPruned, hasPartitionPredicate, selectedTabletIds, - selectedIndexId, indexSelected, preAggStatus, manuallySpecifiedPartitions, - hints, cacheSlotWithSlotName, cachedOutput, tableSample, directMvScan, - colToSubPathsMap, manuallySpecifiedTabletIds, operativeSlots, virtualColumns, - scoreOrderKeys, scoreLimit, scoreRangeInfo, annOrderKeys, annLimit, tableAlias, - partitionPrunablePredicates, scanParams, isReset, isSnapshot)); - } - - @Override - public LogicalOlapTableStreamScan withVirtualColumns(List virtualColumns) { - LogicalProperties logicalProperties = getLogicalProperties(); - List output = Lists.newArrayList(logicalProperties.getOutput()); - output.addAll(virtualColumns.stream().map(NamedExpression::toSlot).collect(Collectors.toList())); - LogicalProperties finalLogicalProperties = new LogicalProperties(() -> output, this::computeDataTrait); - return AbstractPlan.copyWithSameId(this, () -> - new LogicalOlapTableStreamScan(relationId, (Table) table, qualifier, - groupExpression, Optional.of(finalLogicalProperties), - selectedPartitionIds, partitionPruned, hasPartitionPredicate, selectedTabletIds, - selectedIndexId, indexSelected, preAggStatus, manuallySpecifiedPartitions, - hints, cacheSlotWithSlotName, cachedOutput, tableSample, directMvScan, - colToSubPathsMap, manuallySpecifiedTabletIds, operativeSlots, virtualColumns, - scoreOrderKeys, scoreLimit, scoreRangeInfo, annOrderKeys, annLimit, tableAlias, - partitionPrunablePredicates, scanParams, - isReset, isSnapshot)); - } - - /** withTableScanParams */ - @Override - public LogicalOlapTableStreamScan withTableScanParams(TableScanParams scanParams) { - return AbstractPlan.copyWithSameId(this, () -> - new LogicalOlapTableStreamScan(relationId, (Table) table, qualifier, - groupExpression, Optional.of(getLogicalProperties()), - selectedPartitionIds, partitionPruned, hasPartitionPredicate, selectedTabletIds, - selectedIndexId, indexSelected, preAggStatus, manuallySpecifiedPartitions, - hints, cacheSlotWithSlotName, cachedOutput, tableSample, directMvScan, colToSubPathsMap, - manuallySpecifiedTabletIds, operativeSlots, virtualColumns, scoreOrderKeys, scoreLimit, - scoreRangeInfo, annOrderKeys, annLimit, tableAlias, partitionPrunablePredicates, - Optional.of(scanParams), isReset, isSnapshot)); - } - - /** - * withRelationId - */ - @Override - public LogicalOlapTableStreamScan withRelationId(RelationId relationId) { - return AbstractPlan.copyWithSameId(this, () -> - new LogicalOlapTableStreamScan(relationId, (Table) table, qualifier, - groupExpression, Optional.empty(), - selectedPartitionIds, partitionPruned, hasPartitionPredicate, selectedTabletIds, - selectedIndexId, indexSelected, preAggStatus, manuallySpecifiedPartitions, - hints, cacheSlotWithSlotName, cachedOutput, tableSample, directMvScan, colToSubPathsMap, - manuallySpecifiedTabletIds, operativeSlots, virtualColumns, scoreOrderKeys, scoreLimit, - scoreRangeInfo, annOrderKeys, annLimit, tableAlias, partitionPrunablePredicates, - scanParams, isReset, isSnapshot)); - } - - /** - * withIsSnapshot - */ - public LogicalOlapTableStreamScan withIsSnapshot(boolean isSnapshot) { - return AbstractPlan.copyWithSameId(this, () -> - new LogicalOlapTableStreamScan(relationId, (Table) table, qualifier, - groupExpression, Optional.empty(), - selectedPartitionIds, partitionPruned, hasPartitionPredicate, selectedTabletIds, - selectedIndexId, indexSelected, preAggStatus, manuallySpecifiedPartitions, - hints, cacheSlotWithSlotName, cachedOutput, tableSample, directMvScan, colToSubPathsMap, - manuallySpecifiedTabletIds, operativeSlots, virtualColumns, scoreOrderKeys, scoreLimit, - scoreRangeInfo, annOrderKeys, annLimit, tableAlias, partitionPrunablePredicates, - scanParams, isReset, isSnapshot)); - } - - /** - * withIsReset - */ - public LogicalOlapTableStreamScan withIsReset(boolean isReset) { - return AbstractPlan.copyWithSameId(this, () -> - new LogicalOlapTableStreamScan(relationId, (Table) table, qualifier, - groupExpression, Optional.empty(), - selectedPartitionIds, partitionPruned, hasPartitionPredicate, selectedTabletIds, - selectedIndexId, indexSelected, preAggStatus, manuallySpecifiedPartitions, - hints, cacheSlotWithSlotName, cachedOutput, tableSample, directMvScan, colToSubPathsMap, - manuallySpecifiedTabletIds, operativeSlots, virtualColumns, scoreOrderKeys, scoreLimit, - scoreRangeInfo, annOrderKeys, annLimit, tableAlias, partitionPrunablePredicates, - scanParams, isReset, isSnapshot)); - } - - @Override - public R accept(PlanVisitor visitor, C context) { - return visitor.visitLogicalOlapTableStreamScan(this, context); - } - - @Override - public boolean equals(Object o) { - if (this == o) { - return true; - } - if (o == null || getClass() != o.getClass()) { - return false; - } - if (!super.equals(o)) { - return false; - } - LogicalOlapTableStreamScan that = (LogicalOlapTableStreamScan) o; - return Objects.equals(isReset, that.isReset) - && Objects.equals(isSnapshot, that.isSnapshot); - } - - @Override - public OlapTableStreamWrapper getTable() { - return (OlapTableStreamWrapper) super.getTable(); - } - - @Override - public String toString() { - return Utils.toSqlStringSkipNull("LogicalOlapTableStreamScan[" + id.asInt() + "]", - "qualified", qualifiedName(), - "alias", tableAlias, - "indexName", getSelectedMaterializedIndexName().orElse(""), - "selectedIndexId", selectedIndexId, - "preAgg", preAggStatus, - "operativeCol", operativeSlots, - "stats", statistics, - "virtualColumns", virtualColumns, - "isSnapshot", isSnapshot, - "isReset", isReset); - } - - public boolean isSnapshot() { - return isSnapshot; - } - - public boolean isReset() { - return isReset; - } - - public boolean isIncremental() { - return !isSnapshot && !isReset; - } -} From 92f9cb88518e854b2fea0b9295523377d585ade2 Mon Sep 17 00:00:00 2001 From: yujun Date: Sat, 11 Jul 2026 10:33:24 +0800 Subject: [PATCH 5/8] [test](regression) Add stream change type CTE reuse case ### What problem does this PR solve? Issue Number: None Related PR: None Problem Summary: Add a regression case for min_delta stream queries that project the stream change type column through a CTE and reuse that CTE twice in a self-join. The case verifies the output through an auto-generated .out file and keeps the existing stream regression coverage intact. ### Release note None ### Check List (For Author) - Test: Regression test - ./run-regression-test.sh --run -d table_stream_p0 -s test_min_delta_stream - Behavior changed: No - Does this need documentation: No --- .../table_stream_p0/test_min_delta_stream.out | 4 ++ .../test_min_delta_stream.groovy | 60 ++++++++++++++++++- 2 files changed, 62 insertions(+), 2 deletions(-) create mode 100644 regression-test/data/table_stream_p0/test_min_delta_stream.out diff --git a/regression-test/data/table_stream_p0/test_min_delta_stream.out b/regression-test/data/table_stream_p0/test_min_delta_stream.out new file mode 100644 index 00000000000000..b0ed5884f088d2 --- /dev/null +++ b/regression-test/data/table_stream_p0/test_min_delta_stream.out @@ -0,0 +1,4 @@ +-- This file is automatically generated. You should know what you did if you want to edit this +-- !cte_reuse_change_type_join -- +1 11 UPDATE_AFTER 10 UPDATE_BEFORE + diff --git a/regression-test/suites/table_stream_p0/test_min_delta_stream.groovy b/regression-test/suites/table_stream_p0/test_min_delta_stream.groovy index 89ee5adacdc50e..634367f14d4acd 100644 --- a/regression-test/suites/table_stream_p0/test_min_delta_stream.groovy +++ b/regression-test/suites/table_stream_p0/test_min_delta_stream.groovy @@ -46,6 +46,8 @@ suite("test_min_delta_stream", "nonConcurrent") { def ukShowInitialTarget = "md_uk_show_initial_target" def paperBase = "md_paper_base" def paperStream = "md_paper_stream" + def cteReuseBase = "md_cte_reuse_base" + def cteReuseStream = "md_cte_reuse_stream" def incrBase = "md_incr_base" def incrDupBase = "md_incr_dup_base" def incrMowNoHistoryBase = "md_incr_mow_no_history_base" @@ -72,6 +74,8 @@ suite("test_min_delta_stream", "nonConcurrent") { sql "DROP TABLE IF EXISTS ${ukShowInitialTarget}" sql "DROP STREAM IF EXISTS ${paperStream}" sql "DROP TABLE IF EXISTS ${paperBase}" + sql "DROP STREAM IF EXISTS ${cteReuseStream}" + sql "DROP TABLE IF EXISTS ${cteReuseBase}" sql "DROP TABLE IF EXISTS ${incrBase}" sql "DROP TABLE IF EXISTS ${incrDupBase}" sql "DROP TABLE IF EXISTS ${incrMowNoHistoryBase}" @@ -619,7 +623,59 @@ suite("test_min_delta_stream", "nonConcurrent") { assertEquals("Maude", paperRows[4][1].toString()) assertEquals("APPEND", paperRows[4][2].toString()) - // 11) Base table @incr(timestamp-based) queries should support MIN_DELTA / APPEND_ONLY / DETAIL, + // 11) CTE reused twice over a stream should preserve the change_type alias and support + // self-join over different change-type subsets. + sql """ + CREATE TABLE ${cteReuseBase} ( + id BIGINT, + v1 INT + ) ENGINE=OLAP + UNIQUE KEY(id) + DISTRIBUTED BY HASH(id) BUCKETS 1 + PROPERTIES ( + "replication_num" = "1", + "enable_unique_key_merge_on_write" = "true", + "binlog.enable" = "true", + "binlog.format" = "ROW", + "binlog.need_historical_value" = "true" + ) + """ + sql "INSERT INTO ${cteReuseBase} VALUES (1, 10), (2, 20)" + sql """ + CREATE STREAM ${cteReuseStream} + ON TABLE ${cteReuseBase} + PROPERTIES ( + "type" = "min_delta", + "show_initial_rows" = "false" + ) + """ + sql "INSERT INTO ${cteReuseBase} VALUES (1, 11)" + sql "DELETE FROM ${cteReuseBase} WHERE id = 2" + sql "INSERT INTO ${cteReuseBase} VALUES (3, 30)" + sql "sync" + sleep(1200) + + order_qt_cte_reuse_change_type_join """ + WITH cte AS ( + SELECT id, v1, __DORIS_STREAM_CHANGE_TYPE_COL__ AS change_type + FROM ${cteReuseStream} + ) + SELECT t1.id, t1.v1, t1.change_type, t2.v1, t2.change_type + FROM ( + SELECT id, v1, change_type + FROM cte + WHERE change_type IN ('APPEND', 'UPDATE_AFTER') + ) t1 + JOIN ( + SELECT id, v1, change_type + FROM cte + WHERE change_type IN ('DELETE', 'UPDATE_BEFORE') + ) t2 + ON t1.id = t2.id + ORDER BY t1.id + """ + + // 12) Base table @incr(timestamp-based) queries should support MIN_DELTA / APPEND_ONLY / DETAIL, // and DETAIL without startTimestamp should behave the same as default @incr(). sql """ CREATE TABLE ${incrBase} ( @@ -742,7 +798,7 @@ suite("test_min_delta_stream", "nonConcurrent") { """ assertEquals(minDeltaDefaultRows, emptyIncrRows) - // 12) DETAIL incr should support duplicate table and MOW table without historical value. + // 13) DETAIL incr should support duplicate table and MOW table without historical value. sql """ CREATE TABLE ${incrDupBase} ( id BIGINT, From 0db627ec3f8a6b76e2c112c88c3fa1499636f19a Mon Sep 17 00:00:00 2001 From: yujun Date: Sat, 11 Jul 2026 11:31:34 +0800 Subject: [PATCH 6/8] [refactor](fe) Unify stream scan read mode handling ### What problem does this PR solve? Issue Number: None Related PR: None Problem Summary: Stream scan planning had split state for snapshot and reset, while partition pruning still depended on table-specific non-empty checks that did not receive stream read semantics. This made the stream planning path harder to reason about and caused snapshot/reset handling to diverge from incremental reads. This change replaces the dual boolean state on logical stream scan with a single StreamReadMode enum, threads the optional read mode through LogicalCatalogRelation into selectNonEmptyPartitionIds, and lets the stream wrapper choose incremental pruning from stream.hasData while delegating snapshot/reset pruning to the base table path. The result keeps the read-mode semantics continuous from binding through rewrite and pruning, while preserving the convenience isSnapshot/isReset/isIncremental helpers on the scan node. ### Release note None ### Check List (For Author) - Test: Unit Test - ./run-fe-ut.sh --run org.apache.doris.nereids.trees.plans.ExplainTableStreamPlanTest - Behavior changed: Yes (stream scan snapshot/reset partition pruning now follows explicit read-mode plumbing instead of split boolean flags) - Does this need documentation: No --- .../org/apache/doris/catalog/OlapTable.java | 4 +- .../doris/catalog/OlapTableWrapper.java | 7 +- .../stream/OlapTableStreamWrapper.java | 8 +- .../doris/catalog/stream/StreamReadMode.java | 27 +++++++ .../nereids/rules/analysis/BindRelation.java | 7 +- .../exploration/mv/PartitionCompensator.java | 4 +- .../rules/rewrite/PruneEmptyPartition.java | 2 +- .../commands/UpdateMvByPartitionCommand.java | 2 +- .../plans/logical/LogicalCatalogRelation.java | 5 ++ .../logical/LogicalOlapTableStreamScan.java | 81 ++++++++----------- .../apache/doris/planner/OlapScanNode.java | 2 +- .../mv/PartitionCompensatorTest.java | 5 +- 12 files changed, 95 insertions(+), 59 deletions(-) create mode 100644 fe/fe-core/src/main/java/org/apache/doris/catalog/stream/StreamReadMode.java diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/OlapTable.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/OlapTable.java index c8d6c515b29686..89d9970d7fe402 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/catalog/OlapTable.java +++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/OlapTable.java @@ -33,6 +33,7 @@ import org.apache.doris.catalog.Tablet.TabletStatus; import org.apache.doris.catalog.info.IndexType; import org.apache.doris.catalog.stream.BaseTableStream; +import org.apache.doris.catalog.stream.StreamReadMode; import org.apache.doris.clone.TabletScheduler; import org.apache.doris.cloud.catalog.CloudPartition; import org.apache.doris.cloud.catalog.CloudReplica; @@ -1564,7 +1565,8 @@ public void getVersionInBatchForCloudMode(Collection partitionIds) throws // select the non-empty partition ids belonging to this table. // // ATTN: partitions not belonging to this table will be filtered. - public List selectNonEmptyPartitionIds(Collection partitionIds) { + public List selectNonEmptyPartitionIds(Collection partitionIds, + Optional streamReadMode) { if (Config.isCloudMode() && Config.enable_cloud_snapshot_version) { // Assumption: all partitions are CloudPartition. List partitions = partitionIds.stream() diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/OlapTableWrapper.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/OlapTableWrapper.java index 4e075727d27a63..83be627f8ec045 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/catalog/OlapTableWrapper.java +++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/OlapTableWrapper.java @@ -17,6 +17,7 @@ package org.apache.doris.catalog; +import org.apache.doris.catalog.stream.StreamReadMode; import org.apache.doris.common.Pair; import java.util.Collection; @@ -24,6 +25,7 @@ import java.util.List; import java.util.Map; import java.util.Objects; +import java.util.Optional; import java.util.Set; import java.util.concurrent.TimeUnit; @@ -146,8 +148,9 @@ public Collection getPartitions() { } @Override - public List selectNonEmptyPartitionIds(Collection partitionIds) { - return originTable.selectNonEmptyPartitionIds(partitionIds); + public List selectNonEmptyPartitionIds(Collection partitionIds, + Optional streamReadMode) { + return originTable.selectNonEmptyPartitionIds(partitionIds, streamReadMode); } @Override diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/stream/OlapTableStreamWrapper.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/stream/OlapTableStreamWrapper.java index 15119819e909a6..92c9701884af35 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/catalog/stream/OlapTableStreamWrapper.java +++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/stream/OlapTableStreamWrapper.java @@ -34,6 +34,7 @@ import java.util.Collection; import java.util.List; import java.util.Map; +import java.util.Optional; import java.util.stream.Collectors; // runtime-only class for unified query/insert experience, created when bind relation with OlapTableStream @@ -174,7 +175,12 @@ public Collection getPartitions() { } @Override - public List selectNonEmptyPartitionIds(Collection partitionIds) { + public List selectNonEmptyPartitionIds(Collection partitionIds, + Optional streamReadMode) { + StreamReadMode readMode = streamReadMode.orElse(StreamReadMode.INCREMENTAL); + if (readMode == StreamReadMode.SNAPSHOT || readMode == StreamReadMode.RESET) { + return baseTable.selectNonEmptyPartitionIds(partitionIds, Optional.of(readMode)); + } List nonEmptyIds = Lists.newArrayListWithCapacity(partitionIds.size()); for (Long partitionId : partitionIds) { if (stream.hasData(getPartition(partitionId))) { diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/stream/StreamReadMode.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/stream/StreamReadMode.java new file mode 100644 index 00000000000000..e385bdebe5df9e --- /dev/null +++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/stream/StreamReadMode.java @@ -0,0 +1,27 @@ +// 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.catalog.stream; + +/** + * Read mode of a stream scan after binding. + */ +public enum StreamReadMode { + INCREMENTAL, + SNAPSHOT, + RESET +} diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindRelation.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindRelation.java index 28515845d30a61..a7ee0188901310 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindRelation.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindRelation.java @@ -40,6 +40,7 @@ import org.apache.doris.catalog.stream.BaseTableStream.StreamScanType; 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.common.Config; import org.apache.doris.common.IdGenerator; import org.apache.doris.common.Pair; @@ -345,9 +346,9 @@ private LogicalOlapTableStreamScan makeOlapTableStreamScan(OlapTableStream olapT if (unboundRelation.getScanParams() != null) { unboundRelation.getScanParams().validateOlapTableStream(); if (unboundRelation.getScanParams().isSnapshot()) { - scan = scan.withIsSnapshot(true); + scan = scan.withReadMode(StreamReadMode.SNAPSHOT); } else if (unboundRelation.getScanParams().isReset()) { - scan = scan.withIsReset(true); + scan = scan.withReadMode(StreamReadMode.RESET); } } if (!tabletIds.isEmpty()) { @@ -1004,7 +1005,7 @@ private LogicalPlan makeTableStreamScan(TableIf table, UnboundRelation unboundRe OlapTableStream olapTableStream = (OlapTableStream) table; LogicalOlapTableStreamScan scan = makeOlapTableStreamScan(olapTableStream, unboundRelation, qualifier); - if (!scan.isSnapshot() && !scan.isReset() && isScanAppendOnlyTableStream(olapTableStream)) { + if (scan.isIncremental() && isScanAppendOnlyTableStream(olapTableStream)) { LOG.debug("Add append only filter on olap scan if need."); return addAppendOnlyFilter(scan); } diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/PartitionCompensator.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/PartitionCompensator.java index fe2e88cd5046c3..9064049d3b801b 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/PartitionCompensator.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/PartitionCompensator.java @@ -50,6 +50,7 @@ import java.util.List; import java.util.Map; import java.util.Objects; +import java.util.Optional; import java.util.Set; /** @@ -176,7 +177,8 @@ private static Pair>, Pair relatedBaseTablePartitions = partitionMapping.get(mvValidPartition.getName()); if (relatedBaseTablePartitions != null) { mvValidBaseTablePartitionNameSet.addAll(relatedBaseTablePartitions); - if (!mtmv.selectNonEmptyPartitionIds(ImmutableList.of(mvValidPartition.getId())).isEmpty()) { + if (!mtmv.selectNonEmptyPartitionIds(ImmutableList.of(mvValidPartition.getId()), + Optional.empty()).isEmpty()) { mvValidHasDataRelatedBaseTableNameSet.addAll(relatedBaseTablePartitions); } } diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/PruneEmptyPartition.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/PruneEmptyPartition.java index aadd48bbd523be..aa0fa7fed521aa 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/PruneEmptyPartition.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/PruneEmptyPartition.java @@ -49,7 +49,7 @@ public Rule build() { LogicalOlapScan scan = ctx.root; OlapTable table = scan.getTable(); List partitionIdsToPrune = scan.getSelectedPartitionIds(); - List ids = table.selectNonEmptyPartitionIds(partitionIdsToPrune); + List ids = table.selectNonEmptyPartitionIds(partitionIdsToPrune, scan.getStreamReadMode()); if (ctx.connectContext != null && ctx.connectContext.isTxnModel()) { // In transaction load, need to add empty partitions which have invisible data of sub transactions selectNonEmptyPartitionIdsForTxnLoad(ctx.connectContext.getTxnEntry(), table, scan.getSelectedIndexId(), diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/UpdateMvByPartitionCommand.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/UpdateMvByPartitionCommand.java index 74f2bd1bfec007..94be0543496e52 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/UpdateMvByPartitionCommand.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/UpdateMvByPartitionCommand.java @@ -330,7 +330,7 @@ public Plan visitLogicalCatalogRelation(LogicalCatalogRelation catalogRelation, continue; } if (!((OlapTable) targetTable).selectNonEmptyPartitionIds( - Lists.newArrayList(partition.getId())).isEmpty()) { + Lists.newArrayList(partition.getId()), Optional.empty()).isEmpty()) { // Add filter only when partition has data when olap table partitionHasDataItems.add( ((OlapTable) targetTable).getPartitionInfo().getItem(partition.getId())); diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalCatalogRelation.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalCatalogRelation.java index e45516c17c0657..1fc43325d18516 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalCatalogRelation.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalCatalogRelation.java @@ -25,6 +25,7 @@ import org.apache.doris.catalog.constraint.PrimaryKeyConstraint; import org.apache.doris.catalog.constraint.UniqueConstraint; import org.apache.doris.catalog.info.TableNameInfo; +import org.apache.doris.catalog.stream.StreamReadMode; import org.apache.doris.common.IdGenerator; import org.apache.doris.common.util.Util; import org.apache.doris.datasource.CatalogIf; @@ -194,6 +195,10 @@ public String getTableAlias() { return tableAlias; } + public Optional getStreamReadMode() { + return Optional.empty(); + } + @Override public void computeUnique(DataTrait.Builder builder) { Set outputSet = Utils.fastToImmutableSet(getOutputSet()); 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 4d1a52326661f4..b56a92ffccc08a 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 @@ -22,6 +22,7 @@ import org.apache.doris.catalog.OlapTable; import org.apache.doris.catalog.Table; 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.memo.GroupExpression; import org.apache.doris.nereids.properties.LogicalProperties; @@ -57,8 +58,7 @@ * Logical OlapTableStreamScan */ public class LogicalOlapTableStreamScan extends LogicalOlapScan { - private final boolean isReset; - private final boolean isSnapshot; + private final StreamReadMode readMode; /** * LogicalOlapTableStreamScan construct method @@ -66,8 +66,7 @@ public class LogicalOlapTableStreamScan extends LogicalOlapScan { public LogicalOlapTableStreamScan(RelationId id, OlapTable table, List qualifier, List tabletIds, List hints, Optional tableSample, Collection operativeSlots) { super(id, table, qualifier, tabletIds, hints, tableSample, operativeSlots); - this.isReset = false; - this.isSnapshot = false; + this.readMode = StreamReadMode.INCREMENTAL; } /** @@ -77,8 +76,7 @@ public LogicalOlapTableStreamScan(RelationId id, OlapTable table, List q List specifiedPartitions, List tabletIds, List hints, Optional tableSample, List operativeSlots) { super(id, table, qualifier, specifiedPartitions, tabletIds, hints, tableSample, operativeSlots); - this.isReset = false; - this.isSnapshot = false; + this.readMode = StreamReadMode.INCREMENTAL; } /** @@ -101,15 +99,14 @@ public LogicalOlapTableStreamScan(RelationId id, Table table, List quali List annOrderKeys, Optional annLimit, String tableAlias, Optional partitionPrunablePredicates, Optional scanParams, - boolean isReset, boolean isSnapshot) { + StreamReadMode readMode) { super(id, table, qualifier, groupExpression, logicalProperties, selectedPartitionIds, partitionPruned, hasPartitionPredicate, selectedTabletIds, selectedIndexId, indexSelected, preAggStatus, specifiedPartitions, hints, cacheSlotWithSlotName, cachedOutput, tableSample, directMvScan, colToSubPathsMap, specifiedTabletIds, operativeSlots, virtualColumns, scoreOrderKeys, scoreLimit, scoreRangeInfo, annOrderKeys, annLimit, tableAlias, partitionPrunablePredicates, scanParams); - this.isReset = isReset; - this.isSnapshot = isSnapshot; + this.readMode = Objects.requireNonNull(readMode, "readMode can not be null"); } @Override @@ -122,7 +119,7 @@ public LogicalOlapTableStreamScan withManuallySpecifiedTabletIds(List manu hints, cacheSlotWithSlotName, cachedOutput, tableSample, directMvScan, colToSubPathsMap, manuallySpecifiedTabletIds, operativeSlots, virtualColumns, scoreOrderKeys, scoreLimit, scoreRangeInfo, annOrderKeys, annLimit, tableAlias, - partitionPrunablePredicates, scanParams, isReset, isSnapshot)); + partitionPrunablePredicates, scanParams, readMode)); } @Override @@ -132,7 +129,7 @@ public List computeOutput() { } // for reset, we could use get full schema of base table; // otherwise, we only need to get the schema without hidden columns - List baseSchema = table.getBaseSchema(isReset); + List baseSchema = table.getBaseSchema(readMode == StreamReadMode.RESET); List slotFromColumn = createSlotsVectorized(baseSchema); ImmutableList.Builder slots = ImmutableList.builder(); @@ -186,7 +183,7 @@ public LogicalOlapTableStreamScan withSelectedTabletIds(List selectedTable hints, cacheSlotWithSlotName, cachedOutput, tableSample, directMvScan, colToSubPathsMap, manuallySpecifiedTabletIds, operativeSlots, virtualColumns, scoreOrderKeys, scoreLimit, scoreRangeInfo, annOrderKeys, annLimit, tableAlias, partitionPrunablePredicates, - scanParams, isReset, isSnapshot)); + scanParams, readMode)); } /** withCachedOutput */ @@ -200,7 +197,7 @@ public LogicalOlapTableStreamScan withCachedOutput(List outputSlots) { cacheSlotWithSlotName, Optional.of(outputSlots), tableSample, directMvScan, colToSubPathsMap, manuallySpecifiedTabletIds, operativeSlots, virtualColumns, scoreOrderKeys, scoreLimit, scoreRangeInfo, annOrderKeys, annLimit, tableAlias, partitionPrunablePredicates, - scanParams, isReset, isSnapshot)); + scanParams, readMode)); } @Override @@ -213,7 +210,7 @@ public LogicalOlapTableStreamScan withOperativeSlots(Collection operativeS hints, cacheSlotWithSlotName, cachedOutput, tableSample, directMvScan, colToSubPathsMap, manuallySpecifiedTabletIds, operativeSlots, virtualColumns, scoreOrderKeys, scoreLimit, scoreRangeInfo, annOrderKeys, annLimit, tableAlias, partitionPrunablePredicates, - scanParams, isReset, isSnapshot)); + scanParams, readMode)); } /** @@ -229,7 +226,7 @@ public LogicalOlapTableStreamScan withPreAggStatus(PreAggStatus preAggStatus) { hints, cacheSlotWithSlotName, cachedOutput, tableSample, directMvScan, colToSubPathsMap, manuallySpecifiedTabletIds, operativeSlots, virtualColumns, scoreOrderKeys, scoreLimit, scoreRangeInfo, annOrderKeys, annLimit, tableAlias, - partitionPrunablePredicates, scanParams, isReset, isSnapshot)); + partitionPrunablePredicates, scanParams, readMode)); } /** @@ -245,7 +242,7 @@ public LogicalOlapTableStreamScan withGroupExpression(Optional hints, cacheSlotWithSlotName, cachedOutput, tableSample, directMvScan, colToSubPathsMap, manuallySpecifiedTabletIds, operativeSlots, virtualColumns, scoreOrderKeys, scoreLimit, scoreRangeInfo, annOrderKeys, annLimit, tableAlias, - partitionPrunablePredicates, scanParams, isReset, isSnapshot)); + partitionPrunablePredicates, scanParams, readMode)); } /** @@ -270,7 +267,7 @@ public LogicalOlapTableStreamScan withSelectedPartitionIds(List selectedPa hints, cacheSlotWithSlotName, cachedOutput, tableSample, directMvScan, colToSubPathsMap, manuallySpecifiedTabletIds, operativeSlots, virtualColumns, scoreOrderKeys, scoreLimit, scoreRangeInfo, annOrderKeys, annLimit, tableAlias, - partitionPrunablePredicates, scanParams, isReset, isSnapshot)); + partitionPrunablePredicates, scanParams, readMode)); } /** @@ -290,7 +287,7 @@ public LogicalOlapTableStreamScan withPartitionPrunablePredicates( hints, cacheSlotWithSlotName, cachedOutput, tableSample, directMvScan, colToSubPathsMap, manuallySpecifiedTabletIds, operativeSlots, virtualColumns, scoreOrderKeys, scoreLimit, scoreRangeInfo, annOrderKeys, annLimit, tableAlias, - partitionPrunablePredicates, scanParams, isReset, isSnapshot)); + partitionPrunablePredicates, scanParams, readMode)); } @Override @@ -308,7 +305,7 @@ public LogicalOlapTableStreamScan withVirtualColumns(List virtu colToSubPathsMap, manuallySpecifiedTabletIds, operativeSlots, virtualColumns, scoreOrderKeys, scoreLimit, scoreRangeInfo, annOrderKeys, annLimit, tableAlias, partitionPrunablePredicates, scanParams, - isReset, isSnapshot)); + readMode)); } /** withTableScanParams */ @@ -322,7 +319,7 @@ public LogicalOlapTableStreamScan withTableScanParams(TableScanParams scanParams hints, cacheSlotWithSlotName, cachedOutput, tableSample, directMvScan, colToSubPathsMap, manuallySpecifiedTabletIds, operativeSlots, virtualColumns, scoreOrderKeys, scoreLimit, scoreRangeInfo, annOrderKeys, annLimit, tableAlias, partitionPrunablePredicates, - Optional.of(scanParams), isReset, isSnapshot)); + Optional.of(scanParams), readMode)); } /** @@ -338,13 +335,13 @@ public LogicalOlapTableStreamScan withRelationId(RelationId relationId) { hints, cacheSlotWithSlotName, cachedOutput, tableSample, directMvScan, colToSubPathsMap, manuallySpecifiedTabletIds, operativeSlots, virtualColumns, scoreOrderKeys, scoreLimit, scoreRangeInfo, annOrderKeys, annLimit, tableAlias, partitionPrunablePredicates, - scanParams, isReset, isSnapshot)); + scanParams, readMode)); } /** - * withIsSnapshot + * withReadMode */ - public LogicalOlapTableStreamScan withIsSnapshot(boolean isSnapshot) { + public LogicalOlapTableStreamScan withReadMode(StreamReadMode readMode) { return AbstractPlan.copyWithSameId(this, () -> new LogicalOlapTableStreamScan(relationId, (Table) table, qualifier, groupExpression, Optional.empty(), @@ -353,22 +350,7 @@ public LogicalOlapTableStreamScan withIsSnapshot(boolean isSnapshot) { hints, cacheSlotWithSlotName, cachedOutput, tableSample, directMvScan, colToSubPathsMap, manuallySpecifiedTabletIds, operativeSlots, virtualColumns, scoreOrderKeys, scoreLimit, scoreRangeInfo, annOrderKeys, annLimit, tableAlias, partitionPrunablePredicates, - scanParams, isReset, isSnapshot)); - } - - /** - * withIsReset - */ - public LogicalOlapTableStreamScan withIsReset(boolean isReset) { - return AbstractPlan.copyWithSameId(this, () -> - new LogicalOlapTableStreamScan(relationId, (Table) table, qualifier, - groupExpression, Optional.empty(), - selectedPartitionIds, partitionPruned, hasPartitionPredicate, selectedTabletIds, - selectedIndexId, indexSelected, preAggStatus, manuallySpecifiedPartitions, - hints, cacheSlotWithSlotName, cachedOutput, tableSample, directMvScan, colToSubPathsMap, - manuallySpecifiedTabletIds, operativeSlots, virtualColumns, scoreOrderKeys, scoreLimit, - scoreRangeInfo, annOrderKeys, annLimit, tableAlias, partitionPrunablePredicates, - scanParams, isReset, isSnapshot)); + scanParams, readMode)); } @Override @@ -388,8 +370,7 @@ public boolean equals(Object o) { return false; } LogicalOlapTableStreamScan that = (LogicalOlapTableStreamScan) o; - return Objects.equals(isReset, that.isReset) - && Objects.equals(isSnapshot, that.isSnapshot); + return readMode == that.readMode; } @Override @@ -397,6 +378,11 @@ public OlapTableStreamWrapper getTable() { return (OlapTableStreamWrapper) super.getTable(); } + @Override + public Optional getStreamReadMode() { + return Optional.of(readMode); + } + @Override public String toString() { return Utils.toSqlStringSkipNull("LogicalOlapTableStreamScan[" + id.asInt() + "]", @@ -408,19 +394,22 @@ public String toString() { "operativeCol", operativeSlots, "stats", statistics, "virtualColumns", virtualColumns, - "isSnapshot", isSnapshot, - "isReset", isReset); + "readMode", readMode); + } + + public StreamReadMode getReadMode() { + return readMode; } public boolean isSnapshot() { - return isSnapshot; + return readMode == StreamReadMode.SNAPSHOT; } public boolean isReset() { - return isReset; + return readMode == StreamReadMode.RESET; } public boolean isIncremental() { - return !isSnapshot && !isReset; + return readMode == StreamReadMode.INCREMENTAL; } } diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/OlapScanNode.java b/fe/fe-core/src/main/java/org/apache/doris/planner/OlapScanNode.java index 7fb0975d88d0f1..2535961a76ffe9 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/planner/OlapScanNode.java +++ b/fe/fe-core/src/main/java/org/apache/doris/planner/OlapScanNode.java @@ -792,7 +792,7 @@ public void computePartitionInfo() throws AnalysisException { setHasPartitionPredicate(false); selectedPartitionIds = olapTable.getPartitionIds(); } - selectedPartitionIds = olapTable.selectNonEmptyPartitionIds(selectedPartitionIds); + selectedPartitionIds = olapTable.selectNonEmptyPartitionIds(selectedPartitionIds, Optional.empty()); selectedPartitionNum = selectedPartitionIds.size(); for (long id : selectedPartitionIds) { diff --git a/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/exploration/mv/PartitionCompensatorTest.java b/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/exploration/mv/PartitionCompensatorTest.java index 1af0b983fee7a1..96e689cee82694 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/exploration/mv/PartitionCompensatorTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/exploration/mv/PartitionCompensatorTest.java @@ -444,7 +444,8 @@ public void testCalcInvalidPartitionsNoConcurrentModificationWithTwoRelatedTable Mockito.when(mvPctInfo.getPctTables()).thenReturn(ImmutableSet.of(relatedTable1, relatedTable2)); Mockito.when(mvPctInfo.getPctInfos()).thenReturn(ImmutableList.of(colInfo1, colInfo2)); // All MV partitions contain data - Mockito.when(mtmv.selectNonEmptyPartitionIds(ArgumentMatchers.any())).thenReturn(ImmutableList.of(1L)); + Mockito.when(mtmv.selectNonEmptyPartitionIds(ArgumentMatchers.any(), ArgumentMatchers.any())) + .thenReturn(ImmutableList.of(1L)); AsyncMaterializationContext matCtx = Mockito.mock(AsyncMaterializationContext.class); Mockito.when(matCtx.getMtmv()).thenReturn(mtmv); @@ -500,7 +501,7 @@ public void testCalcInvalidPartitionsDoesNotCompensateBasePartitionsUnusedByQuer Mockito.when(mtmv.getName()).thenReturn("mv1"); Mockito.when(mtmv.getId()).thenReturn(100L); Mockito.when(mtmv.getDatabase()).thenReturn(mvDb); - Mockito.when(mtmv.selectNonEmptyPartitionIds(ArgumentMatchers.any())) + Mockito.when(mtmv.selectNonEmptyPartitionIds(ArgumentMatchers.any(), ArgumentMatchers.any())) .thenReturn(ImmutableList.of(1L)); long mvP20260301Id = 101L; From 0e4329c7c3c6cdd1b979e498be0640b9cd022149 Mon Sep 17 00:00:00 2001 From: yujun Date: Sat, 11 Jul 2026 11:40:33 +0800 Subject: [PATCH 7/8] [refactor](fe) Reorder stream scan helper methods ### What problem does this PR solve? Issue Number: None Related PR: #65468 Problem Summary: This follow-up keeps the stream scan helper implementation order aligned with the surrounding builder methods by moving after in . The change is code organization only and does not alter semantics. ### Release note None ### Check List (For Author) - Test: No need to test (with reason) - Method order only, no logic change - Behavior changed: No - Does this need documentation: No --- .../logical/LogicalOlapTableStreamScan.java | 36 +++++++++---------- 1 file changed, 18 insertions(+), 18 deletions(-) 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 b56a92ffccc08a..44c99ad4cbb1e5 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 @@ -290,24 +290,6 @@ public LogicalOlapTableStreamScan withPartitionPrunablePredicates( partitionPrunablePredicates, scanParams, readMode)); } - @Override - public LogicalOlapTableStreamScan withVirtualColumns(List virtualColumns) { - LogicalProperties logicalProperties = getLogicalProperties(); - List output = Lists.newArrayList(logicalProperties.getOutput()); - output.addAll(virtualColumns.stream().map(NamedExpression::toSlot).collect(Collectors.toList())); - LogicalProperties finalLogicalProperties = new LogicalProperties(() -> output, this::computeDataTrait); - return AbstractPlan.copyWithSameId(this, () -> - new LogicalOlapTableStreamScan(relationId, (Table) table, qualifier, - groupExpression, Optional.of(finalLogicalProperties), - selectedPartitionIds, partitionPruned, hasPartitionPredicate, selectedTabletIds, - selectedIndexId, indexSelected, preAggStatus, manuallySpecifiedPartitions, - hints, cacheSlotWithSlotName, cachedOutput, tableSample, directMvScan, - colToSubPathsMap, manuallySpecifiedTabletIds, operativeSlots, virtualColumns, - scoreOrderKeys, scoreLimit, scoreRangeInfo, annOrderKeys, annLimit, tableAlias, - partitionPrunablePredicates, scanParams, - readMode)); - } - /** withTableScanParams */ @Override public LogicalOlapTableStreamScan withTableScanParams(TableScanParams scanParams) { @@ -338,6 +320,24 @@ public LogicalOlapTableStreamScan withRelationId(RelationId relationId) { scanParams, readMode)); } + @Override + public LogicalOlapTableStreamScan withVirtualColumns(List virtualColumns) { + LogicalProperties logicalProperties = getLogicalProperties(); + List output = Lists.newArrayList(logicalProperties.getOutput()); + output.addAll(virtualColumns.stream().map(NamedExpression::toSlot).collect(Collectors.toList())); + LogicalProperties finalLogicalProperties = new LogicalProperties(() -> output, this::computeDataTrait); + return AbstractPlan.copyWithSameId(this, () -> + new LogicalOlapTableStreamScan(relationId, (Table) table, qualifier, + groupExpression, Optional.of(finalLogicalProperties), + selectedPartitionIds, partitionPruned, hasPartitionPredicate, selectedTabletIds, + selectedIndexId, indexSelected, preAggStatus, manuallySpecifiedPartitions, + hints, cacheSlotWithSlotName, cachedOutput, tableSample, directMvScan, + colToSubPathsMap, manuallySpecifiedTabletIds, operativeSlots, virtualColumns, + scoreOrderKeys, scoreLimit, scoreRangeInfo, annOrderKeys, annLimit, tableAlias, + partitionPrunablePredicates, scanParams, + readMode)); + } + /** * withReadMode */ From 1a3a4ab51d177c09c05549a228bc64101c4e5679 Mon Sep 17 00:00:00 2001 From: yujun Date: Mon, 13 Jul 2026 15:55:25 +0800 Subject: [PATCH 8/8] [fix](fe) Preserve stream scan type in OLAP copy paths Keep LogicalOlapTableStreamScan on relation-copy and builder-style paths that were still inherited from LogicalOlapScan. Without these overrides, aliased stream scans and some virtual-column / topn rewrite paths could rebuild the node as a plain LogicalOlapScan, so NormalizeOlapTableStreamScan would no longer match. Key changes: - override withTableAlias to preserve LogicalOlapTableStreamScan and readMode - override remaining LogicalOlapScan rebuild helpers used by stream rewrite paths - add aliased snapshot/reset FE UT coverage for the stream normalization path Unit Test: - ./run-fe-ut.sh --run org.apache.doris.nereids.trees.plans.ExplainTableStreamPlanTest --- .../logical/LogicalOlapTableStreamScan.java | 111 ++++++++++++++++++ .../plans/ExplainTableStreamPlanTest.java | 18 +++ 2 files changed, 129 insertions(+) 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 44c99ad4cbb1e5..503a515e0c6e48 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 @@ -34,6 +34,7 @@ import org.apache.doris.nereids.trees.expressions.SlotReference; import org.apache.doris.nereids.trees.expressions.StatementScopeIdGenerator; import org.apache.doris.nereids.trees.plans.AbstractPlan; +import org.apache.doris.nereids.trees.plans.Plan; import org.apache.doris.nereids.trees.plans.PartitionPrunablePredicate; import org.apache.doris.nereids.trees.plans.PreAggStatus; import org.apache.doris.nereids.trees.plans.RelationId; @@ -245,6 +246,23 @@ public LogicalOlapTableStreamScan withGroupExpression(Optional partitionPrunablePredicates, scanParams, readMode)); } + /** + * withGroupExprLogicalPropChildren + */ + @Override + public Plan withGroupExprLogicalPropChildren(Optional groupExpression, + Optional logicalProperties, List children) { + return AbstractPlan.copyWithSameId(this, () -> + new LogicalOlapTableStreamScan(relationId, (Table) table, qualifier, + groupExpression, logicalProperties, + selectedPartitionIds, partitionPruned, hasPartitionPredicate, selectedTabletIds, + selectedIndexId, indexSelected, preAggStatus, manuallySpecifiedPartitions, + hints, cacheSlotWithSlotName, cachedOutput, tableSample, directMvScan, + colToSubPathsMap, manuallySpecifiedTabletIds, operativeSlots, virtualColumns, + scoreOrderKeys, scoreLimit, scoreRangeInfo, annOrderKeys, annLimit, tableAlias, + partitionPrunablePredicates, scanParams, readMode)); + } + /** * withSelectedPartitionIds */ @@ -290,6 +308,38 @@ public LogicalOlapTableStreamScan withPartitionPrunablePredicates( partitionPrunablePredicates, scanParams, readMode)); } + /** + * with sync materialized index id. + */ + @Override + public LogicalOlapTableStreamScan withMaterializedIndexSelected(long indexId) { + return AbstractPlan.copyWithSameId(this, () -> + new LogicalOlapTableStreamScan(relationId, (Table) table, qualifier, + Optional.empty(), Optional.of(getLogicalProperties()), + selectedPartitionIds, partitionPruned, hasPartitionPredicate, selectedTabletIds, + indexId, true, PreAggStatus.unset(), manuallySpecifiedPartitions, + hints, cacheSlotWithSlotName, cachedOutput, tableSample, directMvScan, + colToSubPathsMap, manuallySpecifiedTabletIds, operativeSlots, virtualColumns, + scoreOrderKeys, scoreLimit, scoreRangeInfo, annOrderKeys, annLimit, tableAlias, + partitionPrunablePredicates, scanParams, readMode)); + } + + /** + * constructor + */ + @Override + public LogicalOlapTableStreamScan withColToSubPathsMap(Map>> colToSubPathsMap) { + return AbstractPlan.copyWithSameId(this, () -> + new LogicalOlapTableStreamScan(relationId, (Table) table, qualifier, + Optional.empty(), Optional.empty(), + selectedPartitionIds, partitionPruned, hasPartitionPredicate, selectedTabletIds, + selectedIndexId, indexSelected, preAggStatus, manuallySpecifiedPartitions, + hints, cacheSlotWithSlotName, cachedOutput, tableSample, directMvScan, + colToSubPathsMap, manuallySpecifiedTabletIds, operativeSlots, virtualColumns, + scoreOrderKeys, scoreLimit, scoreRangeInfo, annOrderKeys, annLimit, tableAlias, + partitionPrunablePredicates, scanParams, readMode)); + } + /** withTableScanParams */ @Override public LogicalOlapTableStreamScan withTableScanParams(TableScanParams scanParams) { @@ -320,6 +370,19 @@ public LogicalOlapTableStreamScan withRelationId(RelationId relationId) { scanParams, readMode)); } + @Override + public LogicalOlapTableStreamScan withTableAlias(String tableAlias) { + return AbstractPlan.copyWithSameId(this, () -> + new LogicalOlapTableStreamScan(relationId, (Table) table, qualifier, + Optional.empty(), Optional.of(getLogicalProperties()), + selectedPartitionIds, partitionPruned, hasPartitionPredicate, selectedTabletIds, + selectedIndexId, indexSelected, preAggStatus, manuallySpecifiedPartitions, + hints, cacheSlotWithSlotName, cachedOutput, tableSample, directMvScan, + colToSubPathsMap, manuallySpecifiedTabletIds, operativeSlots, virtualColumns, + scoreOrderKeys, scoreLimit, scoreRangeInfo, annOrderKeys, annLimit, tableAlias, + partitionPrunablePredicates, scanParams, readMode)); + } + @Override public LogicalOlapTableStreamScan withVirtualColumns(List virtualColumns) { LogicalProperties logicalProperties = getLogicalProperties(); @@ -338,6 +401,54 @@ public LogicalOlapTableStreamScan withVirtualColumns(List virtu readMode)); } + @Override + public LogicalOlapTableStreamScan appendVirtualColumns(List additionalVirtualColumns) { + LogicalProperties logicalProperties = getLogicalProperties(); + List output = Lists.newArrayList(logicalProperties.getOutput()); + output.addAll(additionalVirtualColumns.stream().map(NamedExpression::toSlot) + .collect(Collectors.toList())); + logicalProperties = new LogicalProperties(() -> output, this::computeDataTrait); + List mergedVirtualColumns = ImmutableList.builder() + .addAll(this.virtualColumns) + .addAll(additionalVirtualColumns) + .build(); + return new LogicalOlapTableStreamScan(relationId, (Table) table, qualifier, + groupExpression, Optional.of(logicalProperties), + selectedPartitionIds, partitionPruned, hasPartitionPredicate, selectedTabletIds, + selectedIndexId, indexSelected, preAggStatus, manuallySpecifiedPartitions, + hints, cacheSlotWithSlotName, cachedOutput, tableSample, directMvScan, colToSubPathsMap, + manuallySpecifiedTabletIds, operativeSlots, mergedVirtualColumns, scoreOrderKeys, scoreLimit, + scoreRangeInfo, annOrderKeys, annLimit, tableAlias, partitionPrunablePredicates, scanParams, + readMode); + } + + @Override + public LogicalOlapTableStreamScan appendVirtualColumnsAndTopN( + List additionalVirtualColumns, + List annOrderKeys, + Optional annLimit, + List scoreOrderKeys, + Optional scoreLimit, + Optional scoreRangeInfo) { + LogicalProperties logicalProperties = getLogicalProperties(); + List output = Lists.newArrayList(logicalProperties.getOutput()); + output.addAll(additionalVirtualColumns.stream().map(NamedExpression::toSlot) + .collect(Collectors.toList())); + logicalProperties = new LogicalProperties(() -> output, this::computeDataTrait); + List mergedVirtualColumns = ImmutableList.builder() + .addAll(this.virtualColumns) + .addAll(additionalVirtualColumns) + .build(); + return new LogicalOlapTableStreamScan(relationId, (Table) table, qualifier, + groupExpression, Optional.of(logicalProperties), + selectedPartitionIds, partitionPruned, hasPartitionPredicate, selectedTabletIds, + selectedIndexId, indexSelected, preAggStatus, manuallySpecifiedPartitions, + hints, cacheSlotWithSlotName, cachedOutput, tableSample, directMvScan, colToSubPathsMap, + manuallySpecifiedTabletIds, operativeSlots, mergedVirtualColumns, scoreOrderKeys, scoreLimit, + scoreRangeInfo, annOrderKeys, annLimit, tableAlias, partitionPrunablePredicates, scanParams, + readMode); + } + /** * withReadMode */ diff --git a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/ExplainTableStreamPlanTest.java b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/ExplainTableStreamPlanTest.java index 8afe8d4dfe7a47..740085852fe1da 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/ExplainTableStreamPlanTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/ExplainTableStreamPlanTest.java @@ -472,6 +472,15 @@ public void testAppendOnlyStreamSnapshotCanBePlanned() throws Exception { assertStreamScanCanBePlanned(ctx, "explain select * from test_stream.s_dup@snapshot()"); } + @Test + public void testAppendOnlyAliasedStreamSnapshotCanBePlanned() throws Exception { + ConnectContext ctx = createDefaultCtx(); + ctx.setDatabase("test_stream"); + ctx.getSessionVariable().showHiddenColumns = true; + + assertStreamScanCanBePlanned(ctx, "explain select s.* from test_stream.s_dup@snapshot() as s"); + } + @Test public void testAppendOnlyStreamResetCanBePlanned() throws Exception { ConnectContext ctx = createDefaultCtx(); @@ -481,6 +490,15 @@ public void testAppendOnlyStreamResetCanBePlanned() throws Exception { assertStreamScanCanBePlanned(ctx, "explain select * from test_stream.s_dup@reset()"); } + @Test + public void testAppendOnlyAliasedStreamResetCanBePlanned() throws Exception { + ConnectContext ctx = createDefaultCtx(); + ctx.setDatabase("test_stream"); + ctx.getSessionVariable().showHiddenColumns = true; + + assertStreamScanCanBePlanned(ctx, "explain select s.* from test_stream.s_dup@reset() as s"); + } + @Test public void testMowTimeTravelQualifiedColumnCanBind() { // MOW time-travel goes through a union whose outputs are rebuilt with empty qualifiers.