This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 126bb220312 [fix](be) Preserve condition cache granule base (#65496)
126bb220312 is described below
commit 126bb220312f6d774b3524df379628a1b1069ba4
Author: Gabriel <[email protected]>
AuthorDate: Mon Jul 13 09:27:42 2026 +0800
[fix](be) Preserve condition cache granule base (#65496)
### What problem does this PR solve?
Issue Number: None
Related PR: None
Problem Summary: External file condition-cache entries stored only their
survivor bitmap. On a
later cache hit, the physical reader recomputed the bitmap base from the
current metadata-pruned
scan plan. If pruning selected a different first row group, bitmap
indexes shifted and valid rows
could be silently skipped.
This change stores the bitmap's base granule with the cache value. The
V2 table reader restores the
stored base on cache hits, while V2 Parquet and ORC readers derive a
base from their current plan
only when producing a new cache entry. V1 reader behavior is unchanged.
### Release note
Fix incorrect row filtering when a V2 external file condition-cache hit
uses a different pruned
scan plan from the cache-producing scan.
### Check List (For Author)
- Test: Unit Test
-
`ParquetScanConditionCacheTest.HitKeepsCachedBaseWhenCurrentPlanStartsLater`
- `TableReaderTest.ConditionCacheMissPublishesBitmapAfterReaderEof`
- `NewOrcReaderTest.ConditionCacheHitUsesSplitBaseGranule`
- Behavior changed: Yes. V2 condition-cache hits use the bitmap
coordinate base stored with the
cache entry instead of recomputing it from the current scan plan.
- Does this need documentation: No
---
be/src/format_v2/orc/orc_reader.cpp | 28 ++++++++-
be/src/format_v2/orc/orc_reader.h | 1 +
be/src/format_v2/parquet/parquet_scan.cpp | 12 +++-
be/src/format_v2/table_reader.cpp | 13 +++-
be/src/format_v2/table_reader.h | 5 +-
be/src/storage/segment/condition_cache.cpp | 10 ++-
be/src/storage/segment/condition_cache.h | 26 ++++++--
be/test/format_v2/orc/orc_reader_test.cpp | 69 +++++++++++++++++++--
be/test/format_v2/parquet/parquet_scan_test.cpp | 22 +++++++
be/test/format_v2/table_reader_test.cpp | 82 ++++++++++++++++++++++++-
10 files changed, 244 insertions(+), 24 deletions(-)
diff --git a/be/src/format_v2/orc/orc_reader.cpp
b/be/src/format_v2/orc/orc_reader.cpp
index 32b8aed2a88..7c5b49b1681 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<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) {
diff --git a/be/src/format_v2/orc/orc_reader.h
b/be/src/format_v2/orc/orc_reader.h
index e7de8cf5c1a..279ae7373cc 100644
--- a/be/src/format_v2/orc/orc_reader.h
+++ b/be/src/format_v2/orc/orc_reader.h
@@ -154,6 +154,7 @@ private:
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 99bfcc8e0ab..a7602cd11b2 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_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;
}
diff --git a/be/src/format_v2/table_reader.cpp
b/be/src/format_v2/table_reader.cpp
index 11e2b30df6d..69ae3b562cf 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<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;
}
diff --git a/be/src/format_v2/table_reader.h b/be/src/format_v2/table_reader.h
index da2cb351ff6..ec179abb92f 100644
--- a/be/src/format_v2/table_reader.h
+++ b/be/src/format_v2/table_reader.h
@@ -252,9 +252,10 @@ public:
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 @@ public:
_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 d6499f0eaa8..78ca6a6d053 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 <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;
@@ -46,6 +47,7 @@ void ConditionCache::insert(const KeyType& key,
std::shared_ptr<std::vector<bool
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
diff --git a/be/src/storage/segment/condition_cache.h
b/be/src/storage/segment/condition_cache.h
index a189312ee14..0ef5534a622 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<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 @@ public:
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 @@ public:
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 @@ public:
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 0e36498c41d..c2e7bf63236 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<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;
diff --git a/be/test/format_v2/parquet/parquet_scan_test.cpp
b/be/test/format_v2/parquet/parquet_scan_test.cpp
index edd51c45203..a101936c391 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<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);
diff --git a/be/test/format_v2/table_reader_test.cpp
b/be/test/format_v2/table_reader_test.cpp
index 252bfa5c04d..528da7338fd 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<FileScanRequest> last_request;
std::shared_ptr<ConditionCacheContext> condition_cache_ctx;
@@ -1068,6 +1071,10 @@ public:
}
}
+ 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 @@ public:
void set_condition_cache_context(std::shared_ptr<ConditionCacheContext>
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<FakeFileReaderState>();
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<ColumnDefinition> file_schema;
+ file_schema.push_back(make_file_column(0, "id",
std::make_shared<DataTypeInt32>()));
+
+ std::vector<ColumnDefinition> projected_columns;
+ projected_columns.push_back(make_table_column(0, "id",
std::make_shared<DataTypeInt32>()));
+ set_name_identifiers(&projected_columns);
+
+ auto io_ctx = std::make_shared<io::IOContext>();
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ auto fake_state = std::make_shared<FakeFileReaderState>();
+ 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);
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
