[fix](sort) keep heap TopN pruning for expression order keys

be/src/exec/operator/sort_sink_operator.cpp

@@ -51,10 +51,15 @@ Status SortSinkLocalState::open(RuntimeState* state) {
     }
     switch (p._algorithm) {
     case TSortAlgorithm::HEAP_SORT: {
+        bool disable_local_topn_filter = false;
+        if (state->get_query_ctx()->has_runtime_predicate(p._node_id)) {
+            disable_local_topn_filter =
+                state->get_query_ctx()->get_runtime_predicate(p._node_id)
+                    .all_targets_are_slot();
+        }
         _shared_state->sorter = HeapSorter::create_shared(
                 _ordering_expr_ctxs, state, p._limit, p._offset, p._pool, p._is_asc_order,
-                p._nulls_first, p._child->row_desc(),
-                state->get_query_ctx()->has_runtime_predicate(p._node_id));
+            p._nulls_first, p._child->row_desc(), disable_local_topn_filter);
         break;
     }
     case TSortAlgorithm::TOPN_SORT: {

be/src/runtime/runtime_predicate.h

@@ -17,6 +17,7 @@
 
 #pragma once
 
+#include <algorithm>
 #include <functional>
 #include <memory>
 #include <mutex>
@@ -90,6 +91,13 @@ class RuntimePredicate {
         return _contexts.find(target_node_id)->second.target_is_slot();
     }
 
+    bool all_targets_are_slot() const {
+        std::shared_lock<std::shared_mutex> rlock(_rwlock);
+        return std::ranges::all_of(_contexts, [](const auto& entry) {
+            return entry.second.target_is_slot();
+        });
+    }
+
     const TExpr& get_texpr(int32_t target_node_id) const {
         check_target_node_id(target_node_id);
         return _contexts.find(target_node_id)->second.expr;