[feat](condition cache) Support condition cache for external table

Author: jacktengg

be/src/olap/rowset/segment_v2/condition_cache.cpp

@@ -49,4 +49,32 @@ void ConditionCache::insert(const CacheKey& key, std::shared_ptr<std::vector<boo
                                    result->capacity(), result->capacity(), CachePriority::NORMAL));
 }
 
+bool ConditionCache::lookup(const ExternalCacheKey& key, ConditionCacheHandle* handle) {
+    auto encoded = key.encode();
+    if (encoded.empty()) {
+        return false;
+    }
+    auto* lru_handle = LRUCachePolicy::lookup(encoded);
+    if (lru_handle == nullptr) {
+        return false;
+    }
+    *handle = ConditionCacheHandle(this, lru_handle);
+    return true;
+}
+
+void ConditionCache::insert(const ExternalCacheKey& key,
+                            std::shared_ptr<std::vector<bool>> result) {
+    auto encoded = key.encode();
+    if (encoded.empty()) {
+        return;
+    }
+    std::unique_ptr<ConditionCache::CacheValue> cache_value_ptr =
+            std::make_unique<ConditionCache::CacheValue>();
+    cache_value_ptr->filter_result = result;
+
+    ConditionCacheHandle(this, LRUCachePolicy::insert(encoded, (void*)cache_value_ptr.release(),
+                                                      result->capacity(), result->capacity(),
+                                                      CachePriority::NORMAL));
+}
+
 } // namespace doris::segment_v2

be/src/olap/rowset/segment_v2/condition_cache.h

@@ -69,6 +69,30 @@ class ConditionCache : public LRUCachePolicy {
         std::shared_ptr<std::vector<bool>> filter_result;
     };
 
+    // Cache key for external tables (Hive ORC/Parquet)
+    struct ExternalCacheKey {
+        ExternalCacheKey(const std::string& path_, int64_t modification_time_, int64_t file_size_,
+                         uint64_t digest_)
+                : path(path_),
+                  modification_time(modification_time_),
+                  file_size(file_size_),
+                  digest(digest_) {}
+        std::string path;
+        int64_t modification_time;
+        int64_t file_size;
+        uint64_t digest;
+
+        [[nodiscard]] std::string encode() const {
+            std::string key = path;
+            char buf[24];
+            memcpy(buf, &modification_time, 8);
+            memcpy(buf + 8, &file_size, 8);
+            memcpy(buf + 16, &digest, 8);
+            key.append(buf, 24);
+            return key;
+        }
+    };
+
     // Create global instance of this class
     static ConditionCache* create_global_cache(size_t capacity, uint32_t num_shards = 16) {
         auto* res = new ConditionCache(capacity, num_shards);
@@ -90,6 +114,10 @@ class ConditionCache : public LRUCachePolicy {
     bool lookup(const CacheKey& key, ConditionCacheHandle* handle);
 
     void insert(const CacheKey& key, std::shared_ptr<std::vector<bool>> filter_result);
+
+    bool lookup(const ExternalCacheKey& key, ConditionCacheHandle* handle);
+
+    void insert(const ExternalCacheKey& key, std::shared_ptr<std::vector<bool>> filter_result);
 };
 
 class ConditionCacheHandle {

be/src/olap/rowset/segment_v2/row_ranges.h

@@ -192,6 +192,53 @@ class RowRanges {
         *result = std::move(tmp_range);
     }
 
+    // Calculates the exception (set difference) of the two specified RowRanges objects: left \ right.
+    // The result contains all row indexes that are in the left ranges but NOT in the right ranges.
+    // For example:
+    // [100, 300) \ [150, 200) = [100, 150), [200, 300)
+    // [100, 300) \ [0, 150) = [150, 300)
+    // [100, 300) \ [250, 400) = [100, 250)
+    // [100, 200) \ [200, 300) = [100, 200)
+    // [100, 300) \ [0, 400) = <EMPTY>
+    // [100, 200), [300, 400) \ [150, 350) = [100, 150), [350, 400)
+    static void ranges_exception(const RowRanges& left, const RowRanges& right, RowRanges* result) {
+        RowRanges tmp_range;
+        int right_index = 0;
+        for (auto it1 = left._ranges.begin(); it1 != left._ranges.end(); ++it1) {
+            int64_t current_from = it1->from();
+            int64_t current_to = it1->to();
+            for (int i = right_index; i < right._ranges.size(); ++i) {
+                const RowRange& range2 = right._ranges[i];
+                if (current_from >= current_to) {
+                    // Current range fully consumed
+                    break;
+                }
+                if (current_to <= range2.from()) {
+                    // Current remaining range is entirely before range2, no more subtraction needed
+                    break;
+                }
+                if (current_from >= range2.to()) {
+                    // range2 is entirely before the current remaining range, advance right_index
+                    right_index = i + 1;
+                    continue;
+                }
+                // There is overlap between [current_from, current_to) and range2
+                if (current_from < range2.from()) {
+                    // Left portion before the overlap: [current_from, range2.from())
+                    tmp_range.add(RowRange(current_from, range2.from()));
+                }
+                // Advance current_from past the overlap
+                current_from = range2.to();
+            }
+            // Add whatever remains of the current left range
+            if (current_from < current_to) {
+                tmp_range.add(RowRange(current_from, current_to));
+            }
+        }
+        *result = std::move(tmp_range);
+    }
+
+
     static roaring::Roaring ranges_to_roaring(const RowRanges& ranges) {
         roaring::Roaring result;
         for (auto it = ranges._ranges.begin(); it != ranges._ranges.end(); ++it) {
@@ -275,6 +322,23 @@ class RowRanges {
         _count += range_to_add.count();
     }
 
+    // Returns the row index (within the original row space) of the pos-th element
+    // across all ranges. For example, if ranges are [0,3000) and [8000,11000),
+    // pos=0 returns 0, pos=2999 returns 2999, pos=3000 returns 8000.
+    int64_t get_row_index_by_pos(int64_t pos) const {
+        size_t remaining = pos;
+        for (const auto& range : _ranges) {
+            size_t range_len = range.count();
+            if (remaining < range_len) {
+                return range.from() + remaining;
+            }
+            remaining -= range_len;
+        }
+        // pos is out of bounds; return -1 to indicate invalid
+        DCHECK(false) << "pos " << pos << " is out of bounds for RowRanges with count " << _count;
+        return -1;
+    }
+
     uint64_t get_digest(uint64_t seed) const {
         for (auto range : _ranges) {
             seed = range.get_digest(seed);

be/src/vec/exec/format/generic_reader.h

@@ -35,6 +35,16 @@ namespace doris::vectorized {
 
 class Block;
 class VSlotRef;
+
+// Context passed from FileScanner to readers for condition cache integration.
+// On MISS: readers populate filter_result per-granule during predicate evaluation.
+// On HIT: readers skip granules where filter_result[granule] == false.
+struct ConditionCacheContext {
+    bool is_hit = false;
+    std::shared_ptr<std::vector<bool>> filter_result; // per-granule: true = has surviving rows
+    static constexpr int GRANULE_SIZE = 2048;
+};
+
 // This a reader interface for all file readers.
 // A GenericReader is responsible for reading a file and return
 // a set of blocks with specified schema,
@@ -103,7 +113,16 @@ class GenericReader : public ProfileCollector {
     bool _fill_all_columns = false;
     TPushAggOp::type _push_down_agg_type {};
 
-    // For TopN queries, rows will be read according to row ids produced by TopN result.
+public:
+    // Pass condition cache context to the reader for HIT/MISS tracking.
+    virtual void set_condition_cache_context(std::shared_ptr<ConditionCacheContext> ctx) {}
+
+    // Returns true if this reader has delete operations (e.g. Iceberg position/equality deletes,
+    // Hive ACID deletes). Used to disable condition cache when deletes are present, since cached
+    // granule results may become stale if delete files change between queries.
+    virtual bool has_delete_operations() const { return false; }
+
+protected:
     bool _read_by_rows = false;
     std::list<int64_t> _row_ids;
 

be/src/vec/exec/format/orc/vorc_reader.cpp

@@ -2264,6 +2264,26 @@ Status OrcReader::_get_next_block_impl(Block* block, size_t* read_rows, bool* eo
             // reset decimal_scale_params_index;
             _decimal_scale_params_index = 0;
             try {
+                // Condition cache HIT: skip consecutive false granules before reading
+                if (_condition_cache_ctx && _condition_cache_ctx->is_hit) {
+                    int64_t current_row = _row_reader->getRowNumber();
+                    auto& cache = *_condition_cache_ctx->filter_result;
+                    int64_t granule = current_row / ConditionCacheContext::GRANULE_SIZE;
+                    int64_t max_granule = static_cast<int64_t>(cache.size());
+                    while (granule < max_granule && !cache[granule]) {
+                        granule++;
+                    }
+                    if (granule >= max_granule) {
+                        *eof = true;
+                        *read_rows = 0;
+                        return Status::OK();
+                    }
+                    int64_t target_row = granule * ConditionCacheContext::GRANULE_SIZE;
+                    if (target_row > current_row) {
+                        _row_reader->seekToRow(target_row);
+                    }
+                }
+                _last_batch_row_offset = _row_reader->getRowNumber();
                 rr = _row_reader->nextBatch(*_batch, block);
                 if (rr == 0 || _batch->numElements == 0) {
                     *eof = true;
@@ -2363,6 +2383,26 @@ Status OrcReader::_get_next_block_impl(Block* block, size_t* read_rows, bool* eo
             // reset decimal_scale_params_index;
             _decimal_scale_params_index = 0;
             try {
+                // Condition cache HIT: skip consecutive false granules before reading
+                if (_condition_cache_ctx && _condition_cache_ctx->is_hit) {
+                    int64_t current_row = _row_reader->getRowNumber();
+                    auto& cache = *_condition_cache_ctx->filter_result;
+                    int64_t granule = current_row / ConditionCacheContext::GRANULE_SIZE;
+                    int64_t max_granule = static_cast<int64_t>(cache.size());
+                    while (granule < max_granule && !cache[granule]) {
+                        granule++;
+                    }
+                    if (granule >= max_granule) {
+                        *eof = true;
+                        *read_rows = 0;
+                        return Status::OK();
+                    }
+                    int64_t target_row = granule * ConditionCacheContext::GRANULE_SIZE;
+                    if (target_row > current_row) {
+                        _row_reader->seekToRow(target_row);
+                    }
+                }
+                _last_batch_row_offset = _row_reader->getRowNumber();
                 rr = _row_reader->nextBatch(*_batch, block);
                 if (rr == 0 || _batch->numElements == 0) {
                     *eof = true;
@@ -2480,6 +2520,24 @@ Status OrcReader::_get_next_block_impl(Block* block, size_t* read_rows, bool* eo
                 bool can_filter_all = false;
                 RETURN_IF_ERROR_OR_CATCH_EXCEPTION(VExprContext::execute_conjuncts(
                         filter_conjuncts, &filters, block, &result_filter, &can_filter_all));
+
+                // Condition cache MISS: mark granules with surviving rows (non-lazy path)
+                if (_condition_cache_ctx && !_condition_cache_ctx->is_hit) {
+                    auto& cache = *_condition_cache_ctx->filter_result;
+                    auto* filter_data = result_filter.data();
+                    size_t num_rows = block->rows();
+                    for (size_t i = 0; i < num_rows; i++) {
+                        if (filter_data[i]) {
+                            size_t granule = (_last_batch_row_offset + i) /
+                                             ConditionCacheContext::GRANULE_SIZE;
+                            if (granule >= cache.size()) {
+                                cache.resize(granule + 1, false);
+                            }
+                            cache[granule] = true;
+                        }
+                    }
+                }
+
                 if (can_filter_all) {
                     for (auto& col : columns_to_filter) {
                         std::move(*block->get_by_position(col).column).assume_mutable()->clear();
@@ -2697,6 +2755,21 @@ Status OrcReader::filter(orc::ColumnVectorBatch& data, uint16_t* sel, uint16_t s
         sel[new_size] = i;
         new_size += result_filter_data[i] ? 1 : 0;
     }
+
+    // Condition cache MISS: mark granules with surviving rows
+    if (_condition_cache_ctx && !_condition_cache_ctx->is_hit && new_size > 0) {
+        auto& cache = *_condition_cache_ctx->filter_result;
+        for (uint16_t i = 0; i < size; i++) {
+            if (result_filter_data[i]) {
+                size_t granule = (_last_batch_row_offset + i) / ConditionCacheContext::GRANULE_SIZE;
+                if (granule >= cache.size()) {
+                    cache.resize(granule + 1, false);
+                }
+                cache[granule] = true;
+            }
+        }
+    }
+
     _statistics.lazy_read_filtered_rows += static_cast<int64_t>(size - new_size);
     data.numElements = new_size;
     return Status::OK();

be/src/vec/exec/format/orc/vorc_reader.h

@@ -232,6 +232,16 @@ class OrcReader : public GenericReader {
 
     bool count_read_rows() override { return true; }
 
+    void set_condition_cache_context(std::shared_ptr<ConditionCacheContext> ctx) override {
+        _condition_cache_ctx = std::move(ctx);
+    }
+
+    bool has_delete_operations() const override {
+        return (_position_delete_ordered_rowids != nullptr &&
+                !_position_delete_ordered_rowids->empty()) ||
+               (_delete_rows != nullptr && !_delete_rows->empty());
+    }
+
 protected:
     void _collect_profile_before_close() override;
 
@@ -700,6 +710,9 @@ class OrcReader : public GenericReader {
     std::unique_ptr<orc::ColumnVectorBatch> _batch;
     std::unique_ptr<orc::Reader> _reader = nullptr;
     std::unique_ptr<orc::RowReader> _row_reader;
+
+    int64_t _last_batch_row_offset = -1;
+    std::shared_ptr<ConditionCacheContext> _condition_cache_ctx;
     std::unique_ptr<ORCFilterImpl> _orc_filter;
     orc::RowReaderOptions _row_reader_options;