[fix](predicate) Avoid recomputing predicates (#60484)

Author: Gabriel39

be/src/olap/column_predicate.h

@@ -225,6 +225,8 @@ class ColumnPredicate : public std::enable_shared_from_this<ColumnPredicate> {
     }
 
     virtual double get_ignore_threshold() const { return 0; }
+    // If this predicate acts on the key column, this predicate should be erased.
+    virtual bool could_be_erased() const { return false; }
     // Return the size of value set for IN/NOT IN predicates and 0 for others.
     virtual std::string debug_string() const {
         fmt::memory_buffer debug_string_buffer;

be/src/olap/comparison_predicate.h

@@ -52,6 +52,12 @@ class ComparisonPredicateBase final : public ColumnPredicate {
                        ColumnPredicate::debug_string());
         return fmt::to_string(debug_string_buffer);
     }
+    bool could_be_erased() const override {
+        if ((PT == PredicateType::NE && !_opposite) || (PT == PredicateType::EQ && _opposite)) {
+            return false;
+        }
+        return true;
+    }
 
     PredicateType type() const override { return PT; }
 

be/src/olap/in_list_predicate.h

@@ -145,6 +145,13 @@ class InListPredicateBase final : public ColumnPredicate {
 
     PredicateType type() const override { return PT; }
 
+    bool could_be_erased() const override {
+        if ((PT == PredicateType::NOT_IN_LIST && !_opposite) ||
+            (PT == PredicateType::IN_LIST && _opposite)) {
+            return false;
+        }
+        return true;
+    }
     Status evaluate(const vectorized::IndexFieldNameAndTypePair& name_with_type,
                     IndexIterator* iterator, uint32_t num_rows,
                     roaring::Roaring* result) const override {

be/src/olap/null_predicate.h

@@ -61,6 +61,7 @@ class NullPredicate final : public ColumnPredicate {
                        ColumnPredicate::debug_string(), _is_null);
         return fmt::to_string(debug_string_buffer);
     }
+    bool could_be_erased() const override { return true; }
 
     PredicateType type() const override;
 

be/src/pipeline/exec/olap_scan_operator.cpp

@@ -21,6 +21,7 @@
 
 #include <memory>
 #include <numeric>
+#include <optional>
 
 #include "cloud/cloud_meta_mgr.h"
 #include "cloud/cloud_storage_engine.h"
@@ -883,6 +884,8 @@ Status OlapScanLocalState::_build_key_ranges_and_filters() {
             DCHECK(_slot_id_to_predicates.count(iter->first) > 0);
             const auto& value_range = iter->second;
 
+            std::optional<int> key_to_erase;
+
             RETURN_IF_ERROR(std::visit(
                     [&](auto&& range) {
                         // make a copy or range and pass to extend_scan_key, keep the range unchanged
@@ -894,7 +897,7 @@ Status OlapScanLocalState::_build_key_ranges_and_filters() {
                                     _scan_keys.extend_scan_key(temp_range, p._max_scan_key_num,
                                                                &exact_range, &eos, &should_break));
                             if (exact_range) {
-                                _slot_id_to_value_range.erase(iter->first);
+                                key_to_erase = iter->first;
                             }
                         } else {
                             // if exceed max_pushdown_conditions_per_column, use whole_value_rang instead
@@ -907,6 +910,24 @@ Status OlapScanLocalState::_build_key_ranges_and_filters() {
                         return Status::OK();
                     },
                     value_range));
+
+            // Perform the erase operation after the visit is complete, still under lock
+            if (key_to_erase.has_value()) {
+                _slot_id_to_value_range.erase(*key_to_erase);
+
+                std::vector<std::shared_ptr<ColumnPredicate>> new_predicates;
+                for (const auto& it : _slot_id_to_predicates[*key_to_erase]) {
+                    if (!it->could_be_erased()) {
+                        new_predicates.push_back(it);
+                    }
+                }
+                if (new_predicates.empty()) {
+                    _slot_id_to_predicates.erase(*key_to_erase);
+                } else {
+                    _slot_id_to_predicates[*key_to_erase] = new_predicates;
+                }
+            }
+            // lock is released here when it goes out of scope
         }
         if (eos) {
             _eos = true;