diff --git a/be/src/format_v2/orc/orc_reader.cpp b/be/src/format_v2/orc/orc_reader.cpp index 32b8aed2a888db..7c5b49b16810c2 100644 --- a/be/src/format_v2/orc/orc_reader.cpp +++ b/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(_state->condition_cache_ctx->base_granule)); + const auto required_granules = static_cast( + end_granule - static_cast(_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 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( _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(_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) { diff --git a/be/src/format_v2/orc/orc_reader.h b/be/src/format_v2/orc/orc_reader.h index e7de8cf5c1a6fc..279ae7373ccf7c 100644 --- a/be/src/format_v2/orc/orc_reader.h +++ b/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; diff --git a/be/src/format_v2/parquet/parquet_scan.cpp b/be/src/format_v2/parquet/parquet_scan.cpp index 99bfcc8e0ab7ca..a7602cd11b27d0 100644 --- a/be/src/format_v2/parquet/parquet_scan.cpp +++ b/be/src/format_v2/parquet/parquet_scan.cpp @@ -605,9 +605,17 @@ void ParquetScanScheduler::set_condition_cache_context(std::shared_ptrbase_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(end_granule - _condition_cache_ctx->base_granule)); return; } diff --git a/be/src/format_v2/table_reader.cpp b/be/src/format_v2/table_reader.cpp index 11e2b30df6de23..69ae3b562cf379 100644 --- a/be/src/format_v2/table_reader.cpp +++ b/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(); _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; } diff --git a/be/src/format_v2/table_reader.h b/be/src/format_v2/table_reader.h index da2cb351ff68ce..ec179abb92f49b 100644 --- a/be/src/format_v2/table_reader.h +++ b/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, ¤t_rows, ¤t_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(); diff --git a/be/src/storage/segment/condition_cache.cpp b/be/src/storage/segment/condition_cache.cpp index d6499f0eaa8e24..78ca6a6d05328b 100644 --- a/be/src/storage/segment/condition_cache.cpp +++ b/be/src/storage/segment/condition_cache.cpp @@ -38,7 +38,8 @@ bool ConditionCache::lookup(const KeyType& key, ConditionCacheHandle* handle) { } template -void ConditionCache::insert(const KeyType& key, std::shared_ptr> result) { +void ConditionCache::insert(const KeyType& key, std::shared_ptr> result, + int64_t base_granule) { auto encoded_key = key.encode(); if (encoded_key.empty()) { return; @@ -46,6 +47,7 @@ void ConditionCache::insert(const KeyType& key, std::shared_ptr cache_value_ptr = std::make_unique(); 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(const CacheKey& k template bool ConditionCache::lookup( const ExternalCacheKey& key, ConditionCacheHandle* handle); template void ConditionCache::insert( - const CacheKey& key, std::shared_ptr> filter_result); + const CacheKey& key, std::shared_ptr> filter_result, + int64_t base_granule); template void ConditionCache::insert( - const ExternalCacheKey& key, std::shared_ptr> filter_result); + const ExternalCacheKey& key, std::shared_ptr> filter_result, + int64_t base_granule); } // namespace doris::segment_v2 diff --git a/be/src/storage/segment/condition_cache.h b/be/src/storage/segment/condition_cache.h index a189312ee1427a..0ef5534a622826 100644 --- a/be/src/storage/segment/condition_cache.h +++ b/be/src/storage/segment/condition_cache.h @@ -48,6 +48,7 @@ struct ConditionCacheContext { bool is_hit = false; std::shared_ptr> 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> 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(format_version); + key.append(buf, 41); return key; } }; @@ -135,7 +145,8 @@ class ConditionCache : public LRUCachePolicy { bool lookup(const KeyType& key, ConditionCacheHandle* handle); template - void insert(const KeyType& key, std::shared_ptr> filter_result); + void insert(const KeyType& key, std::shared_ptr> 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; diff --git a/be/test/format_v2/orc/orc_reader_test.cpp b/be/test/format_v2/orc/orc_reader_test.cpp index 0e36498c41d9bf..c2e7bf6323645c 100644 --- a/be/test/format_v2/orc/orc_reader_test.cpp +++ b/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& first_values) { + const std::vector& first_values, + int64_t rows_per_stripe = 200) { auto type = std::unique_ptr<::orc::Type>(::orc::Type::buildTypeFromString("struct")); 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(); ctx->is_hit = true; const auto base_granule = stripe_first_row / ConditionCacheContext::GRANULE_SIZE; + ctx->base_granule = static_cast(base_granule); const auto last_granule = (stripe_first_row + layout[stripe_index].rows - 1) / ConditionCacheContext::GRANULE_SIZE; ctx->filter_result = std::make_shared>(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 schema; + ASSERT_TRUE(reader->get_schema(&schema).ok()); + ASSERT_EQ(schema.size(), 1); + + auto request = std::make_shared(); + request->predicate_columns = {field_projection(0)}; + request->conjuncts.push_back( + VExprContext::create_shared(std::make_shared(0, 5000))); + ASSERT_TRUE(reader->open(request).ok()); + EXPECT_EQ(reader->get_total_rows(), ROWS_PER_STRIPE); + + auto cache_ctx = std::make_shared(); + cache_ctx->filter_result = std::make_shared>( + (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(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; diff --git a/be/test/format_v2/parquet/parquet_scan_test.cpp b/be/test/format_v2/parquet/parquet_scan_test.cpp index edd51c4520307a..a101936c391889 100644 --- a/be/test/format_v2/parquet/parquet_scan_test.cpp +++ b/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(); + ctx->is_hit = true; + ctx->base_granule = 0; + ctx->filter_result = std::make_shared>(std::vector {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); diff --git a/be/test/format_v2/table_reader_test.cpp b/be/test/format_v2/table_reader_test.cpp index 252bfa5c04d73f..528da7338fd160 100644 --- a/be/test/format_v2/table_reader_test.cpp +++ b/be/test/format_v2/table_reader_test.cpp @@ -975,9 +975,12 @@ struct FakeFileReaderState { int close_count = 0; int64_t total_rows = 2; int64_t aggregate_count = -1; + int64_t condition_cache_base_granule = 0; + size_t condition_cache_num_granules = 0; bool eof_with_first_batch = true; bool inject_delete_conjunct = false; bool stop_during_aggregate = false; + bool stop_during_read = false; bool not_found_during_init = false; std::shared_ptr last_request; std::shared_ptr condition_cache_ctx; @@ -1068,6 +1071,10 @@ class FakeFileReader final : public FileReader { } } + if (_state->stop_during_read) { + DORIS_CHECK(_state->io_ctx != nullptr); + _state->io_ctx->should_stop = true; + } _returned_batch = true; *rows = 2; *eof = _state->eof_with_first_batch; @@ -1105,6 +1112,12 @@ class FakeFileReader final : public FileReader { void set_condition_cache_context(std::shared_ptr ctx) override { _state->condition_cache_ctx = std::move(ctx); + if (_state->condition_cache_ctx != nullptr && !_state->condition_cache_ctx->is_hit) { + _state->condition_cache_ctx->base_granule = _state->condition_cache_base_granule; + if (_state->condition_cache_num_granules > 0) { + _state->condition_cache_ctx->num_granules = _state->condition_cache_num_granules; + } + } } int64_t get_total_rows() const override { return _state->total_rows; } @@ -1843,6 +1856,8 @@ TEST(TableReaderTest, ConditionCacheMissPublishesBitmapAfterReaderEof) { RuntimeState state {TQueryOptions(), TQueryGlobals()}; auto fake_state = std::make_shared(); fake_state->total_rows = ConditionCacheContext::GRANULE_SIZE; + fake_state->condition_cache_base_granule = 7; + fake_state->condition_cache_num_granules = 1; FakeTableReader reader(file_schema, fake_state); ASSERT_TRUE(reader.init({ .projected_columns = projected_columns, @@ -1867,13 +1882,20 @@ TEST(TableReaderTest, ConditionCacheMissPublishesBitmapAfterReaderEof) { ASSERT_NE(fake_state->condition_cache_ctx, nullptr); EXPECT_FALSE(fake_state->condition_cache_ctx->is_hit); - segment_v2::ConditionCache::ExternalCacheKey key("fake-table-reader-input", 0, -1, 7, 0, -1); + segment_v2::ConditionCache::ExternalCacheKey legacy_key("fake-table-reader-input", 0, -1, 7, 0, + -1); segment_v2::ConditionCacheHandle handle; + EXPECT_FALSE(cache.get()->lookup(legacy_key, &handle)); + segment_v2::ConditionCache::ExternalCacheKey key( + "fake-table-reader-input", 0, -1, 7, 0, -1, + segment_v2::ConditionCache::ExternalCacheKey::BASE_GRANULE_AWARE_VERSION); ASSERT_TRUE(cache.get()->lookup(key, &handle)); const auto cached_bitmap = handle.get_filter_result(); ASSERT_NE(cached_bitmap, nullptr); ASSERT_FALSE(cached_bitmap->empty()); + EXPECT_EQ(cached_bitmap->size(), 1); EXPECT_TRUE((*cached_bitmap)[0]); + EXPECT_EQ(handle.get_base_granule(), 7); ASSERT_TRUE(reader.close().ok()); } @@ -1919,11 +1941,67 @@ TEST(TableReaderTest, ConditionCacheMissIsDroppedWhenReaderClosesBeforeEof) { EXPECT_FALSE(fake_state->condition_cache_ctx->is_hit); ASSERT_TRUE(reader.close().ok()); - segment_v2::ConditionCache::ExternalCacheKey key("fake-table-reader-input", 0, -1, 7, 0, -1); + segment_v2::ConditionCache::ExternalCacheKey key( + "fake-table-reader-input", 0, -1, 7, 0, -1, + segment_v2::ConditionCache::ExternalCacheKey::BASE_GRANULE_AWARE_VERSION); segment_v2::ConditionCacheHandle handle; EXPECT_FALSE(cache.get()->lookup(key, &handle)); } +// Scenario: a stop request can arrive while a physical read is in progress. Even if the reader +// converts that stop into eof=true, TableReader must not publish the partially visited MISS bitmap. +TEST(TableReaderTest, ConditionCacheMissIsDroppedWhenStopTurnsReadIntoEof) { + ScopedConditionCacheForTest cache; + + std::vector file_schema; + file_schema.push_back(make_file_column(0, "id", std::make_shared())); + + std::vector projected_columns; + projected_columns.push_back(make_table_column(0, "id", std::make_shared())); + set_name_identifiers(&projected_columns); + + auto io_ctx = std::make_shared(); + RuntimeState state {TQueryOptions(), TQueryGlobals()}; + auto fake_state = std::make_shared(); + fake_state->total_rows = ConditionCacheContext::GRANULE_SIZE * 2; + fake_state->condition_cache_num_granules = 2; + fake_state->stop_during_read = true; + fake_state->io_ctx = io_ctx; + FakeTableReader reader(file_schema, fake_state); + ASSERT_TRUE(reader.init({ + .projected_columns = projected_columns, + .conjuncts = {prepared_conjunct( + &state, table_int32_greater_than_expr(0, 0, 0))}, + .format = FileFormat::PARQUET, + .scan_params = nullptr, + .io_ctx = io_ctx, + .runtime_state = &state, + .scanner_profile = nullptr, + .condition_cache_digest = 7, + }) + .ok()); + + SplitReadOptions split_options; + split_options.current_range.__set_path("fake-table-reader-input"); + ASSERT_TRUE(reader.prepare_split(split_options).ok()); + + Block block = build_table_block(projected_columns); + bool eos = false; + ASSERT_TRUE(reader.get_block(&block, &eos).ok()); + EXPECT_TRUE(io_ctx->should_stop); + EXPECT_EQ(block.rows(), 2); + ASSERT_NE(fake_state->condition_cache_ctx, nullptr); + EXPECT_FALSE(fake_state->condition_cache_ctx->is_hit); + + segment_v2::ConditionCache::ExternalCacheKey key( + "fake-table-reader-input", 0, -1, 7, 0, -1, + segment_v2::ConditionCache::ExternalCacheKey::BASE_GRANULE_AWARE_VERSION); + segment_v2::ConditionCacheHandle handle; + EXPECT_FALSE(cache.get()->lookup(key, &handle)); + + ASSERT_TRUE(reader.close().ok()); +} + TEST(TableReaderTest, PushDownCountFromNewParquetReader) { const auto test_dir = std::filesystem::temp_directory_path() / "doris_table_reader_count_test"; std::filesystem::remove_all(test_dir);