Skip to content

[fix](fe) Fix stream scan normalization and snapshot/reset pruning #65468

New issue

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

Sign up for GitHub

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

Already on GitHub? Sign in to your account

Open
yujun777 wants to merge 7 commits into
base: master
Choose a base branch
from
Open

[fix](fe) Fix stream scan normalization and snapshot/reset pruning #65468

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
bc8c6e5
[fix](nereids) Preserve LogicalOlapTableStreamScan type in deep copy …
yujun777 Jul 8, 2026
9bdec4c
[fix](fe) Move stream scan normalization into CTE before-derive pipeline
yujun777 Jul 10, 2026
f42fcb3
add is incremental
yujun777 Jul 10, 2026
44b0ee4
drop temp file
yujun777 Jul 10, 2026
92f9cb8
[test](regression) Add stream change type CTE reuse case
yujun777 Jul 11, 2026
0db627e
[refactor](fe) Unify stream scan read mode handling
yujun777 Jul 11, 2026
0e4329c
[refactor](fe) Reorder stream scan helper methods
yujun777 Jul 11, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 3 additions & 1 deletion fe/fe-core/src/main/java/org/apache/doris/catalog/OlapTable.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -1564,7 +1565,8 @@ public void getVersionInBatchForCloudMode(Collection<Long> partitionIds) throws
// select the non-empty partition ids belonging to this table.
//
// ATTN: partitions not belonging to this table will be filtered.
public List<Long> selectNonEmptyPartitionIds(Collection<Long> partitionIds) {
public List<Long> selectNonEmptyPartitionIds(Collection<Long> partitionIds,
Optional<StreamReadMode> streamReadMode) {
if (Config.isCloudMode() && Config.enable_cloud_snapshot_version) {
// Assumption: all partitions are CloudPartition.
List<CloudPartition> partitions = partitionIds.stream()
Expand Down
7 changes: 5 additions & 2 deletions fe/fe-core/src/main/java/org/apache/doris/catalog/OlapTableWrapper.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,13 +17,15 @@

package org.apache.doris.catalog;

import org.apache.doris.catalog.stream.StreamReadMode;
import org.apache.doris.common.Pair;

import java.util.Collection;
import java.util.HashMap;
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;

Expand Down Expand Up @@ -146,8 +148,9 @@ public Collection<Partition> getPartitions() {
}

@Override
public List<Long> selectNonEmptyPartitionIds(Collection<Long> partitionIds) {
return originTable.selectNonEmptyPartitionIds(partitionIds);
public List<Long> selectNonEmptyPartitionIds(Collection<Long> partitionIds,
Optional<StreamReadMode> streamReadMode) {
return originTable.selectNonEmptyPartitionIds(partitionIds, streamReadMode);
}

@Override
Expand Down
8 changes: 7 additions & 1 deletion fe/fe-core/src/main/java/org/apache/doris/catalog/stream/OlapTableStreamWrapper.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -174,7 +175,12 @@ public Collection<Partition> getPartitions() {
}

@Override
public List<Long> selectNonEmptyPartitionIds(Collection<Long> partitionIds) {
public List<Long> selectNonEmptyPartitionIds(Collection<Long> partitionIds,
Optional<StreamReadMode> streamReadMode) {
StreamReadMode readMode = streamReadMode.orElse(StreamReadMode.INCREMENTAL);
if (readMode == StreamReadMode.SNAPSHOT || readMode == StreamReadMode.RESET) {
return baseTable.selectNonEmptyPartitionIds(partitionIds, Optional.of(readMode));
}
List<Long> nonEmptyIds = Lists.newArrayListWithCapacity(partitionIds.size());
for (Long partitionId : partitionIds) {
if (stream.hasData(getPartition(partitionId))) {
Expand Down
27 changes: 27 additions & 0 deletions fe/fe-core/src/main/java/org/apache/doris/catalog/stream/StreamReadMode.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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
}
12 changes: 3 additions & 9 deletions fe/fe-core/src/main/java/org/apache/doris/nereids/jobs/executor/Rewriter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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()
Expand Down Expand Up @@ -892,14 +894,6 @@ private static List<RewriteJob> getWholeTreeRewriteJobs(
ImmutableSet.of(LogicalCTEAnchor.class),
() -> {
List<RewriteJob> rewriteJobs = Lists.newArrayListWithExpectedSize(300);
if (Config.enable_table_stream) {

@TsukiokaKogane TsukiokaKogane Jul 13, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

maybe move config check inside rule?

@yujun777 yujun777 Jul 13, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

The rule already matches only LogicalOlapTableStreamScan. If such a node exists, it should always be normalized regardless of Config.enable_table_stream. The config controls whether stream tables can be created, but it is not a condition for this normalization rule, so the config check is not needed here.

@yujun777 yujun777 Jul 13, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

Also, a rewrite rule first performs pattern matching and only executes its rewrite function after the pattern matches. If no LogicalOlapTableStreamScan exists, this rule is not matched by any plan node and its rewrite logic is never executed.

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),
Expand Down
7 changes: 4 additions & 3 deletions fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindRelation.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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()) {
Expand Down Expand Up @@ -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);
}
Expand Down
4 changes: 3 additions & 1 deletion ...ore/src/main/java/org/apache/doris/nereids/rules/exploration/mv/PartitionCompensator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -50,6 +50,7 @@
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.Set;

/**
Expand Down Expand Up @@ -176,7 +177,8 @@ private static Pair<Pair<BaseTableInfo, Set<String>>, Pair<BaseColInfo, Set<Stri
Set<String> 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);
}
}
Expand Down
2 changes: 1 addition & 1 deletion fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/PruneEmptyPartition.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -49,7 +49,7 @@ public Rule build() {
LogicalOlapScan scan = ctx.root;
OlapTable table = scan.getTable();
List<Long> partitionIdsToPrune = scan.getSelectedPartitionIds();
List<Long> ids = table.selectNonEmptyPartitionIds(partitionIdsToPrune);
List<Long> 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(),
Expand Down
2 changes: 1 addition & 1 deletion ...c/main/java/org/apache/doris/nereids/trees/plans/commands/UpdateMvByPartitionCommand.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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()));
Expand Down
5 changes: 5 additions & 0 deletions ...re/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalCatalogRelation.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -194,6 +195,10 @@ public String getTableAlias() {
return tableAlias;
}

public Optional<StreamReadMode> getStreamReadMode() {
return Optional.empty();
}

@Override
public void computeUnique(DataTrait.Builder builder) {
Set<Slot> outputSet = Utils.fastToImmutableSet(getOutputSet());
Expand Down
Loading
Loading