[fix](be) Preserve condition cache granule base

be/src/format_v2/orc/orc_reader.cpp

@@ -885,18 +885,43 @@ Status OrcReader::init(RuntimeState* state) {
     return Status::OK();
 }
 
+void OrcReader::_extend_condition_cache_context_for_current_range() {
+    if (_state->condition_cache_ctx == nullptr ||
+        _state->condition_cache_ctx->filter_result == nullptr ||
+        _state->condition_cache_ctx->is_hit) {
+        return;
+    }
+    const auto end_granule = (_state->row_reader_range_first_row + _state->row_reader_range_rows +
+                              ConditionCacheContext::GRANULE_SIZE - 1) /
+                             ConditionCacheContext::GRANULE_SIZE;
+    DORIS_CHECK(end_granule > static_cast<uint64_t>(_state->condition_cache_ctx->base_granule));
+    const auto required_granules = static_cast<size_t>(
+            end_granule - static_cast<uint64_t>(_state->condition_cache_ctx->base_granule));
+    if (_state->condition_cache_ctx->filter_result->size() < required_granules) {
+        _state->condition_cache_ctx->filter_result->resize(required_granules, false);
+    }
+    _state->condition_cache_ctx->num_granules =
+            std::max(_state->condition_cache_ctx->num_granules, required_granules);
+}
+
 void OrcReader::set_condition_cache_context(std::shared_ptr<ConditionCacheContext> ctx) {
     DORIS_CHECK(_state != nullptr);
     _state->condition_cache_ctx = std::move(ctx);
     if (_state->condition_cache_ctx != nullptr &&
-        _state->condition_cache_ctx->filter_result != nullptr) {
+        _state->condition_cache_ctx->filter_result != nullptr &&
+        !_state->condition_cache_ctx->is_hit) {
         _state->condition_cache_ctx->base_granule = static_cast<int64_t>(
                 _state->row_reader_range_first_row / ConditionCacheContext::GRANULE_SIZE);
+        _state->condition_cache_ctx->num_granules = 0;
+        _extend_condition_cache_context_for_current_range();
     }
 }
 
 int64_t OrcReader::get_total_rows() const {
     DORIS_CHECK(_state != nullptr);
+    if (_state->stripe_pruning_applied && _state->selected_stripe_ranges.empty()) {
+        return 0;
+    }
     if (_state->row_reader != nullptr) {
         return cast_set<int64_t>(_state->row_reader_range_rows);
     }
@@ -1543,6 +1568,7 @@ Status OrcReader::_create_row_reader() {
         _state->row_reader_range_end_row =
                 _state->row_reader_range_first_row + _state->row_reader_range_rows;
         _state->condition_cache_next_row = _state->row_reader_range_first_row;
+        _extend_condition_cache_context_for_current_range();
         _state->column_to_selected_batch_index.clear();
         size_t physical_read_column_count = 0;
         for (const auto file_column_id : _state->read_columns) {

be/src/format_v2/orc/orc_reader.h

@@ -154,6 +154,7 @@ class OrcReader final : public format::FileReader {
     bool _filter_has_row_level_predicates() const;
     Status _build_keep_filter(Block* file_block, size_t rows, IColumn::Filter* keep_filter) const;
     Status _filter_block(Block* file_block, size_t* rows) const;
+    void _extend_condition_cache_context_for_current_range();
     void _skip_condition_cache_false_granules(size_t* rows, bool* eof);
     void _mark_condition_cache_surviving_rows(const IColumn::Filter& keep_filter,
                                               size_t rows) const;

be/src/format_v2/parquet/parquet_scan.cpp

@@ -605,9 +605,17 @@ void ParquetScanScheduler::set_condition_cache_context(std::shared_ptr<Condition
         return;
     }
 
-    _condition_cache_ctx->base_granule =
-            _row_group_plans.front().first_file_row / ConditionCacheContext::GRANULE_SIZE;
     if (!_condition_cache_ctx->is_hit) {
+        _condition_cache_ctx->base_granule =
+                _row_group_plans.front().first_file_row / ConditionCacheContext::GRANULE_SIZE;
+        const auto& last_plan = _row_group_plans.back();
+        const int64_t end_granule = (last_plan.first_file_row + last_plan.row_group_rows +
+                                     ConditionCacheContext::GRANULE_SIZE - 1) /
+                                    ConditionCacheContext::GRANULE_SIZE;
+        DORIS_CHECK(end_granule > _condition_cache_ctx->base_granule);
+        _condition_cache_ctx->num_granules =
+                std::min(_condition_cache_ctx->filter_result->size(),
+                         static_cast<size_t>(end_granule - _condition_cache_ctx->base_granule));
         return;
     }
 

be/src/format_v2/table_reader.cpp

@@ -581,7 +581,8 @@ Status TableReader::_init_reader_condition_cache(const FileScanRequest& file_req
     const auto& file = *_current_file_description;
     _condition_cache_key = segment_v2::ConditionCache::ExternalCacheKey(
             file.path, file.mtime, file.file_size, _condition_cache_digest, file.range_start_offset,
-            file.range_size);
+            file.range_size,
+            segment_v2::ConditionCache::ExternalCacheKey::BASE_GRANULE_AWARE_VERSION);
 
     segment_v2::ConditionCacheHandle handle;
     const bool condition_cache_hit = cache->lookup(_condition_cache_key, &handle);
@@ -605,6 +606,10 @@ Status TableReader::_init_reader_condition_cache(const FileScanRequest& file_req
         _condition_cache_ctx = std::make_shared<ConditionCacheContext>();
         _condition_cache_ctx->is_hit = condition_cache_hit;
         _condition_cache_ctx->filter_result = _condition_cache;
+        _condition_cache_ctx->num_granules = _condition_cache->size();
+        if (condition_cache_hit) {
+            _condition_cache_ctx->base_granule = handle.get_base_granule();
+        }
         _data_reader.reader->set_condition_cache_context(_condition_cache_ctx);
     }
     return Status::OK();
@@ -624,8 +629,10 @@ void TableReader::_finalize_reader_condition_cache() {
         _condition_cache_ctx = nullptr;
         return;
     }
-    segment_v2::ConditionCache::instance()->insert(_condition_cache_key,
-                                                   std::move(_condition_cache));
+    DORIS_CHECK(_condition_cache_ctx->num_granules <= _condition_cache->size());
+    _condition_cache->resize(_condition_cache_ctx->num_granules);
+    segment_v2::ConditionCache::instance()->insert(
+            _condition_cache_key, std::move(_condition_cache), _condition_cache_ctx->base_granule);
     _condition_cache = nullptr;
     _condition_cache_ctx = nullptr;
 }

be/src/format_v2/table_reader.h

@@ -252,9 +252,10 @@ class TableReader {
             size_t current_rows = 0;
             RETURN_IF_ERROR(_data_reader.reader->get_block(&_data_reader.block_template,
                                                            &current_rows, &current_eof));
+            const bool stopped_during_read = _io_ctx != nullptr && _io_ctx->should_stop;
             if (current_rows == 0) {
                 if (current_eof) {
-                    _current_reader_reached_eof = true;
+                    _current_reader_reached_eof = !stopped_during_read;
                     RETURN_IF_ERROR(close_current_reader());
                 }
                 continue;
@@ -271,7 +272,7 @@ class TableReader {
                     _check_table_block_columns("after finalize_chunk", block, current_rows));
 #endif
             if (current_eof) {
-                _current_reader_reached_eof = true;
+                _current_reader_reached_eof = !stopped_during_read;
                 RETURN_IF_ERROR(close_current_reader());
             }
             return Status::OK();

be/src/storage/segment/condition_cache.cpp

@@ -38,14 +38,16 @@ bool ConditionCache::lookup(const KeyType& key, ConditionCacheHandle* handle) {
 }
 
 template <typename KeyType>
-void ConditionCache::insert(const KeyType& key, std::shared_ptr<std::vector<bool>> result) {
+void ConditionCache::insert(const KeyType& key, std::shared_ptr<std::vector<bool>> result,
+                            int64_t base_granule) {
     auto encoded_key = key.encode();
     if (encoded_key.empty()) {
         return;
     }
     std::unique_ptr<ConditionCache::CacheValue> cache_value_ptr =
             std::make_unique<ConditionCache::CacheValue>();
     cache_value_ptr->filter_result = result;
+    cache_value_ptr->base_granule = base_granule;
 
     ConditionCacheHandle(this, LRUCachePolicy::insert(encoded_key, (void*)cache_value_ptr.release(),
                                                       result->capacity(), result->capacity(),
@@ -58,8 +60,10 @@ template bool ConditionCache::lookup<ConditionCache::CacheKey>(const CacheKey& k
 template bool ConditionCache::lookup<ConditionCache::ExternalCacheKey>(
         const ExternalCacheKey& key, ConditionCacheHandle* handle);
 template void ConditionCache::insert<ConditionCache::CacheKey>(
-        const CacheKey& key, std::shared_ptr<std::vector<bool>> filter_result);
+        const CacheKey& key, std::shared_ptr<std::vector<bool>> filter_result,
+        int64_t base_granule);
 template void ConditionCache::insert<ConditionCache::ExternalCacheKey>(
-        const ExternalCacheKey& key, std::shared_ptr<std::vector<bool>> filter_result);
+        const ExternalCacheKey& key, std::shared_ptr<std::vector<bool>> filter_result,
+        int64_t base_granule);
 
 } // namespace doris::segment_v2

be/src/storage/segment/condition_cache.h

@@ -48,6 +48,7 @@ struct ConditionCacheContext {
     bool is_hit = false;
     std::shared_ptr<std::vector<bool>> filter_result; // per-granule: true = has surviving rows
     int64_t base_granule = 0;                         // global granule index of filter_result[0]
+    size_t num_granules = 0; // authoritative bitmap length; excludes allocation-only guard bits
     static constexpr int GRANULE_SIZE = 2048;
 };
 
@@ -80,35 +81,44 @@ class ConditionCache : public LRUCachePolicy {
     class CacheValue : public LRUCacheValueBase {
     public:
         std::shared_ptr<std::vector<bool>> filter_result;
+        // The bitmap coordinate system is part of the cached result. A later scan may prune a
+        // different first row group, so it must not derive this origin from its current plan.
+        int64_t base_granule = 0;
     };
 
     // Cache key for external tables (Hive ORC/Parquet)
     struct ExternalCacheKey {
+        static constexpr uint8_t BASE_GRANULE_AWARE_VERSION = 1;
+
         ExternalCacheKey() = default;
         ExternalCacheKey(const std::string& path_, int64_t modification_time_, int64_t file_size_,
-                         uint64_t digest_, int64_t start_offset_, int64_t size_)
+                         uint64_t digest_, int64_t start_offset_, int64_t size_,
+                         uint8_t format_version_ = 0)
                 : path(path_),
                   modification_time(modification_time_),
                   file_size(file_size_),
                   digest(digest_),
                   start_offset(start_offset_),
-                  size(size_) {}
+                  size(size_),
+                  format_version(format_version_) {}
         std::string path;
         int64_t modification_time = 0;
         int64_t file_size = 0;
         uint64_t digest = 0;
         int64_t start_offset = 0;
         int64_t size = 0;
+        uint8_t format_version = 0;
 
         [[nodiscard]] std::string encode() const {
             std::string key = path;
-            char buf[40];
+            char buf[41];
             memcpy(buf, &modification_time, 8);
             memcpy(buf + 8, &file_size, 8);
             memcpy(buf + 16, &digest, 8);
             memcpy(buf + 24, &start_offset, 8);
             memcpy(buf + 32, &size, 8);
-            key.append(buf, 40);
+            buf[40] = static_cast<char>(format_version);
+            key.append(buf, 41);
             return key;
         }
     };
@@ -135,7 +145,8 @@ class ConditionCache : public LRUCachePolicy {
     bool lookup(const KeyType& key, ConditionCacheHandle* handle);
 
     template <typename KeyType>
-    void insert(const KeyType& key, std::shared_ptr<std::vector<bool>> filter_result);
+    void insert(const KeyType& key, std::shared_ptr<std::vector<bool>> filter_result,
+                int64_t base_granule = 0);
 };
 
 class ConditionCacheHandle {
@@ -172,6 +183,11 @@ class ConditionCacheHandle {
         return ((ConditionCache::CacheValue*)_cache->value(_handle))->filter_result;
     }
 
+    int64_t get_base_granule() const {
+        DORIS_CHECK(_cache != nullptr);
+        return ((ConditionCache::CacheValue*)_cache->value(_handle))->base_granule;
+    }
+
 private:
     LRUCachePolicy* _cache = nullptr;
     Cache::Handle* _handle = nullptr;

be/test/format_v2/orc/orc_reader_test.cpp

@@ -3419,7 +3419,8 @@ void write_multi_stripe_orc_int_file(const std::string& file_path,
 }
 
 void write_multi_stripe_orc_int_only_file(const std::string& file_path,
-                                          const std::vector<int64_t>& first_values) {
+                                          const std::vector<int64_t>& first_values,
+                                          int64_t rows_per_stripe = 200) {
     auto type = std::unique_ptr<::orc::Type>(::orc::Type::buildTypeFromString("struct<id:int>"));
 
     MemoryOutputStream memory_stream(1024 * 1024);
@@ -3430,15 +3431,14 @@ void write_multi_stripe_orc_int_only_file(const std::string& file_path,
     auto writer = ::orc::createWriter(*type, &memory_stream, options);
 
     auto add_batch = [&](int64_t first_value) {
-        constexpr int64_t ROWS_PER_STRIPE = 200;
-        auto batch = writer->createRowBatch(ROWS_PER_STRIPE);
+        auto batch = writer->createRowBatch(rows_per_stripe);
         auto& struct_batch = dynamic_cast<::orc::StructVectorBatch&>(*batch);
         auto& id_batch = dynamic_cast<::orc::LongVectorBatch&>(*struct_batch.fields[0]);
-        for (int64_t row = 0; row < ROWS_PER_STRIPE; ++row) {
+        for (int64_t row = 0; row < rows_per_stripe; ++row) {
             id_batch.data[row] = first_value + row;
         }
-        struct_batch.numElements = ROWS_PER_STRIPE;
-        id_batch.numElements = ROWS_PER_STRIPE;
+        struct_batch.numElements = rows_per_stripe;
+        id_batch.numElements = rows_per_stripe;
         writer->add(*batch);
     };
 
@@ -5455,6 +5455,7 @@ TEST_F(NewOrcReaderTest, ConditionCacheHitUsesSplitBaseGranule) {
     auto ctx = std::make_shared<ConditionCacheContext>();
     ctx->is_hit = true;
     const auto base_granule = stripe_first_row / ConditionCacheContext::GRANULE_SIZE;
+    ctx->base_granule = static_cast<int64_t>(base_granule);
     const auto last_granule = (stripe_first_row + layout[stripe_index].rows - 1) /
                               ConditionCacheContext::GRANULE_SIZE;
     ctx->filter_result = std::make_shared<std::vector<bool>>(last_granule - base_granule + 1, true);
@@ -8724,6 +8725,59 @@ TEST_F(NewOrcReaderTest, SargConjunctReadsNonAdjacentStripeRangesAfterPruning) {
     EXPECT_EQ(reader->reader_statistics().filtered_group_rows, 200);
 }
 
+TEST_F(NewOrcReaderTest, ConditionCacheMissExtendsAcrossNonAdjacentStripeRanges) {
+    constexpr int64_t ROWS_PER_STRIPE = 2500;
+    const auto multi_stripe_file_path =
+            (_test_dir / "condition_cache_non_adjacent_stripes.orc").string();
+    write_multi_stripe_orc_int_only_file(multi_stripe_file_path, {1, 10000, 3000}, ROWS_PER_STRIPE);
+    ASSERT_EQ(get_orc_stripe_count(multi_stripe_file_path), 3);
+
+    auto reader = create_reader_for_path(multi_stripe_file_path);
+    RuntimeState state {TQueryOptions(), TQueryGlobals()};
+    ASSERT_TRUE(reader->init(&state).ok());
+
+    std::vector<format::ColumnDefinition> schema;
+    ASSERT_TRUE(reader->get_schema(&schema).ok());
+    ASSERT_EQ(schema.size(), 1);
+
+    auto request = std::make_shared<format::FileScanRequest>();
+    request->predicate_columns = {field_projection(0)};
+    request->conjuncts.push_back(
+            VExprContext::create_shared(std::make_shared<NullableInt32LessThanExpr>(0, 5000)));
+    ASSERT_TRUE(reader->open(request).ok());
+    EXPECT_EQ(reader->get_total_rows(), ROWS_PER_STRIPE);
+
+    auto cache_ctx = std::make_shared<ConditionCacheContext>();
+    cache_ctx->filter_result = std::make_shared<std::vector<bool>>(
+            (reader->get_total_rows() + ConditionCacheContext::GRANULE_SIZE - 1) /
+                            ConditionCacheContext::GRANULE_SIZE +
+                    1,
+            false);
+    reader->set_condition_cache_context(cache_ctx);
+    EXPECT_EQ(cache_ctx->base_granule, 0);
+    EXPECT_EQ(cache_ctx->num_granules, 2);
+
+    bool eof = false;
+    size_t result_rows = 0;
+    while (!eof) {
+        Block block = build_file_block(schema);
+        size_t rows = 0;
+        ASSERT_TRUE(reader->get_block(&block, &rows, &eof).ok());
+        result_rows += rows;
+    }
+
+    EXPECT_EQ(result_rows, 4500);
+    ASSERT_NE(cache_ctx->filter_result, nullptr);
+    EXPECT_EQ(cache_ctx->num_granules, 4);
+    ASSERT_EQ(cache_ctx->filter_result->size(), 4);
+    EXPECT_TRUE((*cache_ctx->filter_result)[0]);
+    EXPECT_TRUE((*cache_ctx->filter_result)[1]);
+    EXPECT_TRUE((*cache_ctx->filter_result)[2]);
+    EXPECT_TRUE((*cache_ctx->filter_result)[3]);
+    EXPECT_EQ(reader->reader_statistics().filtered_row_groups, 1);
+    EXPECT_EQ(reader->reader_statistics().filtered_group_rows, ROWS_PER_STRIPE);
+}
+
 TEST_F(NewOrcReaderTest, SargConjunctReturnsEofWhenAllStripesArePruned) {
     const auto multi_stripe_file_path = (_test_dir / "all_pruned_stripes.orc").string();
     write_multi_stripe_orc_int_file(multi_stripe_file_path);
@@ -8742,6 +8796,9 @@ TEST_F(NewOrcReaderTest, SargConjunctReturnsEofWhenAllStripesArePruned) {
     request->conjuncts.push_back(
             VExprContext::create_shared(std::make_shared<NullableInt32GreaterThanExpr>(0, 5000)));
     ASSERT_TRUE(reader->open(request).ok());
+    // TableReader uses this value to decide whether a condition-cache MISS bitmap should be
+    // created. When SARG pruning removes every stripe, there is no row-reader range to cache.
+    EXPECT_EQ(reader->get_total_rows(), 0);
 
     Block block = build_file_block(schema);
     size_t rows = 0;

be/test/format_v2/parquet/parquet_scan_test.cpp

@@ -537,6 +537,28 @@ TEST_F(ParquetScanTest, PlanRowGroupsSelectsAllRowGroupsWithoutFilters) {
     }
 }
 
+TEST(ParquetScanConditionCacheTest, HitKeepsCachedBaseWhenCurrentPlanStartsLater) {
+    format::parquet::RowGroupScanPlan plan;
+    plan.row_groups.push_back(
+            {.row_group_id = 1,
+             .first_file_row = ConditionCacheContext::GRANULE_SIZE,
+             .row_group_rows = ConditionCacheContext::GRANULE_SIZE,
+             .selected_ranges = {{.start = 0, .length = ConditionCacheContext::GRANULE_SIZE}},
+             .page_skip_plans = {}});
+
+    format::parquet::ParquetScanScheduler scheduler;
+    scheduler.set_plan(std::move(plan));
+    auto ctx = std::make_shared<ConditionCacheContext>();
+    ctx->is_hit = true;
+    ctx->base_granule = 0;
+    ctx->filter_result = std::make_shared<std::vector<bool>>(std::vector<bool> {false});
+    scheduler.set_condition_cache_context(ctx);
+
+    EXPECT_FALSE(scheduler.empty());
+    EXPECT_EQ(scheduler.condition_cache_filtered_rows(), 0);
+    EXPECT_EQ(ctx->base_granule, 0);
+}
+
 TEST_F(ParquetScanTest, PageIndexIntersectsMultipleFiltersAndBuildsSkipPlan) {
     write_page_index_pair_parquet_file(_file_path);
     auto parquet_file_reader = ::parquet::ParquetFileReader::OpenFile(_file_path, false);