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/jobs/executor/Rewriter.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/jobs/executor/Rewriter.java index 4728abaf805181..549d299692e742 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; @@ -408,6 +407,7 @@ public class Rewriter extends AbstractBatchJobExecutor { topDown( new PruneOlapScanPartition(), new PruneEmptyPartition(), + new NormalizeOlapTableStreamScan(), new PruneFileScanPartition(), new PushDownFilterIntoSchemaScan() ) @@ -728,6 +728,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() @@ -892,14 +895,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), 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 f1e20a0dae522a..dd7c06792c63e4 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; @@ -34,6 +35,7 @@ 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.Plan; import org.apache.doris.nereids.trees.plans.PreAggStatus; import org.apache.doris.nereids.trees.plans.RelationId; import org.apache.doris.nereids.trees.plans.ScoreRangeInfo; @@ -41,6 +43,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,13 +53,13 @@ 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; + private final StreamReadMode readMode; /** * LogicalOlapTableStreamScan construct method @@ -64,8 +67,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; } /** @@ -75,8 +77,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; } /** @@ -99,15 +100,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 @@ -120,7 +120,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 @@ -130,7 +130,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(); @@ -158,7 +158,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())); @@ -184,7 +184,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 */ @@ -198,7 +198,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 @@ -211,7 +211,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)); } /** @@ -227,7 +227,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)); } /** @@ -243,7 +243,24 @@ 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)); + } + + /** + * 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)); } /** @@ -268,7 +285,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)); } /** @@ -288,7 +305,39 @@ 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)); + } + + /** + * 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 */ @@ -302,7 +351,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)); } /** @@ -318,28 +367,92 @@ 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 - */ - public LogicalOlapTableStreamScan withIsSnapshot(boolean isSnapshot) { + @Override + public LogicalOlapTableStreamScan withTableAlias(String tableAlias) { return AbstractPlan.copyWithSameId(this, () -> new LogicalOlapTableStreamScan(relationId, (Table) table, qualifier, - groupExpression, Optional.empty(), + 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)); + 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(); + 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)); + } + + @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); } /** - * withIsReset + * withReadMode */ - public LogicalOlapTableStreamScan withIsReset(boolean isReset) { + public LogicalOlapTableStreamScan withReadMode(StreamReadMode readMode) { return AbstractPlan.copyWithSameId(this, () -> new LogicalOlapTableStreamScan(relationId, (Table) table, qualifier, groupExpression, Optional.empty(), @@ -348,7 +461,7 @@ public LogicalOlapTableStreamScan withIsReset(boolean isReset) { hints, cacheSlotWithSlotName, cachedOutput, tableSample, directMvScan, colToSubPathsMap, manuallySpecifiedTabletIds, operativeSlots, virtualColumns, scoreOrderKeys, scoreLimit, scoreRangeInfo, annOrderKeys, annLimit, tableAlias, partitionPrunablePredicates, - scanParams, isReset, isSnapshot)); + scanParams, readMode)); } @Override @@ -368,8 +481,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 @@ -377,6 +489,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() + "]", @@ -388,15 +505,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 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; 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..9feaba171e5231 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 @@ -48,6 +48,7 @@ import org.apache.doris.nereids.trees.expressions.StatementScopeIdGenerator; import org.apache.doris.nereids.trees.expressions.literal.VarcharLiteral; import org.apache.doris.nereids.trees.plans.commands.ExplainCommand; +import org.apache.doris.nereids.trees.plans.logical.LogicalOlapTableStreamScan; import org.apache.doris.nereids.trees.plans.logical.LogicalPlan; import org.apache.doris.nereids.trees.plans.logical.LogicalProject; import org.apache.doris.nereids.trees.plans.physical.PhysicalPlan; @@ -472,6 +473,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 +491,46 @@ 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 testRecordPlanForMvPreRewriteNormalizesStreamScanInsideCte() throws Exception { + ConnectContext ctx = createDefaultCtx(); + ctx.setDatabase("test_stream"); + ctx.getSessionVariable().showHiddenColumns = true; + ctx.getSessionVariable().enableMaterializedViewRewrite = true; + + String sql = "explain with cte as (" + + "select k1, __DORIS_STREAM_CHANGE_TYPE_COL__ as change_type from test_stream.s2" + + ") " + + "select t1.k1 " + + "from (select k1, change_type from cte where change_type in ('APPEND', 'UPDATE_AFTER')) t1 " + + "join (select k1, change_type from cte where change_type in ('DELETE', 'UPDATE_BEFORE')) t2 " + + "on t1.k1 = t2.k1"; + + StatementScopeIdGenerator.clear(); + StatementContext statementContext = MemoTestUtils.createStatementContext(ctx, sql); + statementContext.setForceRecordTmpPlan(true); + NereidsPlanner planner = new NereidsPlanner(statementContext); + LogicalPlan logicalPlan = (LogicalPlan) ((Explainable) (((ExplainCommand) parser.parseSingle(sql)) + .getLogicalPlan())).getExplainPlan(ctx); + planner.planWithLock(logicalPlan, PhysicalProperties.ANY); + + List tmpPlans = planner.getCascadesContext().getStatementContext().getTmpPlanForMvRewrite(); + Assertions.assertFalse(tmpPlans.isEmpty()); + for (Plan tmpPlan : tmpPlans) { + Assertions.assertFalse(containsLogicalStreamScan(tmpPlan), + "tmp plan for mv pre rewrite should have normalized stream scans inside cte children"); + } + } + @Test public void testMowTimeTravelQualifiedColumnCanBind() { // MOW time-travel goes through a union whose outputs are rebuilt with empty qualifiers. @@ -527,6 +577,18 @@ private LogicalProject findFirstLogicalProject(Plan plan) { return null; } + private boolean containsLogicalStreamScan(Plan plan) { + if (plan instanceof LogicalOlapTableStreamScan) { + return true; + } + for (Plan child : plan.children()) { + if (containsLogicalStreamScan(child)) { + return true; + } + } + return false; + } + private void assertStreamScanCanBePlanned(ConnectContext ctx, String sql) { PlanFragment fragment = Assertions.assertDoesNotThrow(() -> getFragment(ctx, sql)); 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,