diff --git a/be/src/exec/operator/scan_operator.cpp b/be/src/exec/operator/scan_operator.cpp index 0874c525c98cb1..791712998810b1 100644 --- a/be/src/exec/operator/scan_operator.cpp +++ b/be/src/exec/operator/scan_operator.cpp @@ -1108,7 +1108,8 @@ Status ScanLocalState::_get_topn_filters(RuntimeState* state) { for (auto id : get_topn_filter_source_node_ids(state, false)) { const auto& pred = state->get_query_ctx()->get_runtime_predicate(id); VExprSPtr topn_pred; - RETURN_IF_ERROR(VTopNPred::create_vtopn_pred(pred.get_texpr(p.node_id()), id, topn_pred)); + RETURN_IF_ERROR(VTopNPred::create_vtopn_pred(pred.get_texpr(p.node_id()), id, p.node_id(), + topn_pred)); VExprContextSPtr conjunct = VExprContext::create_shared(topn_pred); RETURN_IF_ERROR(conjunct->prepare( @@ -1120,7 +1121,8 @@ Status ScanLocalState::_get_topn_filters(RuntimeState* state) { for (auto id : get_topn_filter_source_node_ids(state, true)) { const auto& pred = state->get_query_ctx()->get_runtime_predicate(id); VExprSPtr topn_pred; - RETURN_IF_ERROR(VTopNPred::create_vtopn_pred(pred.get_texpr(p.node_id()), id, topn_pred)); + RETURN_IF_ERROR(VTopNPred::create_vtopn_pred(pred.get_texpr(p.node_id()), id, p.node_id(), + topn_pred)); VExprContextSPtr conjunct = VExprContext::create_shared(topn_pred); RETURN_IF_ERROR(conjunct->prepare( diff --git a/be/src/exprs/vtopn_pred.cpp b/be/src/exprs/vtopn_pred.cpp index 0ef432cde952d9..526d42869d46a7 100644 --- a/be/src/exprs/vtopn_pred.cpp +++ b/be/src/exprs/vtopn_pred.cpp @@ -76,7 +76,18 @@ void execute_topn_raw_comparison(const uint8_t* values, size_t num_values, const template 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 @@ -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, int target_node_id) { + if (predicate.is_asc()) { + return predicate.is_strict(target_node_id) ? RawComparisonOp::LT : RawComparisonOp::LE; + } + return predicate.is_strict(target_node_id) ? RawComparisonOp::GT : RawComparisonOp::GE; } } // namespace @@ -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, _target_node_id); switch (primitive_type) { case TYPE_INT: execute_topn_raw_comparison(values, num_values, bound, op, matches); @@ -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, _target_node_id); for (size_t row = 0; row < num_values; ++row) { matches[row] &= topn_string_matches(values[row].compare(bound_ref), op) ? 1 : 0; } @@ -299,6 +317,7 @@ 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, _target_node_id); for (const Field& value : dictionary->values) { if (value.is_null()) { continue; @@ -306,7 +325,7 @@ ZoneMapFilterResult VTopNPred::evaluate_dictionary_filter(const DictionaryEvalCo 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; } } diff --git a/be/src/exprs/vtopn_pred.h b/be/src/exprs/vtopn_pred.h index 612e411427f4a3..e08e13d1e1ee2e 100644 --- a/be/src/exprs/vtopn_pred.h +++ b/be/src/exprs/vtopn_pred.h @@ -39,14 +39,17 @@ class VTopNPred : public VExpr { ENABLE_FACTORY_CREATOR(VTopNPred); public: - VTopNPred(const TExprNode& node, int source_node_id, VExprContextSPtr target_ctx) + VTopNPred(const TExprNode& node, int source_node_id, int target_node_id, + VExprContextSPtr target_ctx) : VExpr(node), _source_node_id(source_node_id), + _target_node_id(target_node_id), _expr_name(fmt::format("VTopNPred(source_node_id={})", _source_node_id)), _target_ctx(std::move(target_ctx)) {} bool is_topn_filter() const override { return true; } - static Status create_vtopn_pred(const TExpr& target_expr, int source_node_id, VExprSPtr& expr) { + static Status create_vtopn_pred(const TExpr& target_expr, int source_node_id, + int target_node_id, VExprSPtr& expr) { VExprContextSPtr target_ctx; RETURN_IF_ERROR(VExpr::create_expr_tree(target_expr, target_ctx)); @@ -54,7 +57,7 @@ class VTopNPred : public VExpr { node.__set_node_type(TExprNodeType::FUNCTION_CALL); node.__set_type(create_type_desc(PrimitiveType::TYPE_BOOLEAN)); node.__set_is_nullable(target_ctx->root()->is_nullable()); - expr = VTopNPred::create_shared(node, source_node_id, target_ctx); + expr = VTopNPred::create_shared(node, source_node_id, target_node_id, target_ctx); DCHECK(target_ctx->root() != nullptr); expr->add_child(target_ctx->root()); @@ -71,7 +74,8 @@ class VTopNPred : public VExpr { Status clone_node(VExprSPtr* cloned_expr) const override { DORIS_CHECK(cloned_expr != nullptr); - *cloned_expr = VTopNPred::create_shared(clone_texpr_node(), _source_node_id, nullptr); + *cloned_expr = VTopNPred::create_shared(clone_texpr_node(), _source_node_id, + _target_node_id, nullptr); return Status::OK(); } @@ -85,7 +89,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(_target_node_id), argument_template, _data_type, {}, state->be_exec_version()); if (!_function) { return Status::InternalError("get function failed"); @@ -192,7 +196,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(_target_node_id)); fn.__set_name(fn_name); fn.__set_binary_type(TFunctionBinaryType::BUILTIN); std::vector arg_types; @@ -210,7 +214,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(_target_node_id)); texpr_node.__set_fn(fn); texpr_node.__set_num_children(2); texpr_node.__set_is_nullable(is_nullable()); @@ -230,7 +234,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; { @@ -302,6 +306,7 @@ class VTopNPred : public VExpr { } int _source_node_id; + int _target_node_id; std::string _expr_name; RuntimePredicate* _predicate = nullptr; FunctionBasePtr _function; diff --git a/be/src/runtime/fragment_mgr.cpp b/be/src/runtime/fragment_mgr.cpp index ac3684c12f5625..f16f57be1e0a7a 100644 --- a/be/src/runtime/fragment_mgr.cpp +++ b/be/src/runtime/fragment_mgr.cpp @@ -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); } } @@ -1362,6 +1363,15 @@ Status FragmentMgr::rerun_fragment(const std::shared_ptr& 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( 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 diff --git a/be/src/runtime/query_context.cpp b/be/src/runtime/query_context.cpp index 316c7cd5c55772..2e8cb85c657fd1 100644 --- a/be/src/runtime/query_context.cpp +++ b/be/src/runtime/query_context.cpp @@ -66,6 +66,145 @@ class DelayReleaseToken : public Runnable { std::unique_ptr token_; }; +namespace { + +struct TopnPlanNode { + const TPlanNode* node; + TPlanNodeId parent; +}; + +using TopnPlanNodes = std::unordered_map; + +bool topn_join_allows_strict_bounds(TJoinOp::type join) { + switch (join) { + case TJoinOp::INNER_JOIN: + case TJoinOp::CROSS_JOIN: + case TJoinOp::LEFT_OUTER_JOIN: + case TJoinOp::LEFT_SEMI_JOIN: + case TJoinOp::LEFT_ANTI_JOIN: + case TJoinOp::NULL_AWARE_LEFT_ANTI_JOIN: + case TJoinOp::NULL_AWARE_LEFT_SEMI_JOIN: + case TJoinOp::ASOF_LEFT_INNER_JOIN: + case TJoinOp::ASOF_LEFT_OUTER_JOIN: + return true; + default: + // Removing a probe can create an unmatched build row or change its mark. + return false; + } +} + +bool topn_node_allows_strict_bounds(const TPlanNode& node) { + if (node.limit >= 0) { + // Removing a tie can let a different row pass an intermediate LIMIT. + return false; + } + switch (node.node_type) { + case TPlanNodeType::OLAP_SCAN_NODE: { + const auto& scan = node.olap_scan_node; + // Use the selected index's model. Storage aggregation, MOR version selection, + // and changelog folding must retain every contribution at the boundary. + return scan.__isset.keyType && !(scan.__isset.read_changelog && scan.read_changelog) && + (scan.keyType == TKeysType::DUP_KEYS || + (scan.keyType == TKeysType::UNIQUE_KEYS && + scan.__isset.enable_unique_key_merge_on_write && + scan.enable_unique_key_merge_on_write)); + } + case TPlanNodeType::HASH_JOIN_NODE: + return topn_join_allows_strict_bounds(node.hash_join_node.join_op); + case TPlanNodeType::CROSS_JOIN_NODE: + return topn_join_allows_strict_bounds(node.nested_loop_join_node.join_op); + case TPlanNodeType::FILE_SCAN_NODE: + case TPlanNodeType::ODBC_SCAN_NODE: + case TPlanNodeType::SELECT_NODE: + case TPlanNodeType::SORT_NODE: + case TPlanNodeType::EXCHANGE_NODE: + case TPlanNodeType::LOCAL_EXCHANGE_NODE: + case TPlanNodeType::UNION_NODE: + case TPlanNodeType::INTERSECT_NODE: + case TPlanNodeType::EXCEPT_NODE: + case TPlanNodeType::TABLE_FUNCTION_NODE: + case TPlanNodeType::MATERIALIZATION_NODE: + return true; + default: + // Aggregates need complete groups. Repeat, windows, and row-count assertions + // can change surviving rows or create new ones. Unknown operators need proof. + return false; + } +} + +TopnPlanNodes collect_topn_plan_nodes(const std::vector& fragment_params) { + TopnPlanNodes nodes; + for (const auto& params : fragment_params) { + std::vector> parents; + // An exchange ends the serialized tree. Its sending fragment continues the + // same source-to-target path, possibly with an aggregate below the exchange. + const auto& sink = params.fragment.output_sink; + const TPlanNodeId destination = + sink.__isset.stream_sink ? sink.stream_sink.dest_node_id : -1; + for (const auto& node : params.fragment.plan.nodes) { + while (!parents.empty() && parents.back().second == 0) { + parents.pop_back(); + } + const auto parent = parents.empty() ? destination : parents.back().first; + nodes.insert_or_assign(node.node_id, TopnPlanNode {.node = &node, .parent = parent}); + if (!parents.empty()) { + --parents.back().second; + } + if (node.num_children > 0) { + parents.emplace_back(node.node_id, node.num_children); + } + } + } + return nodes; +} + +bool topn_target_can_use_strict_bounds(const TopnPlanNodes& nodes, TPlanNodeId source_id, + TPlanNodeId target_id) { + const auto source = nodes.find(source_id); + if (source == nodes.end() || source->second.node->node_type != TPlanNodeType::SORT_NODE) { + // A backend request can omit the source fragment. There is then no proof of + // either its algorithm or its complete path to this target. + return false; + } + const auto& sort = source->second.node->sort_node; + if (!sort.__isset.algorithm || sort.algorithm != TSortAlgorithm::HEAP_SORT || + sort.sort_info.ordering_exprs.size() != 1) { + // Missing algorithms execute as FULL_SORT. Multiple keys must keep ties: + // (1, 2) can still beat (1, 9), even after the heap has LIMIT + OFFSET rows. + return false; + } + for (auto current = nodes.find(target_id); current != nodes.end(); + current = nodes.find(current->second.parent)) { + if (current->first == source_id) { + return true; + } + if (!topn_node_allows_strict_bounds(*current->second.node)) { + return false; + } + } + // An unavailable sender fragment or a multicast boundary leaves the path unknown. + return false; +} + +} // namespace + +void QueryContext::init_runtime_predicates( + const std::vector& topn_filter_descs, + const std::vector& fragment_params) { + const auto nodes = collect_topn_plan_nodes(fragment_params); + // Run before publishing QueryContext: a target fragment can prepare its cached comparison + // function before the source SortSink is initialized. Keep the mode immutable thereafter. + for (const auto& desc : topn_filter_descs) { + std::unordered_set strict_targets; + for (const auto& [target_id, expression] : desc.target_node_id_to_target_expr) { + if (topn_target_can_use_strict_bounds(nodes, desc.source_node_id, target_id)) { + strict_targets.insert(target_id); + } + } + _runtime_predicates.try_emplace(desc.source_node_id, desc, strict_targets); + } +} + const std::string toString(QuerySource queryType) { switch (queryType) { case QuerySource::INTERNAL_FRONTEND: diff --git a/be/src/runtime/query_context.h b/be/src/runtime/query_context.h index b2b0184589a52f..7c975af21329fa 100644 --- a/be/src/runtime/query_context.h +++ b/be/src/runtime/query_context.h @@ -164,11 +164,8 @@ class QueryContext : public std::enable_shared_from_this { return _runtime_predicates.find(source_node_id)->second; } - void init_runtime_predicates(const std::vector& 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& topn_filter_descs, + const std::vector& fragment_params); Status set_workload_group(WorkloadGroupPtr& wg); diff --git a/be/src/runtime/runtime_predicate.cpp b/be/src/runtime/runtime_predicate.cpp index 37eddc61926012..72858e6212d9f2 100644 --- a/be/src/runtime/runtime_predicate.cpp +++ b/be/src/runtime/runtime_predicate.cpp @@ -29,11 +29,23 @@ namespace doris { -RuntimePredicate::RuntimePredicate(const TTopnFilterDesc& desc) +RuntimePredicate::RuntimePredicate(const TTopnFilterDesc& desc, + const std::unordered_set& strict_target_node_ids) : _nulls_first(desc.null_first), _is_asc(desc.is_asc) { 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; + for (const auto& [target_id, expression] : desc.target_node_id_to_target_expr) { + auto& ctx = _contexts[target_id]; + ctx.expr = expression; + ctx.is_strict = strict_target_node_ids.contains(target_id); + // A sibling scan can discard ties, but an aggregate target must retain every + // contribution at the boundary: SUM(10, 20) must not become SUM(10). + if (ctx.is_strict) { + ctx.pred_constructor = _is_asc ? create_comparison_predicate + : create_comparison_predicate; + } else { + ctx.pred_constructor = _is_asc ? create_comparison_predicate + : create_comparison_predicate; + } } _type = thrift_to_type(desc.target_node_id_to_target_expr.begin() @@ -46,13 +58,19 @@ RuntimePredicate::RuntimePredicate(const TTopnFilterDesc& desc) throw Exception(ErrorCode::INTERNAL_ERROR, "meet invalid type, type={}, expr={}", 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 - : create_comparison_predicate; +void RuntimePredicate::reset_bounds() { + std::unique_lock 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(ctx.predicate.get())->reset_for_new_execution(); + } + } } Status RuntimePredicate::init_target( @@ -110,8 +128,7 @@ Status RuntimePredicate::update(const Field& value) { if (!updated) { return Status::OK(); } - for (auto p : _contexts) { - auto ctx = p.second; + for (const auto& [target_id, ctx] : _contexts) { if (ctx.predicate == nullptr) { // 1. `init_target` will not create predicate. example : `order by abs(col1) limit x;` // So don't need create new `ColumnPredicate`, @@ -130,8 +147,8 @@ Status RuntimePredicate::update(const Field& value) { continue; } std::shared_ptr pred = - _pred_constructor(ctx.predicate->column_id(), ctx.col_name, ctx.col_data_type, - _orderby_extrem, false); + ctx.pred_constructor(ctx.predicate->column_id(), ctx.col_name, ctx.col_data_type, + _orderby_extrem, false); // For NULLS FIRST, wrap a AcceptNullPredicate to return true for NULL // since ORDER BY ASC/DESC should get NULL first but pred returns NULL diff --git a/be/src/runtime/runtime_predicate.h b/be/src/runtime/runtime_predicate.h index a56cf7efbe7404..5403f1a75b2609 100644 --- a/be/src/runtime/runtime_predicate.h +++ b/be/src/runtime/runtime_predicate.h @@ -22,6 +22,7 @@ #include #include #include +#include #include "common/status.h" #include "core/arena.h" @@ -40,7 +41,8 @@ class ColumnPredicate; class RuntimePredicate { public: - RuntimePredicate(const TTopnFilterDesc& desc); + RuntimePredicate(const TTopnFilterDesc& desc, + const std::unordered_set& strict_target_node_ids); Status init_target(int32_t target_node_id, phmap::flat_hash_map slot_id_to_slot_desc, @@ -54,10 +56,13 @@ class RuntimePredicate { void set_detected_source() { std::unique_lock 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 get_predicate(int32_t target_node_id) { std::shared_lock rlock(_rwlock); check_target_node_id(target_node_id); @@ -83,6 +88,25 @@ class RuntimePredicate { bool is_asc() const { return _is_asc; } + bool is_strict(int32_t target_node_id) const { + check_target_node_id(target_node_id); + return _contexts.find(target_node_id)->second.is_strict; + } + + const char* comparison_function(int32_t target_node_id) const { + if (_is_asc) { + return is_strict(target_node_id) ? "lt" : "le"; + } + return is_strict(target_node_id) ? "gt" : "ge"; + } + + TExprOpcode::type comparison_opcode(int32_t target_node_id) const { + if (_is_asc) { + return is_strict(target_node_id) ? TExprOpcode::LT : TExprOpcode::LE; + } + return is_strict(target_node_id) ? TExprOpcode::GT : TExprOpcode::GE; + } + bool nulls_first() const { return _nulls_first; } bool target_is_slot(int32_t target_node_id) const { @@ -114,9 +138,14 @@ class RuntimePredicate { } struct TargetContext { TExpr expr; + // Fixed before QueryContext is published. Targets share a bound, not its inclusiveness. + bool is_strict = false; std::string col_name; DataTypePtr col_data_type; std::shared_ptr predicate; + std::function(const int, const std::string&, + const DataTypePtr&, const Field&, bool)> + pred_constructor; bool target_is_slot() const { return expr.nodes[0].node_type == TExprNodeType::SLOT_REF && @@ -133,10 +162,6 @@ class RuntimePredicate { std::map _contexts; Field _orderby_extrem {PrimitiveType::TYPE_NULL}; - std::function(const int cid, const std::string& col_name, - const DataTypePtr& data_type, const Field& value, - bool opposite)> - _pred_constructor; bool _detected_source = false; bool _detected_target = false; bool _has_value = false; diff --git a/be/src/storage/predicate/shared_predicate.h b/be/src/storage/predicate/shared_predicate.h index 6d1a4aead0885c..620eb5397363bb 100644 --- a/be/src/storage/predicate/shared_predicate.h +++ b/be/src/storage/predicate/shared_predicate.h @@ -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 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 lock(*_mtx); diff --git a/be/test/exec/operator/sort_operator_test.cpp b/be/test/exec/operator/sort_operator_test.cpp index b96c21922ddc2d..08ed0b696ad640 100644 --- a/be/test/exec/operator/sort_operator_test.cpp +++ b/be/test/exec/operator/sort_operator_test.cpp @@ -25,6 +25,9 @@ #include "exec/operator/repeat_operator.h" #include "exec/operator/sort_sink_operator.h" #include "exec/operator/sort_source_operator.h" +#include "exec/pipeline/thrift_builder.h" +#include "runtime/query_context.h" +#include "runtime/runtime_predicate.h" #include "testutil/column_helper.h" #include "testutil/mock/mock_descriptors.h" #include "testutil/mock/mock_runtime_state.h" @@ -60,6 +63,57 @@ struct SortOperatorTest : public ::testing::Test { _child_op = std::make_unique(); } + void check_runtime_predicate_mode(TSortAlgorithm::type algorithm, int mode) { + constexpr int target_node_id = 20; + SCOPED_TRACE(algorithm); + SCOPED_TRACE(mode); + MockRuntimeState runtime_state; + const auto actual_algorithm = mode == 1 ? TSortAlgorithm::FULL_SORT : algorithm; + SortSinkOperatorX sort_sink(&pool, actual_algorithm, 3, 2); + const auto source_node_id = sort_sink.node_id(); + const auto target_expr = TRuntimeFilterDescBuilder::get_default_expr(); + TTopnFilterDesc filter_desc; + filter_desc.__set_source_node_id(source_node_id); + filter_desc.__set_is_asc(true); + filter_desc.__set_null_first(false); + filter_desc.__set_target_node_id_to_target_expr({{target_node_id, target_expr}}); + TSortInfo sort_info; + const size_t key_count = mode == 3 ? 2 : 1; + sort_info.__set_ordering_exprs(std::vector(key_count, target_expr)); + sort_info.__set_is_asc_order(std::vector(key_count, true)); + sort_info.__set_nulls_first(std::vector(key_count, false)); + TSortNode sort_node; + sort_node.__set_sort_info(sort_info); + if (mode != 1) { + sort_node.__set_algorithm(algorithm); + } + TPlanNode plan_node; + plan_node.__set_node_id(source_node_id); + plan_node.__set_node_type(TPlanNodeType::SORT_NODE); + plan_node.__set_sort_node(sort_node); + plan_node.__set_num_children(1); + TPlanNode scan_node; + scan_node.__set_node_id(target_node_id); + scan_node.__set_node_type(TPlanNodeType::FILE_SCAN_NODE); + scan_node.__set_limit(-1); + TPipelineFragmentParams source_params; + if (mode != 0) { + source_params.fragment.plan.__set_nodes({plan_node, scan_node}); + } + // The request can list the target fragment before the source fragment. + runtime_state.get_query_ctx()->init_runtime_predicates( + {filter_desc}, {TPipelineFragmentParams {}, source_params}); + auto& predicate = runtime_state.get_query_ctx()->get_runtime_predicate(source_node_id); + ASSERT_TRUE(predicate.init_target(target_node_id, {}, -1).ok()); + const bool expected_strict = mode == 2 && algorithm == TSortAlgorithm::HEAP_SORT; + EXPECT_EQ(predicate.is_strict(target_node_id), expected_strict); + EXPECT_STREQ(predicate.comparison_function(target_node_id), expected_strict ? "lt" : "le"); + ASSERT_TRUE(sort_sink.init(plan_node, &runtime_state).ok()); + + EXPECT_EQ(predicate.enable(), actual_algorithm == TSortAlgorithm::HEAP_SORT); + EXPECT_EQ(predicate.is_strict(target_node_id), expected_strict); + } + void create_operator(TSortAlgorithm::type type, int64_t limit, int64_t offset) { sink = std::make_unique(&pool, type, limit, offset); @@ -177,6 +231,55 @@ TEST_F(SortOperatorTest, test) { } } +TEST_F(SortOperatorTest, QueryContextSelectsModeBeforeSourceInitialization) { + for (const auto algorithm : + {TSortAlgorithm::HEAP_SORT, TSortAlgorithm::TOPN_SORT, TSortAlgorithm::FULL_SORT}) { + // 0: absent source fragment; 1: absent algorithm; 2/3: one/two ordering keys. + for (const int mode : {0, 1, 2, 3}) { + ASSERT_NO_FATAL_FAILURE(check_runtime_predicate_mode(algorithm, mode)); + } + } +} + +TEST_F(SortOperatorTest, QueryContextMatchesEachFilterToItsSourcePlan) { + const auto target_expr = TRuntimeFilterDescBuilder::get_default_expr(); + std::vector filters; + std::vector params; + for (const auto source_id : {10, 20, 30}) { + TTopnFilterDesc filter; + filter.__set_source_node_id(source_id); + filter.__set_is_asc(true); + filter.__set_null_first(false); + filter.__set_target_node_id_to_target_expr({{source_id + 100, target_expr}}); + filters.push_back(filter); + + TSortNode sort; + sort.__set_algorithm(TSortAlgorithm::HEAP_SORT); + sort.sort_info.__set_ordering_exprs( + std::vector(source_id == 20 ? 1 : 2, target_expr)); + TPlanNode node; + node.__set_node_id(source_id); + node.__set_node_type(TPlanNodeType::SORT_NODE); + node.__set_sort_node(sort); + node.__set_num_children(1); + TPlanNode scan; + scan.__set_node_id(source_id + 100); + scan.__set_node_type(TPlanNodeType::FILE_SCAN_NODE); + scan.__set_limit(-1); + TPipelineFragmentParams fragment; + fragment.fragment.plan.__set_nodes({node, scan}); + params.push_back(fragment); + } + // Descriptor order is independent of the preorder node order in the fragment plan. + std::ranges::reverse(filters); + state->get_query_ctx()->init_runtime_predicates(filters, params); + for (const auto source_id : {10, 20, 30}) { + EXPECT_EQ( + state->get_query_ctx()->get_runtime_predicate(source_id).is_strict(source_id + 100), + source_id == 20); + } +} + TEST_F(SortOperatorTest, test_dep) { create_operator(TSortAlgorithm::HEAP_SORT, 10, 0); diff --git a/be/test/exec/sort/heap_sorter_test.cpp b/be/test/exec/sort/heap_sorter_test.cpp index 9c91db2e5e3833..e6f7c57f60c627 100644 --- a/be/test/exec/sort/heap_sorter_test.cpp +++ b/be/test/exec/sort/heap_sorter_test.cpp @@ -117,4 +117,25 @@ TEST_F(HeapSorterTest, test_topn_sorter1) { } } -} // namespace doris \ No newline at end of file +TEST_F(HeapSorterTest, RuntimeFilterBoundWaitsForLimitPlusOffset) { + sorter = HeapSorter::create_unique(ordering_expr_ctxs, &_state, 3, 2, &pool, is_asc_order, + nulls_first, *row_desc); + sorter->init_profile(&_profile); + + auto first = ColumnHelper::create_block({1, 2, 3, 4}); + ASSERT_TRUE(sorter->append_block(&first).ok()); + EXPECT_TRUE(sorter->get_top_value().is_null()); + + auto second = ColumnHelper::create_block({5, 5}); + ASSERT_TRUE(sorter->append_block(&second).ok()); + EXPECT_EQ(sorter->get_top_value(), Field::create_field(5)); + ASSERT_TRUE(sorter->prepare_for_read(false).ok()); + + Block output; + bool eos = false; + ASSERT_TRUE(sorter->get_next(&_state, &output, &eos).ok()); + EXPECT_TRUE(ColumnHelper::block_equal(output, + ColumnHelper::create_block({3, 4, 5}))); +} + +} // namespace doris diff --git a/be/test/format_v2/parquet/parquet_scan_test.cpp b/be/test/format_v2/parquet/parquet_scan_test.cpp index 0a5339cbccd2e9..bf9146c323afda 100644 --- a/be/test/format_v2/parquet/parquet_scan_test.cpp +++ b/be/test/format_v2/parquet/parquet_scan_test.cpp @@ -989,7 +989,7 @@ PreparedTopNConjunct create_topn_conjunct(RuntimeState* state, int column_id, desc.__set_target_node_id_to_target_expr({{20, std::move(target_expr)}}); auto query_context = MockQueryContext::create(); state->_query_ctx = query_context.get(); - query_context->init_runtime_predicates({desc}); + query_context->init_runtime_predicates({desc}, {}); auto& predicate = query_context->get_runtime_predicate(10); predicate.set_detected_source(); DORIS_CHECK(predicate.init_target(20, {}, -1).ok()); @@ -1000,7 +1000,7 @@ PreparedTopNConjunct create_topn_conjunct(RuntimeState* state, int column_id, TExprNode topn_node; topn_node.__set_type(std::make_shared()->to_thrift()); topn_node.__set_is_nullable(data_type->is_nullable()); - auto topn = VTopNPred::create_shared(topn_node, 10, nullptr); + auto topn = VTopNPred::create_shared(topn_node, 10, 20, nullptr); topn->add_child(VSlotRef::create_shared(column_id, column_id, -1, data_type, "topn_column")); auto conjunct = VExprContext::create_shared(std::move(topn)); DORIS_CHECK(conjunct->prepare(state, RowDescriptor()).ok()); diff --git a/be/test/format_v2/table/iceberg_reader_test.cpp b/be/test/format_v2/table/iceberg_reader_test.cpp index d5dcea52cde4dc..a8b954c7bbba6c 100644 --- a/be/test/format_v2/table/iceberg_reader_test.cpp +++ b/be/test/format_v2/table/iceberg_reader_test.cpp @@ -1898,14 +1898,14 @@ EvolutionTopNFilter make_evolution_topn_filter(RuntimeState* state, bool ascendi desc.__set_target_node_id_to_target_expr({{20, target}}); auto query = MockQueryContext::create(); state->_query_ctx = query.get(); - query->init_runtime_predicates({desc}); + query->init_runtime_predicates({desc}, {}); auto& predicate = query->get_runtime_predicate(10); predicate.set_detected_source(); DORIS_CHECK(predicate.init_target(20, {}, -1).ok()); TExprNode topn_node; topn_node.__set_type(std::make_shared()->to_thrift()); topn_node.__set_is_nullable(true); - auto topn = VTopNPred::create_shared(topn_node, 10, nullptr); + auto topn = VTopNPred::create_shared(topn_node, 10, 20, nullptr); topn->add_child(VSlotRef::create_shared(0, 0, -1, type, "k")); return {.query_context = std::move(query), .conjunct = prepared_conjunct(state, topn)}; } diff --git a/be/test/runtime/fragment_mgr_cross_cluster_cancel_test.cpp b/be/test/runtime/fragment_mgr_cross_cluster_cancel_test.cpp index ce13fbb345920c..08faee44acc9b6 100644 --- a/be/test/runtime/fragment_mgr_cross_cluster_cancel_test.cpp +++ b/be/test/runtime/fragment_mgr_cross_cluster_cancel_test.cpp @@ -21,11 +21,13 @@ #include "common/config.h" #include "exec/pipeline/pipeline_fragment_context.h" +#include "exec/pipeline/thrift_builder.h" #include "runtime/descriptor_helper.h" #include "runtime/exec_env.h" #include "runtime/fragment_mgr.h" #include "runtime/frontend_info.h" #include "runtime/index_policy/index_policy_mgr.h" +#include "runtime/runtime_predicate.h" #include "runtime/workload_group/workload_group.h" #include "runtime/workload_group/workload_group_manager.h" #include "storage/id_manager.h" @@ -60,6 +62,24 @@ class FragmentMgrCrossClusterCancelTest : public testing::Test { } protected: + static void _add_topn_bounds(QueryContext* query_ctx, TPipelineFragmentParams& params) { + for (int node_id : {10, 11}) { + TTopnFilterDesc filter; + filter.__set_source_node_id(node_id); + filter.__set_is_asc(true); + filter.__set_null_first(false); + filter.__set_target_node_id_to_target_expr( + {{20, TRuntimeFilterDescBuilder::get_default_expr()}}); + query_ctx->init_runtime_predicates({filter}, {}); + ASSERT_TRUE(query_ctx->get_runtime_predicate(node_id) + .update(Field::create_field(3)) + .ok()); + } + TPlanNode source; + source.__set_node_id(10); + params.fragment.plan.nodes.push_back(source); + } + static TDescriptorTable _make_min_desc_tbl() { TDescriptorTableBuilder dtb; TTupleDescriptorBuilder tuple_builder; @@ -269,6 +289,7 @@ TEST_F(FragmentMgrCrossClusterCancelTest, RebuildRestoresFragmentExecutingCount) query_ctx) .ok()); ASSERT_NE(query_ctx, nullptr); + ASSERT_NO_FATAL_FAILURE(_add_topn_bounds(query_ctx.get(), params)); { std::lock_guard lock(fragment_mgr->_rerunnable_params_lock); auto& info = fragment_mgr->_rerunnable_params_map[{query_id, fragment_id}]; @@ -282,6 +303,8 @@ TEST_F(FragmentMgrCrossClusterCancelTest, RebuildRestoresFragmentExecutingCount) ASSERT_TRUE(st.ok()) << st.to_string(); EXPECT_EQ(fragment_mgr->_pipeline_map.num_items(), 1); EXPECT_EQ(get_fragment_executing_count(), count_before_rebuild + 1); + EXPECT_FALSE(query_ctx->get_runtime_predicate(10).has_value()); + EXPECT_TRUE(query_ctx->get_runtime_predicate(11).has_value()); fragment_mgr->remove_pipeline_context({query_id, fragment_id}); EXPECT_EQ(get_fragment_executing_count(), count_before_rebuild); diff --git a/be/test/runtime/runtime_predicate_test.cpp b/be/test/runtime/runtime_predicate_test.cpp index d02fd187c1932a..ff3e7d5bac9784 100644 --- a/be/test/runtime/runtime_predicate_test.cpp +++ b/be/test/runtime/runtime_predicate_test.cpp @@ -20,6 +20,7 @@ #include #include +#include #include "core/data_type/data_type_factory.hpp" #include "core/data_type/data_type_number.h" @@ -31,6 +32,7 @@ #include "exprs/vtopn_pred.h" #include "runtime/descriptors.h" #include "runtime/query_context.h" +#include "storage/index/zone_map/zone_map_index.h" #include "testutil/mock/mock_runtime_state.h" namespace doris { @@ -61,29 +63,68 @@ SlotDescriptor create_int_slot_descriptor() { return slot_desc; } +TPlanNode topn_plan_node(int id, TPlanNodeType::type type, int children = 0) { + TPlanNode node; + node.__set_node_id(id); + node.__set_node_type(type); + node.__set_num_children(children); + node.__set_limit(-1); + return node; +} + +TPlanNode heap_topn_plan_node() { + auto node = topn_plan_node(SOURCE_NODE_ID, TPlanNodeType::SORT_NODE, 1); + node.sort_node.__set_algorithm(TSortAlgorithm::HEAP_SORT); + node.sort_node.sort_info.__set_ordering_exprs({TRuntimeFilterDescBuilder::get_default_expr()}); + return node; +} + +VExprContextSPtr prepare_topn_target(MockRuntimeState* state, const DataTypePtr& data_type, + int target_node_id) { + TExprNode node; + node.__set_type(create_type_desc(PrimitiveType::TYPE_BOOLEAN)); + node.__set_is_nullable(data_type->is_nullable()); + auto expr = VTopNPred::create_shared(node, SOURCE_NODE_ID, target_node_id, nullptr); + expr->add_child(VSlotRef::create_shared(SLOT_ID, 0, -1, data_type, "topn_column")); + auto context = VExprContext::create_shared(std::move(expr)); + DORIS_CHECK(context->prepare(state, RowDescriptor()).ok()); + DORIS_CHECK(context->open(state).ok()); + return context; +} + VExprContextSPtr create_prepared_topn_expr(MockRuntimeState* state, const DataTypePtr& data_type, const Field& top_value = Field(), bool is_asc = true, - bool nulls_first = false) { + bool nulls_first = false, bool is_strict = false, + bool init_source = true) { DORIS_CHECK(state != nullptr); DORIS_CHECK(data_type != nullptr); auto desc = create_topn_filter_desc(data_type->get_primitive_type(), is_asc, nulls_first); - state->get_query_ctx()->init_runtime_predicates({desc}); + TSortNode sort; + sort.__set_algorithm(TSortAlgorithm::HEAP_SORT); + sort.sort_info.__set_ordering_exprs(std::vector( + is_strict ? 1 : 2, desc.target_node_id_to_target_expr.at(TARGET_NODE_ID))); + TPlanNode source; + source.__set_node_id(SOURCE_NODE_ID); + source.__set_node_type(TPlanNodeType::SORT_NODE); + source.__set_sort_node(sort); + source.__set_num_children(1); + TPlanNode scan; + scan.__set_node_id(TARGET_NODE_ID); + scan.__set_node_type(TPlanNodeType::FILE_SCAN_NODE); + scan.__set_limit(-1); + TPipelineFragmentParams params; + params.fragment.plan.__set_nodes({source, scan}); + state->get_query_ctx()->init_runtime_predicates({desc}, {params}); auto& predicate = state->get_query_ctx()->get_runtime_predicate(SOURCE_NODE_ID); - predicate.set_detected_source(); + if (init_source) { + predicate.set_detected_source(); + } DORIS_CHECK(predicate.init_target(TARGET_NODE_ID, {}, -1).ok()); if (!top_value.is_null()) { DORIS_CHECK(predicate.update(top_value).ok()); } - TExprNode node; - node.__set_type(create_type_desc(PrimitiveType::TYPE_BOOLEAN)); - node.__set_is_nullable(data_type->is_nullable()); - auto expr = VTopNPred::create_shared(node, SOURCE_NODE_ID, nullptr); - expr->add_child(VSlotRef::create_shared(SLOT_ID, 0, -1, data_type, "topn_column")); - auto context = VExprContext::create_shared(std::move(expr)); - DORIS_CHECK(context->prepare(state, RowDescriptor()).ok()); - DORIS_CHECK(context->open(state).ok()); - return context; + return prepare_topn_target(state, data_type, TARGET_NODE_ID); } void check_topn_uuid_scanner_predicate(bool is_asc, bool nulls_first) { @@ -99,7 +140,7 @@ void check_topn_uuid_scanner_predicate(bool is_asc, bool nulls_first) { SCOPED_TRACE(is_asc); SCOPED_TRACE(nulls_first); - RuntimePredicate predicate(create_topn_filter_desc(TYPE_UUID, is_asc, nulls_first)); + RuntimePredicate predicate(create_topn_filter_desc(TYPE_UUID, is_asc, nulls_first), {}); predicate.set_detected_source(); SlotDescriptor slot_desc; slot_desc._id = SLOT_ID; @@ -132,10 +173,472 @@ void check_topn_uuid_scanner_predicate(bool is_asc, bool nulls_first) { : (std::array {false, false, false, true, true, nulls_first})); } +void check_strict_scanner_predicate(bool ascending, bool nulls_first) { + RuntimePredicate predicate(create_topn_filter_desc(TYPE_INT, ascending, nulls_first), + {TARGET_NODE_ID}); + predicate.set_detected_source(); + auto slot = create_int_slot_descriptor(); + ASSERT_TRUE(predicate.init_target(TARGET_NODE_ID, {{SLOT_ID, &slot}}, 0).ok()); + const auto scanner_predicate = predicate.get_predicate(TARGET_NODE_ID); + auto column = ColumnInt32::create(); + for (int value : {2, 3, 4}) { + column->insert_value(value); + } + std::array matches {}; + scanner_predicate->evaluate_vec(*column, matches.size(), matches.data()); + EXPECT_EQ(matches, (std::array {true, true, true})); + ASSERT_TRUE(predicate.update(Field::create_field(3)).ok()); + scanner_predicate->evaluate_vec(*column, matches.size(), matches.data()); + EXPECT_EQ(matches, ascending ? (std::array {true, false, false}) + : (std::array {false, false, true})); + segment_v2::ZoneMap zone_map; + zone_map.has_not_null = true; + zone_map.min_value = Field::create_field(3); + zone_map.max_value = Field::create_field(3); + EXPECT_FALSE(scanner_predicate->evaluate_and(zone_map)); + zone_map.has_null = true; + EXPECT_EQ(scanner_predicate->evaluate_and(zone_map), nulls_first); +} + +void check_comparison_mode(bool ascending, bool strict, const char* expected_function) { + RuntimePredicate predicate( + create_topn_filter_desc(TYPE_INT, ascending, false), + strict ? std::unordered_set {TARGET_NODE_ID} : std::unordered_set {}); + EXPECT_EQ(predicate.is_strict(TARGET_NODE_ID), strict); + predicate.set_detected_source(); + EXPECT_EQ(predicate.is_strict(TARGET_NODE_ID), strict); + EXPECT_STREQ(predicate.comparison_function(TARGET_NODE_ID), expected_function); + const auto bound = Field::create_field(3); + ASSERT_TRUE(predicate.update(bound).ok()); + predicate.set_detected_source(); + EXPECT_TRUE(predicate.has_value()); + EXPECT_EQ(predicate.get_value(), bound); +} + +void check_scan_prepared_before_source(bool ascending) { + MockRuntimeState state; + const auto type = std::make_shared(); + auto context = create_prepared_topn_expr(&state, type, Field(), ascending, false, true, false); + auto& predicate = state.get_query_ctx()->get_runtime_predicate(SOURCE_NODE_ID); + auto slot = create_int_slot_descriptor(); + ASSERT_TRUE(predicate.init_target(TARGET_NODE_ID, {{SLOT_ID, &slot}}, 0).ok()); + const auto scanner_predicate = predicate.get_predicate(TARGET_NODE_ID); + ASSERT_NE(scanner_predicate, nullptr); + + Block block; + auto column = ColumnInt32::create(); + for (int value : {2, 3, 4}) { + column->insert_value(value); + } + block.insert({std::move(column), type, "topn_column"}); + ColumnPtr result; + ASSERT_TRUE(context->root()->execute_column(context.get(), &block, nullptr, 3, result).ok()); + for (size_t row = 0; row < 3; ++row) { + EXPECT_TRUE(result->get_bool(row)); + } + + predicate.set_detected_source(); + ASSERT_TRUE(predicate.update(Field::create_field(3)).ok()); + ASSERT_TRUE(context->root()->execute_column(context.get(), &block, nullptr, 3, result).ok()); + const std::array values {2, 3, 4}; + IColumn::Filter matches(3, 1); + ASSERT_TRUE(context->root() + ->execute_on_raw_fixed_values( + reinterpret_cast(values.data()), values.size(), + sizeof(int32_t), type, 0, matches.data()) + .ok()); + const IColumn::Filter expected = + ascending ? IColumn::Filter {1, 0, 0} : IColumn::Filter {0, 0, 1}; + std::array scanner_matches {}; + scanner_predicate->evaluate_vec(*block.get_by_position(0).column, 3, scanner_matches.data()); + EXPECT_EQ(matches, expected); + for (size_t row = 0; row < 3; ++row) { + EXPECT_EQ(result->get_bool(row), expected[row]); + EXPECT_EQ(scanner_matches[row], expected[row]); + } + context->close(); +} + +void check_bounds_reset(bool ascending, bool strict) { + RuntimePredicate predicate( + create_topn_filter_desc(TYPE_INT, ascending, false), + strict ? std::unordered_set {TARGET_NODE_ID} : std::unordered_set {}); + predicate.set_detected_source(); + auto slot = create_int_slot_descriptor(); + ASSERT_TRUE(predicate.init_target(TARGET_NODE_ID, {{SLOT_ID, &slot}}, 0).ok()); + const auto scanner_predicate = predicate.get_predicate(TARGET_NODE_ID); + ASSERT_TRUE(predicate.update(Field::create_field(3)).ok()); + + predicate.reset_bounds(); + EXPECT_FALSE(predicate.has_value()); + EXPECT_EQ(predicate.get_predicate(TARGET_NODE_ID), scanner_predicate); + + auto column = ColumnInt32::create(); + const int new_bound = ascending ? 5 : 1; + for (int value : {new_bound - 1, new_bound, new_bound + 1}) { + column->insert_value(value); + } + std::array matches {}; + scanner_predicate->evaluate_vec(*column, 3, matches.data()); + EXPECT_EQ(matches, (std::array {true, true, true})); + + // The new heap can have a worse frontier than the previous round's heap. + predicate.set_detected_source(); + ASSERT_TRUE(predicate.update(Field::create_field(new_bound)).ok()); + EXPECT_EQ(predicate.get_value(), Field::create_field(new_bound)); + scanner_predicate->evaluate_vec(*column, 3, matches.data()); + EXPECT_EQ(matches, ascending ? (std::array {true, !strict, false}) + : (std::array {false, !strict, true})); +} + +void check_inclusive_bounds(bool ascending) { + RuntimePredicate predicate(create_topn_filter_desc(TYPE_INT, ascending, false), {}); + predicate.set_detected_source(); + EXPECT_FALSE(predicate.is_strict(TARGET_NODE_ID)); + EXPECT_STREQ(predicate.comparison_function(TARGET_NODE_ID), ascending ? "le" : "ge"); + EXPECT_EQ(predicate.comparison_opcode(TARGET_NODE_ID), + ascending ? TExprOpcode::LE : TExprOpcode::GE); + + auto slot = create_int_slot_descriptor(); + ASSERT_TRUE(predicate.init_target(TARGET_NODE_ID, {{SLOT_ID, &slot}}, 0).ok()); + ASSERT_TRUE(predicate.enable()); + ASSERT_TRUE(predicate.update(Field::create_field(3)).ok()); + const auto scanner_predicate = predicate.get_predicate(TARGET_NODE_ID); + auto column = ColumnInt32::create(); + for (const int value : {2, 3, 4}) { + column->insert_value(value); + } + std::array matches {}; + scanner_predicate->evaluate_vec(*column, matches.size(), matches.data()); + EXPECT_EQ(matches, ascending ? (std::array {true, true, false}) + : (std::array {false, true, true})); + segment_v2::ZoneMap zone_map; + zone_map.has_not_null = true; + zone_map.min_value = Field::create_field(3); + zone_map.max_value = Field::create_field(3); + EXPECT_TRUE(scanner_predicate->evaluate_and(zone_map)); +} + +void check_mixed_target_evaluation(RuntimePredicate& predicate, const VExprContextSPtr& context, + int target, bool ascending) { + const auto type = std::make_shared(); + const bool strict = target == 30; + std::array values {}; + auto column = ColumnInt32::create(); + for (size_t row = 0; row < values.size(); ++row) { + values[row] = 2 + row % 3; + column->insert_value(values[row]); + } + Block block; + block.insert({std::move(column), type, "topn_column"}); + ColumnPtr result; + ASSERT_TRUE(context->root() + ->execute_column(context.get(), &block, nullptr, values.size(), result) + .ok()); + IColumn::Filter matches(values.size(), 1); + ASSERT_TRUE(context->root() + ->execute_on_raw_fixed_values( + reinterpret_cast(values.data()), values.size(), + sizeof(int32_t), type, 0, matches.data()) + .ok()); + std::array storage_matches {}; + predicate.get_predicate(target)->evaluate_vec(*block.get_by_position(0).column, values.size(), + storage_matches.data()); + for (size_t row = 0; row < values.size(); ++row) { + const bool expected = + (values[row] == 3 && !strict) || (ascending ? values[row] < 3 : values[row] > 3); + EXPECT_EQ(result->get_bool(row), expected); + EXPECT_EQ(matches[row], expected); + EXPECT_EQ(storage_matches[row], expected); + } + DictionaryEvalContext dictionary; + dictionary.slots.emplace( + 0, DictionaryEvalContext::SlotDictionary { + .data_type = type, .values = {Field::create_field(3)}}); + EXPECT_EQ(context->root()->evaluate_dictionary_filter(dictionary), + strict ? ZoneMapFilterResult::kNoMatch : ZoneMapFilterResult::kMayMatch); + VExprSPtr binary; + ASSERT_TRUE(assert_cast(context->root().get())->get_binary_expr(binary)); + EXPECT_EQ(binary->op(), predicate.comparison_opcode(target)); +} + +void check_mixed_target_lifecycle(bool ascending) { + MockRuntimeState state; + auto desc = create_topn_filter_desc(TYPE_INT, ascending); + desc.target_node_id_to_target_expr.emplace(30, desc.target_node_id_to_target_expr.at(20)); + TPipelineFragmentParams params; + params.fragment.plan.__set_nodes({heap_topn_plan_node(), + topn_plan_node(11, TPlanNodeType::UNION_NODE, 2), + topn_plan_node(21, TPlanNodeType::AGGREGATION_NODE, 1), + topn_plan_node(20, TPlanNodeType::FILE_SCAN_NODE), + topn_plan_node(30, TPlanNodeType::FILE_SCAN_NODE)}); + state.get_query_ctx()->init_runtime_predicates({desc}, {params}); + auto& predicate = state.get_query_ctx()->get_runtime_predicate(SOURCE_NODE_ID); + const auto type = std::make_shared(); + auto slot = create_int_slot_descriptor(); + std::vector> expressions; + for (const auto target : {20, 30}) { + ASSERT_TRUE(predicate.init_target(target, {{SLOT_ID, &slot}}, 0).ok()); + auto expression = prepare_topn_target(&state, type, target); + VExprContextSPtr clone; + ASSERT_TRUE(expression->clone(&state, clone).ok()); + expressions.emplace_back(target, expression); + expressions.emplace_back(target, clone); + } + // Prepare both targets (and clones) before detecting the source and publishing a bound. + predicate.set_detected_source(); + ASSERT_TRUE(predicate.update(Field::create_field(3)).ok()); + for (const auto& [target, expression] : expressions) { + ASSERT_NO_FATAL_FAILURE(check_mixed_target_evaluation(predicate, expression, target, ascending)); + } + predicate.reset_bounds(); + EXPECT_FALSE(predicate.is_strict(20)); + EXPECT_TRUE(predicate.is_strict(30)); + EXPECT_FALSE(predicate.has_value()); + for (auto& [target, expression] : expressions) { + expression->close(); + } +} + +void check_intermediate_target_mode(TPlanNode middle, bool strict) { + MockRuntimeState state; + TPipelineFragmentParams params; + params.fragment.plan.__set_nodes( + {heap_topn_plan_node(), middle, + topn_plan_node(TARGET_NODE_ID, TPlanNodeType::FILE_SCAN_NODE)}); + if (middle.num_children == 2) { + params.fragment.plan.nodes.push_back(topn_plan_node(31, TPlanNodeType::EMPTY_SET_NODE)); + } + state.get_query_ctx()->init_runtime_predicates({create_topn_filter_desc()}, {params}); + EXPECT_EQ( + state.get_query_ctx()->get_runtime_predicate(SOURCE_NODE_ID).is_strict(TARGET_NODE_ID), + strict); +} + } // namespace +TEST(RuntimePredicateTest, InclusiveBoundsRetainEqualRows) { + for (const bool ascending : {false, true}) { + ASSERT_NO_FATAL_FAILURE(check_inclusive_bounds(ascending)); + } +} + +TEST(RuntimePredicateTest, SourcePreservesComparisonModeAndPublishedBounds) { + ASSERT_NO_FATAL_FAILURE(check_comparison_mode(true, false, "le")); + ASSERT_NO_FATAL_FAILURE(check_comparison_mode(true, true, "lt")); + ASSERT_NO_FATAL_FAILURE(check_comparison_mode(false, false, "ge")); + ASSERT_NO_FATAL_FAILURE(check_comparison_mode(false, true, "gt")); +} + +TEST(RuntimePredicateTest, TargetPathsKeepAggregateTiesAndPruneSiblingTies) { + for (const auto aggregation : + {TPlanNodeType::AGGREGATION_NODE, TPlanNodeType::BUCKETED_AGGREGATION_NODE}) { + MockRuntimeState state; + auto desc = create_topn_filter_desc(); + desc.target_node_id_to_target_expr.emplace(30, desc.target_node_id_to_target_expr.at(20)); + desc.target_node_id_to_target_expr.emplace(99, desc.target_node_id_to_target_expr.at(20)); + TPipelineFragmentParams params; + params.fragment.plan.__set_nodes({heap_topn_plan_node(), + topn_plan_node(11, TPlanNodeType::UNION_NODE, 2), + topn_plan_node(21, aggregation, 1), + topn_plan_node(20, TPlanNodeType::FILE_SCAN_NODE), + topn_plan_node(30, TPlanNodeType::FILE_SCAN_NODE)}); + state.get_query_ctx()->init_runtime_predicates({desc}, {params}); + auto& predicate = state.get_query_ctx()->get_runtime_predicate(SOURCE_NODE_ID); + EXPECT_FALSE(predicate.is_strict(20)); + EXPECT_TRUE(predicate.is_strict(30)); + EXPECT_FALSE(predicate.is_strict(99)); // Target plan is absent from this request. + } +} + +TEST(RuntimePredicateTest, TargetModesFollowExchangeSendersWithoutGuessingMissingPaths) { + for (const bool include_source : {false, true}) { + for (const bool connect_sibling : {false, true}) { + MockRuntimeState state; + auto desc = create_topn_filter_desc(); + desc.target_node_id_to_target_expr.emplace(30, + desc.target_node_id_to_target_expr.at(20)); + TPipelineFragmentParams source; + source.fragment.plan.__set_nodes({heap_topn_plan_node(), + topn_plan_node(11, TPlanNodeType::UNION_NODE, 2), + topn_plan_node(12, TPlanNodeType::EXCHANGE_NODE), + topn_plan_node(13, TPlanNodeType::EXCHANGE_NODE)}); + TPipelineFragmentParams aggregate; + aggregate.fragment.plan.__set_nodes( + {topn_plan_node(21, TPlanNodeType::AGGREGATION_NODE, 1), + topn_plan_node(20, TPlanNodeType::FILE_SCAN_NODE)}); + TDataStreamSink sink; + sink.__set_dest_node_id(12); + aggregate.fragment.output_sink.__set_stream_sink(sink); + TPipelineFragmentParams sibling; + sibling.fragment.plan.__set_nodes({topn_plan_node(30, TPlanNodeType::FILE_SCAN_NODE)}); + if (connect_sibling) { + sink.__set_dest_node_id(13); + sibling.fragment.output_sink.__set_stream_sink(sink); + } + std::vector params {aggregate, sibling}; + if (include_source) { + params.push_back(source); // Target fragments precede their source. + } + state.get_query_ctx()->init_runtime_predicates({desc}, params); + auto& predicate = state.get_query_ctx()->get_runtime_predicate(SOURCE_NODE_ID); + EXPECT_FALSE(predicate.is_strict(20)); + EXPECT_EQ(predicate.is_strict(30), include_source && connect_sibling); + } + } +} + +TEST(RuntimePredicateTest, AggregateAboveSourceDoesNotChangeItsTargetMode) { + MockRuntimeState state; + TPipelineFragmentParams params; + params.fragment.plan.__set_nodes( + {topn_plan_node(1, TPlanNodeType::AGGREGATION_NODE, 1), heap_topn_plan_node(), + topn_plan_node(TARGET_NODE_ID, TPlanNodeType::FILE_SCAN_NODE)}); + state.get_query_ctx()->init_runtime_predicates({create_topn_filter_desc()}, {params}); + EXPECT_TRUE( + state.get_query_ctx()->get_runtime_predicate(SOURCE_NODE_ID).is_strict(TARGET_NODE_ID)); +} + +TEST(RuntimePredicateTest, IntermediateLimitsAndRowChangingOperatorsKeepTies) { + auto select = topn_plan_node(11, TPlanNodeType::SELECT_NODE, 1); + ASSERT_NO_FATAL_FAILURE(check_intermediate_target_mode(select, true)); + select.__set_limit(1); + ASSERT_NO_FATAL_FAILURE(check_intermediate_target_mode(select, false)); + for (const auto type : {TPlanNodeType::ASSERT_NUM_ROWS_NODE, TPlanNodeType::REPEAT_NODE, + TPlanNodeType::ANALYTIC_EVAL_NODE, TPlanNodeType::PARTITION_SORT_NODE, + TPlanNodeType::REC_CTE_NODE}) { + ASSERT_NO_FATAL_FAILURE(check_intermediate_target_mode(topn_plan_node(11, type, 1), false)); + } +} + +TEST(RuntimePredicateTest, JoinsThatCanCreateBuildRowsKeepTies) { + for (const auto join : + {TJoinOp::INNER_JOIN, TJoinOp::LEFT_OUTER_JOIN, TJoinOp::LEFT_SEMI_JOIN, + TJoinOp::LEFT_ANTI_JOIN, TJoinOp::RIGHT_OUTER_JOIN, TJoinOp::FULL_OUTER_JOIN, + TJoinOp::RIGHT_SEMI_JOIN, TJoinOp::RIGHT_ANTI_JOIN, TJoinOp::ASOF_RIGHT_OUTER_JOIN}) { + const bool strict = join == TJoinOp::INNER_JOIN || join == TJoinOp::LEFT_OUTER_JOIN || + join == TJoinOp::LEFT_SEMI_JOIN || join == TJoinOp::LEFT_ANTI_JOIN; + auto hash_join = topn_plan_node(11, TPlanNodeType::HASH_JOIN_NODE, 2); + hash_join.hash_join_node.__set_join_op(join); + ASSERT_NO_FATAL_FAILURE(check_intermediate_target_mode(hash_join, strict)); + auto nested_join = topn_plan_node(11, TPlanNodeType::CROSS_JOIN_NODE, 2); + nested_join.nested_loop_join_node.__set_join_op(join); + ASSERT_NO_FATAL_FAILURE(check_intermediate_target_mode(nested_join, strict)); + } +} + +TEST(RuntimePredicateTest, NativeMergeTargetsUseInclusiveBounds) { + for (const auto keys : {TKeysType::DUP_KEYS, TKeysType::AGG_KEYS, TKeysType::UNIQUE_KEYS}) { + for (const bool mow : {false, true}) { + for (const bool changelog : {false, true}) { + MockRuntimeState state; + auto scan = topn_plan_node(TARGET_NODE_ID, TPlanNodeType::OLAP_SCAN_NODE); + scan.olap_scan_node.__set_keyType(keys); + scan.olap_scan_node.__set_enable_unique_key_merge_on_write(mow); + scan.olap_scan_node.__set_read_changelog(changelog); + TPipelineFragmentParams params; + params.fragment.plan.__set_nodes({heap_topn_plan_node(), scan}); + state.get_query_ctx()->init_runtime_predicates({create_topn_filter_desc()}, + {params}); + EXPECT_EQ(state.get_query_ctx() + ->get_runtime_predicate(SOURCE_NODE_ID) + .is_strict(TARGET_NODE_ID), + !changelog && (keys == TKeysType::DUP_KEYS || + (keys == TKeysType::UNIQUE_KEYS && mow))); + } + } + } + MockRuntimeState state; + TPipelineFragmentParams params; + params.fragment.plan.__set_nodes( + {heap_topn_plan_node(), topn_plan_node(TARGET_NODE_ID, TPlanNodeType::OLAP_SCAN_NODE)}); + state.get_query_ctx()->init_runtime_predicates({create_topn_filter_desc()}, {params}); + EXPECT_FALSE(state.get_query_ctx() + ->get_runtime_predicate(SOURCE_NODE_ID) + .is_strict(TARGET_NODE_ID)); // Missing storage model. +} + +TEST(RuntimePredicateTest, MixedTargetsKeepTheirModesAcrossEvaluationCloneAndReset) { + ASSERT_NO_FATAL_FAILURE(check_mixed_target_lifecycle(false)); + ASSERT_NO_FATAL_FAILURE(check_mixed_target_lifecycle(true)); +} + +TEST(RuntimePredicateTest, ScanPreparedBeforeSourceUsesBackendModeAcrossEvaluationPaths) { + ASSERT_NO_FATAL_FAILURE(check_scan_prepared_before_source(true)); + ASSERT_NO_FATAL_FAILURE(check_scan_prepared_before_source(false)); +} + +TEST(RuntimePredicateTest, NewExecutionClearsBoundsAndAcceptsAWorseFrontier) { + ASSERT_NO_FATAL_FAILURE(check_bounds_reset(true, false)); + ASSERT_NO_FATAL_FAILURE(check_bounds_reset(true, true)); + ASSERT_NO_FATAL_FAILURE(check_bounds_reset(false, false)); + ASSERT_NO_FATAL_FAILURE(check_bounds_reset(false, true)); +} + +TEST(RuntimePredicateTest, StrictBoundsRejectEqualRowsAndEqualZoneMaps) { + for (const bool ascending : {false, true}) { + for (const bool nulls_first : {false, true}) { + ASSERT_NO_FATAL_FAILURE(check_strict_scanner_predicate(ascending, nulls_first)); + } + } +} + +TEST(RuntimePredicateTest, StrictBoundsAgreeAcrossColumnRawAndDictionaryEvaluation) { + for (const bool ascending : {false, true}) { + MockRuntimeState state; + const auto type = std::make_shared(); + auto context = create_prepared_topn_expr(&state, type, Field::create_field(3), + ascending, false, true); + const std::array values {2, 3, 4}; + IColumn::Filter matches(3, 1); + ASSERT_TRUE(context->root() + ->execute_on_raw_fixed_values( + reinterpret_cast(values.data()), values.size(), + sizeof(int32_t), type, 0, matches.data()) + .ok()); + EXPECT_EQ(matches, ascending ? (IColumn::Filter {1, 0, 0}) : (IColumn::Filter {0, 0, 1})); + Block block; + auto column = ColumnInt32::create(); + for (const auto value : values) { + column->insert_value(value); + } + block.insert({std::move(column), type, "topn_column"}); + ColumnPtr result; + ASSERT_TRUE( + context->root()->execute_column(context.get(), &block, nullptr, 3, result).ok()); + for (size_t row = 0; row < 3; ++row) { + EXPECT_EQ(result->get_bool(row), matches[row]); + } + DictionaryEvalContext dictionary; + dictionary.slots.emplace( + 0, DictionaryEvalContext::SlotDictionary { + .data_type = type, .values = {Field::create_field(3)}}); + EXPECT_EQ(context->root()->evaluate_dictionary_filter(dictionary), + ZoneMapFilterResult::kNoMatch); + context->close(); + } +} + +TEST(RuntimePredicateTest, StrictBinaryBoundsRejectEquality) { + for (const bool ascending : {false, true}) { + MockRuntimeState state; + const auto type = DataTypeFactory::instance().create_data_type(TYPE_STRING, false); + auto context = create_prepared_topn_expr( + &state, type, Field::create_field("middle"), ascending, false, true); + const std::array values {StringRef("alpha"), StringRef("middle"), + StringRef("zulu")}; + IColumn::Filter matches(3, 1); + ASSERT_TRUE(context->root() + ->execute_on_raw_binary_values(values.data(), values.size(), type, 0, + matches.data()) + .ok()); + EXPECT_EQ(matches, ascending ? (IColumn::Filter {1, 0, 0}) : (IColumn::Filter {0, 0, 1})); + context->close(); + } +} + TEST(RuntimePredicateTest, init_target_creates_column_predicate_for_valid_column_id) { - RuntimePredicate predicate(create_topn_filter_desc()); + RuntimePredicate predicate(create_topn_filter_desc(), {}); predicate.set_detected_source(); auto slot_desc = create_int_slot_descriptor(); @@ -150,7 +653,7 @@ TEST(RuntimePredicateTest, init_target_creates_column_predicate_for_valid_column } TEST(RuntimePredicateTest, init_target_without_column_predicate_still_enables_runtime_filter) { - RuntimePredicate predicate(create_topn_filter_desc()); + RuntimePredicate predicate(create_topn_filter_desc(), {}); predicate.set_detected_source(); phmap::flat_hash_map slot_id_to_slot_desc; diff --git a/be/test/storage/segment/segment_iterator_runtime_common_expr_test.cpp b/be/test/storage/segment/segment_iterator_runtime_common_expr_test.cpp index a50713c484d195..225520b6eb2523 100644 --- a/be/test/storage/segment/segment_iterator_runtime_common_expr_test.cpp +++ b/be/test/storage/segment/segment_iterator_runtime_common_expr_test.cpp @@ -108,7 +108,8 @@ VExprContextSPtr make_topn_filter_ctx() { node.__set_node_type(TExprNodeType::FUNCTION_CALL); node.__set_type(create_type_desc(PrimitiveType::TYPE_BOOLEAN)); node.__set_is_nullable(true); - auto topn_pred = VTopNPred::create_shared(node, /*source_node_id=*/0, nullptr); + auto topn_pred = + VTopNPred::create_shared(node, /*source_node_id=*/0, /*target_node_id=*/1, nullptr); topn_pred->add_child(make_struct_slot_ref()); return VExprContext::create_shared(topn_pred); } 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 0efae9d7df8ad2..7b7d1d91d2e907 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 @@ -24,12 +24,15 @@ import org.apache.doris.nereids.trees.expressions.Slot; 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.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.PhysicalCTEAnchor; import org.apache.doris.nereids.trees.plans.physical.PhysicalCTEProducer; import org.apache.doris.nereids.trees.plans.physical.PhysicalFileScan; +import org.apache.doris.nereids.trees.plans.physical.PhysicalHashAggregate; 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; @@ -37,6 +40,7 @@ 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.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.PhysicalTopN; import org.apache.doris.nereids.trees.plans.physical.PhysicalWindow; @@ -48,6 +52,7 @@ import java.util.List; import java.util.Map; +import java.util.Set; /** * push down topn filter @@ -82,8 +87,6 @@ public PushDownContext withNewProbeExpression(Expression newProbe) { @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. boolean pushed = false; for (Plan child : plan.children()) { pushed |= child.accept(this, ctx); @@ -171,6 +174,25 @@ public Boolean visitPhysicalWindow(PhysicalWindow window, PushDo return false; } + @Override + public Boolean visitPhysicalHashAggregate(PhysicalHashAggregate aggregate, + PushDownContext ctx) { + // BE uses inclusive bounds for this target path. A candidate group's key passes + // every earlier, wider bound, so all of its contributions remain. Do not substitute + // aggregate results with input values: SUM(v) <= bound does not imply v <= bound. + if (!aggregate.getGroupByExpressions().containsAll(ctx.probeExpr.getInputSlots())) { + return false; + } + return aggregate.child().accept(this, ctx); + } + + @Override + public Boolean visitPhysicalRepeat(PhysicalRepeat repeat, PushDownContext ctx) { + // GROUPING SETS can replace a non-NULL input key with NULL in another output group. + // Removing the aggregate barrier must not admit a filter with a different output key. + return false; + } + private PushDownContext adjustProbeExprNullableThroughOuterJoin(Plan leftChild, PushDownContext ctx) { Expression probeExpr = ctx.probeExpr; Map replaceMap = Maps.newHashMap(); @@ -191,6 +213,25 @@ private PushDownContext adjustProbeExprNullableThroughOuterJoin(Plan leftChild, return ctx; } + private boolean canPushDownToLeftChild(AbstractPhysicalJoin join, + Expression probeExpr) { + JoinType joinType = join.getJoinType(); + if (joinType.isRightAntiJoin() || (joinType.isRightSemiJoin() && join.isMarkJoin())) { + // Removing a probe can create an unmatched right row or change its mark value. + // This also applies to constant probes, which have no input slots. + return false; + } + if (joinType.isRightOuterJoin() || joinType.isFullOuterJoin() || joinType.isAsofRightOuterJoin()) { + // A right-side ordering key keeps its value when a match is removed, but the left + // payload changes. Strict bounds can then return a row that the original join would + // not produce. Slot-free expressions have the same problem. Check the original + // probe before substituting a right equality key with its left counterpart. + Set inputSlots = probeExpr.getInputSlots(); + return !inputSlots.isEmpty() && join.left().getOutputSet().containsAll(inputSlots); + } + return true; + } + @Override public Boolean visitPhysicalHashJoin(PhysicalHashJoin join, PushDownContext ctx) { @@ -201,6 +242,9 @@ public Boolean visitPhysicalHashJoin(PhysicalHashJoin TSOTimestamp.composePhysicalTimestamp(System.currentTimeMillis())); + checker.analyze().rewrite().implement(); + PhysicalPlan plan = new PlanPostProcessors(checker.getCascadesContext()) + .process(checker.getPhysicalPlan()); + // BE detects the storage merge even when there is no SQL aggregate node. + Assertions.assertFalse(plan.anyMatch(PhysicalHashAggregate.class::isInstance), + plan.treeString()); + List filters = checker.getCascadesContext().getTopnFilterContext().getTopnFilters(); + Assertions.assertEquals(1, filters.size(), plan.treeString()); + TopnFilter filter = filters.get(0); + Assertions.assertEquals(2, filter.targets.size(), plan.treeString()); + Assertions.assertTrue(filter.targets.keySet().stream().anyMatch(target -> + ((PhysicalOlapScan) target).getTable().getName().equals("topn_storage_dup"))); + new PhysicalPlanTranslator(new PlanTranslatorContext(checker.getCascadesContext())) + .translatePlan(plan); + Assertions.assertEquals(2, filter.toThrift().getTargetNodeIdToTargetExprSize()); + } + } + } + } finally { + connectContext.getSessionVariable().topnFilterRatio = previousRatio; + connectContext.getSessionVariable().forceSortAlgorithm = previousAlgorithm; + connectContext.getSessionVariable().setDisableNereidsRules(previousRules); + } + } + + @Test + public void testTopNFilterRetainsSelectedIndexThroughLazyScan() { + double previousRatio = connectContext.getSessionVariable().topnFilterRatio; + String previousRules = String.join(",", connectContext.getSessionVariable().getDisableNereidsRuleNames()); + connectContext.getSessionVariable().topnFilterRatio = 1000000; + connectContext.getSessionVariable().setDisableNereidsRules(previousRules + ",PRUNE_EMPTY_PARTITION"); + try { + PlanChecker checker = PlanChecker.from(connectContext) + .analyze("SELECT k,v FROM topn_storage_dup ORDER BY k LIMIT 1").rewrite().implement(); + PhysicalPlan plan = checker.getPhysicalPlan(); + PhysicalOlapScan scan = plan.collectToList(PhysicalOlapScan.class::isInstance).get(0); + PhysicalTopN topn = plan.>collectToList(PhysicalTopN.class::isInstance).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); + for (long indexId : List.of(table.getBaseIndexId(), 10002L)) { + PhysicalOlapScan selectedScan = Mockito.spy(scan); + Mockito.doReturn(table).when(selectedScan).getTable(); + Mockito.doReturn(indexId).when(selectedScan).getSelectedIndexId(); + for (PhysicalOlapScan target : List.of(selectedScan, new PhysicalLazyMaterializeOlapScan( + selectedScan, Mockito.mock(SlotReference.class), List.of()))) { + TopnFilterContext context = new TopnFilterContext(); + boolean accepted = target.accept(new TopnFilterPushDownVisitor(context), + new PushDownContext(topn, scan.getOutput().get(0), false)); + Assertions.assertTrue(accepted); + Assertions.assertEquals(1, context.getTopnFilters().size()); + } + } + } finally { + connectContext.getSessionVariable().topnFilterRatio = previousRatio; + connectContext.getSessionVariable().setDisableNereidsRules(previousRules); + } + } + @Test public void testUseTopNRf() { String sql = "select * from customer order by c_custkey limit 5"; @@ -74,6 +194,31 @@ public void testUseTopNRf() { PhysicalTopN localTopN = (PhysicalTopN) rfSource; Assertions.assertTrue(checker.getCascadesContext().getTopnFilterContext().isTopnFilterSource(localTopN)); + new PhysicalPlanTranslator(new PlanTranslatorContext(checker.getCascadesContext())).translatePlan(plan); + TTopnFilterDesc desc = checker.getCascadesContext().getTopnFilterContext() + .getTopnFilter(localTopN).toThrift(); + Assertions.assertFalse(desc.getTargetNodeIdToTargetExpr().isEmpty()); + } + + @Test + public void testSerializedSortAlgorithmAndOrderKeys() { + String previousAlgorithm = connectContext.getSessionVariable().forceSortAlgorithm; + double previousRatio = connectContext.getSessionVariable().topnFilterRatio; + connectContext.getSessionVariable().topnFilterRatio = 1000000; + try { + for (String algorithm : List.of("heap", "topn", "full")) { + connectContext.getSessionVariable().forceSortAlgorithm = algorithm; + TSortAlgorithm expectedAlgorithm = algorithm.equals("heap") ? TSortAlgorithm.HEAP_SORT + : algorithm.equals("topn") ? TSortAlgorithm.TOPN_SORT : TSortAlgorithm.FULL_SORT; + assertSerializedSortMetadata("SELECT id,k FROM topn_safety ORDER BY k LIMIT 3 OFFSET 2", expectedAlgorithm, 1); + assertSerializedSortMetadata("SELECT id,k FROM topn_safety ORDER BY k,id LIMIT 3 OFFSET 2", expectedAlgorithm, 2); + assertSerializedSortMetadata("SELECT id,k FROM topn_safety ORDER BY k,k LIMIT 3 OFFSET 2", expectedAlgorithm, 1); + assertSerializedSortMetadata("SELECT id,k FROM topn_safety ORDER BY k,id,k LIMIT 3 OFFSET 2", expectedAlgorithm, 2); + } + } finally { + connectContext.getSessionVariable().forceSortAlgorithm = previousAlgorithm; + connectContext.getSessionVariable().topnFilterRatio = previousRatio; + } } @Test @@ -99,7 +244,7 @@ public void testTopNFilterDoesNotRequireTotalSecondaryExpressions() { } @Test - public void testTopNFilterCrossesGroupedKeyAggregation() { + public void testTopNFilterCrossesAggregationOnGroupingKeys() { double previousRatio = connectContext.getSessionVariable().topnFilterRatio; boolean previousPushTopn = connectContext.getSessionVariable().pushTopnToAgg; boolean previousJoinReorder = connectContext.getSessionVariable().isDisableJoinReorder(); @@ -113,11 +258,13 @@ public void testTopNFilterCrossesGroupedKeyAggregation() { try { for (String sql : List.of( "SELECT k, count(1) FROM topn_safety GROUP BY k ORDER BY k LIMIT 1", + "SELECT k, sum(ts + 1) FROM topn_safety GROUP BY k ORDER BY abs(k) LIMIT 1", + "SELECT k, count(DISTINCT ts) FROM topn_safety GROUP BY k ORDER BY k LIMIT 1", "SELECT t.k, t.total FROM (SELECT k, count(1) AS total FROM topn_safety " + "GROUP BY k HAVING total > 10) t JOIN customer c ON t.k = c.c_custkey " + "ORDER BY t.k LIMIT 1", "SELECT k, sum(ts) FROM topn_safety GROUP BY GROUPING SETS((k)) ORDER BY k LIMIT 2")) { - // Exercise the runtime filter across aggregation, independently of TopN-to-agg pushdown. + // Inclusive target bounds retain all contributions at the boundary. PlanChecker.from(connectContext).checkPlannerResult(sql, planner -> { String explain = planner.getFragments().stream() .map(fragment -> fragment.getExplainString(TExplainLevel.NORMAL)) @@ -125,6 +272,12 @@ public void testTopNFilterCrossesGroupedKeyAggregation() { Assertions.assertTrue(explain.contains("TOPN OPT:"), explain); }); } + for (String sql : List.of( + "SELECT k, sum(ts) AS s FROM topn_safety GROUP BY k ORDER BY s LIMIT 1", + "SELECT k, count(*) FROM topn_safety GROUP BY GROUPING SETS((k), ()) " + + "ORDER BY k NULLS FIRST LIMIT 1")) { + assertTopNFilterPresence(sql, false); + } } finally { connectContext.getSessionVariable().topnFilterRatio = previousRatio; connectContext.getSessionVariable().pushTopnToAgg = previousPushTopn; @@ -133,6 +286,73 @@ public void testTopNFilterCrossesGroupedKeyAggregation() { } } + @Test + public void testTopNFilterPreservesAggregateAndUnionTargets() { + double previousRatio = connectContext.getSessionVariable().topnFilterRatio; + String previousAlgorithm = connectContext.getSessionVariable().forceSortAlgorithm; + boolean previousPushTopn = connectContext.getSessionVariable().pushTopnToAgg; + String previousDisabledRules = String.join(",", + connectContext.getSessionVariable().getDisableNereidsRuleNames()); + connectContext.getSessionVariable().topnFilterRatio = 1000000; + connectContext.getSessionVariable().forceSortAlgorithm = "heap"; + connectContext.getSessionVariable().pushTopnToAgg = false; + connectContext.getSessionVariable().setDisableNereidsRules(previousDisabledRules + + ",PRUNE_EMPTY_PARTITION,PUSH_DOWN_TOP_N_THROUGH_UNION"); + try { + for (boolean grouped : List.of(false, true)) { + for (String innerOrder : List.of("", " ORDER BY k LIMIT 2")) { + for (String order : List.of("ASC NULLS FIRST", "DESC NULLS FIRST", + "ASC NULLS LAST", "DESC NULLS LAST")) { + String sql = "SELECT k,s FROM (SELECT " + (grouped ? "k" : "CAST(1 AS INT)") + + " AS k, sum(ts + 1) AS s FROM (SELECT k,ts FROM topn_safety" + + innerOrder + ") a" + (grouped ? " GROUP BY k" : "") + + " UNION ALL SELECT k,ts AS s FROM topn_safety) u ORDER BY abs(k) " + + order + " LIMIT 1"; + PlanChecker checker = PlanChecker.from(connectContext).analyze(sql).rewrite().implement(); + PhysicalPlan plan = new PlanPostProcessors(checker.getCascadesContext()) + .process(checker.getPhysicalPlan()); + List> aggregates = plan.collectToList( + PhysicalHashAggregate.class::isInstance); + Assertions.assertFalse(aggregates.isEmpty(), plan.treeString()); + List aggregateScans = aggregates.stream() + .flatMap(aggregate -> aggregate.collectToList( + PhysicalRelation.class::isInstance).stream()) + .distinct().collect(Collectors.toList()); + Assertions.assertEquals(1, aggregateScans.size(), plan.treeString()); + List filters = checker.getCascadesContext().getTopnFilterContext().getTopnFilters(); + List outerFilters = filters.stream() + .filter(filter -> aggregates.stream().noneMatch(aggregate -> + aggregate.anyMatch(node -> node == filter.topn))) + .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. + 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() + .anyMatch(aggregateScans::contains), plan.treeString()); + // The independent TopN inside the aggregate can still filter its own scan. + long innerFilters = filters.stream() + .filter(filter -> aggregates.stream().anyMatch(aggregate -> + aggregate.anyMatch(node -> node == filter.topn))) + .filter(filter -> filter.targets.keySet().stream().anyMatch(aggregateScans::contains)) + .count(); + Assertions.assertEquals(innerOrder.isEmpty() ? 0 : 1, innerFilters, plan.treeString()); + new PhysicalPlanTranslator(new PlanTranslatorContext(checker.getCascadesContext())) + .translatePlan(plan); + Assertions.assertEquals(expectedTargets, outerFilters.get(0).legacyTargets.size()); + Assertions.assertFalse(outerFilters.get(0).toThrift().getTargetNodeIdToTargetExpr().isEmpty()); + } + } + } + } finally { + connectContext.getSessionVariable().topnFilterRatio = previousRatio; + connectContext.getSessionVariable().forceSortAlgorithm = previousAlgorithm; + connectContext.getSessionVariable().pushTopnToAgg = previousPushTopn; + connectContext.getSessionVariable().setDisableNereidsRules(previousDisabledRules); + } + } + @Test public void testTopNFilterSupportsNestedOrderKeyWithSiblingPredicate() { double previousRatio = connectContext.getSessionVariable().topnFilterRatio; @@ -320,4 +540,208 @@ public void testProbeExprNullableThroughRightOuterJoin() { new IntegerLiteral(5)); Assertions.assertEquals(expectedProbeExpr, probeExpr); } + + private void assertSerializedSortMetadata(String sql, TSortAlgorithm expectedAlgorithm, int expectedKeyCount) { + PlanChecker checker = PlanChecker.from(connectContext).analyze(sql).rewrite().implement(); + PhysicalPlan plan = new PlanPostProcessors(checker.getCascadesContext()) + .process(checker.getPhysicalPlan()); + TopnFilterContext context = checker.getCascadesContext().getTopnFilterContext(); + Assertions.assertFalse(context.getTopnFilters().isEmpty(), plan.treeString()); + new PhysicalPlanTranslator(new PlanTranslatorContext(checker.getCascadesContext())).translatePlan(plan); + for (TopnFilter filter : context.getTopnFilters()) { + TSortNode sort = filter.legacySortNode.treeToThrift().getNodes().get(0).getSortNode(); + Assertions.assertTrue(sort.isSetAlgorithm()); + Assertions.assertEquals(expectedAlgorithm, sort.getAlgorithm(), plan.treeString()); + Assertions.assertEquals(expectedKeyCount, sort.getSortInfo().getOrderingExprsSize()); + } + } + + @Test + public void testTopNFilterPreservesRightJoinRows() { + double previousRatio = connectContext.getSessionVariable().topnFilterRatio; + String previousAlgorithm = connectContext.getSessionVariable().forceSortAlgorithm; + boolean previousJoinReorder = connectContext.getSessionVariable().isDisableJoinReorder(); + String previousDisabledRules = String.join(",", + connectContext.getSessionVariable().getDisableNereidsRuleNames()); + connectContext.getSessionVariable().topnFilterRatio = 1000000; + connectContext.getSessionVariable().forceSortAlgorithm = "heap"; + connectContext.getSessionVariable().setDisableJoinReorder(true); + connectContext.getSessionVariable().setDisableNereidsRules(previousDisabledRules + + ",PRUNE_EMPTY_PARTITION,PUSH_DOWN_TOP_N_THROUGH_UNION,PUSH_DOWN_TOP_N_THROUGH_JOIN"); + try { + for (JoinType joinType : List.of(JoinType.RIGHT_OUTER_JOIN, JoinType.FULL_OUTER_JOIN, + JoinType.RIGHT_ANTI_JOIN, JoinType.RIGHT_SEMI_JOIN, JoinType.INNER_JOIN, + JoinType.LEFT_OUTER_JOIN)) { + String joinSql = joinType.name().replace('_', ' '); + String payload = joinType.isRightAntiJoin() || joinType.isRightSemiJoin() ? "b.ts" : "a.ts"; + for (String condition : List.of("=", "<")) { + for (String rank : List.of("b.k", "CAST(1 AS INT)", "CAST(sleep(0) AS INT)")) { + for (String order : List.of("ASC NULLS FIRST", "DESC NULLS FIRST", + "ASC NULLS LAST", "DESC NULLS LAST")) { + String sql = "SELECT k,payload FROM (SELECT /*+ leading(a b) */ " + rank + + " AS k," + payload + " AS payload FROM topn_safety a " + joinSql + + " topn_safety b ON a.k " + condition + " b.k UNION ALL " + + "SELECT k,ts AS payload FROM topn_safety) u ORDER BY k " + order + " LIMIT 1"; + PlanChecker checker = PlanChecker.from(connectContext).analyze(sql).rewrite().implement(); + PhysicalPlan plan = new PlanPostProcessors(checker.getCascadesContext()) + .process(checker.getPhysicalPlan()); + AbstractPhysicalJoin join = getOnlyJoin(plan); + Assertions.assertEquals(joinType, join.getJoinType(), plan.treeString()); + Assertions.assertEquals(condition.equals("="), join instanceof PhysicalHashJoin, + plan.treeString()); + boolean safeJoin = joinType.isInnerJoin() || joinType.isLeftOuterJoin() + || joinType.isRightSemiJoin(); + boolean safeNullOrder = !joinType.isLeftOuterJoin() || order.contains("LAST"); + boolean canResolveProbe = condition.equals("=") || rank.startsWith("CAST"); + assertSubtreeTopNTargets(checker, plan, join, safeJoin && safeNullOrder && canResolveProbe); + } + } + } + } + } finally { + connectContext.getSessionVariable().topnFilterRatio = previousRatio; + connectContext.getSessionVariable().forceSortAlgorithm = previousAlgorithm; + connectContext.getSessionVariable().setDisableJoinReorder(previousJoinReorder); + connectContext.getSessionVariable().setDisableNereidsRules(previousDisabledRules); + } + } + + @Test + public void testTopNFilterPreservesRightSemiMark() { + double previousRatio = connectContext.getSessionVariable().topnFilterRatio; + String previousAlgorithm = connectContext.getSessionVariable().forceSortAlgorithm; + String previousDisabledRules = String.join(",", + connectContext.getSessionVariable().getDisableNereidsRuleNames()); + connectContext.getSessionVariable().topnFilterRatio = 1000000; + connectContext.getSessionVariable().forceSortAlgorithm = "heap"; + connectContext.getSessionVariable().setDisableNereidsRules(previousDisabledRules + + ",PRUNE_EMPTY_PARTITION,PUSH_DOWN_TOP_N_THROUGH_UNION,PUSH_DOWN_TOP_N_THROUGH_JOIN"); + try { + for (String order : List.of("ASC NULLS FIRST", "DESC NULLS FIRST", + "ASC NULLS LAST", "DESC NULLS LAST")) { + String sql = "SELECT k,m FROM (SELECT CAST(1 AS INT) AS k, " + + "EXISTS(SELECT 1 FROM topn_safety a WHERE a.k = b.k) AS m FROM topn_safety b " + + "UNION ALL SELECT k,k IS NOT NULL AS m FROM topn_safety) u ORDER BY k " + order + " LIMIT 1"; + PlanChecker checker = PlanChecker.from(connectContext).analyze(sql).rewrite().implement(); + // JoinCommute permits this hash mark alternative. Exercise it independently of costing. + PhysicalPlan plan = (PhysicalPlan) checker.getPhysicalPlan().rewriteUp(node -> { + if (node instanceof PhysicalHashJoin && ((PhysicalHashJoin) node).isMarkJoin()) { + PhysicalHashJoin join = (PhysicalHashJoin) node; + Assertions.assertFalse(join.getHashJoinConjuncts().isEmpty()); + Assertions.assertTrue(join.getMarkJoinConjuncts().isEmpty()); + if (join.getJoinType().isLeftSemiJoin()) { + PhysicalHashJoin swapped = new PhysicalHashJoin<>(JoinType.RIGHT_SEMI_JOIN, + join.getHashJoinConjuncts(), join.getOtherJoinConjuncts(), + join.getMarkJoinConjuncts(), join.getDistributeHint(), + join.getMarkJoinSlotReference(), join.getLogicalProperties(), + join.right(), join.left()); + swapped = swapped.withPhysicalPropertiesAndStats(join.getPhysicalProperties(), join.getStats()); + swapped.addFilterConjuncts(join.getFilterConjuncts()); + swapped.setShouldTranslateOutput(join.isShouldTranslateOutput()); + return swapped; + } + } + return node; + }); + plan = new PlanPostProcessors(checker.getCascadesContext()).process(plan); + AbstractPhysicalJoin join = getOnlyJoin(plan); + Assertions.assertEquals(JoinType.RIGHT_SEMI_JOIN, join.getJoinType(), plan.treeString()); + Assertions.assertTrue(join.isMarkJoin(), plan.treeString()); + assertSubtreeTopNTargets(checker, plan, join, false); + } + } finally { + connectContext.getSessionVariable().topnFilterRatio = previousRatio; + connectContext.getSessionVariable().forceSortAlgorithm = previousAlgorithm; + connectContext.getSessionVariable().setDisableNereidsRules(previousDisabledRules); + } + } + + @Test + public void testTopNFilterDoesNotReevaluateVolatileProbe() { + double previousRatio = connectContext.getSessionVariable().topnFilterRatio; + String previousAlgorithm = connectContext.getSessionVariable().forceSortAlgorithm; + String previousDisabledRules = String.join(",", + connectContext.getSessionVariable().getDisableNereidsRuleNames()); + connectContext.getSessionVariable().topnFilterRatio = 1000000; + connectContext.getSessionVariable().forceSortAlgorithm = "heap"; + connectContext.getSessionVariable().setDisableNereidsRules(previousDisabledRules + + ",PRUNE_EMPTY_PARTITION,PUSH_DOWN_TOP_N_THROUGH_UNION"); + try { + Map ranks = Map.of( + "if(rand(3) < 0.5, 0, 1)", false, + "if(rand(3) < 0.5, 0, 1) + k", false, + "cast(if(rand(3) < 0.5, 0, 1) + k AS BIGINT)", false, + "length(uuid())", false, + "abs(k)", true, + "CAST(1 AS INT)", true); + for (Map.Entry rank : ranks.entrySet()) { + for (String order : List.of("ASC NULLS FIRST", "DESC NULLS FIRST", + "ASC NULLS LAST", "DESC NULLS LAST")) { + String sql = "SELECT k,payload FROM (SELECT " + rank.getKey() + + " AS k,ts AS payload FROM topn_safety UNION ALL " + + "SELECT k,ts AS payload FROM topn_safety) u ORDER BY k " + order + " LIMIT 1"; + PlanChecker checker = PlanChecker.from(connectContext).analyze(sql).rewrite().implement(); + PhysicalPlan plan = new PlanPostProcessors(checker.getCascadesContext()) + .process(checker.getPhysicalPlan()); + List unions = plan.collectToList(PhysicalUnion.class::isInstance); + Assertions.assertEquals(1, unions.size(), plan.treeString()); + Plan branch = unions.get(0).child(0); + Assertions.assertEquals(1, branch.collectToList(PhysicalRelation.class::isInstance).size()); + Assertions.assertEquals(!rank.getValue(), branch.anyMatch(node -> ((Plan) node).getExpressions() + .stream().anyMatch(Expression::containsVolatileExpression)), plan.treeString()); + // Alias expansion must reject the volatile branch while retaining the deterministic sibling. + assertSubtreeTopNTargets(checker, plan, branch, rank.getValue()); + if (!rank.getValue()) { + // Without a sibling, the source must not register any scan filter at all. + assertTopNFilterPresence("SELECT " + rank.getKey() + " AS rk,ts FROM topn_safety " + + "ORDER BY rk " + order + " LIMIT 1", false); + } + } + } + } finally { + connectContext.getSessionVariable().topnFilterRatio = previousRatio; + connectContext.getSessionVariable().forceSortAlgorithm = previousAlgorithm; + connectContext.getSessionVariable().setDisableNereidsRules(previousDisabledRules); + } + } + + private AbstractPhysicalJoin getOnlyJoin(PhysicalPlan plan) { + List> joins = plan.collectToList(AbstractPhysicalJoin.class::isInstance); + Assertions.assertEquals(1, joins.size(), plan.treeString()); + return joins.get(0); + } + + private void assertSubtreeTopNTargets(PlanChecker checker, PhysicalPlan plan, Plan subtree, + boolean allowSubtreeTarget) { + List filters = checker.getCascadesContext().getTopnFilterContext().getTopnFilters(); + List outerFilters = filters.stream() + .filter(filter -> !subtree.anyMatch(node -> node == filter.topn)).collect(Collectors.toList()); + Assertions.assertEquals(1, outerFilters.size(), plan.treeString()); + TopnFilter filter = outerFilters.get(0); + List subtreeScans = subtree.collectToList(PhysicalRelation.class::isInstance); + // The safe UNION sibling still supplies a bound; check the other branch independently. + Assertions.assertEquals(allowSubtreeTarget ? 2 : 1, filter.targets.size(), plan.treeString()); + Assertions.assertEquals(allowSubtreeTarget, + filter.targets.keySet().stream().anyMatch(subtreeScans::contains), plan.treeString()); + new PhysicalPlanTranslator(new PlanTranslatorContext(checker.getCascadesContext())).translatePlan(plan); + Assertions.assertEquals(allowSubtreeTarget ? 2 : 1, filter.legacyTargets.size()); + Assertions.assertEquals(allowSubtreeTarget ? 2 : 1, + filter.toThrift().getTargetNodeIdToTargetExprSize()); + } + + private void assertTopNFilterPresence(String sql, boolean expected) { + PlanChecker checker = PlanChecker.from(connectContext).analyze(sql).rewrite().implement(); + PhysicalPlan plan = new PlanPostProcessors(checker.getCascadesContext()) + .process(checker.getPhysicalPlan()); + Assertions.assertEquals(expected, + !checker.getCascadesContext().getTopnFilterContext().getTopnFilters().isEmpty(), + plan.treeString()); + if (expected) { + new PhysicalPlanTranslator(new PlanTranslatorContext(checker.getCascadesContext())).translatePlan(plan); + for (TopnFilter filter : checker.getCascadesContext().getTopnFilterContext().getTopnFilters()) { + TTopnFilterDesc desc = filter.toThrift(); + Assertions.assertEquals(filter.legacyTargets.size(), desc.getTargetNodeIdToTargetExprSize()); + } + } + } } diff --git a/regression-test/data/changelog_p0/topn_runtime_filter_changelog.out b/regression-test/data/changelog_p0/topn_runtime_filter_changelog.out new file mode 100644 index 00000000000000..53535b70cb0171 --- /dev/null +++ b/regression-test/data/changelog_p0/topn_runtime_filter_changelog.out @@ -0,0 +1,25 @@ +-- This file is automatically generated. You should know what you did if you want to edit this +-- !slot_asc_disabled -- +1 999 + +-- !slot_asc_enabled -- +1 999 + +-- !slot_desc_disabled -- +1 999 + +-- !slot_desc_enabled -- +1 999 + +-- !expr_asc_disabled -- +1 999 + +-- !expr_asc_enabled -- +1 999 + +-- !expr_desc_disabled -- +1 999 + +-- !expr_desc_enabled -- +1 999 + diff --git a/regression-test/data/topn_optimize/topn_runtime_filter_fe_guards.out b/regression-test/data/topn_optimize/topn_runtime_filter_fe_guards.out new file mode 100644 index 00000000000000..c75546978f9b5a --- /dev/null +++ b/regression-test/data/topn_optimize/topn_runtime_filter_fe_guards.out @@ -0,0 +1,73 @@ +-- This file is automatically generated. You should know what you did if you want to edit this +-- !scalar_sum_disabled -- +1 true + +-- !scalar_sum_enabled -- +1 true + +-- !grouped_sum_disabled -- +1 true + +-- !grouped_sum_enabled -- +1 true + +-- !right_anti_disabled -- +1 999 + +-- !right_anti_enabled -- +1 999 + +-- !right_anti_constant_disabled -- +1 999 + +-- !right_anti_constant_enabled -- +1 999 + +-- !right_outer_disabled -- +1 true + +-- !right_outer_enabled -- +1 true + +-- !right_outer_constant_disabled -- +1 true + +-- !right_outer_constant_enabled -- +1 true + +-- !full_outer_disabled -- +1 true + +-- !full_outer_enabled -- +1 true + +-- !full_outer_constant_disabled -- +1 true + +-- !full_outer_constant_enabled -- +1 true + +-- !volatile_disabled -- +0 true + +-- !volatile_enabled -- +0 true + +-- !volatile_with_slot_disabled -- +0 true + +-- !volatile_with_slot_enabled -- +0 true + +-- !inner_control_disabled -- +1 true + +-- !inner_control_enabled -- +1 true + +-- !mark_disabled -- +1 true + +-- !mark_enabled -- +1 true + diff --git a/regression-test/data/topn_optimize/topn_runtime_filter_safety.out b/regression-test/data/topn_optimize/topn_runtime_filter_safety.out new file mode 100644 index 00000000000000..1b0686a475a0db --- /dev/null +++ b/regression-test/data/topn_optimize/topn_runtime_filter_safety.out @@ -0,0 +1,13 @@ +-- This file is automatically generated. You should know what you did if you want to edit this +-- !recursive_baseline -- +0 0 +1 10 +2 10 +3 10 + +-- !recursive_enabled -- +0 0 +1 10 +2 10 +3 10 + diff --git a/regression-test/data/topn_optimize/topn_runtime_filter_storage_merge.out b/regression-test/data/topn_optimize/topn_runtime_filter_storage_merge.out new file mode 100644 index 00000000000000..41b0c08e14c089 --- /dev/null +++ b/regression-test/data/topn_optimize/topn_runtime_filter_storage_merge.out @@ -0,0 +1,73 @@ +-- This file is automatically generated. You should know what you did if you want to edit this +-- !agg_slot_asc_disabled -- +1 true + +-- !agg_slot_asc_enabled -- +1 true + +-- !agg_slot_desc_disabled -- +1 true + +-- !agg_slot_desc_enabled -- +1 true + +-- !agg_expr_asc_disabled -- +1 true + +-- !agg_expr_asc_enabled -- +1 true + +-- !agg_expr_desc_disabled -- +1 true + +-- !agg_expr_desc_enabled -- +1 true + +-- !mor_slot_asc_disabled -- +1 true + +-- !mor_slot_asc_enabled -- +1 true + +-- !mor_slot_desc_disabled -- +1 true + +-- !mor_slot_desc_enabled -- +1 true + +-- !mor_expr_asc_disabled -- +1 true + +-- !mor_expr_asc_enabled -- +1 true + +-- !mor_expr_desc_disabled -- +1 true + +-- !mor_expr_desc_enabled -- +1 true + +-- !mow_slot_asc_disabled -- +1 true + +-- !mow_slot_asc_enabled -- +1 true + +-- !mow_slot_desc_disabled -- +1 true + +-- !mow_slot_desc_enabled -- +1 true + +-- !mow_expr_asc_disabled -- +1 true + +-- !mow_expr_asc_enabled -- +1 true + +-- !mow_expr_desc_disabled -- +1 true + +-- !mow_expr_desc_enabled -- +1 true + diff --git a/regression-test/data/topn_optimize/topn_runtime_filter_target_mode.out b/regression-test/data/topn_optimize/topn_runtime_filter_target_mode.out new file mode 100644 index 00000000000000..17a7f0b64f8c09 --- /dev/null +++ b/regression-test/data/topn_optimize/topn_runtime_filter_target_mode.out @@ -0,0 +1,67 @@ +-- This file is automatically generated. You should know what you did if you want to edit this +-- !grouped_asc_nulls_first_disabled -- +\N true + +-- !grouped_asc_nulls_first_enabled -- +\N true + +-- !storage_asc_nulls_first_disabled -- +\N true + +-- !storage_asc_nulls_first_enabled -- +\N true + +-- !grouped_asc_nulls_last_disabled -- +1 true + +-- !grouped_asc_nulls_last_enabled -- +1 true + +-- !storage_asc_nulls_last_disabled -- +1 true + +-- !storage_asc_nulls_last_enabled -- +1 true + +-- !grouped_desc_nulls_first_disabled -- +\N true + +-- !grouped_desc_nulls_first_enabled -- +\N true + +-- !storage_desc_nulls_first_disabled -- +\N true + +-- !storage_desc_nulls_first_enabled -- +\N true + +-- !grouped_desc_nulls_last_disabled -- +2 true + +-- !grouped_desc_nulls_last_enabled -- +2 true + +-- !storage_desc_nulls_last_disabled -- +2 true + +-- !storage_desc_nulls_last_enabled -- +2 true + +-- !group_expression_disabled -- +1 true + +-- !group_expression_enabled -- +1 true + +-- !storage_expression_disabled -- +1 true + +-- !storage_expression_enabled -- +1 true + +-- !scalar_constant_disabled -- +1 true + +-- !scalar_constant_enabled -- +1 true + diff --git a/regression-test/data/topn_optimize/topn_strict_runtime_predicate.out b/regression-test/data/topn_optimize/topn_strict_runtime_predicate.out new file mode 100644 index 00000000000000..2680fd05576c56 --- /dev/null +++ b/regression-test/data/topn_optimize/topn_strict_runtime_predicate.out @@ -0,0 +1,435 @@ +-- This file is automatically generated. You should know what you did if you want to edit this +-- !asc_baseline -- + + + + + + + + + + + + + + + + + + + + + +-- !desc_baseline -- +zulu +zulu +zulu +zulu +zulu +zulu +zulu +zulu +zulu +zulu +zulu +zulu +zulu +zulu +zulu +zulu +zulu +zulu +zulu +zulu + +-- !nulls_first_baseline -- +\N +\N +\N +\N +\N +\N +\N +\N +\N +\N +\N +\N +\N +\N +\N +\N +\N +\N +\N +\N + +-- !multi_baseline -- +\N 119 +\N 136 +\N 153 +\N 170 +\N 187 +\N 204 +\N 221 +\N 238 +\N 255 +\N 272 +\N 289 +\N 306 +\N 323 +\N 340 +\N 357 +\N 374 +\N 391 +\N 408 +\N 425 +\N 442 + +-- !expr_asc_baseline -- +0 +0 +0 +0 +0 +0 +0 +0 +0 +0 +0 +0 +0 +0 +0 +0 +0 +0 +0 +0 + +-- !expr_desc_baseline -- +6 +6 +6 +6 +6 +6 +6 +6 +6 +6 +6 +6 +6 +6 +6 +6 +6 +6 +6 +6 + +-- !join_asc_baseline -- + + + + + + + + + + + + + + + + + + + + + +-- !join_desc_baseline -- +zulu +zulu +zulu +zulu +zulu +zulu +zulu +zulu +zulu +zulu +zulu +zulu +zulu +zulu +zulu +zulu +zulu +zulu +zulu +zulu + +-- !join_multi_baseline -- +\N 51 6 +\N 51 100001 +\N 68 3 +\N 68 8 +\N 85 0 +\N 85 5 +\N 102 2 +\N 102 7 +\N 119 4 +\N 119 9 +\N 136 1 +\N 136 6 +\N 136 100001 +\N 153 3 +\N 153 8 +\N 170 0 +\N 170 5 +\N 187 2 +\N 187 7 +\N 204 4 + +-- !right_join_last_baseline -- + + +middle +middle +middle + +-- !right_join_first_baseline -- +\N + + + +middle + +-- !grouped_baseline -- + 31373 +middle 31372 +zulu 31372 + +-- !asc_strict -- + + + + + + + + + + + + + + + + + + + + + +-- !desc_strict -- +zulu +zulu +zulu +zulu +zulu +zulu +zulu +zulu +zulu +zulu +zulu +zulu +zulu +zulu +zulu +zulu +zulu +zulu +zulu +zulu + +-- !nulls_first_strict -- +\N +\N +\N +\N +\N +\N +\N +\N +\N +\N +\N +\N +\N +\N +\N +\N +\N +\N +\N +\N + +-- !multi_inclusive -- +\N 119 +\N 136 +\N 153 +\N 170 +\N 187 +\N 204 +\N 221 +\N 238 +\N 255 +\N 272 +\N 289 +\N 306 +\N 323 +\N 340 +\N 357 +\N 374 +\N 391 +\N 408 +\N 425 +\N 442 + +-- !expr_asc_strict -- +0 +0 +0 +0 +0 +0 +0 +0 +0 +0 +0 +0 +0 +0 +0 +0 +0 +0 +0 +0 + +-- !expr_desc_strict -- +6 +6 +6 +6 +6 +6 +6 +6 +6 +6 +6 +6 +6 +6 +6 +6 +6 +6 +6 +6 + +-- !join_asc_strict -- + + + + + + + + + + + + + + + + + + + + + +-- !join_desc_strict -- +zulu +zulu +zulu +zulu +zulu +zulu +zulu +zulu +zulu +zulu +zulu +zulu +zulu +zulu +zulu +zulu +zulu +zulu +zulu +zulu + +-- !join_multi_inclusive -- +\N 51 6 +\N 51 100001 +\N 68 3 +\N 68 8 +\N 85 0 +\N 85 5 +\N 102 2 +\N 102 7 +\N 119 4 +\N 119 9 +\N 136 1 +\N 136 6 +\N 136 100001 +\N 153 3 +\N 153 8 +\N 170 0 +\N 170 5 +\N 187 2 +\N 187 7 +\N 204 4 + +-- !right_join_last_strict -- + + +middle +middle +middle + +-- !right_join_first_no_filter -- +\N + + + +middle + +-- !grouped_strict -- + 31373 +middle 31372 +zulu 31372 + diff --git a/regression-test/suites/changelog_p0/topn_runtime_filter_changelog.groovy b/regression-test/suites/changelog_p0/topn_runtime_filter_changelog.groovy new file mode 100644 index 00000000000000..debf780301ccca --- /dev/null +++ b/regression-test/suites/changelog_p0/topn_runtime_filter_changelog.groovy @@ -0,0 +1,60 @@ +// 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. + +suite("topn_runtime_filter_changelog") { + sql "DROP TABLE IF EXISTS topn_storage_merge_log" + sql """CREATE TABLE topn_storage_merge_log (k INT NOT NULL, v BIGINT NOT NULL) + UNIQUE KEY(k) DISTRIBUTED BY HASH(k) BUCKETS 1 + PROPERTIES ('replication_num'='1', 'enable_unique_key_merge_on_write'='true', + 'disable_auto_compaction'='true', 'changelog.enable'='true', + 'changelog.need_historical_value'='true')""" + sql "INSERT INTO topn_storage_merge_log VALUES (1,10)" + sql "DELETE FROM topn_storage_merge_log WHERE k=1" + + sql "DROP TABLE IF EXISTS topn_changelog_control" + sql """CREATE TABLE topn_changelog_control (k INT NOT NULL, v BIGINT NOT NULL) + DUPLICATE KEY(k) DISTRIBUTED BY HASH(k) BUCKETS 1 + PROPERTIES ('replication_num'='1')""" + sql "INSERT INTO topn_changelog_control VALUES (1,999)" + + sql "SET enable_sql_cache=false" + sql "SET enable_query_cache=false" + sql "SET force_sort_algorithm='heap'" + sql "SET disable_nereids_rules='PUSH_DOWN_TOP_N_THROUGH_UNION'" + + for (String key : ["k", "abs(k)"]) { + for (String direction : ["ASC", "DESC"]) { + String tag = "${key == 'k' ? 'slot' : 'expr'}_${direction.toLowerCase()}" + // INSERT followed by DELETE cancels out. Only (1,999) is valid. + String query = """SELECT k,v FROM ( + SELECT k,v FROM topn_storage_merge_log@incr('incrementType'='MIN_DELTA') + UNION ALL SELECT k,v FROM topn_changelog_control + ) u ORDER BY ${key} ${direction} NULLS LAST LIMIT 1""" + sql "SET topn_filter_ratio=0" + "order_qt_${tag}_disabled"(query) + sql "SET topn_filter_ratio=1000000" + explain { + sql query + contains "UNION" + contains "TOP-N" + // Inclusive bounds retain every event at the changelog boundary. + multiContains "TOPN OPT:", 2 + } + "order_qt_${tag}_enabled"(query) + } + } +} diff --git a/regression-test/suites/topn_optimize/topn_runtime_filter_fe_guards.groovy b/regression-test/suites/topn_optimize/topn_runtime_filter_fe_guards.groovy new file mode 100644 index 00000000000000..57fa2f60971300 --- /dev/null +++ b/regression-test/suites/topn_optimize/topn_runtime_filter_fe_guards.groovy @@ -0,0 +1,149 @@ +// 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. + +suite("topn_runtime_filter_fe_guards") { + sql "DROP TABLE IF EXISTS topn_runtime_filter_fe_guards" + sql """CREATE TABLE topn_runtime_filter_fe_guards ( + id INT NOT NULL, k INT NOT NULL, v BIGINT NOT NULL + ) DUPLICATE KEY(id) DISTRIBUTED BY HASH(id) BUCKETS 1 + PROPERTIES ('replication_num'='1')""" + sql "INSERT INTO topn_runtime_filter_fe_guards VALUES (1,1,1), (2,1,1), (9,1,999)" + + sql "SET enable_sql_cache=false" + sql "SET enable_query_cache=false" + sql "SET disable_join_reorder=true" + sql "SET parallel_pipeline_task_num=1" + sql "SET force_sort_algorithm='heap'" + sql "SET push_topn_to_agg=false" + sql "SET disable_nereids_rules='PUSH_DOWN_TOP_N_THROUGH_UNION,PUSH_DOWN_TOP_N_THROUGH_JOIN'" + + // Check complete-row validity without adding a second ordering key, which would + // disable strict bounds. Either original boundary tie is a valid Top1 result. + def validateRow = { String query, String predicate -> + "SELECT k, (${predicate}) AS valid_row FROM (${query}) selected" + } + def checkQuery = { String name, String query, String operator, int targets -> + sql "SET topn_filter_ratio=0" + "order_qt_${name}_disabled"(query) + + sql "SET topn_filter_ratio=1000000" + explain { + sql query + contains "UNION" + contains "TOP-N" + contains operator + // Verify the eligible paths, including inclusive aggregate targets. + multiContains "TOPN OPT:", targets + } + "order_qt_${name}_enabled"(query) + } + + def scalarQuery = """SELECT k,s FROM ( + SELECT CAST(1 AS INT) AS k, SUM(v + 1) AS s + FROM topn_runtime_filter_fe_guards WHERE id=1 + UNION ALL + SELECT k,v AS s FROM topn_runtime_filter_fe_guards WHERE id=9 + ) u ORDER BY k NULLS LAST LIMIT 1""" + // Emptying SUM's input can synthesize (1,NULL), absent from the original result. + checkQuery("scalar_sum", validateRow(scalarQuery, "s <=> 2 OR s <=> 999"), "AGGREGATE", 2) + + def groupedQuery = """SELECT k,s FROM ( + SELECT k,SUM(v + 1) AS s FROM ( + SELECT k,v FROM topn_runtime_filter_fe_guards WHERE id=1 + UNION ALL + SELECT k,v FROM topn_runtime_filter_fe_guards WHERE id=2 + ) d GROUP BY k + UNION ALL + SELECT k,v AS s FROM topn_runtime_filter_fe_guards WHERE id=9 + ) u ORDER BY ABS(k) NULLS LAST LIMIT 1""" + // Separate inputs expose a group that can consume a prefix before a bound arrives. + checkQuery("grouped_sum", validateRow(groupedQuery, "s <=> 4 OR s <=> 999"), "AGGREGATE", 3) + + def antiQuery = """SELECT k,p FROM ( + SELECT b.k,b.v AS p + FROM (SELECT k FROM topn_runtime_filter_fe_guards WHERE id=1) a + RIGHT ANTI JOIN (SELECT k,v FROM topn_runtime_filter_fe_guards WHERE id=2) b + ON a.k=b.k + UNION ALL + SELECT k,v AS p FROM topn_runtime_filter_fe_guards WHERE id=9 + ) u ORDER BY k NULLS LAST LIMIT 1""" + // The anti branch is empty. Removing its left match would invent (1,1). + checkQuery("right_anti", antiQuery, "RIGHT ANTI JOIN", 1) + def constantAntiQuery = antiQuery.replace("SELECT b.k,b.v AS p", "SELECT CAST(1 AS INT) AS k,b.v AS p") + .replace("ON a.k=b.k", "ON a.k 1 OR p <=> 999"), "${joinType} JOIN", 1) + + // An empty input-column set cannot prove that an ordering key comes from the left. + String constantQuery = query.replace("SELECT b.k,a.v AS p", "SELECT CAST(1 AS INT) AS k,a.v AS p") + .replace("ON a.k=b.k", "ON a.k 1 OR p <=> 999"), + "${joinType} JOIN", 1) + } + + def volatileQuery = """SELECT k,id FROM ( + SELECT IF(RAND(3)<0.5,CAST(0 AS INT),CAST(1 AS INT)) AS k,id + FROM topn_runtime_filter_fe_guards WHERE id IN (1,2) + UNION ALL + SELECT k,id FROM topn_runtime_filter_fe_guards WHERE id=9 + ) u ORDER BY k NULLS LAST LIMIT 1""" + // Retain id in the output expression so the scan must return actual rows. + checkQuery("volatile", validateRow(volatileQuery, "id IN (1,2)"), "UNION", 1) + String slotVolatileQuery = volatileQuery.replace("CAST(1 AS INT)) AS k,id", "CAST(1 AS INT))+k-1 AS k,id") + checkQuery("volatile_with_slot", validateRow(slotVolatileQuery, "id IN (1,2)"), "UNION", 1) + + // Positive control: a safe join keeps its scan target as well as the UNION sibling. + String innerQuery = outerQuery.replace("RIGHT OUTER JOIN", "INNER JOIN") + .replace("SELECT b.k,a.v AS p", "SELECT a.k,a.v AS p") + checkQuery("inner_control", validateRow(innerQuery, "p <=> 1 OR p <=> 999"), "INNER JOIN", 2) + + // EXISTS can use either mark-join orientation. The FE unit test constructs the + // RIGHT SEMI alternative explicitly; this result check does not fix a CBO choice. + def markQuery = """SELECT k,m FROM ( + SELECT CAST(1 AS INT) AS k, + EXISTS (SELECT 1 FROM topn_runtime_filter_fe_guards a + WHERE a.id=1 AND a.k=b.k) AS m + FROM topn_runtime_filter_fe_guards b WHERE b.id=2 + UNION ALL + SELECT k,v>0 AS m FROM topn_runtime_filter_fe_guards WHERE id=9 + ) u ORDER BY k NULLS LAST LIMIT 1""" + sql "SET disable_join_reorder=false" + sql "SET topn_filter_ratio=0" + order_qt_mark_disabled markQuery + sql "SET topn_filter_ratio=1000000" + explain { + sql markQuery + verbose true + contains "UNION" + contains "isMarkJoin: true" + contains "TOPN OPT:" + } + order_qt_mark_enabled markQuery +} diff --git a/regression-test/suites/topn_optimize/topn_runtime_filter_safety.groovy b/regression-test/suites/topn_optimize/topn_runtime_filter_safety.groovy new file mode 100644 index 00000000000000..1733ca1edc5ddd --- /dev/null +++ b/regression-test/suites/topn_optimize/topn_runtime_filter_safety.groovy @@ -0,0 +1,42 @@ +// 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. + +suite("topn_runtime_filter_safety") { + sql "DROP TABLE IF EXISTS topn_runtime_filter_safety_a" + sql """CREATE TABLE topn_runtime_filter_safety_a ( + id INT NOT NULL, k INT NULL + ) DUPLICATE KEY(id) DISTRIBUTED BY HASH(id) BUCKETS 1 + PROPERTIES ('replication_num'='1')""" + sql """INSERT INTO topn_runtime_filter_safety_a + SELECT CAST(number AS INT), 10 FROM numbers('number'='1000000')""" + + def recursiveQuery = """WITH RECURSIVE r(depth, v) AS ( + SELECT CAST(0 AS INT), CAST(0 AS BIGINT) + UNION ALL + SELECT CAST(r.depth + 1 AS INT), s.v + FROM r CROSS JOIN ( + SELECT abs(k) AS v FROM topn_runtime_filter_safety_a + ORDER BY v NULLS FIRST LIMIT 1 + ) s WHERE r.depth < 3 + ) SELECT depth, v FROM r ORDER BY depth""" + + sql "SET topn_filter_ratio=0" + qt_recursive_baseline recursiveQuery + + sql "SET topn_filter_ratio=1000000" + qt_recursive_enabled recursiveQuery +} diff --git a/regression-test/suites/topn_optimize/topn_runtime_filter_storage_merge.groovy b/regression-test/suites/topn_optimize/topn_runtime_filter_storage_merge.groovy new file mode 100644 index 00000000000000..c9c787c9ca6a2f --- /dev/null +++ b/regression-test/suites/topn_optimize/topn_runtime_filter_storage_merge.groovy @@ -0,0 +1,84 @@ +// 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. + +suite("topn_runtime_filter_storage_merge") { + sql "DROP TABLE IF EXISTS topn_storage_merge_agg" + sql """CREATE TABLE topn_storage_merge_agg (k INT NOT NULL, v BIGINT SUM NOT NULL) + AGGREGATE KEY(k) DISTRIBUTED BY HASH(k) BUCKETS 1 + PROPERTIES ('replication_num'='1', 'disable_auto_compaction'='true')""" + // Separate rowsets keep the two contributions visible to the storage merge. + sql "INSERT INTO topn_storage_merge_agg VALUES (1,10)" + sql "INSERT INTO topn_storage_merge_agg VALUES (1,20)" + + sql "DROP TABLE IF EXISTS topn_storage_merge_control" + sql """CREATE TABLE topn_storage_merge_control (k INT NOT NULL, v BIGINT NOT NULL) + DUPLICATE KEY(k) DISTRIBUTED BY HASH(k) BUCKETS 1 + PROPERTIES ('replication_num'='1')""" + sql "INSERT INTO topn_storage_merge_control VALUES (1,999)" + + sql "DROP TABLE IF EXISTS topn_storage_merge_mor" + sql """CREATE TABLE topn_storage_merge_mor (k INT NOT NULL, v BIGINT NOT NULL) + UNIQUE KEY(k) DISTRIBUTED BY HASH(k) BUCKETS 1 + PROPERTIES ('replication_num'='1', 'enable_unique_key_merge_on_write'='false', + 'disable_auto_compaction'='true')""" + sql "INSERT INTO topn_storage_merge_mor VALUES (1,10)" + sql "INSERT INTO topn_storage_merge_mor VALUES (1,20)" + + sql "DROP TABLE IF EXISTS topn_storage_merge_mow" + sql """CREATE TABLE topn_storage_merge_mow (k INT NOT NULL, v BIGINT NOT NULL) + UNIQUE KEY(k) DISTRIBUTED BY HASH(k) BUCKETS 1 + PROPERTIES ('replication_num'='1', 'enable_unique_key_merge_on_write'='true')""" + sql "INSERT INTO topn_storage_merge_mow VALUES (1,10)" + sql "INSERT INTO topn_storage_merge_mow VALUES (1,20)" + + sql "SET enable_sql_cache=false" + sql "SET enable_query_cache=false" + sql "SET force_sort_algorithm='heap'" + sql "SET disable_nereids_rules='PUSH_DOWN_TOP_N_THROUGH_UNION'" + + def checkMerge = { String name, String source, String validPayload -> + for (String key : ["k", "abs(k)"]) { + for (String direction : ["ASC", "DESC"]) { + String tag = "${name}_${key == 'k' ? 'slot' : 'expr'}_${direction.toLowerCase()}" + // A second ordering key would disable strict bounds. Accept either complete tie, + // but reject partial SUM values, stale unique rows, or invented changelog events. + String query = """SELECT k, (${validPayload}) AS valid_row FROM ( + SELECT k,v FROM ( + SELECT k,v FROM ${source} + UNION ALL + SELECT k,v FROM topn_storage_merge_control + ) u ORDER BY ${key} ${direction} NULLS LAST LIMIT 1 + ) selected""" + sql "SET topn_filter_ratio=0" + "order_qt_${tag}_disabled"(query) + sql "SET topn_filter_ratio=1000000" + explain { + sql query + contains "UNION" + contains "TOP-N" + // Keep both targets. BE retains ties on the merging scan only. + multiContains "TOPN OPT:", 2 + } + "order_qt_${tag}_enabled"(query) + } + } + } + checkMerge("agg", "topn_storage_merge_agg", "v <=> 30 OR v <=> 999") + checkMerge("mor", "topn_storage_merge_mor", "v <=> 20 OR v <=> 999") + // Ordinary MOW reads do not merge contributions at query time and remain eligible. + checkMerge("mow", "topn_storage_merge_mow", "v <=> 20 OR v <=> 999") +} 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 new file mode 100644 index 00000000000000..3d4ca2b036a609 --- /dev/null +++ b/regression-test/suites/topn_optimize/topn_runtime_filter_target_mode.groovy @@ -0,0 +1,98 @@ +// 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. + +suite("topn_runtime_filter_target_mode") { + sql "DROP TABLE IF EXISTS topn_target_input" + sql """CREATE TABLE topn_target_input (k INT NULL, v BIGINT NOT NULL) + DUPLICATE KEY(k) DISTRIBUTED BY HASH(k) BUCKETS 1 + PROPERTIES ('replication_num'='1')""" + sql "INSERT INTO topn_target_input VALUES (1,10),(1,20),(2,40),(2,50),(NULL,7),(NULL,8)" + + sql "DROP TABLE IF EXISTS topn_target_aggregate" + sql """CREATE TABLE topn_target_aggregate (k INT NULL, v BIGINT SUM NOT NULL) + AGGREGATE KEY(k) DISTRIBUTED BY HASH(k) BUCKETS 1 + PROPERTIES ('replication_num'='1', 'disable_auto_compaction'='true')""" + // Keep separate rowsets so the scan must combine stored contributions. + sql "INSERT INTO topn_target_aggregate VALUES (1,10),(2,40),(NULL,7)" + sql "INSERT INTO topn_target_aggregate VALUES (1,20),(2,50),(NULL,8)" + + sql "DROP TABLE IF EXISTS topn_target_control" + sql """CREATE TABLE topn_target_control (k INT NULL, v BIGINT NOT NULL) + DUPLICATE KEY(k) DISTRIBUTED BY HASH(k) BUCKETS 1 + PROPERTIES ('replication_num'='1')""" + sql "INSERT INTO topn_target_control VALUES (1,999),(2,999),(NULL,999)" + + sql "SET enable_sql_cache=false" + sql "SET enable_query_cache=false" + sql "SET force_sort_algorithm='heap'" + sql "SET parallel_pipeline_task_num=1" + sql "SET push_topn_to_agg=false" + sql "SET disable_nereids_rules='PUSH_DOWN_TOP_N_THROUGH_UNION'" + + def checkQuery = { String name, String query, String expectedValue, boolean sqlAggregate -> + // A second ordering key would hide the single-key strict-bound behavior. + // Check the complete payload while accepting either legal boundary tie. + String checked = "SELECT k, (v <=> (${expectedValue}) OR v <=> 999) AS valid_row " + + "FROM (${query}) selected" + sql "SET topn_filter_ratio=0" + "order_qt_${name}_disabled"(checked) + sql "SET topn_filter_ratio=1000000" + explain { + sql checked + contains "UNION" + contains "TOP-N" + if (sqlAggregate) { + contains "AGGREGATE" + } + // Both targets remain enabled. BE keeps equality only on the aggregate path. + multiContains "TOPN OPT:", 2 + } + "order_qt_${name}_enabled"(checked) + } + + String groupValue = "CASE WHEN k IS NULL THEN 15 WHEN k=1 THEN 30 ELSE 90 END" + for (String order : ["ASC NULLS FIRST", "ASC NULLS LAST", "DESC NULLS FIRST", "DESC NULLS LAST"]) { + String suffix = order.toLowerCase().replace(" ", "_") + checkQuery("grouped_${suffix}", """SELECT k,v FROM ( + SELECT k,SUM(v) AS v FROM topn_target_input GROUP BY k + UNION ALL SELECT k,v FROM topn_target_control + ) u ORDER BY k ${order} LIMIT 1""", groupValue, true) + checkQuery("storage_${suffix}", """SELECT k,v FROM ( + SELECT k,v FROM topn_target_aggregate + UNION ALL SELECT k,v FROM topn_target_control + ) u ORDER BY k ${order} LIMIT 1""", groupValue, false) + } + + checkQuery("group_expression", """SELECT k,v FROM ( + SELECT k,SUM(v) AS v FROM topn_target_input GROUP BY k + UNION ALL SELECT k,v FROM topn_target_control + ) u ORDER BY ABS(k) NULLS LAST LIMIT 1""", groupValue, true) + checkQuery("storage_expression", """SELECT k,v FROM ( + SELECT k,v FROM topn_target_aggregate + UNION ALL SELECT k,v FROM topn_target_control + ) u ORDER BY ABS(k) NULLS LAST LIMIT 1""", groupValue, false) + checkQuery("scalar_constant", """SELECT k,v FROM ( + SELECT CAST(1 AS INT) AS k,SUM(v) AS v FROM topn_target_input + UNION ALL SELECT k,v FROM topn_target_control + ) u ORDER BY k NULLS LAST LIMIT 1""", "135", true) + + // 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" + notContains "TOPN OPT:" + } +} diff --git a/regression-test/suites/topn_optimize/topn_strict_runtime_predicate.groovy b/regression-test/suites/topn_optimize/topn_strict_runtime_predicate.groovy new file mode 100644 index 00000000000000..367b31150af995 --- /dev/null +++ b/regression-test/suites/topn_optimize/topn_strict_runtime_predicate.groovy @@ -0,0 +1,106 @@ +// 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. + +suite("topn_strict_runtime_predicate") { + sql "DROP TABLE IF EXISTS topn_strict_runtime_predicate" + sql """CREATE TABLE topn_strict_runtime_predicate ( + id BIGINT NOT NULL, k VARCHAR(32) NULL + ) DUPLICATE KEY(id) DISTRIBUTED BY HASH(id) BUCKETS 1 + PROPERTIES ('replication_num'='1')""" + sql """INSERT INTO topn_strict_runtime_predicate + SELECT number, CASE WHEN number % 17 = 0 THEN NULL + WHEN number % 3 = 0 THEN '' + WHEN number % 3 = 1 THEN 'middle' ELSE 'zulu' END + FROM numbers('number'='100000')""" + + sql "DROP TABLE IF EXISTS topn_strict_runtime_predicate_join" + sql """CREATE TABLE topn_strict_runtime_predicate_join ( + id BIGINT NOT NULL + ) DUPLICATE KEY(id) DISTRIBUTED BY HASH(id) BUCKETS 1 + PROPERTIES ('replication_num'='1')""" + sql """INSERT INTO topn_strict_runtime_predicate_join + SELECT number FROM numbers('number'='10') UNION ALL SELECT 100001""" + + // Keep the offset candidates and the requested number of equal boundary values. + sql "SET topn_filter_ratio=0" + qt_asc_baseline "SELECT k FROM topn_strict_runtime_predicate ORDER BY k ASC NULLS LAST LIMIT 20 OFFSET 7" + qt_desc_baseline "SELECT k FROM topn_strict_runtime_predicate ORDER BY k DESC NULLS LAST LIMIT 20 OFFSET 7" + qt_nulls_first_baseline "SELECT k FROM topn_strict_runtime_predicate ORDER BY k NULLS FIRST LIMIT 20 OFFSET 7" + qt_multi_baseline "SELECT k,id FROM topn_strict_runtime_predicate ORDER BY k,id LIMIT 20 OFFSET 7" + qt_expr_asc_baseline """SELECT abs(id % 13 - 6) AS v FROM topn_strict_runtime_predicate + ORDER BY v LIMIT 20 OFFSET 7""" + qt_expr_desc_baseline """SELECT abs(id % 13 - 6) AS v FROM topn_strict_runtime_predicate + ORDER BY v DESC LIMIT 20 OFFSET 7""" + qt_join_asc_baseline """SELECT /*+ leading(a b) */ a.k + FROM topn_strict_runtime_predicate a JOIN topn_strict_runtime_predicate_join b + ON a.id % 5 = b.id % 5 ORDER BY a.k ASC NULLS LAST LIMIT 20 OFFSET 7""" + qt_join_desc_baseline """SELECT /*+ leading(a b) */ a.k + FROM topn_strict_runtime_predicate a JOIN topn_strict_runtime_predicate_join b + ON a.id % 5 = b.id % 5 ORDER BY a.k DESC NULLS LAST LIMIT 20 OFFSET 7""" + qt_join_multi_baseline """SELECT /*+ leading(a b) */ a.k,a.id,b.id + FROM topn_strict_runtime_predicate a JOIN topn_strict_runtime_predicate_join b + ON a.id % 5 = b.id % 5 ORDER BY a.k,a.id,b.id LIMIT 20 OFFSET 7""" + qt_right_join_last_baseline """SELECT /*+ leading(a b) */ a.k + FROM topn_strict_runtime_predicate a RIGHT OUTER JOIN topn_strict_runtime_predicate_join b + ON a.id = b.id ORDER BY a.k NULLS LAST LIMIT 5 OFFSET 1""" + qt_right_join_first_baseline """SELECT /*+ leading(a b) */ a.k + FROM topn_strict_runtime_predicate a RIGHT OUTER JOIN topn_strict_runtime_predicate_join b + ON a.id = b.id ORDER BY a.k NULLS FIRST LIMIT 5 OFFSET 1""" + qt_grouped_baseline """SELECT k,count(*) FROM topn_strict_runtime_predicate + GROUP BY k ORDER BY k NULLS LAST LIMIT 3""" + + sql "SET topn_filter_ratio=1000000" + qt_asc_strict "SELECT k FROM topn_strict_runtime_predicate ORDER BY k ASC NULLS LAST LIMIT 20 OFFSET 7" + qt_desc_strict "SELECT k FROM topn_strict_runtime_predicate ORDER BY k DESC NULLS LAST LIMIT 20 OFFSET 7" + qt_nulls_first_strict "SELECT k FROM topn_strict_runtime_predicate ORDER BY k NULLS FIRST LIMIT 20 OFFSET 7" + qt_multi_inclusive "SELECT k,id FROM topn_strict_runtime_predicate ORDER BY k,id LIMIT 20 OFFSET 7" + qt_expr_asc_strict """SELECT abs(id % 13 - 6) AS v FROM topn_strict_runtime_predicate + ORDER BY v LIMIT 20 OFFSET 7""" + qt_expr_desc_strict """SELECT abs(id % 13 - 6) AS v FROM topn_strict_runtime_predicate + ORDER BY v DESC LIMIT 20 OFFSET 7""" + qt_join_asc_strict """SELECT /*+ leading(a b) */ a.k + FROM topn_strict_runtime_predicate a JOIN topn_strict_runtime_predicate_join b + ON a.id % 5 = b.id % 5 ORDER BY a.k ASC NULLS LAST LIMIT 20 OFFSET 7""" + qt_join_desc_strict """SELECT /*+ leading(a b) */ a.k + FROM topn_strict_runtime_predicate a JOIN topn_strict_runtime_predicate_join b + ON a.id % 5 = b.id % 5 ORDER BY a.k DESC NULLS LAST LIMIT 20 OFFSET 7""" + qt_join_multi_inclusive """SELECT /*+ leading(a b) */ a.k,a.id,b.id + FROM topn_strict_runtime_predicate a JOIN topn_strict_runtime_predicate_join b + ON a.id % 5 = b.id % 5 ORDER BY a.k,a.id,b.id LIMIT 20 OFFSET 7""" + qt_right_join_last_strict """SELECT /*+ leading(a b) */ a.k + FROM topn_strict_runtime_predicate a RIGHT OUTER JOIN topn_strict_runtime_predicate_join b + ON a.id = b.id ORDER BY a.k NULLS LAST LIMIT 5 OFFSET 1""" + qt_right_join_first_no_filter """SELECT /*+ leading(a b) */ a.k + FROM topn_strict_runtime_predicate a RIGHT OUTER JOIN topn_strict_runtime_predicate_join b + ON a.id = b.id ORDER BY a.k NULLS FIRST LIMIT 5 OFFSET 1""" + qt_grouped_strict """SELECT k,count(*) FROM topn_strict_runtime_predicate + GROUP BY k ORDER BY k NULLS LAST LIMIT 3""" + + // Group-key targets keep inclusive bounds, so boundary groups retain all contributions. + explain { + sql """SELECT /*+ SET_VAR(push_topn_to_agg=false) */ k,count(*) + FROM topn_strict_runtime_predicate GROUP BY k ORDER BY k NULLS LAST LIMIT 3""" + contains "AGGREGATE" + contains "TOPN OPT:" + } + explain { + sql """SELECT /*+ SET_VAR(push_topn_to_agg=false) */ k,sum(id + 1) + FROM topn_strict_runtime_predicate GROUP BY k ORDER BY k NULLS LAST LIMIT 3""" + contains "AGGREGATE" + contains "TOPN OPT:" + } +}