github-actions[bot] commented on code in PR #65482:
URL: https://github.com/apache/doris/pull/65482#discussion_r3569683545
##########
be/test/exec/pipeline/query_cache_test.cpp:
##########
@@ -287,4 +307,542 @@ TEST_F(QueryCacheTest, insert_and_lookup) {
// ./run-be-ut.sh --run --filter=DataQueueTest.*
+namespace {
+
+std::vector<TScanRangeParams> make_scan_ranges(int64_t tablet_id, const
std::string& version) {
+ std::vector<TScanRangeParams> scan_ranges;
+ TScanRangeParams scan_range;
+ TPaloScanRange palo_scan_range;
+ palo_scan_range.__set_tablet_id(tablet_id);
+ palo_scan_range.__set_version(version);
+ scan_range.scan_range.__set_palo_scan_range(palo_scan_range);
+ scan_ranges.push_back(scan_range);
+ return scan_ranges;
+}
+
+TQueryCacheParam make_cache_param(int64_t tablet_id) {
+ TQueryCacheParam cache_param;
+ cache_param.__set_digest("runtime_test_digest");
+ cache_param.tablet_to_range.insert({tablet_id, "range"});
+ // FE always sets the entry limits; keep them roomy so decision tests do
+ // not trip the write-back feasibility check unintentionally.
+ cache_param.__set_entry_max_bytes(1024 * 1024);
+ cache_param.__set_entry_max_rows(100000);
+ return cache_param;
+}
+
+void insert_entry(QueryCache* cache, const std::string& cache_key, int64_t
version,
+ int64_t delta_count) {
+ CacheResult result;
+ result.push_back(std::make_unique<Block>());
+ *result.back() = ColumnHelper::create_block<DataTypeInt64>({1, 2, 3});
+ cache->insert(cache_key, version, result, {0}, 1, delta_count);
+}
+
+} // namespace
+
+TEST_F(QueryCacheTest, runtime_decision_miss_and_idempotent) {
+ std::unique_ptr<QueryCache> cache(QueryCache::create_global_cache(1024 *
1024));
+ auto scan_ranges = make_scan_ranges(42, "100");
+ QueryCacheRuntime runtime(make_cache_param(42), cache.get());
+
+ auto decision = runtime.get_or_make_decision(scan_ranges);
+ EXPECT_EQ(decision->mode, QueryCacheInstanceDecision::Mode::MISS);
+ EXPECT_TRUE(decision->key_valid);
+ EXPECT_EQ(decision->current_version, 100);
+ EXPECT_FALSE(decision->handle.valid());
+
+ // Idempotent: the second caller (e.g. the other operator) observes the
+ // same decision object.
+ auto decision2 = runtime.get_or_make_decision(scan_ranges);
+ EXPECT_EQ(decision.get(), decision2.get());
+}
+
+TEST_F(QueryCacheTest, runtime_decision_invalid_key) {
+ std::unique_ptr<QueryCache> cache(QueryCache::create_global_cache(1024 *
1024));
+ auto scan_ranges = make_scan_ranges(42, "100");
+ // Tablet 42 is missing from tablet_to_range, so build_cache_key fails and
+ // the query degrades to an uncached scan instead of failing.
+ TQueryCacheParam cache_param;
+ cache_param.__set_digest("runtime_test_digest");
+ cache_param.tablet_to_range.insert({43, "range"});
+ QueryCacheRuntime runtime(cache_param, cache.get());
+
+ auto decision = runtime.get_or_make_decision(scan_ranges);
+ EXPECT_EQ(decision->mode, QueryCacheInstanceDecision::Mode::MISS);
+ EXPECT_FALSE(decision->key_valid);
+ // Every caller shares one immutable invalid decision (and one log line).
+ EXPECT_EQ(decision.get(), runtime.get_or_make_decision(scan_ranges).get());
+}
+
+TEST_F(QueryCacheTest, runtime_decision_hit) {
+ std::unique_ptr<QueryCache> cache(QueryCache::create_global_cache(1024 *
1024));
+ auto scan_ranges = make_scan_ranges(42, "100");
+ auto cache_param = make_cache_param(42);
+
+ std::string cache_key;
+ int64_t version = 0;
+ EXPECT_TRUE(QueryCache::build_cache_key(scan_ranges, cache_param,
&cache_key, &version).ok());
+ insert_entry(cache.get(), cache_key, 100, 0);
+
+ QueryCacheRuntime runtime(cache_param, cache.get());
+ auto decision = runtime.get_or_make_decision(scan_ranges);
+ EXPECT_EQ(decision->mode, QueryCacheInstanceDecision::Mode::HIT);
+ EXPECT_TRUE(decision->handle.valid());
+ EXPECT_EQ(decision->handle.get_cache_version(), 100);
+}
+
+TEST_F(QueryCacheTest, runtime_decision_force_refresh) {
+ std::unique_ptr<QueryCache> cache(QueryCache::create_global_cache(1024 *
1024));
+ auto scan_ranges = make_scan_ranges(42, "100");
+ auto cache_param = make_cache_param(42);
+
+ std::string cache_key;
+ int64_t version = 0;
+ EXPECT_TRUE(QueryCache::build_cache_key(scan_ranges, cache_param,
&cache_key, &version).ok());
+ insert_entry(cache.get(), cache_key, 100, 0);
+
+ cache_param.__set_force_refresh_query_cache(true);
+ QueryCacheRuntime runtime(cache_param, cache.get());
+ auto decision = runtime.get_or_make_decision(scan_ranges);
+ // Recompute and write back even though a fresh entry exists.
+ EXPECT_EQ(decision->mode, QueryCacheInstanceDecision::Mode::MISS);
+ EXPECT_TRUE(decision->key_valid);
+ EXPECT_FALSE(decision->handle.valid());
+}
+
+TEST_F(QueryCacheTest, runtime_decision_binlog_scan) {
+ std::unique_ptr<QueryCache> cache(QueryCache::create_global_cache(1024 *
1024));
+ auto scan_ranges = make_scan_ranges(42, "100");
+ auto cache_param = make_cache_param(42);
+
+ std::string cache_key;
+ int64_t version = 0;
+ EXPECT_TRUE(QueryCache::build_cache_key(scan_ranges, cache_param,
&cache_key, &version).ok());
+ insert_entry(cache.get(), cache_key, 100, 0);
+
+ QueryCacheRuntime runtime(cache_param, cache.get());
+ runtime.disable_for_binlog_scan();
+ auto decision = runtime.get_or_make_decision(scan_ranges);
+ // A binlog scan must neither serve the cached entry nor write back.
+ EXPECT_EQ(decision->mode, QueryCacheInstanceDecision::Mode::MISS);
+ EXPECT_FALSE(decision->key_valid);
+}
+
+TEST_F(QueryCacheTest, runtime_decision_stale_without_incremental) {
+ std::unique_ptr<QueryCache> cache(QueryCache::create_global_cache(1024 *
1024));
+ auto scan_ranges = make_scan_ranges(42, "100");
+ auto cache_param = make_cache_param(42);
+
+ std::string cache_key;
+ int64_t version = 0;
+ EXPECT_TRUE(QueryCache::build_cache_key(scan_ranges, cache_param,
&cache_key, &version).ok());
+ insert_entry(cache.get(), cache_key, 50, 0);
+
+ // allow_incremental unset: a stale entry is a plain miss (full recompute).
+ QueryCacheRuntime runtime(cache_param, cache.get());
+ auto decision = runtime.get_or_make_decision(scan_ranges);
+ EXPECT_EQ(decision->mode, QueryCacheInstanceDecision::Mode::MISS);
+ EXPECT_TRUE(decision->key_valid);
+ EXPECT_FALSE(decision->handle.valid());
+}
+
+TEST_F(QueryCacheTest, runtime_decision_stale_incremental_fallbacks) {
+ std::unique_ptr<QueryCache> cache(QueryCache::create_global_cache(1024 *
1024));
+ auto scan_ranges = make_scan_ranges(42, "100");
+ auto cache_param = make_cache_param(42);
+ cache_param.__set_allow_incremental(true);
+
+ std::string cache_key;
+ int64_t version = 0;
+ EXPECT_TRUE(QueryCache::build_cache_key(scan_ranges, cache_param,
&cache_key, &version).ok());
+
+ {
+ // The entry is newer than the version this replica is asked to read:
+ // never usable, fall back to a full scan of the requested version.
+ insert_entry(cache.get(), cache_key, 200, 0);
+ QueryCacheRuntime runtime(cache_param, cache.get());
+ auto decision = runtime.get_or_make_decision(scan_ranges);
+ EXPECT_EQ(decision->mode, QueryCacheInstanceDecision::Mode::MISS);
+ EXPECT_TRUE(decision->key_valid);
+ EXPECT_EQ(decision->incremental_fallback_reason, "cached entry is
newer");
+ }
+ {
+ // Too many accumulated incremental merges: force a full recompute to
+ // compact the entry. (Checked before any tablet access.)
+ insert_entry(cache.get(), cache_key, 50,
config::query_cache_max_incremental_merge_count);
+ QueryCacheRuntime runtime(cache_param, cache.get());
+ auto decision = runtime.get_or_make_decision(scan_ranges);
+ EXPECT_EQ(decision->mode, QueryCacheInstanceDecision::Mode::MISS);
+ EXPECT_TRUE(decision->key_valid);
+ EXPECT_EQ(decision->incremental_fallback_reason,
+ "delta count reached compaction threshold");
+ }
+ {
+ // Use a tablet id that exists nowhere, so fetching the tablet fails
+ // regardless of whether some other suite left a storage engine behind
+ // in this test process, and the decision safely falls back to MISS.
+ auto missing_scan_ranges = make_scan_ranges(424242424242, "100");
+ auto missing_cache_param = make_cache_param(424242424242);
+ missing_cache_param.__set_allow_incremental(true);
+ std::string missing_cache_key;
+ int64_t missing_version = 0;
+ EXPECT_TRUE(QueryCache::build_cache_key(missing_scan_ranges,
missing_cache_param,
+ &missing_cache_key,
&missing_version)
+ .ok());
+ insert_entry(cache.get(), missing_cache_key, 50, 0);
+ QueryCacheRuntime runtime(missing_cache_param, cache.get());
+ auto decision = runtime.get_or_make_decision(missing_scan_ranges);
+ EXPECT_EQ(decision->mode, QueryCacheInstanceDecision::Mode::MISS);
+ EXPECT_TRUE(decision->key_valid);
+ EXPECT_FALSE(decision->handle.valid());
+ EXPECT_EQ(decision->incremental_fallback_reason, "tablet not found");
+ }
+}
+
+TEST_F(QueryCacheTest, lookup_any_version_and_delta_count) {
+ std::unique_ptr<QueryCache> cache(QueryCache::create_global_cache(1024 *
1024));
+ auto scan_ranges = make_scan_ranges(42, "100");
+ auto cache_param = make_cache_param(42);
+
+ std::string cache_key;
+ int64_t version = 0;
+ EXPECT_TRUE(QueryCache::build_cache_key(scan_ranges, cache_param,
&cache_key, &version).ok());
+
+ QueryCacheHandle handle;
+ EXPECT_FALSE(cache->lookup_any_version(cache_key, &handle));
+
+ int64_t write_backs_before =
DorisMetrics::instance()->query_cache_write_back_total->value();
+ insert_entry(cache.get(), cache_key, 50, 3);
+ EXPECT_EQ(DorisMetrics::instance()->query_cache_write_back_total->value(),
+ write_backs_before + 1);
+ QueryCacheHandle handle2;
+ EXPECT_TRUE(cache->lookup_any_version(cache_key, &handle2));
+ EXPECT_EQ(handle2.get_cache_version(), 50);
+ EXPECT_EQ(handle2.get_cache_delta_count(), 3);
+ // insert_entry stores one 3-row block with cache_size 1.
+ EXPECT_EQ(handle2.get_cache_total_rows(), 3);
+ EXPECT_EQ(handle2.get_cache_total_bytes(), 1);
+
+ // Exact-version lookup still rejects the stale entry.
+ QueryCacheHandle handle3;
+ EXPECT_FALSE(cache->lookup(cache_key, 100, &handle3));
+}
+
+TEST_F(QueryCacheTest, runtime_default_global_cache) {
+ // The single-argument constructor falls back to the global instance
+ // (whatever it is in this test environment).
+ QueryCacheRuntime runtime(make_cache_param(42));
+ EXPECT_EQ(runtime.cache(), QueryCache::instance());
+}
+
+TEST_F(QueryCacheTest, take_delta_read_source) {
+ auto decision = std::make_shared<QueryCacheInstanceDecision>();
+ decision->_delta_read_sources[42] = std::make_unique<TabletReadSource>();
+
+ EXPECT_EQ(decision->take_delta_read_source(41), nullptr);
+ auto source = decision->take_delta_read_source(42);
+ EXPECT_NE(source, nullptr);
+ // Each read source can be consumed exactly once.
+ EXPECT_EQ(decision->take_delta_read_source(42), nullptr);
+}
+
+TEST_F(QueryCacheTest, runtime_decision_stale_incremental_cloud_mode) {
+ std::unique_ptr<QueryCache> cache(QueryCache::create_global_cache(1024 *
1024));
+ auto scan_ranges = make_scan_ranges(42, "100");
+ auto cache_param = make_cache_param(42);
+ cache_param.__set_allow_incremental(true);
+
+ std::string cache_key;
+ int64_t version = 0;
+ EXPECT_TRUE(QueryCache::build_cache_key(scan_ranges, cache_param,
&cache_key, &version).ok());
+ insert_entry(cache.get(), cache_key, 50, 0);
+
+ // Incremental merge only supports local storage for now.
+ std::string saved_deploy_mode = config::deploy_mode;
+ config::deploy_mode = "cloud";
+ QueryCacheRuntime runtime(cache_param, cache.get());
+ auto decision = runtime.get_or_make_decision(scan_ranges);
+ config::deploy_mode = saved_deploy_mode;
+
+ EXPECT_EQ(decision->mode, QueryCacheInstanceDecision::Mode::MISS);
+ EXPECT_TRUE(decision->key_valid);
+ EXPECT_EQ(decision->incremental_fallback_reason, "cloud mode");
+}
+
+// Exercises the per-tablet part of the incremental decision against a real
+// (metadata-only) tablet registered in a real storage engine: capturing the
+// delta read source never touches segment files, so no data is needed.
+class QueryCacheIncrementalTest : public testing::Test {
+protected:
+ static constexpr int64_t kTabletId = 15673;
+ static constexpr const char* kTestDir =
"/ut_dir/query_cache_incremental_test";
+
+ void SetUp() override {
+ char buffer[1024];
+ EXPECT_NE(getcwd(buffer, sizeof(buffer)), nullptr);
+ _absolute_dir = std::string(buffer) + kTestDir;
+
EXPECT_TRUE(io::global_local_filesystem()->delete_directory(_absolute_dir).ok());
+
EXPECT_TRUE(io::global_local_filesystem()->create_directory(_absolute_dir).ok());
+
+ auto engine = std::make_unique<StorageEngine>(EngineOptions {});
+ _engine = engine.get();
+ ExecEnv::GetInstance()->set_storage_engine(std::move(engine));
+ _data_dir = std::make_unique<DataDir>(*_engine, _absolute_dir);
+ static_cast<void>(_data_dir->init());
Review Comment:
Please do not discard these setup `Status` values. If `_data_dir->init()`
fails, the fixture still proceeds to register tablets and the later
incremental/fallback assertions can run against an invalid test environment.
The same pattern appears below for `tablet_meta->add_rs_meta(rs_meta)` and
`tablet->init()`. Check each status with a form valid for the enclosing
function, for example a fatal assertion in `SetUp()` and either a non-fatal
check plus early failure return, or a refactor to a void/fatal helper, inside
`create_tablet()`, before using the constructed tablet state.
##########
be/test/exec/pipeline/query_cache_test.cpp:
##########
@@ -287,4 +307,542 @@ TEST_F(QueryCacheTest, insert_and_lookup) {
// ./run-be-ut.sh --run --filter=DataQueueTest.*
+namespace {
+
+std::vector<TScanRangeParams> make_scan_ranges(int64_t tablet_id, const
std::string& version) {
+ std::vector<TScanRangeParams> scan_ranges;
+ TScanRangeParams scan_range;
+ TPaloScanRange palo_scan_range;
+ palo_scan_range.__set_tablet_id(tablet_id);
+ palo_scan_range.__set_version(version);
+ scan_range.scan_range.__set_palo_scan_range(palo_scan_range);
+ scan_ranges.push_back(scan_range);
+ return scan_ranges;
+}
+
+TQueryCacheParam make_cache_param(int64_t tablet_id) {
+ TQueryCacheParam cache_param;
+ cache_param.__set_digest("runtime_test_digest");
+ cache_param.tablet_to_range.insert({tablet_id, "range"});
+ // FE always sets the entry limits; keep them roomy so decision tests do
+ // not trip the write-back feasibility check unintentionally.
+ cache_param.__set_entry_max_bytes(1024 * 1024);
+ cache_param.__set_entry_max_rows(100000);
+ return cache_param;
+}
+
+void insert_entry(QueryCache* cache, const std::string& cache_key, int64_t
version,
+ int64_t delta_count) {
+ CacheResult result;
+ result.push_back(std::make_unique<Block>());
+ *result.back() = ColumnHelper::create_block<DataTypeInt64>({1, 2, 3});
+ cache->insert(cache_key, version, result, {0}, 1, delta_count);
+}
+
+} // namespace
+
+TEST_F(QueryCacheTest, runtime_decision_miss_and_idempotent) {
+ std::unique_ptr<QueryCache> cache(QueryCache::create_global_cache(1024 *
1024));
+ auto scan_ranges = make_scan_ranges(42, "100");
+ QueryCacheRuntime runtime(make_cache_param(42), cache.get());
+
+ auto decision = runtime.get_or_make_decision(scan_ranges);
+ EXPECT_EQ(decision->mode, QueryCacheInstanceDecision::Mode::MISS);
+ EXPECT_TRUE(decision->key_valid);
+ EXPECT_EQ(decision->current_version, 100);
+ EXPECT_FALSE(decision->handle.valid());
+
+ // Idempotent: the second caller (e.g. the other operator) observes the
+ // same decision object.
+ auto decision2 = runtime.get_or_make_decision(scan_ranges);
+ EXPECT_EQ(decision.get(), decision2.get());
+}
+
+TEST_F(QueryCacheTest, runtime_decision_invalid_key) {
+ std::unique_ptr<QueryCache> cache(QueryCache::create_global_cache(1024 *
1024));
+ auto scan_ranges = make_scan_ranges(42, "100");
+ // Tablet 42 is missing from tablet_to_range, so build_cache_key fails and
+ // the query degrades to an uncached scan instead of failing.
+ TQueryCacheParam cache_param;
+ cache_param.__set_digest("runtime_test_digest");
+ cache_param.tablet_to_range.insert({43, "range"});
+ QueryCacheRuntime runtime(cache_param, cache.get());
+
+ auto decision = runtime.get_or_make_decision(scan_ranges);
+ EXPECT_EQ(decision->mode, QueryCacheInstanceDecision::Mode::MISS);
+ EXPECT_FALSE(decision->key_valid);
+ // Every caller shares one immutable invalid decision (and one log line).
+ EXPECT_EQ(decision.get(), runtime.get_or_make_decision(scan_ranges).get());
+}
+
+TEST_F(QueryCacheTest, runtime_decision_hit) {
+ std::unique_ptr<QueryCache> cache(QueryCache::create_global_cache(1024 *
1024));
+ auto scan_ranges = make_scan_ranges(42, "100");
+ auto cache_param = make_cache_param(42);
+
+ std::string cache_key;
+ int64_t version = 0;
+ EXPECT_TRUE(QueryCache::build_cache_key(scan_ranges, cache_param,
&cache_key, &version).ok());
+ insert_entry(cache.get(), cache_key, 100, 0);
+
+ QueryCacheRuntime runtime(cache_param, cache.get());
+ auto decision = runtime.get_or_make_decision(scan_ranges);
+ EXPECT_EQ(decision->mode, QueryCacheInstanceDecision::Mode::HIT);
+ EXPECT_TRUE(decision->handle.valid());
+ EXPECT_EQ(decision->handle.get_cache_version(), 100);
+}
+
+TEST_F(QueryCacheTest, runtime_decision_force_refresh) {
+ std::unique_ptr<QueryCache> cache(QueryCache::create_global_cache(1024 *
1024));
+ auto scan_ranges = make_scan_ranges(42, "100");
+ auto cache_param = make_cache_param(42);
+
+ std::string cache_key;
+ int64_t version = 0;
+ EXPECT_TRUE(QueryCache::build_cache_key(scan_ranges, cache_param,
&cache_key, &version).ok());
+ insert_entry(cache.get(), cache_key, 100, 0);
+
+ cache_param.__set_force_refresh_query_cache(true);
+ QueryCacheRuntime runtime(cache_param, cache.get());
+ auto decision = runtime.get_or_make_decision(scan_ranges);
+ // Recompute and write back even though a fresh entry exists.
+ EXPECT_EQ(decision->mode, QueryCacheInstanceDecision::Mode::MISS);
+ EXPECT_TRUE(decision->key_valid);
+ EXPECT_FALSE(decision->handle.valid());
+}
+
+TEST_F(QueryCacheTest, runtime_decision_binlog_scan) {
+ std::unique_ptr<QueryCache> cache(QueryCache::create_global_cache(1024 *
1024));
+ auto scan_ranges = make_scan_ranges(42, "100");
+ auto cache_param = make_cache_param(42);
+
+ std::string cache_key;
+ int64_t version = 0;
+ EXPECT_TRUE(QueryCache::build_cache_key(scan_ranges, cache_param,
&cache_key, &version).ok());
+ insert_entry(cache.get(), cache_key, 100, 0);
+
+ QueryCacheRuntime runtime(cache_param, cache.get());
+ runtime.disable_for_binlog_scan();
+ auto decision = runtime.get_or_make_decision(scan_ranges);
+ // A binlog scan must neither serve the cached entry nor write back.
+ EXPECT_EQ(decision->mode, QueryCacheInstanceDecision::Mode::MISS);
+ EXPECT_FALSE(decision->key_valid);
+}
+
+TEST_F(QueryCacheTest, runtime_decision_stale_without_incremental) {
+ std::unique_ptr<QueryCache> cache(QueryCache::create_global_cache(1024 *
1024));
+ auto scan_ranges = make_scan_ranges(42, "100");
+ auto cache_param = make_cache_param(42);
+
+ std::string cache_key;
+ int64_t version = 0;
+ EXPECT_TRUE(QueryCache::build_cache_key(scan_ranges, cache_param,
&cache_key, &version).ok());
+ insert_entry(cache.get(), cache_key, 50, 0);
+
+ // allow_incremental unset: a stale entry is a plain miss (full recompute).
+ QueryCacheRuntime runtime(cache_param, cache.get());
+ auto decision = runtime.get_or_make_decision(scan_ranges);
+ EXPECT_EQ(decision->mode, QueryCacheInstanceDecision::Mode::MISS);
+ EXPECT_TRUE(decision->key_valid);
+ EXPECT_FALSE(decision->handle.valid());
+}
+
+TEST_F(QueryCacheTest, runtime_decision_stale_incremental_fallbacks) {
+ std::unique_ptr<QueryCache> cache(QueryCache::create_global_cache(1024 *
1024));
+ auto scan_ranges = make_scan_ranges(42, "100");
+ auto cache_param = make_cache_param(42);
+ cache_param.__set_allow_incremental(true);
+
+ std::string cache_key;
+ int64_t version = 0;
+ EXPECT_TRUE(QueryCache::build_cache_key(scan_ranges, cache_param,
&cache_key, &version).ok());
+
+ {
+ // The entry is newer than the version this replica is asked to read:
+ // never usable, fall back to a full scan of the requested version.
+ insert_entry(cache.get(), cache_key, 200, 0);
+ QueryCacheRuntime runtime(cache_param, cache.get());
+ auto decision = runtime.get_or_make_decision(scan_ranges);
+ EXPECT_EQ(decision->mode, QueryCacheInstanceDecision::Mode::MISS);
+ EXPECT_TRUE(decision->key_valid);
+ EXPECT_EQ(decision->incremental_fallback_reason, "cached entry is
newer");
+ }
+ {
+ // Too many accumulated incremental merges: force a full recompute to
+ // compact the entry. (Checked before any tablet access.)
+ insert_entry(cache.get(), cache_key, 50,
config::query_cache_max_incremental_merge_count);
+ QueryCacheRuntime runtime(cache_param, cache.get());
+ auto decision = runtime.get_or_make_decision(scan_ranges);
+ EXPECT_EQ(decision->mode, QueryCacheInstanceDecision::Mode::MISS);
+ EXPECT_TRUE(decision->key_valid);
+ EXPECT_EQ(decision->incremental_fallback_reason,
+ "delta count reached compaction threshold");
+ }
+ {
+ // Use a tablet id that exists nowhere, so fetching the tablet fails
+ // regardless of whether some other suite left a storage engine behind
+ // in this test process, and the decision safely falls back to MISS.
+ auto missing_scan_ranges = make_scan_ranges(424242424242, "100");
+ auto missing_cache_param = make_cache_param(424242424242);
+ missing_cache_param.__set_allow_incremental(true);
+ std::string missing_cache_key;
+ int64_t missing_version = 0;
+ EXPECT_TRUE(QueryCache::build_cache_key(missing_scan_ranges,
missing_cache_param,
+ &missing_cache_key,
&missing_version)
+ .ok());
+ insert_entry(cache.get(), missing_cache_key, 50, 0);
+ QueryCacheRuntime runtime(missing_cache_param, cache.get());
+ auto decision = runtime.get_or_make_decision(missing_scan_ranges);
+ EXPECT_EQ(decision->mode, QueryCacheInstanceDecision::Mode::MISS);
+ EXPECT_TRUE(decision->key_valid);
+ EXPECT_FALSE(decision->handle.valid());
+ EXPECT_EQ(decision->incremental_fallback_reason, "tablet not found");
+ }
+}
+
+TEST_F(QueryCacheTest, lookup_any_version_and_delta_count) {
+ std::unique_ptr<QueryCache> cache(QueryCache::create_global_cache(1024 *
1024));
+ auto scan_ranges = make_scan_ranges(42, "100");
+ auto cache_param = make_cache_param(42);
+
+ std::string cache_key;
+ int64_t version = 0;
+ EXPECT_TRUE(QueryCache::build_cache_key(scan_ranges, cache_param,
&cache_key, &version).ok());
+
+ QueryCacheHandle handle;
+ EXPECT_FALSE(cache->lookup_any_version(cache_key, &handle));
+
+ int64_t write_backs_before =
DorisMetrics::instance()->query_cache_write_back_total->value();
+ insert_entry(cache.get(), cache_key, 50, 3);
+ EXPECT_EQ(DorisMetrics::instance()->query_cache_write_back_total->value(),
+ write_backs_before + 1);
+ QueryCacheHandle handle2;
+ EXPECT_TRUE(cache->lookup_any_version(cache_key, &handle2));
+ EXPECT_EQ(handle2.get_cache_version(), 50);
+ EXPECT_EQ(handle2.get_cache_delta_count(), 3);
+ // insert_entry stores one 3-row block with cache_size 1.
+ EXPECT_EQ(handle2.get_cache_total_rows(), 3);
+ EXPECT_EQ(handle2.get_cache_total_bytes(), 1);
+
+ // Exact-version lookup still rejects the stale entry.
+ QueryCacheHandle handle3;
+ EXPECT_FALSE(cache->lookup(cache_key, 100, &handle3));
+}
+
+TEST_F(QueryCacheTest, runtime_default_global_cache) {
+ // The single-argument constructor falls back to the global instance
+ // (whatever it is in this test environment).
+ QueryCacheRuntime runtime(make_cache_param(42));
+ EXPECT_EQ(runtime.cache(), QueryCache::instance());
+}
+
+TEST_F(QueryCacheTest, take_delta_read_source) {
+ auto decision = std::make_shared<QueryCacheInstanceDecision>();
+ decision->_delta_read_sources[42] = std::make_unique<TabletReadSource>();
+
+ EXPECT_EQ(decision->take_delta_read_source(41), nullptr);
+ auto source = decision->take_delta_read_source(42);
+ EXPECT_NE(source, nullptr);
+ // Each read source can be consumed exactly once.
+ EXPECT_EQ(decision->take_delta_read_source(42), nullptr);
+}
+
+TEST_F(QueryCacheTest, runtime_decision_stale_incremental_cloud_mode) {
+ std::unique_ptr<QueryCache> cache(QueryCache::create_global_cache(1024 *
1024));
+ auto scan_ranges = make_scan_ranges(42, "100");
+ auto cache_param = make_cache_param(42);
+ cache_param.__set_allow_incremental(true);
+
+ std::string cache_key;
+ int64_t version = 0;
+ EXPECT_TRUE(QueryCache::build_cache_key(scan_ranges, cache_param,
&cache_key, &version).ok());
+ insert_entry(cache.get(), cache_key, 50, 0);
+
+ // Incremental merge only supports local storage for now.
+ std::string saved_deploy_mode = config::deploy_mode;
+ config::deploy_mode = "cloud";
+ QueryCacheRuntime runtime(cache_param, cache.get());
+ auto decision = runtime.get_or_make_decision(scan_ranges);
+ config::deploy_mode = saved_deploy_mode;
+
+ EXPECT_EQ(decision->mode, QueryCacheInstanceDecision::Mode::MISS);
+ EXPECT_TRUE(decision->key_valid);
+ EXPECT_EQ(decision->incremental_fallback_reason, "cloud mode");
+}
+
+// Exercises the per-tablet part of the incremental decision against a real
+// (metadata-only) tablet registered in a real storage engine: capturing the
+// delta read source never touches segment files, so no data is needed.
+class QueryCacheIncrementalTest : public testing::Test {
+protected:
+ static constexpr int64_t kTabletId = 15673;
+ static constexpr const char* kTestDir =
"/ut_dir/query_cache_incremental_test";
+
+ void SetUp() override {
+ char buffer[1024];
+ EXPECT_NE(getcwd(buffer, sizeof(buffer)), nullptr);
+ _absolute_dir = std::string(buffer) + kTestDir;
+
EXPECT_TRUE(io::global_local_filesystem()->delete_directory(_absolute_dir).ok());
+
EXPECT_TRUE(io::global_local_filesystem()->create_directory(_absolute_dir).ok());
+
+ auto engine = std::make_unique<StorageEngine>(EngineOptions {});
+ _engine = engine.get();
+ ExecEnv::GetInstance()->set_storage_engine(std::move(engine));
+ _data_dir = std::make_unique<DataDir>(*_engine, _absolute_dir);
Review Comment:
Please do not discard these setup `Status` values. If `_data_dir->init()`
fails, the fixture still proceeds to register tablets and the later
incremental/fallback assertions can run against an invalid test environment.
The same pattern appears below for `tablet_meta->add_rs_meta(rs_meta)` and
`tablet->init()`. Check each status with a form valid for the enclosing
function, for example a fatal assertion in `SetUp()` and either a non-fatal
check plus early failure return, or a refactor to a void/fatal helper, inside
`create_tablet()`, before using the constructed tablet state.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
