asdf2014 commented on code in PR #65482: URL: https://github.com/apache/doris/pull/65482#discussion_r3570880374
########## 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: Aligned in 26484d76c5a: table names are hardcoded now, five `order_qt_*` snapshots backed by a generated `.out` (via `-forceGenOut` against a real cluster) anchor the deterministic results at each phase, and the final `DROP TABLE` statements are removed so the tables stay in place for debugging. The cache-on/off double-run comparison is kept on top as the behavioral check. ########## 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: Added in 26484d76c5a. On local storage the suite now proves through BE metrics that the incremental path really fires: the first append round on both the duplicate-key table and the merge-on-write table must strictly increase `query_cache_stale_hit_total`. These rounds are chosen because the hot partition holds only two data rowsets at that point, so neither the merge-count threshold nor a background compaction can interfere. The two designed fallback phases (a delete predicate in the delta, and a MoW backfill that rewrites history) must each increase `query_cache_incremental_fallback_total`, so each intended fallback phase is asserted separately. Both counters only move when `enable_query_cache_incremental` is on (default off, and this suite is the only one that sets it), and the assertions are one-sided deltas summed across all backends, so concurrent suites cannot break them. Cloud mode keeps the plain consistency checks since it always falls back by design. Verified on a 1 FE + 1 BE cluster: the counters do move at all four asserted points. ########## 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: Fixed in 26484d76c5a: `SetUp()` now uses fatal assertions and prints the failed `Status`; `create_tablet()` checks `add_rs_meta()` and `tablet->init()` with a non-fatal check plus an early `return nullptr` before the tablet is registered into the tablet map, and the two call sites that dereference the result assert it is not null. The two discarded booleans in `init_rs_meta()` (`JsonToProtoMessage`, `init_from_pb`) are asserted as well. `QueryCache*` UTs pass 34/34 with the change. -- 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]
