This is an automated email from the ASF dual-hosted git repository.
englefly pushed a commit to branch topn-rf-opt
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/topn-rf-opt by this push:
new 3d807efe64b [improvement](nereids) Apply conservative source
eligibility to strict TopN filters
3d807efe64b is described below
commit 3d807efe64b2221f1f82e9822f06e05b550d8d51
Author: englefly <[email protected]>
AuthorDate: Thu Nov 12 17:54:23 2026 +0800
[improvement](nereids) Apply conservative source eligibility to strict TopN
filters
### Background
TopN runtime filters use a populated heap's frontier to prune scan input.
The development branch supports strict bounds per target, with eligibility
derived from translated plan nodes.
### What problem does this PR solve?
Issue Number: None
Related PR: None
PR Type: Enhancement
Problem Summary:
Restrict strict bounds to single-key TopN sources whose actual scan targets
and complete paths are all proven safe. Preserve existing inclusive filters
when a source, expression, operator or reader cannot be certified, and show the
selected comparison in EXPLAIN.
### Details
- **Change map:** 284 changed lines in five production Java files and 411
in four test files (three FE classes and one regression suite); no generated or
documentation files. Production code carries eligibility through Nereids,
aggregates target paths and checks final execution metadata. Tests cover
planner/serialization behavior and EXPLAIN. Obsolete translated-plan proof
commentary is removed.
- **Before and after:** Previously translated node classes could authorize
computed probes, external scans, ASOF/null-aware paths and only some targets of
a source. Now immutable pushdown contexts downgrade monotonically, registration
ANDs every actual path, and a source emits either all target IDs or an empty
strict-target list. Explicitly handle the storage-aggregate visitor bypass and
unwrap lazy scans before checking the selected storage model.
- **Configs and defaults:** No new settings. Retain the existing source
phase, type whitelist, filter ratio and closed-filter discovery. Check positive
LIMIT, nonnegative OFFSET and overflow without imposing uniqueness, LIMIT=1 or
OFFSET=0.
- **Impact on existing users:** Safe direct-slot DUP_KEYS/MOW scans can
prune boundary ties. Aggregates, storage merges, computed keys, external
readers and unproved join paths retain inclusive bounds across every target of
their source.
- **Behavior change:** Strict eligibility is source-wide and conservative.
At serialization recheck the final heap algorithm, ordering keys, candidate
count, reader models and every translated target path, including repeated and
disconnected targets.
- **Compatibility:** Reuse strict_target_node_ids without changing Thrift,
BE or persisted data formats. Empty lists retain inclusive bounds; older BEs
ignore the existing optional field.
### Testing
- ./run-fe-ut.sh --run
org.apache.doris.nereids.postprocess.TopNRuntimeFilterTest,org.apache.doris.nereids.postprocess.TopnFilterStrictBoundsTest,org.apache.doris.nereids.postprocess.TopnFilterPushDownVisitorTest:
passed locally; 35 tests, zero failures/errors/skips. Includes complete
planner/translation tests in both directions and direct visitor/final-plan
tests.
- ./build.sh --fe --exclude-obs-dependencies -j4: passed locally, including
full Checkstyle and FE packaging. The default build was stopped at unavailable
Huawei OBS artifact-source verification; the supported exclusion flag omits
that optional module. The temporary ASAN build setting was restored afterward.
- Standalone fe-core Checkstyle passed with zero violations; git diff
--cached --check passed.
- Added regression EXPLAIN checks to topn_runtime_filter_target_mode.
Cluster regression was not run; no golden result files were edited.
### Notes for Reviewers
Self-review checkpoints:
- Goal, scope and reuse: strict qualification refines existing filters;
shared contexts and join/scan helpers cover parallel paths without widening
pushdown.
- Correctness and conditions: preserve grouping contributions, storage
merges, key bindings and LIMIT+OFFSET candidates. Do not alter visibility,
delete bitmaps, error behavior or NULL semantics.
- Concurrency and lifecycle: state remains query-local; branch contexts are
immutable. No new threads, locks, static initialization, resources or
persistence are introduced.
- Configuration, protocols and writes: no new settings, schema fields,
transaction or data-write paths. Compatibility uses the branch's existing
optional target list.
- Coverage and results: source/target conjunction, child order, aliases,
unsafe operators, joins, lazy mappings, selected indexes, final sort metadata
and inclusive fallback have FE coverage. Regression EXPLAIN assertions do not
require changed golden results.
- Observability and performance: EXPLAIN displays <, <=, > or >= from final
eligibility; qualification runs during planning/serialization, with no new
execution-time work or metrics framework.
### Release note
Strict TopN runtime filters now require all targets of a source to follow
proven-safe direct-column OLAP paths. Other paths keep inclusive bounds, and
EXPLAIN shows the chosen comparison.
### Check List (For Author)
- Behavior changed: Yes (see Details)
- Does this need documentation: No
- Pre-submission status: Partially verified: FE tests, Checkstyle and FE
build passed; optional OBS packaging and cluster regression were not verified.
### AI Disclosure
- Tool: Codex (GPT-6)
- Generated: Code, tests and this description
- Human review: Pending
---
.../doris/nereids/processor/post/TopNScanOpt.java | 17 +-
.../nereids/processor/post/TopnFilterContext.java | 8 +-
.../processor/post/TopnFilterPushDownVisitor.java | 97 +++++++-
.../nereids/trees/plans/physical/TopnFilter.java | 151 ++++---------
.../java/org/apache/doris/planner/SortNode.java | 11 +-
.../nereids/postprocess/TopNRuntimeFilterTest.java | 78 ++++++-
.../postprocess/TopnFilterPushDownVisitorTest.java | 245 +++++++++++++++++++++
.../postprocess/TopnFilterStrictBoundsTest.java | 69 ++++--
.../topn_runtime_filter_target_mode.groovy | 19 +-
9 files changed, 554 insertions(+), 141 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/TopNScanOpt.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/TopNScanOpt.java
index 705562a45b3..f954ed2bef7 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/TopNScanOpt.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/TopNScanOpt.java
@@ -19,11 +19,13 @@ package org.apache.doris.nereids.processor.post;
import org.apache.doris.nereids.CascadesContext;
import
org.apache.doris.nereids.processor.post.TopnFilterPushDownVisitor.PushDownContext;
+import org.apache.doris.nereids.trees.expressions.Slot;
import org.apache.doris.nereids.trees.plans.Plan;
import org.apache.doris.nereids.trees.plans.SortPhase;
import org.apache.doris.nereids.trees.plans.algebra.TopN;
import org.apache.doris.nereids.trees.plans.physical.PhysicalTopN;
import org.apache.doris.nereids.types.DataType;
+import org.apache.doris.nereids.util.Utils;
/**
* topN opt
@@ -42,12 +44,23 @@ public class TopNScanOpt extends PlanPostProcessor {
TopnFilterPushDownVisitor pusher = new
TopnFilterPushDownVisitor(ctx.getTopnFilterContext());
TopnFilterPushDownVisitor.PushDownContext pushdownContext = new
PushDownContext(topN,
topN.getOrderKeys().get(0).getExpr(),
- topN.getOrderKeys().get(0).isNullFirst());
+ topN.getOrderKeys().get(0).isNullFirst(),
checkStrictTopN(topN));
topN.accept(pusher, pushdownContext);
}
return topN;
}
+ // Strict eligibility refines the existing filter; multiple keys still get
inclusive bounds.
+ static boolean checkStrictTopN(TopN topN) {
+ return topN instanceof PhysicalTopN
+ && ((PhysicalTopN<?>) topN).getSortPhase() ==
SortPhase.LOCAL_SORT
+ && topN.getOrderKeys().size() == 1
+ && topN.getOrderKeys().get(0).getExpr() instanceof Slot
+ &&
isSupportedTopNRuntimeFilterType(topN.getOrderKeys().get(0).getExpr().getDataType())
+ && topN.getLimit() > 0 && topN.getOffset() >= 0
+ && !Utils.addOverflows(topN.getLimit(), topN.getOffset());
+ }
+
boolean checkTopN(TopN topN) {
if (!(topN instanceof PhysicalTopN)) {
return false;
@@ -65,7 +78,7 @@ public class TopNScanOpt extends PlanPostProcessor {
return isSupportedTopNRuntimeFilterType(firstKeyType);
}
- private boolean isSupportedTopNRuntimeFilterType(DataType dataType) {
+ private static boolean isSupportedTopNRuntimeFilterType(DataType dataType)
{
return dataType.isBooleanType()
|| dataType.isIntegralType()
|| dataType.isDecimalLikeType()
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/TopnFilterContext.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/TopnFilterContext.java
index d52c5b71ac4..cb52caf6dfa 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/TopnFilterContext.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/TopnFilterContext.java
@@ -45,12 +45,12 @@ public class TopnFilterContext {
/**
* add topN filter
*/
- public void addTopnFilter(TopN topn, PhysicalRelation scan, Expression
expr) {
+ public void addTopnFilter(TopN topn, PhysicalRelation scan, Expression
expr, boolean strictEligible) {
TopnFilter filter = filters.get(topn.getObjectId());
if (filter == null) {
- filters.put(topn.getObjectId(), new TopnFilter(topn, scan, expr));
+ filters.put(topn.getObjectId(), new TopnFilter(topn, scan, expr,
strictEligible));
} else {
- filter.addTarget(scan, expr);
+ filter.addTarget(scan, expr, strictEligible);
}
}
@@ -97,7 +97,7 @@ public class TopnFilterContext {
return;
}
filter.legacySortNode = sortNode;
- sortNode.setUseTopnOpt(true);
+ sortNode.setTopnFilter(filter);
Preconditions.checkArgument(!filter.legacyTargets.isEmpty(), "missing
targets: " + filter);
for (ScanNode scan : filter.legacyTargets.keySet()) {
scan.addTopnFilterSortNode(sortNode);
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/TopnFilterPushDownVisitor.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/TopnFilterPushDownVisitor.java
index 772655e50b4..9e3a2fec898 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/TopnFilterPushDownVisitor.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/TopnFilterPushDownVisitor.java
@@ -17,6 +17,9 @@
package org.apache.doris.nereids.processor.post;
+import org.apache.doris.catalog.KeysType;
+import org.apache.doris.catalog.MaterializedIndexMeta;
+import org.apache.doris.catalog.OlapChangelogTableWrapper;
import
org.apache.doris.nereids.processor.post.TopnFilterPushDownVisitor.PushDownContext;
import org.apache.doris.nereids.trees.expressions.Alias;
import org.apache.doris.nereids.trees.expressions.Expression;
@@ -26,22 +29,27 @@ import
org.apache.doris.nereids.trees.expressions.functions.scalar.Nullable;
import org.apache.doris.nereids.trees.expressions.visitor.ExpressionVisitors;
import org.apache.doris.nereids.trees.plans.JoinType;
import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.algebra.SetOperation.Qualifier;
import org.apache.doris.nereids.trees.plans.algebra.TopN;
import org.apache.doris.nereids.trees.plans.algebra.Union;
import org.apache.doris.nereids.trees.plans.physical.AbstractPhysicalJoin;
import org.apache.doris.nereids.trees.plans.physical.PhysicalAssertNumRows;
import org.apache.doris.nereids.trees.plans.physical.PhysicalCTEAnchor;
import org.apache.doris.nereids.trees.plans.physical.PhysicalCTEProducer;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalDistribute;
import org.apache.doris.nereids.trees.plans.physical.PhysicalFileScan;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalFilter;
import org.apache.doris.nereids.trees.plans.physical.PhysicalHashJoin;
import
org.apache.doris.nereids.trees.plans.physical.PhysicalLazyMaterializeOlapScan;
import org.apache.doris.nereids.trees.plans.physical.PhysicalNestedLoopJoin;
import org.apache.doris.nereids.trees.plans.physical.PhysicalOdbcScan;
import org.apache.doris.nereids.trees.plans.physical.PhysicalOlapScan;
import org.apache.doris.nereids.trees.plans.physical.PhysicalProject;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalQuickSort;
import org.apache.doris.nereids.trees.plans.physical.PhysicalRelation;
import org.apache.doris.nereids.trees.plans.physical.PhysicalRepeat;
import org.apache.doris.nereids.trees.plans.physical.PhysicalSetOperation;
+import
org.apache.doris.nereids.trees.plans.physical.PhysicalStorageLayerAggregate;
import org.apache.doris.nereids.trees.plans.physical.PhysicalTopN;
import org.apache.doris.nereids.trees.plans.physical.PhysicalWindow;
import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
@@ -72,30 +80,65 @@ public class TopnFilterPushDownVisitor extends
PlanVisitor<Boolean, PushDownCont
final Expression probeExpr;
final TopN topn;
final boolean nullsFirst;
+ final boolean strictEligible;
public PushDownContext(TopN topn, Expression probeExpr, boolean
nullsFirst) {
+ this(topn, probeExpr, nullsFirst,
TopNScanOpt.checkStrictTopN(topn));
+ }
+
+ public PushDownContext(TopN topn, Expression probeExpr, boolean
nullsFirst, boolean strictEligible) {
this.topn = topn;
this.probeExpr = probeExpr;
this.nullsFirst = nullsFirst;
+ this.strictEligible = strictEligible;
}
public PushDownContext withNewProbeExpression(Expression newProbe) {
- return new PushDownContext(topn, newProbe, nullsFirst);
+ return new PushDownContext(topn, newProbe, nullsFirst,
strictEligible
+ && newProbe instanceof Slot &&
newProbe.getDataType().equals(probeExpr.getDataType()));
}
+ public PushDownContext disableStrict() {
+ return new PushDownContext(topn, probeExpr, nullsFirst, false);
+ }
}
@Override
public Boolean visit(Plan plan, PushDownContext ctx) {
// A bound on a grouping key can exclude whole groups. Stopping at
unlisted operators
// here would disable existing scan filters below aggregates and their
join inputs.
+ // A late strict bound could remove only part of a group and change
its aggregate values.
+ // Unlisted operators therefore preserve closed pushdown without
certifying strict paths.
boolean pushed = false;
+ PushDownContext inclusiveContext = ctx.disableStrict();
for (Plan child : plan.children()) {
- pushed |= child.accept(this, ctx);
+ pushed |= child.accept(this, inclusiveContext);
}
return pushed;
}
+ @Override
+ public Boolean visitPhysicalFilter(PhysicalFilter<? extends Plan> filter,
PushDownContext ctx) {
+ return filter.child().accept(this, ctx);
+ }
+
+ @Override
+ public Boolean visitPhysicalDistribute(PhysicalDistribute<? extends Plan>
distribute, PushDownContext ctx) {
+ return distribute.child().accept(this, ctx);
+ }
+
+ @Override
+ public Boolean visitPhysicalQuickSort(PhysicalQuickSort<? extends Plan>
sort, PushDownContext ctx) {
+ // PhysicalQuickSort is a full sort; truncation and OFFSET are
represented by other nodes.
+ return sort.child().accept(this, ctx);
+ }
+
+ @Override
+ public Boolean
visitPhysicalStorageLayerAggregate(PhysicalStorageLayerAggregate aggregate,
PushDownContext ctx) {
+ // The base visitor unwraps this relation directly, bypassing
visit(Plan, ...).
+ return aggregate.getRelation().accept(this, ctx.disableStrict());
+ }
+
@Override
public Boolean visitPhysicalProject(
PhysicalProject<? extends Plan> project, PushDownContext ctx) {
@@ -118,6 +161,9 @@ public class TopnFilterPushDownVisitor extends
PlanVisitor<Boolean, PushDownCont
@Override
public Boolean visitPhysicalSetOperation(
PhysicalSetOperation setOperation, PushDownContext ctx) {
+ if (!(setOperation instanceof Union) || setOperation.getQualifier() !=
Qualifier.ALL) {
+ ctx = ctx.disableStrict();
+ }
boolean pushedDown = false;
if (setOperation.arity() > 0) {
pushedDown = pushDownFilterToSetOperatorChild(setOperation, ctx,
0);
@@ -233,6 +279,26 @@ public class TopnFilterPushDownVisitor extends
PlanVisitor<Boolean, PushDownCont
return true;
}
+ private PushDownContext strictJoinContext(AbstractPhysicalJoin<? extends
Plan, ? extends Plan> join,
+ PushDownContext ctx) {
+ if (join.isMarkJoin() || !(ctx.probeExpr instanceof Slot)
+ || !join.left().getOutputSet().contains(ctx.probeExpr)) {
+ return ctx.disableStrict();
+ }
+ // Existing pushdown only targets the left input. Equality-key
substitution from the right
+ // and all ASOF, null-aware, mark and non-preserved-side paths remain
inclusive.
+ switch (join.getJoinType()) {
+ case INNER_JOIN:
+ case CROSS_JOIN:
+ case LEFT_OUTER_JOIN:
+ case LEFT_SEMI_JOIN:
+ case LEFT_ANTI_JOIN:
+ return ctx;
+ default:
+ return ctx.disableStrict();
+ }
+ }
+
@Override
public Boolean visitPhysicalHashJoin(PhysicalHashJoin<? extends Plan, ?
extends Plan> join,
PushDownContext ctx) {
@@ -246,6 +312,7 @@ public class TopnFilterPushDownVisitor extends
PlanVisitor<Boolean, PushDownCont
if (!canPushDownToLeftChild(join, ctx.probeExpr)) {
return false;
}
+ ctx = strictJoinContext(join, ctx);
if
(join.left().getOutputSet().containsAll(ctx.probeExpr.getInputSlots())) {
PushDownContext childPushDownContext =
adjustProbeExprNullableThroughOuterJoin(join.left(), ctx);
return join.left().accept(this, childPushDownContext);
@@ -282,7 +349,7 @@ public class TopnFilterPushDownVisitor extends
PlanVisitor<Boolean, PushDownCont
// push to left child. right child is blocking operator, do not need
topn-filter
if
(join.left().getOutputSet().containsAll(ctx.probeExpr.getInputSlots())
&& canPushDownToLeftChild(join, ctx.probeExpr)) {
- return join.left().accept(this, ctx);
+ return join.left().accept(this, strictJoinContext(join, ctx));
}
return false;
}
@@ -301,13 +368,35 @@ public class TopnFilterPushDownVisitor extends
PlanVisitor<Boolean, PushDownCont
|| Math.max(relation.getStats().getRowCount(), 1)
*
ConnectContext.get().getSessionVariable().topnFilterRatio > ctx.topn.getLimit()
+ ctx.topn.getOffset()) {
- topnFilterContext.addTopnFilter(ctx.topn, relation,
ctx.probeExpr);
+ topnFilterContext.addTopnFilter(ctx.topn, relation,
ctx.probeExpr,
+ ctx.strictEligible && ctx.probeExpr instanceof Slot &&
allowsStrictScan(relation));
return true;
}
}
return false;
}
+ private boolean allowsStrictScan(PhysicalRelation relation) {
+ if (relation instanceof PhysicalLazyMaterializeOlapScan) {
+ relation = ((PhysicalLazyMaterializeOlapScan) relation).getScan();
+ }
+ if (!(relation instanceof PhysicalOlapScan)) {
+ return false;
+ }
+ PhysicalOlapScan scan = (PhysicalOlapScan) relation;
+ if (scan.getTable() instanceof OlapChangelogTableWrapper ||
scan.getScanParams().isPresent()
+ || scan.getTableSample().isPresent() ||
scan.getAnnLimit().isPresent()
+ || scan.getScoreLimit().isPresent()) {
+ return false;
+ }
+ // The selected rollup can aggregate even when the base table uses
DUP_KEYS.
+ MaterializedIndexMeta index =
scan.getTable().getIndexMetaByIndexId(scan.getSelectedIndexId());
+ KeysType keysType = index.getKeysType();
+ return !index.isChangelogIndex()
+ && (keysType == KeysType.DUP_KEYS || (keysType ==
KeysType.UNIQUE_KEYS
+ && scan.getTable().getEnableUniqueKeyMergeOnWrite()));
+ }
+
@Override
public Boolean
visitPhysicalLazyMaterializeOlapScan(PhysicalLazyMaterializeOlapScan lazyScan,
PushDownContext ctx) {
return visitPhysicalRelation(lazyScan, ctx);
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/TopnFilter.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/TopnFilter.java
index 27ad9eb0b6b..d9285f1df81 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/TopnFilter.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/TopnFilter.java
@@ -20,28 +20,28 @@ package org.apache.doris.nereids.trees.plans.physical;
import org.apache.doris.analysis.Expr;
import org.apache.doris.analysis.ExprToThriftVisitor;
import org.apache.doris.catalog.KeysType;
-import org.apache.doris.datasource.scan.FileScanNode;
import org.apache.doris.nereids.trees.expressions.Expression;
import org.apache.doris.nereids.trees.plans.algebra.TopN;
+import org.apache.doris.nereids.util.Utils;
import org.apache.doris.planner.ExchangeNode;
import org.apache.doris.planner.JoinNodeBase;
import org.apache.doris.planner.LocalExchangeNode;
import org.apache.doris.planner.MaterializationNode;
import org.apache.doris.planner.OlapScanNode;
import org.apache.doris.planner.PlanNode;
+import org.apache.doris.planner.RepeatNode;
import org.apache.doris.planner.ScanNode;
import org.apache.doris.planner.SelectNode;
import org.apache.doris.planner.SetOperationNode;
import org.apache.doris.planner.SortNode;
-import org.apache.doris.planner.TableFunctionNode;
import org.apache.doris.thrift.TSortAlgorithm;
import org.apache.doris.thrift.TTopnFilterDesc;
import com.google.common.collect.Maps;
import java.util.ArrayList;
-import java.util.List;
import java.util.Map;
+import java.util.stream.Collectors;
/**
* topn filter
@@ -52,15 +52,25 @@ public class TopnFilter {
public Map<PhysicalRelation, Expression> targets = Maps.newHashMap();
public Map<ScanNode, Expr> legacyTargets = Maps.newHashMap();
- public TopnFilter(TopN topn, PhysicalRelation rel, Expression expr) {
+ private boolean strictEligible;
+
+ public TopnFilter(TopN topn, PhysicalRelation rel, Expression expr,
boolean strictEligible) {
this.topn = topn;
+ this.strictEligible = strictEligible;
targets.put(rel, expr);
}
- public void addTarget(PhysicalRelation rel, Expression expr) {
+ public void addTarget(PhysicalRelation rel, Expression expr, boolean
strictEligible) {
+ // AND every registered path, including repeated registrations of the
same scan.
+ this.strictEligible &= strictEligible;
targets.put(rel, expr);
}
+ /** Eligibility accumulated during pushdown; translated execution metadata
is checked separately. */
+ public boolean isStrictEligible() {
+ return strictEligible;
+ }
+
public boolean hasTargetRelation(PhysicalRelation rel) {
return targets.containsKey(rel);
}
@@ -68,7 +78,7 @@ public class TopnFilter {
@Override
public String toString() {
StringBuilder builder = new StringBuilder();
- builder.append(topn).append("->[ ");
+ builder.append(topn).append("
strictEligible=").append(strictEligible).append("->[ ");
for (PhysicalRelation rel : targets.keySet()) {
builder.append("(").append(rel).append(":").append(targets.get(rel)).append(")
");
}
@@ -88,74 +98,48 @@ public class TopnFilter {
tFilter.putToTargetNodeIdToTargetExpr(scan.getId().asInt(),
ExprToThriftVisitor.treeToThrift(legacyTargets.get(scan)));
}
- List<Integer> strictTargets = new ArrayList<>();
- // Scheduling can change the sort algorithm after translation. Inspect
the final
- // execution plan here, after
ThriftPlansBuilder.setRuntimePredicateIfNeed().
- // Multiple keys must keep ties: (1, 2) can still beat (1, 9) with the
same leading key.
- if (legacySortNode.getSortAlgorithm() == TSortAlgorithm.HEAP_SORT
- && legacySortNode.getSortInfo().getOrderingExprs().size() ==
1) {
- for (PlanNode child : legacySortNode.getChildren()) {
- collectStrictTargets(child, strictTargets);
- }
- }
- tFilter.setStrictTargetNodeIds(strictTargets);
+ // Reuse the branch's per-target wire format, but upgrade a source
only as a whole.
+ tFilter.setStrictTargetNodeIds(allowsStrictBounds()
+ ? legacyTargets.keySet().stream().map(scan ->
scan.getId().asInt())
+ .collect(Collectors.toList()) : new ArrayList<>());
return tFilter;
}
- private void collectStrictTargets(PlanNode node, List<Integer>
strictTargets) {
- if (node.hasLimit() || !allowsStrictBounds(node)) {
- // Removing a tie below an intermediate LIMIT can let a different
row pass it.
- return;
+ /** Check the final execution plan, since scheduling can change the sort
algorithm and limits. */
+ public boolean allowsStrictBounds() {
+ if (!strictEligible || legacyTargets.isEmpty()
+ || legacySortNode.getSortAlgorithm() !=
TSortAlgorithm.HEAP_SORT
+ || legacySortNode.getSortInfo().getOrderingExprs().size() != 1
+ || legacySortNode.getLimit() <= 0 ||
legacySortNode.getOffset() < 0
+ || Utils.addOverflows(legacySortNode.getLimit(),
legacySortNode.getOffset())) {
+ return false;
}
+ Map<ScanNode, Boolean> strictTargets = Maps.newHashMap();
+ for (PlanNode child : legacySortNode.getChildren()) {
+ collectStrictTargets(child, true, strictTargets);
+ }
+ return strictTargets.size() == legacyTargets.size()
+ && strictTargets.values().stream().allMatch(allowed ->
allowed);
+ }
+
+ private void collectStrictTargets(PlanNode node, boolean pathStrict,
Map<ScanNode, Boolean> strictTargets) {
+ // Removing a tie below an intermediate LIMIT/OFFSET can let a
different row pass it.
+ pathStrict &= !node.hasLimit() && node.getOffset() == 0 &&
allowsStrictNode(node);
if (legacyTargets.containsKey(node)) {
- strictTargets.add(node.getId().asInt());
+ strictTargets.merge((ScanNode) node, pathStrict, (previous,
current) -> previous && current);
}
// FE retains the sending fragment below ExchangeNode, unlike the
serialized BE tree.
for (PlanNode child : node.getChildren()) {
- collectStrictTargets(child, strictTargets);
+ collectStrictTargets(child, pathStrict, strictTargets);
}
}
- /**
- * Whether this node preserves the safety of discarding boundary ties at a
descendant scan.
- *
- * <p>This refines an already accepted TopN filter; it does not authorize
new predicate pushdown.
- * The caller checks the final single-key HEAP_SORT algorithm and rejects
intermediate LIMITs.
- * Once the source heap holds LIMIT + OFFSET candidates, another equal key
cannot improve that
- * ordering. However, dropping an input tie must not change a surviving
output row, create a new
- * output row, suppress an error, or remove input needed to produce a
better-ranked row later.
- * This property must hold at every node between the source and the target.
- *
- * <p>The examples below explain local operator semantics, not necessarily
plans that current
- * pushdown rules can produce: some shapes are already rejected upstream.
Unless stated otherwise,
- * use ascending order with boundary b = 1, so the additional rows
rejected by {@code k < b}
- * instead of {@code k <= b} have k = 1. A bound may arrive after some
input has been consumed,
- * for example from a safe sibling of a UNION. Thus a merge can lose only
part of a boundary group.
- * Descending order has the same issue with the comparison reversed.
- *
- * <p>This is a whitelist: false means that the path retains its inclusive
filter, not that every
- * instance of the operator is unsafe. New operator/model variants need
their own proof before
- * being admitted, even if a particular example would happen to produce
the same result.
- */
- private boolean allowsStrictBounds(PlanNode node) {
+ // Nereids proves expression bindings and complete path semantics before
registration.
+ // Recheck translated readers and limits, which can change after that
proof.
+ private boolean allowsStrictNode(PlanNode node) {
if (node instanceof OlapScanNode) {
OlapScanNode scan = (OlapScanNode) node;
KeysType keysType = scan.getSelectedIndexKeyType();
- // Use the selected index's model: a DUP_KEYS base table may have
an AGG_KEYS rollup.
- // Rejected storage cases, even when the SQL plan contains no
AggregationNode:
- // - Changelog: +(k=1,v=10), -(k=1,v=10) should cancel when
folded. If the insert was
- // consumed before the bound and the delete is filtered
afterwards, a spurious insert
- // survives. All changelog scan modes retain inclusive bounds;
this rule does not try
- // to prove individual modes such as DETAIL independent of
storage folding.
- // - AGG_KEYS: stored SUM contributions (k=1,v=10), (k=1,v=20)
must produce (1,30).
- // Reading 10 before the bound and dropping 20 afterwards would
instead produce (1,10).
- // - UNIQUE_KEYS without MOW (MOR): old (k=1,v=10), new (k=1,v=20)
must resolve to (1,20).
- // Dropping the new version after the old one was read can
expose the obsolete (1,10).
- // - PRIMARY_KEYS/UNKNOWN, or a future model: even a scan
containing just (k=1,v=10)
- // remains inclusive. These labels do not establish the
DUP_KEYS/MOW guarantee of
- // finalized, independently discardable rows in this whitelist;
this is conservative.
- // Ordinary DUP_KEYS rows and resolved MOW rows need no such
read-time merge of versions
- // or contributions, so their boundary ties can be discarded when
the rest of the path allows it.
return !scan.isReadChangelog() && (keysType == KeysType.DUP_KEYS
|| (keysType == KeysType.UNIQUE_KEYS &&
scan.getOlapTable().getEnableUniqueKeyMergeOnWrite()));
}
@@ -166,55 +150,14 @@ public class TopnFilter {
case LEFT_OUTER_JOIN:
case LEFT_SEMI_JOIN:
case LEFT_ANTI_JOIN:
- case NULL_AWARE_LEFT_ANTI_JOIN:
- case NULL_AWARE_LEFT_SEMI_JOIN:
- case ASOF_LEFT_INNER_JOIN:
- case ASOF_LEFT_OUTER_JOIN:
return true;
default:
- // The registered scan filter is pushed into the
left/probe side. For equality
- // joins, take L={(k=1,v=10)}, R={(k=1)} and remove the
sole matching L row:
- // - RIGHT_OUTER_JOIN: the result changes from
(L.v=10,R.k=1) to (NULL,1).
- // - FULL_OUTER_JOIN: the same example creates the
previously absent (NULL,1).
- // - RIGHT_ANTI_JOIN: the empty result becomes R.k=1,
since R is now unmatched.
- // - RIGHT_SEMI_JOIN: plain semi output {R.k=1} becomes
empty; with a mark it
- // changes (R.k=1,mark=true) to (1,false). The mark case
changes a surviving row.
- // Plain right-semi is also excluded conservatively; no
separate proof for its
- // build-side output is made here. This does not claim
all plain semi cases fail.
- // ASOF examples use the nearest L.t <= R.t, with R.t=2:
- // - ASOF_RIGHT_INNER_JOIN: L.t={1,2} originally selects
2. Dropping L.t=2
- // (boundary b=2 on L.t) selects 1 instead, creating a
different joined row.
- // - ASOF_RIGHT_OUTER_JOIN: with only L.t=2, dropping that
boundary row changes
- // (L.t=2,R.t=2) to (NULL,2). With L.t={1,2}, it can
also switch matches as above.
- // Future join types stay excluded until their
probe-removal semantics are proved.
return false;
}
}
- // Rejected operator families (all fall through the whitelist below):
- // - AggregationNode/BucketedAggregationNode: GROUP BY k with SUM(v)
over (1,10), (1,20)
- // must return (1,30), not (1,10) if the second contribution is
dropped by a late bound.
- // A scalar COUNT(*) is also sensitive: removing its two inputs
changes the result 2 to 0.
- // - RepeatNode: for grouping sets ((k),()), input k=1 expands to k=1
and k=NULL. Dropping
- // that input also removes the NULL copy, which could rank before
the boundary with NULLS FIRST.
- // - AnalyticEvalNode: SUM(v) OVER () on (k,v)={(1,10),(0,20)} gives
the k=0 row a sum of 30.
- // Dropping k=1 changes that surviving, better-ranked row's sum to
20.
- // - AssertNumRowsNode: an at-most-one-row assertion must fail on
input k={1,1}. Filtering
- // both rows hides the error; scalar-subquery handling may also
synthesize a new NULL row.
- // - PartitionSortNode: in one partition, ORDER BY t with partition
limit 1 chooses (k=1,t=0)
- // over (k=0,t=1). Filtering k=1 first makes the previously
unselected k=0 row survive.
- // - RecursiveCteNode/RecursiveCteScanNode: seed k=1 followed by k-1
while k>0 produces {1,0}.
- // Dropping the boundary seed loses the later k=0 row, which would
beat the bound.
- // - CTEScanNode/shared CTE paths need consumer-specific proof. For
CTE input {k=1}, pushing
- // one consumer's strict bound into the shared producer could change
another's COUNT(*) from 1 to 0.
- // - Other unlisted nodes, including unsupported scans and future
operators, have no proof here.
- // For example, TopN -> SchemaScanNode is not an accepted target
path; EmptySetNode has no rows
- // or useful scan target to mark. Neither needs a strict mode. A new
operator could instead
- // behave like the window example above, so it must not be accepted
merely for having children.
- // The admitted nodes preserve the established pushdown semantics
without requiring additional
- // boundary contributions. Set-operation targets and join probe sides
were selected upstream.
- return node instanceof FileScanNode || node instanceof SelectNode ||
node instanceof SortNode
+ // Repeat is admitted only after the common grouping-key check in
Nereids.
+ return node instanceof SelectNode || node instanceof SortNode || node
instanceof RepeatNode
|| node instanceof ExchangeNode || node instanceof
LocalExchangeNode
- || node instanceof SetOperationNode || node instanceof
TableFunctionNode
- || node instanceof MaterializationNode;
+ || node instanceof SetOperationNode || node instanceof
MaterializationNode;
}
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/SortNode.java
b/fe/fe-core/src/main/java/org/apache/doris/planner/SortNode.java
index ae094e88db8..40bb0b1fd03 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/SortNode.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/planner/SortNode.java
@@ -26,6 +26,7 @@ import org.apache.doris.analysis.SortInfo;
import org.apache.doris.analysis.ToSqlParams;
import org.apache.doris.common.Pair;
import org.apache.doris.nereids.glue.translator.PlanTranslatorContext;
+import org.apache.doris.nereids.trees.plans.physical.TopnFilter;
import org.apache.doris.planner.LocalExchangeNode.LocalExchangeType;
import org.apache.doris.planner.LocalExchangeNode.LocalExchangeTypeRequire;
import org.apache.doris.qe.ConnectContext;
@@ -52,6 +53,7 @@ public class SortNode extends PlanNode {
private final SortInfo info;
private final boolean useTopN;
private boolean useTopnOpt = false;
+ private TopnFilter topnFilter;
private boolean hasRuntimePredicate = false;
// If mergeByexchange is set to true, the sort information is pushed to the
@@ -142,8 +144,9 @@ public class SortNode extends PlanNode {
}
}
- public void setUseTopnOpt(boolean useTopnOpt) {
- this.useTopnOpt = useTopnOpt;
+ public void setTopnFilter(TopnFilter topnFilter) {
+ this.topnFilter = topnFilter;
+ this.useTopnOpt = true;
}
@Override
@@ -170,6 +173,10 @@ public class SortNode extends PlanNode {
if (useTopnOpt) {
output.append(detailPrefix + "TOPN filter targets:
").append(topnFilterTargets).append("\n");
+ boolean strict = topnFilter.allowsStrictBounds();
+ output.append(detailPrefix).append("TOPN filter boundary: ")
+ .append(info.getIsAscOrder().get(0) ? (strict ? "<" :
"<=") : (strict ? ">" : ">="))
+ .append("\n");
}
output.append(detailPrefix + "algorithm: ");
if (algorithm == TSortAlgorithm.HEAP_SORT) {
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/nereids/postprocess/TopNRuntimeFilterTest.java
b/fe/fe-core/src/test/java/org/apache/doris/nereids/postprocess/TopNRuntimeFilterTest.java
index f459b25e999..4d9b4ec6710 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/nereids/postprocess/TopNRuntimeFilterTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/nereids/postprocess/TopNRuntimeFilterTest.java
@@ -52,7 +52,6 @@ import
org.apache.doris.nereids.trees.plans.physical.TopnFilter;
import org.apache.doris.nereids.util.MemoPatternMatchSupported;
import org.apache.doris.nereids.util.PlanChecker;
import org.apache.doris.nereids.util.PlanConstructor;
-import org.apache.doris.planner.OlapScanNode;
import org.apache.doris.planner.ScanNode;
import org.apache.doris.thrift.TExplainLevel;
import org.apache.doris.thrift.TSortAlgorithm;
@@ -142,12 +141,9 @@ public class TopNRuntimeFilterTest extends SSBTestBase
implements MemoPatternMat
.translatePlan(plan);
TTopnFilterDesc desc = filter.toThrift();
Assertions.assertEquals(2,
desc.getTargetNodeIdToTargetExprSize());
- for (ScanNode target : filter.legacyTargets.keySet()) {
- String tableName = ((OlapScanNode)
target).getOlapTable().getName();
- boolean strict =
tableName.equals("topn_storage_dup") || source.equals("topn_storage_mow");
- Assertions.assertEquals(strict,
desc.getStrictTargetNodeIds().contains(target.getId().asInt()),
- sql + "\n" + plan.treeString());
- }
+ boolean strict = source.equals("topn_storage_mow") &&
key.equals("k");
+ Assertions.assertEquals(strict ? 2 : 0,
desc.getStrictTargetNodeIdsSize(),
+ sql + "\n" + plan.treeString());
}
}
}
@@ -169,7 +165,8 @@ public class TopNRuntimeFilterTest extends SSBTestBase
implements MemoPatternMat
.analyze("SELECT k,v FROM topn_storage_dup ORDER BY k
LIMIT 1").rewrite().implement();
PhysicalPlan plan = checker.getPhysicalPlan();
PhysicalOlapScan scan =
plan.<PhysicalOlapScan>collectToList(PhysicalOlapScan.class::isInstance).get(0);
- PhysicalTopN<?> topn =
plan.<PhysicalTopN<?>>collectToList(PhysicalTopN.class::isInstance).get(0);
+ PhysicalTopN<?> topn = plan.<PhysicalTopN<?>>collectToList(node ->
node instanceof PhysicalTopN
+ && ((PhysicalTopN<?>) node).getSortPhase() ==
SortPhase.LOCAL_SORT).get(0);
OlapTable table = PlanConstructor.newOlapTable(10001,
"topn_index_model", 0, KeysType.DUP_KEYS);
table.setIndexMeta(10002, "aggregate_index",
table.getFullSchema(), 0, 0, (short) 0,
TStorageType.COLUMN, KeysType.AGG_KEYS);
@@ -184,6 +181,8 @@ public class TopNRuntimeFilterTest extends SSBTestBase
implements MemoPatternMat
new PushDownContext(topn, scan.getOutput().get(0),
false));
Assertions.assertTrue(accepted);
Assertions.assertEquals(1,
context.getTopnFilters().size());
+ Assertions.assertEquals(indexId == table.getBaseIndexId(),
+
context.getTopnFilters().get(0).isStrictEligible());
}
}
} finally {
@@ -232,6 +231,55 @@ public class TopNRuntimeFilterTest extends SSBTestBase
implements MemoPatternMat
}
}
+ @Test
+ public void testStrictBoundaryExplainAndMixedUnionPaths() {
+ double previousRatio =
connectContext.getSessionVariable().topnFilterRatio;
+ String previousAlgorithm =
connectContext.getSessionVariable().forceSortAlgorithm;
+ boolean previousPushTopn =
connectContext.getSessionVariable().pushTopnToAgg;
+ String previousRules = String.join(",",
connectContext.getSessionVariable().getDisableNereidsRuleNames());
+ connectContext.getSessionVariable().topnFilterRatio = 1000000;
+ connectContext.getSessionVariable().forceSortAlgorithm = "heap";
+ connectContext.getSessionVariable().pushTopnToAgg = false;
+
connectContext.getSessionVariable().setDisableNereidsRules(previousRules
+ + ",PRUNE_EMPTY_PARTITION,PUSH_DOWN_TOP_N_THROUGH_UNION");
+ try {
+ String scan = "SELECT k,ts AS v FROM topn_safety";
+ String aggregate = "SELECT k,SUM(ts) AS v FROM topn_safety GROUP
BY k HAVING SUM(ts)>0";
+ Map<String, Boolean> queries = Map.of(
+ "SELECT k AS alias_key,ts FROM topn_safety WHERE ts=999
ORDER BY alias_key %s LIMIT 1 OFFSET 1", true,
+ "SELECT k,ts FROM topn_safety ORDER BY ABS(k) %s LIMIT 1",
false,
+ "SELECT k,ts FROM topn_safety ORDER BY k %s,ts LIMIT 1",
false,
+ "SELECT k,v FROM (" + scan + " UNION ALL " + scan + ") u
ORDER BY k %s LIMIT 1", true,
+ "SELECT k,v FROM (" + scan + " UNION ALL " + aggregate +
") u ORDER BY k %s LIMIT 1", false,
+ "SELECT k,v FROM (" + aggregate + " UNION ALL " + scan +
") u ORDER BY k %s LIMIT 1", false);
+ for (Map.Entry<String, Boolean> query : queries.entrySet()) {
+ for (String direction : List.of("ASC", "DESC")) {
+
PlanChecker.from(connectContext).checkPlannerResult(String.format(query.getKey(),
direction),
+ planner -> {
+ String explain =
planner.getFragments().stream()
+ .map(fragment ->
fragment.getExplainString(TExplainLevel.NORMAL))
+ .collect(Collectors.joining("\n"));
+ Assertions.assertTrue(explain.contains("TOPN
OPT:"), explain);
+ String boundary = direction.equals("ASC") ?
"<" : ">";
+ boundary += query.getValue() ? "" : "=";
+ Assertions.assertTrue(explain.contains("TOPN
filter boundary: " + boundary + "\n"),
+ explain);
+ for (TopnFilter filter :
planner.getTopnFilters()) {
+ TTopnFilterDesc desc = filter.toThrift();
+ Assertions.assertEquals(query.getValue() ?
filter.legacyTargets.size() : 0,
+ desc.getStrictTargetNodeIdsSize(),
explain);
+ }
+ });
+ }
+ }
+ } finally {
+ connectContext.getSessionVariable().topnFilterRatio =
previousRatio;
+ connectContext.getSessionVariable().forceSortAlgorithm =
previousAlgorithm;
+ connectContext.getSessionVariable().pushTopnToAgg =
previousPushTopn;
+
connectContext.getSessionVariable().setDisableNereidsRules(previousRules);
+ }
+ }
+
@Test
public void testTopNFilterDoesNotRequireTotalSecondaryExpressions() {
double previousRatio =
connectContext.getSessionVariable().topnFilterRatio;
@@ -420,7 +468,7 @@ public class TopNRuntimeFilterTest extends SSBTestBase
implements MemoPatternMat
.collect(Collectors.toList());
Assertions.assertEquals(1, outerFilters.size(),
plan.treeString());
// An inner TopN remains a barrier for the outer
source. Otherwise BE
- // keeps equality on the aggregate target and may
prune ties on its sibling.
+ // keeps equality on every target of the source.
int expectedTargets = innerOrder.isEmpty() ? 2 : 1;
Assertions.assertEquals(expectedTargets,
outerFilters.get(0).targets.size(), plan.treeString());
Assertions.assertEquals(innerOrder.isEmpty(),
outerFilters.get(0).targets.keySet().stream()
@@ -436,6 +484,7 @@ public class TopNRuntimeFilterTest extends SSBTestBase
implements MemoPatternMat
.translatePlan(plan);
Assertions.assertEquals(expectedTargets,
outerFilters.get(0).legacyTargets.size());
Assertions.assertFalse(outerFilters.get(0).toThrift().getTargetNodeIdToTargetExpr().isEmpty());
+
Assertions.assertTrue(outerFilters.get(0).toThrift().getStrictTargetNodeIds().isEmpty());
}
}
}
@@ -534,7 +583,7 @@ public class TopNRuntimeFilterTest extends SSBTestBase
implements MemoPatternMat
Expression probeExpr = filter.targets.get(lazyScan);
TopnFilterContext topnFilterContext = new TopnFilterContext();
- topnFilterContext.addTopnFilter(filter.topn, lazyScan.getScan(),
probeExpr);
+ topnFilterContext.addTopnFilter(filter.topn, lazyScan.getScan(),
probeExpr, false);
PlanTranslatorContext translatorContext = new PlanTranslatorContext();
probeExpr.getInputSlots().forEach(slot -> translatorContext
.createSlotDesc(translatorContext.generateTupleDesc(),
(SlotReference) slot));
@@ -542,6 +591,10 @@ public class TopNRuntimeFilterTest extends SSBTestBase
implements MemoPatternMat
topnFilterContext.translateTarget(lazyScan, legacyScan,
translatorContext);
Assertions.assertTrue(topnFilterContext.getTopnFilter(filter.topn).legacyTargets.containsKey(legacyScan));
+ topnFilterContext.addTopnFilter(filter.topn, lazyScan, probeExpr,
true);
+ topnFilterContext.translateTarget(lazyScan, legacyScan,
translatorContext);
+
Assertions.assertFalse(topnFilterContext.getTopnFilter(filter.topn).isStrictEligible());
+ Assertions.assertEquals(1,
topnFilterContext.getTopnFilter(filter.topn).legacyTargets.size());
}
@Test
@@ -647,8 +700,13 @@ public class TopNRuntimeFilterTest extends SSBTestBase
implements MemoPatternMat
Assertions.assertTrue(sort.isSetAlgorithm());
Assertions.assertEquals(expectedAlgorithm, sort.getAlgorithm(),
plan.treeString());
Assertions.assertEquals(expectedKeyCount,
sort.getSortInfo().getOrderingExprsSize());
+ // The local heap must retain the final LIMIT + OFFSET candidates
(3 + 2).
+ Assertions.assertEquals(5L, filter.legacySortNode.getLimit() +
filter.legacySortNode.getOffset());
TTopnFilterDesc desc = filter.toThrift();
Assertions.assertTrue(desc.isSetStrictTargetNodeIds());
+ boolean strict = expectedAlgorithm == TSortAlgorithm.HEAP_SORT &&
expectedKeyCount == 1;
+
Assertions.assertTrue(filter.legacySortNode.getNodeExplainString("",
TExplainLevel.NORMAL)
+ .contains("TOPN filter boundary: " + (strict ? "<\n" :
"<=\n")));
Assertions.assertEquals(expectedAlgorithm ==
TSortAlgorithm.HEAP_SORT && expectedKeyCount == 1
? desc.getTargetNodeIdToTargetExprSize() : 0,
desc.getStrictTargetNodeIdsSize());
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/nereids/postprocess/TopnFilterPushDownVisitorTest.java
b/fe/fe-core/src/test/java/org/apache/doris/nereids/postprocess/TopnFilterPushDownVisitorTest.java
new file mode 100644
index 00000000000..77f11a1f9e3
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/nereids/postprocess/TopnFilterPushDownVisitorTest.java
@@ -0,0 +1,245 @@
+// 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.postprocess;
+
+import org.apache.doris.catalog.OlapTable;
+import org.apache.doris.nereids.processor.post.TopnFilterContext;
+import org.apache.doris.nereids.processor.post.TopnFilterPushDownVisitor;
+import
org.apache.doris.nereids.processor.post.TopnFilterPushDownVisitor.PushDownContext;
+import org.apache.doris.nereids.properties.DataTrait;
+import org.apache.doris.nereids.properties.DistributionSpecGather;
+import org.apache.doris.nereids.properties.LogicalProperties;
+import org.apache.doris.nereids.properties.OrderKey;
+import org.apache.doris.nereids.trees.expressions.Add;
+import org.apache.doris.nereids.trees.expressions.Alias;
+import org.apache.doris.nereids.trees.expressions.EqualTo;
+import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.expressions.SlotReference;
+import org.apache.doris.nereids.trees.expressions.literal.IntegerLiteral;
+import org.apache.doris.nereids.trees.plans.JoinType;
+import org.apache.doris.nereids.trees.plans.LimitPhase;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.PreAggStatus;
+import org.apache.doris.nereids.trees.plans.SortPhase;
+import org.apache.doris.nereids.trees.plans.algebra.SetOperation.Qualifier;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalAssertNumRows;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalCTEAnchor;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalCTEProducer;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalDistribute;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalFilter;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalHashJoin;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalLimit;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalNestedLoopJoin;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalOlapScan;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalProject;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalQuickSort;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalRepeat;
+import
org.apache.doris.nereids.trees.plans.physical.PhysicalStorageLayerAggregate;
+import
org.apache.doris.nereids.trees.plans.physical.PhysicalStorageLayerAggregate.PushDownAggOp;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalTopN;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalUnion;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalWindow;
+import org.apache.doris.nereids.types.IntegerType;
+import org.apache.doris.nereids.util.PlanConstructor;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import java.util.BitSet;
+import java.util.List;
+import java.util.Optional;
+import java.util.Set;
+
+public class TopnFilterPushDownVisitorTest {
+ private final SlotReference key = new SlotReference("k",
IntegerType.INSTANCE);
+
+ private PhysicalOlapScan scan(SlotReference slot) {
+ OlapTable table = PlanConstructor.newOlapTable(10001, "strict_scan",
0);
+ LogicalProperties properties = new LogicalProperties(() ->
List.of(slot), () -> DataTrait.EMPTY_TRAIT);
+ return new PhysicalOlapScan(PlanConstructor.getNextRelationId(), table,
+ List.of("test"), -1L, List.of(), List.of(), null,
PreAggStatus.on(), List.of(), Optional.empty(),
+ properties, Optional.empty(), List.of(), List.of(), List.of(),
Optional.empty(), Optional.empty(),
+ List.of(), Optional.empty());
+ }
+
+ private PhysicalTopN<Plan> source(Slot slot) {
+ PhysicalTopN<Plan> topn = Mockito.mock(PhysicalTopN.class);
+ Mockito.when(topn.getOrderKeys()).thenReturn(List.of(new
OrderKey(slot, true, false)));
+ Mockito.when(topn.getSortPhase()).thenReturn(SortPhase.LOCAL_SORT);
+ Mockito.when(topn.getLimit()).thenReturn(3L);
+ Mockito.when(topn.getOffset()).thenReturn(2L);
+ return topn;
+ }
+
+ private TopnFilterContext push(Plan plan, PhysicalTopN<?> source) {
+ TopnFilterContext context = new TopnFilterContext();
+ plan.accept(new TopnFilterPushDownVisitor(context),
+ new PushDownContext(source,
source.getOrderKeys().get(0).getExpr(), false));
+ return context;
+ }
+
+ private void assertEligibility(Plan plan, PhysicalTopN<?> source, boolean
strict) {
+ TopnFilterContext context = push(plan, source);
+ Assertions.assertEquals(1, context.getTopnFilters().size());
+ Assertions.assertEquals(strict,
context.getTopnFilters().get(0).isStrictEligible());
+ }
+
+ @Test
+ public void testSourceCandidateCountAndOrderKeys() {
+ PhysicalOlapScan scan = scan(key);
+ PhysicalTopN<?> topn = source(key);
+ assertEligibility(scan, topn, true);
+ for (long[] counts : List.of(new long[] {0, 0}, new long[] {1, -1},
+ new long[] {Long.MAX_VALUE, 1}, new long[] {3, 2}, new long[]
{Long.MAX_VALUE, 0})) {
+ Mockito.when(topn.getLimit()).thenReturn(counts[0]);
+ Mockito.when(topn.getOffset()).thenReturn(counts[1]);
+ assertEligibility(scan, topn, counts[0] > 0 && counts[1] >= 0
+ && counts[0] <= Long.MAX_VALUE - counts[1]);
+ }
+ Mockito.when(topn.getOrderKeys()).thenReturn(List.of(new OrderKey(key,
true, false),
+ new OrderKey(new IntegerLiteral(1), true, false)));
+ assertEligibility(scan, topn, false);
+ Mockito.when(topn.getOrderKeys()).thenReturn(List.of(new OrderKey(new
Add(key, new IntegerLiteral(1)),
+ true, false)));
+ assertEligibility(scan, topn, false);
+ Mockito.when(topn.getOrderKeys()).thenReturn(List.of(new OrderKey(key,
true, false)));
+ Mockito.when(topn.getSortPhase()).thenReturn(SortPhase.MERGE_SORT);
+ assertEligibility(scan, topn, false);
+ }
+
+ @Test
+ public void testSafeUnaryNodesAndExactAliases() {
+ PhysicalOlapScan scan = scan(key);
+ Alias alias = new Alias(key, "alias");
+ Plan plan = new PhysicalProject<>(List.of(alias), null,
+ new PhysicalFilter<>(Set.of(new EqualTo(key, new
IntegerLiteral(1))), null,
+ new
PhysicalDistribute<>(DistributionSpecGather.INSTANCE,
+ new PhysicalQuickSort<>(List.of(new
OrderKey(key, true, false)),
+ SortPhase.LOCAL_SORT, null, scan))));
+ assertEligibility(plan, source(alias.toSlot()), true);
+ Alias computed = new Alias(new Add(key, new IntegerLiteral(1)),
"computed");
+ assertEligibility(new PhysicalProject<>(List.of(computed), null,
scan), source(computed.toSlot()), false);
+ }
+
+ @Test
+ public void testUnknownLimitAndHiddenStorageAggregateKeepClosedFilters() {
+ PhysicalOlapScan scan = scan(key);
+ PhysicalTopN<?> topn = source(key);
+ assertEligibility(new PhysicalLimit<>(10, 0, LimitPhase.LOCAL, null,
scan), topn, false);
+ assertEligibility(new PhysicalLimit<>(10, 1, LimitPhase.LOCAL, null,
scan), topn, false);
+ // No child is exposed by this wrapper: PlanVisitor unwraps its
relation directly.
+ assertEligibility(new PhysicalStorageLayerAggregate(scan,
PushDownAggOp.MIN_MAX), topn, false);
+ Plan unknown = Mockito.mock(Plan.class);
+ Mockito.when(unknown.children()).thenReturn(List.of(scan));
+ TopnFilterContext context = new TopnFilterContext();
+ new TopnFilterPushDownVisitor(context).visit(unknown, new
PushDownContext(topn, key, false));
+
Assertions.assertFalse(context.getTopnFilters().get(0).isStrictEligible());
+ // Rebinding an alias after a downgrade must not re-enable strict
bounds.
+ Alias alias = new Alias(key, "alias");
+ PhysicalProject<?> project = new PhysicalProject<>(List.of(alias),
null, scan);
+ context = new TopnFilterContext();
+ project.accept(new TopnFilterPushDownVisitor(context),
+ new PushDownContext(source(alias.toSlot()), alias.toSlot(),
false).disableStrict());
+
Assertions.assertFalse(context.getTopnFilters().get(0).isStrictEligible());
+ }
+
+ @Test
+ public void testRepeatOnlyForCommonGroupingKeys() {
+ PhysicalOlapScan scan = scan(key);
+ SlotReference groupingId = new SlotReference("grouping_id",
IntegerType.INSTANCE);
+ PhysicalRepeat<?> common = new PhysicalRepeat<>(List.of(List.of(key),
List.of(key, groupingId)),
+ List.of(key, groupingId), groupingId, null, scan);
+ assertEligibility(common, source(key), true);
+ PhysicalRepeat<?> erased = new PhysicalRepeat<>(List.of(List.of(key),
List.of()),
+ List.of(key, groupingId), groupingId, null, scan);
+ Assertions.assertTrue(push(erased,
source(key)).getTopnFilters().isEmpty());
+ }
+
+ @Test
+ public void testExistingBarriersStayClosedToPushdown() {
+ for (Class<? extends Plan> type : List.of(PhysicalTopN.class,
PhysicalWindow.class,
+ PhysicalAssertNumRows.class, PhysicalCTEAnchor.class,
PhysicalCTEProducer.class)) {
+ Plan barrier = Mockito.mock(type);
+ Mockito.doCallRealMethod().when(barrier).accept(Mockito.any(),
Mockito.any());
+ Assertions.assertTrue(push(barrier,
source(key)).getTopnFilters().isEmpty(), type.getSimpleName());
+ }
+ }
+
+ @Test
+ public void testUnionActualTargetsAndChildOrder() {
+ SlotReference other = new SlotReference("other", IntegerType.INSTANCE);
+ SlotReference output = new SlotReference("output",
IntegerType.INSTANCE);
+ PhysicalOlapScan first = scan(key);
+ PhysicalOlapScan second = scan(other);
+ for (boolean reverse : List.of(false, true)) {
+ for (boolean unsafe : List.of(false, true)) {
+ Plan firstBranch = unsafe ? new
PhysicalStorageLayerAggregate(first, PushDownAggOp.MIN_MAX) : first;
+ PhysicalUnion union = new PhysicalUnion(Qualifier.ALL,
List.of(output),
+ reverse ? List.of(List.of(other), List.of(key)) :
List.of(List.of(key), List.of(other)),
+ List.of(), null, reverse ? List.of(second,
firstBranch) : List.of(firstBranch, second));
+ TopnFilterContext context = push(union, source(output));
+ Assertions.assertEquals(2,
context.getTopnFilters().get(0).targets.size());
+ Assertions.assertEquals(!unsafe,
context.getTopnFilters().get(0).isStrictEligible());
+ }
+ }
+ PhysicalWindow<?> window = Mockito.mock(PhysicalWindow.class);
+ Mockito.doCallRealMethod().when(window).accept(Mockito.any(),
Mockito.any());
+ Mockito.when(window.getAllChildrenTypes()).thenReturn(new BitSet());
+ PhysicalUnion union = new PhysicalUnion(Qualifier.ALL, List.of(output),
+ List.of(List.of(key), List.of(other)), List.of(), null,
List.of(first, window));
+ TopnFilterContext context = push(union, source(output));
+ Assertions.assertEquals(1,
context.getTopnFilters().get(0).targets.size());
+
Assertions.assertTrue(context.getTopnFilters().get(0).isStrictEligible());
+ }
+
+ @Test
+ public void testJoinTypesAndEqualitySubstitution() {
+ SlotReference rightKey = new SlotReference("right_k",
IntegerType.INSTANCE);
+ PhysicalOlapScan left = scan(key);
+ PhysicalOlapScan right = scan(rightKey);
+ Set<JoinType> strictJoins = Set.of(JoinType.INNER_JOIN,
JoinType.CROSS_JOIN, JoinType.LEFT_OUTER_JOIN,
+ JoinType.LEFT_SEMI_JOIN, JoinType.LEFT_ANTI_JOIN);
+ for (JoinType type : JoinType.values()) {
+ for (boolean mark : List.of(false, true)) {
+ PhysicalHashJoin<Plan, Plan> hash =
Mockito.mock(PhysicalHashJoin.class);
+ Mockito.when(hash.getJoinType()).thenReturn(type);
+ Mockito.when(hash.isMarkJoin()).thenReturn(mark);
+ Mockito.when(hash.left()).thenReturn(left);
+ Mockito.when(hash.right()).thenReturn(right);
+
Mockito.when(hash.getHashJoinConjuncts()).thenReturn(List.of(new EqualTo(key,
rightKey)));
+ Mockito.doCallRealMethod().when(hash).accept(Mockito.any(),
Mockito.any());
+ PhysicalNestedLoopJoin<Plan, Plan> nested =
Mockito.mock(PhysicalNestedLoopJoin.class);
+ Mockito.when(nested.getJoinType()).thenReturn(type);
+ Mockito.when(nested.isMarkJoin()).thenReturn(mark);
+ Mockito.when(nested.left()).thenReturn(left);
+ Mockito.doCallRealMethod().when(nested).accept(Mockito.any(),
Mockito.any());
+ for (Plan join : List.of(hash, nested)) {
+ TopnFilterContext context = push(join, source(key));
+ if (!context.getTopnFilters().isEmpty()) {
+ Assertions.assertEquals(!mark &&
strictJoins.contains(type),
+
context.getTopnFilters().get(0).isStrictEligible(), type + " mark=" + mark);
+ }
+ }
+ // Retain existing equality-substituted filters, but never
grant them strict bounds.
+ TopnFilterContext substituted = push(hash, source(rightKey));
+
Assertions.assertTrue(substituted.getTopnFilters().stream().noneMatch(f ->
f.isStrictEligible()));
+ }
+ }
+ }
+}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/nereids/postprocess/TopnFilterStrictBoundsTest.java
b/fe/fe-core/src/test/java/org/apache/doris/nereids/postprocess/TopnFilterStrictBoundsTest.java
index cb439ff467f..a33362589d7 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/nereids/postprocess/TopnFilterStrictBoundsTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/nereids/postprocess/TopnFilterStrictBoundsTest.java
@@ -43,7 +43,6 @@ import org.apache.doris.planner.OlapScanNode;
import org.apache.doris.planner.PartitionSortNode;
import org.apache.doris.planner.PlanNode;
import org.apache.doris.planner.PlanNodeId;
-import org.apache.doris.planner.RepeatNode;
import org.apache.doris.planner.ScanContext;
import org.apache.doris.planner.ScanNode;
import org.apache.doris.planner.SelectNode;
@@ -69,15 +68,23 @@ public class TopnFilterStrictBoundsTest {
return node;
}
+ private OlapScanNode scan(int id) {
+ OlapScanNode scan = node(OlapScanNode.class, id);
+
Mockito.when(scan.getSelectedIndexKeyType()).thenReturn(KeysType.DUP_KEYS);
+ return scan;
+ }
+
private TopnFilter filter(PlanNode child, ScanNode... targets) {
TopN topn = Mockito.mock(TopN.class);
Mockito.when(topn.getOrderKeys()).thenReturn(List.of(new OrderKey(new
IntegerLiteral(1), true, false)));
- TopnFilter filter = new TopnFilter(topn,
Mockito.mock(PhysicalRelation.class), new IntegerLiteral(1));
+ TopnFilter filter = new TopnFilter(topn,
Mockito.mock(PhysicalRelation.class), new IntegerLiteral(1), true);
SortNode source = node(SortNode.class, 10, child);
SortInfo sortInfo = Mockito.mock(SortInfo.class);
Mockito.when(sortInfo.getOrderingExprs()).thenReturn(List.<Expr>of(new
IntLiteral(1)));
Mockito.when(source.getSortInfo()).thenReturn(sortInfo);
Mockito.when(source.getSortAlgorithm()).thenReturn(TSortAlgorithm.HEAP_SORT);
+ Mockito.when(source.getLimit()).thenReturn(3L);
+ Mockito.when(source.getOffset()).thenReturn(2L);
filter.legacySortNode = source;
for (ScanNode target : targets) {
filter.legacyTargets.put(target, new IntLiteral(1));
@@ -86,21 +93,49 @@ public class TopnFilterStrictBoundsTest {
}
@Test
- public void
testCompletePathsAcrossExchangesKeepAggregateTiesAndPruneSiblingTies() {
- ScanNode aggregateScan = node(FileQueryScanNode.class, 20);
- ScanNode siblingScan = node(FileQueryScanNode.class, 30);
- ScanNode disconnectedScan = node(FileQueryScanNode.class, 99);
+ public void testEveryActualTargetMustAllowStrictBounds() {
+ ScanNode aggregateScan = scan(20);
+ ScanNode siblingScan = scan(30);
+ ScanNode disconnectedScan = scan(99);
PlanNode aggregate = node(AggregationNode.class, 21, aggregateScan);
PlanNode union = node(UnionNode.class, 11, node(ExchangeNode.class,
12, aggregate),
node(ExchangeNode.class, 13, siblingScan));
TopnFilter filter = filter(union, aggregateScan, siblingScan,
disconnectedScan);
- Assertions.assertEquals(List.of(30),
filter.toThrift().getStrictTargetNodeIds());
+
Assertions.assertTrue(filter.toThrift().getStrictTargetNodeIds().isEmpty());
Assertions.assertEquals(3,
filter.toThrift().getTargetNodeIdToTargetExprSize());
}
+ @Test
+ public void testExternalTargetsAndUnreachableTargetsKeepSourceInclusive() {
+ ScanNode scan = scan(20);
+ ScanNode file = node(FileQueryScanNode.class, 30);
+ Assertions.assertTrue(filter(file,
file).toThrift().getStrictTargetNodeIds().isEmpty());
+ Assertions.assertTrue(filter(node(UnionNode.class, 11, scan, file),
scan, file)
+ .toThrift().getStrictTargetNodeIds().isEmpty());
+ Assertions.assertTrue(filter(scan, scan,
scan(99)).toThrift().getStrictTargetNodeIds().isEmpty());
+ // A non-target branch must not disable the only actual target.
+ Assertions.assertEquals(List.of(20), filter(node(UnionNode.class, 11,
scan,
+ node(AggregationNode.class, 31, scan(30))),
scan).toThrift().getStrictTargetNodeIds());
+ }
+
+ @Test
+ public void testRepeatedTargetRegistrationNeverRestoresStrictEligibility()
{
+ ScanNode scan = scan(20);
+ for (boolean unsafeFirst : List.of(false, true)) {
+ TopnFilter filter = filter(scan, scan);
+ PhysicalRelation target =
filter.targets.keySet().iterator().next();
+ filter.addTarget(target, new IntegerLiteral(1), !unsafeFirst);
+ filter.addTarget(target, new IntegerLiteral(1), unsafeFirst);
+ filter.addTarget(target, new IntegerLiteral(1), true);
+ Assertions.assertFalse(filter.isStrictEligible());
+
Assertions.assertTrue(filter.toThrift().getStrictTargetNodeIds().isEmpty());
+ Assertions.assertEquals(1, filter.targets.size());
+ }
+ }
+
@Test
public void testFinalAlgorithmAndOrderingKeys() {
- ScanNode scan = node(FileQueryScanNode.class, 20);
+ ScanNode scan = scan(20);
TopnFilter filter = filter(scan, scan);
for (TSortAlgorithm algorithm : TSortAlgorithm.values()) {
Mockito.when(filter.legacySortNode.getSortAlgorithm()).thenReturn(algorithm);
@@ -122,7 +157,7 @@ public class TopnFilterStrictBoundsTest {
@Test
public void testLimitsAndRowChangingOperatorsRetainTies() {
- ScanNode scan = node(FileQueryScanNode.class, 20);
+ ScanNode scan = scan(20);
PlanNode select = node(SelectNode.class, 11, scan);
TopnFilter filter = filter(select, scan);
Assertions.assertEquals(List.of(20),
filter.toThrift().getStrictTargetNodeIds());
@@ -132,7 +167,15 @@ public class TopnFilterStrictBoundsTest {
Mockito.when(scan.hasLimit()).thenReturn(true);
Assertions.assertTrue(filter.toThrift().getStrictTargetNodeIds().isEmpty());
Mockito.when(scan.hasLimit()).thenReturn(false);
- for (Class<? extends PlanNode> type : List.of(AggregationNode.class,
RepeatNode.class,
+ Mockito.when(select.getOffset()).thenReturn(1L);
+
Assertions.assertTrue(filter.toThrift().getStrictTargetNodeIds().isEmpty());
+ Mockito.when(select.getOffset()).thenReturn(0L);
+ PlanNode limitedPath = node(SelectNode.class, 12, scan);
+ Mockito.when(limitedPath.hasLimit()).thenReturn(true);
+ // The same target reached by an unsafe path cannot be rescued by its
safe occurrence.
+ Assertions.assertTrue(filter(node(UnionNode.class, 13, select,
limitedPath), scan)
+ .toThrift().getStrictTargetNodeIds().isEmpty());
+ for (Class<? extends PlanNode> type : List.of(AggregationNode.class,
AnalyticEvalNode.class, AssertNumRowsNode.class,
PartitionSortNode.class, PlanNode.class)) {
Assertions.assertTrue(filter(node(type, 11, scan),
scan).toThrift().getStrictTargetNodeIds().isEmpty(),
type.getSimpleName());
@@ -142,12 +185,10 @@ public class TopnFilterStrictBoundsTest {
@Test
public void testJoinModes() {
Set<JoinOperator> allowed = Set.of(JoinOperator.INNER_JOIN,
JoinOperator.CROSS_JOIN,
- JoinOperator.LEFT_OUTER_JOIN, JoinOperator.LEFT_SEMI_JOIN,
JoinOperator.LEFT_ANTI_JOIN,
- JoinOperator.NULL_AWARE_LEFT_ANTI_JOIN,
JoinOperator.NULL_AWARE_LEFT_SEMI_JOIN,
- JoinOperator.ASOF_LEFT_INNER_JOIN,
JoinOperator.ASOF_LEFT_OUTER_JOIN);
+ JoinOperator.LEFT_OUTER_JOIN, JoinOperator.LEFT_SEMI_JOIN,
JoinOperator.LEFT_ANTI_JOIN);
for (Class<? extends JoinNodeBase> type : List.of(HashJoinNode.class,
NestedLoopJoinNode.class)) {
for (JoinOperator join : JoinOperator.values()) {
- ScanNode scan = node(FileQueryScanNode.class, 20);
+ ScanNode scan = scan(20);
JoinNodeBase joinNode = node(type, 11, scan,
node(PlanNode.class, 31));
Mockito.when(joinNode.getJoinOp()).thenReturn(join);
Assertions.assertEquals(allowed.contains(join) ? List.of(20) :
List.of(),
diff --git
a/regression-test/suites/topn_optimize/topn_runtime_filter_target_mode.groovy
b/regression-test/suites/topn_optimize/topn_runtime_filter_target_mode.groovy
index 3d4ca2b036a..267ea070d50 100644
---
a/regression-test/suites/topn_optimize/topn_runtime_filter_target_mode.groovy
+++
b/regression-test/suites/topn_optimize/topn_runtime_filter_target_mode.groovy
@@ -58,8 +58,9 @@ suite("topn_runtime_filter_target_mode") {
if (sqlAggregate) {
contains "AGGREGATE"
}
- // Both targets remain enabled. BE keeps equality only on the
aggregate path.
+ // Both targets remain enabled, and an aggregate target keeps the
entire source inclusive.
multiContains "TOPN OPT:", 2
+ contains "TOPN filter boundary: ${query.contains('DESC') ? '>=' :
'<='}"
}
"order_qt_${name}_enabled"(checked)
}
@@ -90,6 +91,22 @@ suite("topn_runtime_filter_target_mode") {
UNION ALL SELECT k,v FROM topn_target_control
) u ORDER BY k NULLS LAST LIMIT 1""", "135", true)
+ for (String direction : ["ASC", "DESC"]) {
+ explain {
+ sql "SELECT k AS alias_key,v FROM topn_target_control WHERE v=999
" +
+ "ORDER BY alias_key ${direction} NULLS LAST LIMIT 1 OFFSET
1"
+ contains "TOPN OPT:"
+ contains "TOPN filter boundary: ${direction == 'ASC' ? '<' :
'>'}\n"
+ }
+ for (String orderKey : ["k ${direction},v", "ABS(k) ${direction}"]) {
+ explain {
+ sql "SELECT k,v FROM topn_target_control ORDER BY ${orderKey}
LIMIT 1"
+ contains "TOPN OPT:"
+ contains "TOPN filter boundary: ${direction == 'ASC' ? '<=' :
'>='}"
+ }
+ }
+ }
+
// An aggregate result is not a grouping key. It cannot become a scan
predicate.
explain {
sql "SELECT k,SUM(v) AS s FROM topn_target_input GROUP BY k ORDER BY s
LIMIT 1"
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
