Skip to content

[improvement](topn) Use strict bounds for single-key TopN runtime filters #69776

New issue

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

Sign up for GitHub

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

Already on GitHub? Sign in to your account

Open
jacktengg wants to merge 2 commits into
base: master
Choose a base branch
from
+2,047 −47
Open

[improvement](topn) Use strict bounds for single-key TopN runtime filters #69776

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
ff6d1d3
[improvement](be) Use strict bounds for single-key TopN runtime filters
jacktengg Oct 27, 2026
329db1a
[fix](fe) Exclude merging OLAP scans from TopN runtime filters
jacktengg Nov 5, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
29 changes: 24 additions & 5 deletions be/src/exprs/vtopn_pred.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -76,7 +76,18 @@ void execute_topn_raw_comparison(const uint8_t* values, size_t num_values, const

template <typename T>
bool topn_raw_scalar_matches(const T& value, const T& bound, RawComparisonOp op) {
return op == RawComparisonOp::LE ? value <= bound : value >= bound;
switch (op) {
case RawComparisonOp::LT:
return value < bound;
case RawComparisonOp::LE:
return value <= bound;
case RawComparisonOp::GT:
return value > bound;
case RawComparisonOp::GE:
return value >= bound;
default:
throw Exception(ErrorCode::INTERNAL_ERROR, "Invalid TopN comparison operator");
}
}

template <PrimitiveType PT>
Expand All @@ -92,7 +103,14 @@ void execute_topn_raw_scalar(const uint8_t* values, size_t num_values, const Fie
}

bool topn_string_matches(int comparison, RawComparisonOp op) {
return op == RawComparisonOp::LE ? comparison <= 0 : comparison >= 0;
return topn_raw_scalar_matches(comparison, 0, op);
}

RawComparisonOp topn_comparison_op(const RuntimePredicate& predicate) {
if (predicate.is_asc()) {
return predicate.is_strict() ? RawComparisonOp::LT : RawComparisonOp::LE;
}
return predicate.is_strict() ? RawComparisonOp::GT : RawComparisonOp::GE;
}

} // namespace
Expand Down Expand Up @@ -172,7 +190,7 @@ Status VTopNPred::execute_on_raw_fixed_values(const uint8_t* values, size_t num_
type_to_string(bound.get_type()),
type_to_string(primitive_type));
}
const auto op = _predicate->is_asc() ? RawComparisonOp::LE : RawComparisonOp::GE;
const auto op = topn_comparison_op(*_predicate);
switch (primitive_type) {
case TYPE_INT:
execute_topn_raw_comparison<int32_t, TYPE_INT>(values, num_values, bound, op, matches);
Expand Down Expand Up @@ -261,7 +279,7 @@ Status VTopNPred::execute_on_raw_binary_values(const StringRef* values, size_t n
type_to_string(bound.get_type()),
type_to_string(primitive_type));
}
const auto op = _predicate->is_asc() ? RawComparisonOp::LE : RawComparisonOp::GE;
const auto op = topn_comparison_op(*_predicate);
for (size_t row = 0; row < num_values; ++row) {
matches[row] &= topn_string_matches(values[row].compare(bound_ref), op) ? 1 : 0;
}
Expand Down Expand Up @@ -299,14 +317,15 @@ ZoneMapFilterResult VTopNPred::evaluate_dictionary_filter(const DictionaryEvalCo
if (!expr_zonemap::field_types_compatible(bound.get_type(), primitive_type)) {
return ZoneMapFilterResult::kUnsupported;
}
const auto op = topn_comparison_op(*_predicate);
for (const Field& value : dictionary->values) {
if (value.is_null()) {
continue;
}
if (!expr_zonemap::field_types_compatible(value.get_type(), primitive_type)) {
return ZoneMapFilterResult::kUnsupported;
}
if ((_predicate->is_asc() && value <= bound) || (!_predicate->is_asc() && value >= bound)) {
if (topn_raw_scalar_matches(value, bound, op)) {
return ZoneMapFilterResult::kMayMatch;
}
}
Expand Down
8 changes: 4 additions & 4 deletions be/src/exprs/vtopn_pred.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -85,7 +85,7 @@ class VTopNPred : public VExpr {
argument_template.emplace_back(nullptr, _children[0]->data_type(), "topn value");

_function = SimpleFunctionFactory::instance().get_function(
_predicate->is_asc() ? "le" : "ge", argument_template, _data_type, {},
_predicate->comparison_function(), argument_template, _data_type, {},
state->be_exec_version());
if (!_function) {
return Status::InternalError("get function failed");
Expand Down Expand Up @@ -192,7 +192,7 @@ class VTopNPred : public VExpr {
TFunction fn;
TFunctionName fn_name;
fn_name.__set_db_name("");
fn_name.__set_function_name(_predicate->is_asc() ? "le" : "ge");
fn_name.__set_function_name(_predicate->comparison_function());
fn.__set_name(fn_name);
fn.__set_binary_type(TFunctionBinaryType::BUILTIN);
std::vector<TTypeDesc> arg_types;
Expand All @@ -210,7 +210,7 @@ class VTopNPred : public VExpr {
TExprNode texpr_node;
texpr_node.__set_type(create_type_desc(PrimitiveType::TYPE_BOOLEAN));
texpr_node.__set_node_type(TExprNodeType::BINARY_PRED);
texpr_node.__set_opcode(_predicate->is_asc() ? TExprOpcode::LE : TExprOpcode::GE);
texpr_node.__set_opcode(_predicate->comparison_opcode());
texpr_node.__set_fn(fn);
texpr_node.__set_num_children(2);
texpr_node.__set_is_nullable(is_nullable());
Expand All @@ -230,7 +230,7 @@ class VTopNPred : public VExpr {
new_root->add_child(VLiteral::create_shared(node));
}

// Since the normal greater than or less than relationship does not consider the relationship of null values, the generated `col >=/<= xxx OR col is null.`
// A comparison rejects SQL NULL. Add IS NULL to preserve NULLS FIRST for either bound type.
if (_predicate->nulls_first()) {
VExprSPtr col_is_null_node;
{
Expand Down
12 changes: 11 additions & 1 deletion be/src/runtime/fragment_mgr.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -574,7 +574,8 @@ Status FragmentMgr::_get_or_create_query_ctx(const TPipelineFragmentParams& para
}
}
if (info.__isset.topn_filter_descs) {
query_ctx->init_runtime_predicates(info.topn_filter_descs);
query_ctx->init_runtime_predicates(info.topn_filter_descs,
parent.params_list);
}
}

Expand Down Expand Up @@ -1362,6 +1363,15 @@ Status FragmentMgr::rerun_fragment(const std::shared_ptr<brpc::ClosureGuard>& gu
info = it->second;
}

// WAIT_FOR_DESTROY has drained the old recursive tasks. A rebuilt TopN has an empty
// heap, so a bound such as abs(k) < 1 from the previous round no longer has candidates
// to support it. Clear this source's bounds before preparing its new fragment.
for (const auto& node : info.params.fragment.plan.nodes) {
if (q_ctx->has_runtime_predicate(node.node_id)) {
q_ctx->get_runtime_predicate(node.node_id).reset_bounds();
}
}

auto context = std::make_shared<PipelineFragmentContext>(
q_ctx->query_id(), info.params, q_ctx, _exec_env, info.finish_callback);
// Propagate the recursion stage so that runtime filters created by this PFC
Expand Down
27 changes: 27 additions & 0 deletions be/src/runtime/query_context.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -66,6 +66,33 @@ class DelayReleaseToken : public Runnable {
std::unique_ptr<ThreadPoolToken> token_;
};

void QueryContext::init_runtime_predicates(
const std::vector<TTopnFilterDesc>& topn_filter_descs,
const std::vector<TPipelineFragmentParams>& fragment_params) {
std::unordered_set<TPlanNodeId> strict_sources;
for (const auto& params : fragment_params) {
for (const auto& node : params.fragment.plan.nodes) {
if (node.node_type == TPlanNodeType::SORT_NODE) {
const auto& sort = node.sort_node;
// A heap retains LIMIT + OFFSET candidates before publishing a bound. With one
// key, another boundary tie cannot improve them. Multiple keys must keep ties:
// (1, 2) can still beat (1, 9). Missing algorithms execute as FULL_SORT.
if (sort.__isset.algorithm && sort.algorithm == TSortAlgorithm::HEAP_SORT &&
sort.sort_info.ordering_exprs.size() == 1) {

@github-actions github-actions Bot Nov 5, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

[P1] Require sender opt-in before enabling strict TopN bounds

An older FE already sends HEAP_SORT and one order key, so this branch enables strict mode for its descriptors even though it lacks this PR's aggregate/join/volatile safety guards. This is reachable during the documented BE-before-FE upgrade sequence.

For example, the pre-change FE can produce this reduced plan with TopN-through-UNION/JOIN rewrites disabled:

TopN(k ASC NULLS LAST, LIMIT 1)
  UnionAll
    Scan(sibling: k=1, payload=999)
    RIGHT ANTI JOIN(a.k=b.k)
      Scan(a: k=1)
      Scan(b: k=1, payload=7)

That FE substitutes b.k with a.k and registers the left scan target. If the sibling publishes bound 1 first, the new BE applies a.k < 1, removes the match and invents (1,7) from the originally empty anti branch. HeapSorter accepts that equal-key block and can evict the retained sibling while trimming, returning the invalid row. The previous inclusive <= 1 preserves the match; the new FE's guard prevents the descriptor, but an old FE still sends it. Default older descriptors to inclusive comparisons and require an explicit safe-sender capability or equivalent version gate; add an old-FE descriptor/new-BE regression.

strict_sources.insert(node.node_id);
}
}
}
}
// Run before publishing QueryContext: a target fragment can prepare its cached comparison
// function before the source SortSink is initialized. Keep the mode immutable thereafter.
// If the source fragment is absent from this request, retain inclusive comparisons.
for (const auto& desc : topn_filter_descs) {
_runtime_predicates.try_emplace(desc.source_node_id, desc,
strict_sources.contains(desc.source_node_id));
}
}

const std::string toString(QuerySource queryType) {
switch (queryType) {
case QuerySource::INTERNAL_FRONTEND:
Expand Down
7 changes: 2 additions & 5 deletions be/src/runtime/query_context.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -164,11 +164,8 @@ class QueryContext : public std::enable_shared_from_this<QueryContext> {
return _runtime_predicates.find(source_node_id)->second;
}

void init_runtime_predicates(const std::vector<TTopnFilterDesc>& topn_filter_descs) {
for (auto desc : topn_filter_descs) {
_runtime_predicates.try_emplace(desc.source_node_id, desc);
}
}
void init_runtime_predicates(const std::vector<TTopnFilterDesc>& topn_filter_descs,
const std::vector<TPipelineFragmentParams>& fragment_params);

Status set_workload_group(WorkloadGroupPtr& wg);

Expand Down
31 changes: 23 additions & 8 deletions be/src/runtime/runtime_predicate.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,8 +29,8 @@

namespace doris {

RuntimePredicate::RuntimePredicate(const TTopnFilterDesc& desc)
: _nulls_first(desc.null_first), _is_asc(desc.is_asc) {
RuntimePredicate::RuntimePredicate(const TTopnFilterDesc& desc, bool is_strict)
: _nulls_first(desc.null_first), _is_asc(desc.is_asc), _is_strict(is_strict) {
DCHECK(!desc.target_node_id_to_target_expr.empty());
for (auto p : desc.target_node_id_to_target_expr) {
_contexts[p.first].expr = p.second;
Expand All @@ -47,12 +47,27 @@ RuntimePredicate::RuntimePredicate(const TTopnFilterDesc& desc)
type_to_string(_type), ss.str());
}

// For ASC sort, create runtime predicate col_name <= max_top_value
// since values that > min_top_value are large than any value in current topn values
// For DESC sort, create runtime predicate col_name >= min_top_value
// since values that < min_top_value are less than any value in current topn values
_pred_constructor = _is_asc ? create_comparison_predicate<PredicateType::LE>
: create_comparison_predicate<PredicateType::GE>;
// QueryContext fixes the mode before scanners prepare their comparison functions.
if (_is_strict) {
_pred_constructor = _is_asc ? create_comparison_predicate<PredicateType::LT>
: create_comparison_predicate<PredicateType::GT>;
} else {
_pred_constructor = _is_asc ? create_comparison_predicate<PredicateType::LE>
: create_comparison_predicate<PredicateType::GE>;
}
}

void RuntimePredicate::reset_bounds() {
std::unique_lock<std::shared_mutex> wlock(_rwlock);
_orderby_extrem = Field(PrimitiveType::TYPE_NULL);
_has_value = false;
for (auto& [node_id, ctx] : _contexts) {
// Expression targets have no storage predicate. Slot targets may already have been
// prepared for the new round, so clear their bound without replacing the shared wrapper.
if (ctx.predicate != nullptr) {
static_cast<SharedPredicate*>(ctx.predicate.get())->reset_for_new_execution();
}
}
}

Status RuntimePredicate::init_target(
Expand Down
24 changes: 22 additions & 2 deletions be/src/runtime/runtime_predicate.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,7 +40,7 @@ class ColumnPredicate;

class RuntimePredicate {
public:
RuntimePredicate(const TTopnFilterDesc& desc);
RuntimePredicate(const TTopnFilterDesc& desc, bool is_strict);

Status init_target(int32_t target_node_id,
phmap::flat_hash_map<int, SlotDescriptor*> slot_id_to_slot_desc,
Expand All @@ -54,10 +54,13 @@ class RuntimePredicate {

void set_detected_source() {
std::unique_lock<std::shared_mutex> wlock(_rwlock);
_orderby_extrem = Field(PrimitiveType::TYPE_NULL);
// Source detection does not reset bounds. Recursive rebuilds call reset_bounds().
_detected_source = true;
}

// Recursive fragment rebuild calls this after the old tasks finish, before new tasks run.
void reset_bounds();

std::shared_ptr<ColumnPredicate> get_predicate(int32_t target_node_id) {
std::shared_lock<std::shared_mutex> rlock(_rwlock);
check_target_node_id(target_node_id);
Expand All @@ -83,6 +86,22 @@ class RuntimePredicate {

bool is_asc() const { return _is_asc; }

bool is_strict() const { return _is_strict; }

const char* comparison_function() const {
if (_is_asc) {
return _is_strict ? "lt" : "le";
}
return _is_strict ? "gt" : "ge";
}

TExprOpcode::type comparison_opcode() const {
if (_is_asc) {
return is_strict() ? TExprOpcode::LT : TExprOpcode::LE;
}
return is_strict() ? TExprOpcode::GT : TExprOpcode::GE;
}

bool nulls_first() const { return _nulls_first; }

bool target_is_slot(int32_t target_node_id) const {
Expand Down Expand Up @@ -130,6 +149,7 @@ class RuntimePredicate {

bool _nulls_first;
bool _is_asc;
const bool _is_strict;
std::map<int32_t, TargetContext> _contexts;

Field _orderby_extrem {PrimitiveType::TYPE_NULL};
Expand Down
6 changes: 6 additions & 0 deletions be/src/storage/predicate/shared_predicate.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -77,6 +77,12 @@ class SharedPredicate final : public ColumnPredicate {
_nested = nested;
}

// Only reset between executions, after the old scanners have finished.
void reset_for_new_execution() {
std::unique_lock<std::shared_mutex> lock(*_mtx);
_nested.reset();
}

Status evaluate(const IndexFieldNameAndTypePair& name_with_type, IndexIterator* iterator,
uint32_t num_rows, roaring::Roaring* bitmap) const override {
std::shared_lock<std::shared_mutex> lock(*_mtx);
Expand Down
Loading
Loading