github-actions[bot] commented on code in PR #65482: URL: https://github.com/apache/doris/pull/65482#discussion_r3569677239
########## regression-test/suites/query_p0/cache/query_cache_incremental.groovy: ########## @@ -0,0 +1,183 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +// Correctness of the query cache incremental merge: when a partition keeps +// receiving hourly loads, a stale cache entry is reused by scanning only the +// delta rowsets since the cached version and merging them with the cached +// partial aggregation blocks. Every query below is checked against the same +// query with the cache disabled, so the suite passes in any environment +// (including those where incremental merge falls back to a full recompute, +// e.g. cloud mode) while exercising the incremental path on local storage. +suite("query_cache_incremental") { + def tableName = "test_query_cache_incremental" Review Comment: This new regression does not follow the local regression-test standards: ordinary single test tables should use hardcoded names instead of `def tableName`, determined result checks should be generated through `qt`/`order_qt` style expected output rather than only ad hoc `assertEquals`, and the tables should be dropped before use but left in place after the suite for debugging. Please align the suite with those standards while keeping any extra cache-on/cache-off behavioral checks you need. ########## regression-test/suites/query_p0/cache/query_cache_incremental.groovy: ########## @@ -0,0 +1,183 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +// Correctness of the query cache incremental merge: when a partition keeps +// receiving hourly loads, a stale cache entry is reused by scanning only the +// delta rowsets since the cached version and merging them with the cached +// partial aggregation blocks. Every query below is checked against the same +// query with the cache disabled, so the suite passes in any environment +// (including those where incremental merge falls back to a full recompute, +// e.g. cloud mode) while exercising the incremental path on local storage. +suite("query_cache_incremental") { + def tableName = "test_query_cache_incremental" + def uniqueTableName = "test_query_cache_incremental_mow" + def querySql = """ + SELECT + url, + SUM(cost) AS total_cost, + COUNT(*) AS cnt + FROM ${tableName} + WHERE dt >= '2026-01-01' + AND dt < '2026-01-15' + GROUP BY url + """ + + def normalize = { rows -> + return rows.collect { row -> row.collect { col -> String.valueOf(col) }.join("|") }.sort() + } + + // Compare the cached query result against the uncached one, twice: the + // first cached run may fill or incrementally merge the entry, the second + // one should serve it. + def checkConsistency = { String sqlText -> Review Comment: This helper only proves that the cache-enabled result matches the uncached result. The suite even documents that it passes when incremental merge falls back to a full recompute, so a regression where the real SQL path never takes `Mode::INCREMENTAL` would still pass. Please add an assertion that at least one post-insert local-storage query actually records a stale incremental hit, for example via `HitCacheStale` / `IncrementalDeltaVersions` in the profile or a monotonic increase in `query_cache_stale_hit_total`, and assert the intended fallback phase separately. ########## 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]
