github-actions[bot] commented on code in PR #66812:
URL: https://github.com/apache/doris/pull/66812#discussion_r3794796709
##########
be/src/exprs/function/dictionary_factory.h:
##########
@@ -88,65 +110,121 @@ class DictionaryFactory : private boost::noncopyable {
"Version ID is not equal to the refreshing version ID. {}
: {}", version_id,
refresh_version_id);
}
- {
- // commit the dictionary
- if (_dict_id_to_version_id_map.contains(dict_id)) {
- // check version_id
- if (version_id <= _dict_id_to_version_id_map[dict_id]) {
- LOG_WARNING(
- "DictionaryFactory Failed to commit dictionary
because version ID "
- "is not greater than the existing version ID")
- .tag("dict_id", dict_id)
- .tag("version_id", version_id)
- .tag("dict name", dict->dict_name())
- .tag("existing version ID",
_dict_id_to_version_id_map[dict_id]);
- return Status::InvalidArgument(
- "Version ID is not greater than the existing
version ID for the "
- "dictionary. {} : {}",
- version_id, _dict_id_to_version_id_map[dict_id]);
- }
+ auto& versioned_map = _dict_id_to_versioned_map[dict_id];
+ if (!versioned_map.empty()) {
+ int64_t latest = versioned_map.rbegin()->first;
+ if (version_id <= latest) {
+ LOG_WARNING(
+ "DictionaryFactory Failed to commit dictionary because
version ID "
+ "is not greater than the existing version ID")
+ .tag("dict_id", dict_id)
+ .tag("version_id", version_id)
+ .tag("dict name", dict->dict_name())
+ .tag("existing version ID", latest);
+ return Status::InvalidArgument(
+ "Version ID is not greater than the existing version
ID for the "
+ "dictionary. {} : {}",
+ version_id, latest);
}
- LOG_INFO("DictionaryFactory Successfully commit dictionary")
- .tag("dict_id", dict_id)
- .tag("version_id", version_id)
- .tag("dict name", dict->dict_name());
- _dict_id_to_dict_map[dict_id] = dict;
- _dict_id_to_version_id_map[dict_id] = version_id;
- _refreshing_dict_map.erase(dict_id);
}
+ LOG_INFO("DictionaryFactory Successfully commit dictionary")
+ .tag("dict_id", dict_id)
+ .tag("version_id", version_id)
+ .tag("dict name", dict->dict_name());
+ dict->set_commit_time_ms(UnixMillis());
+ versioned_map[version_id] = dict;
+ _refreshing_dict_map.erase(dict_id);
+ lc.unlock();
+ gc_if_needed();
return Status::OK();
}
Status delete_dict(int64_t dict_id) {
VLOG_DEBUG << "DictionaryFactory delete dictionary, dict_id: " <<
dict_id;
std::unique_lock lc(_mutex);
- if (!_dict_id_to_dict_map.contains(dict_id)) {
- LOG_WARNING("DictionaryFactory Failed to delete
dictionary").tag("dict_id", dict_id);
+ auto it = _dict_id_to_versioned_map.find(dict_id);
+ if (it == _dict_id_to_versioned_map.end()) {
return Status::OK();
}
- auto dict = _dict_id_to_dict_map[dict_id];
- LOG_INFO("DictionaryFactory Successfully delete dictionary")
- .tag("dict_id", dict_id)
- .tag("dict name", dict->dict_name());
- _dict_id_to_dict_map.erase(dict_id);
- _dict_id_to_version_id_map.erase(dict_id);
+ if (it->second.empty()) {
+ LOG_WARNING("DictionaryFactory delete dictionary with empty
version map")
+ .tag("dict_id", dict_id);
+ } else {
+ auto latest_it = it->second.rbegin();
+ LOG_INFO("DictionaryFactory Successfully delete dictionary")
+ .tag("dict_id", dict_id)
+ .tag("dict name", latest_it->second->dict_name())
+ .tag("latest version_id", latest_it->first);
+ }
+ _dict_id_to_versioned_map.erase(it);
return Status::OK();
}
std::shared_ptr<MemTrackerLimiter> mem_tracker() const { return
_mem_tracker; }
+ // unified GC entry: count-based + ttl-based, with interval protection
+ void gc_if_needed() {
+ int64_t gc_interval_ms = std::max(1,
config::dictionary_gc_interval_seconds) * 1000LL;
+ int64_t now = UnixMillis();
+ if (now - _last_gc_time_ms.load(std::memory_order_relaxed) <
gc_interval_ms) {
Review Comment:
[P1] Enforce the version cap independently of the GC interval
This return skips count enforcement as well as TTL sweeping. After one pass
sets `_last_gc_time_ms`, every refresh in the next 60 seconds can retain
another full dictionary despite `dictionary_max_versions=2`; if those are the
last commits, neither the count excess nor future TTL expiry is ever revisited.
Since the load limit is per version and the factory tracker has no aggregate
byte limit, retained memory is not bounded by the configured maximum. After
adding query-safe pinning so count GC cannot worsen the pre-open race, please
enforce the changed dictionary's count on every commit and give TTL a periodic
terminal trigger, retiring pointers outside `_mutex`.
##########
be/src/exprs/function/dictionary_factory.cpp:
##########
@@ -32,30 +32,34 @@ DictionaryFactory::DictionaryFactory()
DictionaryFactory::~DictionaryFactory() {
SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER(_mem_tracker);
- _dict_id_to_dict_map.clear();
- _dict_id_to_version_id_map.clear();
+ _dict_id_to_versioned_map.clear();
}
void DictionaryFactory::get_dictionary_status(std::vector<TDictionaryStatus>&
result,
std::vector<int64_t> dict_ids) {
+ // only report the latest version per dict; historical versions are
invisible to FE
std::shared_lock lc(_mutex);
- if (dict_ids.empty()) { // empty means ALL
- for (const auto& [dict_id, dict] : _dict_id_to_dict_map) {
- TDictionaryStatus status;
- status.__set_dictionary_id(dict_id);
- status.__set_version_id(_dict_id_to_version_id_map[dict_id]);
- status.__set_dictionary_memory_size(dict->allocated_bytes());
- result.emplace_back(std::move(status));
+ auto build_latest_status = [&result](int64_t dict_id,
+ const std::map<int64_t,
DictionaryPtr>& versioned_map) {
+ if (versioned_map.empty()) {
+ return;
+ }
+ const auto& [version_id, dict] = *versioned_map.rbegin();
+ TDictionaryStatus status;
+ status.__set_dictionary_id(dict_id);
+ status.__set_version_id(version_id);
+ status.__set_dictionary_memory_size(dict->allocated_bytes());
Review Comment:
[P2] Report memory retained by every committed version
Before multi-version retention, this field was the complete committed
payload footprint for the dictionary. The new map retains two versions by
default (and can retain more because of the interval issue), but this reports
only `rbegin()`'s bytes. FE exposes the value as `memoryBytes` and the BE
endpoint labels it `Memory Size`, so operators can see roughly half—or much
less—of the per-dictionary memory actually retained; the aggregate factory
tracker cannot restore that attribution. Please sum `allocated_bytes()` across
retained committed versions, or expose explicit latest/total/per-version
fields, and test with differently sized versions.
##########
be/src/exprs/function/dictionary_factory.h:
##########
@@ -37,11 +41,29 @@ class DictionaryFactory : private boost::noncopyable {
// Returns nullptr if failed
std::shared_ptr<const IDictionary> get(int64_t dict_id, int64_t
version_id) {
- std::unique_lock lc(_mutex);
- // dict_id and version_id must match
- if (_dict_id_to_dict_map.contains(dict_id) &&
- _dict_id_to_version_id_map[dict_id] == version_id) {
- return _dict_id_to_dict_map[dict_id];
+ // simulate slow query holding old version_id
+ DBUG_EXECUTE_IF("dict_get_delay", {
+ int sleep_sec = dp->param<int>("sleep_sec", 10);
+ LOG(INFO) << "debug point dict_get_delay: sleeping " << sleep_sec
+ << "s before get dict_id=" << dict_id << " version_id="
<< version_id;
+ sleep(sleep_sec);
+ });
+ std::shared_lock lc(_mutex);
+ auto it = _dict_id_to_versioned_map.find(dict_id);
+ if (it != _dict_id_to_versioned_map.end()) {
+ auto vit = it->second.find(version_id);
+ if (vit != it->second.end()) {
+ return vit->second;
+ }
+ }
+ // fallback to staging: version may have been increased by FE but not
yet committed
+ auto rit = _refreshing_dict_map.find(dict_id);
Review Comment:
[P1] Keep published staging versions available to stamped plans
FE journals N+1 before the commit RPC and immediately stamps it into plans,
but those plans may not reach BE `open()` yet. If commit fails, abort erases
N+1 on the failed BE; if the master fails instead, replay restores FE version
N+1 with `OUT_OF_DATE` status and the recovery load stages N+2, replacing the
sole N+1 entry here. Either path leaves an N+1-stamped plan with neither a
committed nor staged match and reproduces `dictionary not found` (partial
commit also makes the result BE-dependent). Please retain published staging
entries by version until stamped plans can no longer reach them, or use a
recoverable publication/commit decision, and test both abort-before-open and
journal-to-commit failover.
##########
be/src/exprs/function/dictionary_factory.h:
##########
@@ -88,65 +110,121 @@ class DictionaryFactory : private boost::noncopyable {
"Version ID is not equal to the refreshing version ID. {}
: {}", version_id,
refresh_version_id);
}
- {
- // commit the dictionary
- if (_dict_id_to_version_id_map.contains(dict_id)) {
- // check version_id
- if (version_id <= _dict_id_to_version_id_map[dict_id]) {
- LOG_WARNING(
- "DictionaryFactory Failed to commit dictionary
because version ID "
- "is not greater than the existing version ID")
- .tag("dict_id", dict_id)
- .tag("version_id", version_id)
- .tag("dict name", dict->dict_name())
- .tag("existing version ID",
_dict_id_to_version_id_map[dict_id]);
- return Status::InvalidArgument(
- "Version ID is not greater than the existing
version ID for the "
- "dictionary. {} : {}",
- version_id, _dict_id_to_version_id_map[dict_id]);
- }
+ auto& versioned_map = _dict_id_to_versioned_map[dict_id];
+ if (!versioned_map.empty()) {
+ int64_t latest = versioned_map.rbegin()->first;
+ if (version_id <= latest) {
+ LOG_WARNING(
+ "DictionaryFactory Failed to commit dictionary because
version ID "
+ "is not greater than the existing version ID")
+ .tag("dict_id", dict_id)
+ .tag("version_id", version_id)
+ .tag("dict name", dict->dict_name())
+ .tag("existing version ID", latest);
+ return Status::InvalidArgument(
+ "Version ID is not greater than the existing version
ID for the "
+ "dictionary. {} : {}",
+ version_id, latest);
}
- LOG_INFO("DictionaryFactory Successfully commit dictionary")
- .tag("dict_id", dict_id)
- .tag("version_id", version_id)
- .tag("dict name", dict->dict_name());
- _dict_id_to_dict_map[dict_id] = dict;
- _dict_id_to_version_id_map[dict_id] = version_id;
- _refreshing_dict_map.erase(dict_id);
}
+ LOG_INFO("DictionaryFactory Successfully commit dictionary")
+ .tag("dict_id", dict_id)
+ .tag("version_id", version_id)
+ .tag("dict name", dict->dict_name());
+ dict->set_commit_time_ms(UnixMillis());
+ versioned_map[version_id] = dict;
+ _refreshing_dict_map.erase(dict_id);
+ lc.unlock();
+ gc_if_needed();
return Status::OK();
}
Status delete_dict(int64_t dict_id) {
VLOG_DEBUG << "DictionaryFactory delete dictionary, dict_id: " <<
dict_id;
std::unique_lock lc(_mutex);
- if (!_dict_id_to_dict_map.contains(dict_id)) {
- LOG_WARNING("DictionaryFactory Failed to delete
dictionary").tag("dict_id", dict_id);
+ auto it = _dict_id_to_versioned_map.find(dict_id);
+ if (it == _dict_id_to_versioned_map.end()) {
return Status::OK();
}
- auto dict = _dict_id_to_dict_map[dict_id];
- LOG_INFO("DictionaryFactory Successfully delete dictionary")
- .tag("dict_id", dict_id)
- .tag("dict name", dict->dict_name());
- _dict_id_to_dict_map.erase(dict_id);
- _dict_id_to_version_id_map.erase(dict_id);
+ if (it->second.empty()) {
+ LOG_WARNING("DictionaryFactory delete dictionary with empty
version map")
+ .tag("dict_id", dict_id);
+ } else {
+ auto latest_it = it->second.rbegin();
+ LOG_INFO("DictionaryFactory Successfully delete dictionary")
+ .tag("dict_id", dict_id)
+ .tag("dict name", latest_it->second->dict_name())
+ .tag("latest version_id", latest_it->first);
+ }
+ _dict_id_to_versioned_map.erase(it);
return Status::OK();
}
std::shared_ptr<MemTrackerLimiter> mem_tracker() const { return
_mem_tracker; }
+ // unified GC entry: count-based + ttl-based, with interval protection
+ void gc_if_needed() {
+ int64_t gc_interval_ms = std::max(1,
config::dictionary_gc_interval_seconds) * 1000LL;
+ int64_t now = UnixMillis();
+ if (now - _last_gc_time_ms.load(std::memory_order_relaxed) <
gc_interval_ms) {
+ return;
+ }
+ std::unique_lock<std::shared_mutex> lc(_mutex);
+ // re-check under lock to avoid duplicate GC
+ if (now - _last_gc_time_ms.load(std::memory_order_relaxed) <
gc_interval_ms) {
+ return;
+ }
+ _last_gc_time_ms.store(now, std::memory_order_relaxed);
+ _gc_all_no_lock(now);
+ }
+
void get_dictionary_status(std::vector<TDictionaryStatus>& result,
std::vector<int64_t> dict_ids);
private:
- std::map<int64_t, DictionaryPtr> _dict_id_to_dict_map;
- std::map<int64_t, int64_t> _dict_id_to_version_id_map;
+ // GC all dicts: first count-based, then ttl-based. Always keeps the
latest version.
+ void _gc_all_no_lock(int64_t now) {
+ int32_t max_versions = std::max(1, config::dictionary_max_versions);
+ int64_t ttl_ms =
static_cast<int64_t>(config::dictionary_version_ttl_seconds) * 1000;
+ int64_t threshold_ms = ttl_ms > 0 ? now - ttl_ms : 0;
+ for (auto& [dict_id, versioned_map] : _dict_id_to_versioned_map) {
+ if (versioned_map.size() <= 1) {
+ continue;
+ }
+ // count-based: drop oldest while exceeding max_versions
+ while (versioned_map.size() > static_cast<size_t>(max_versions)) {
+ auto it = versioned_map.begin();
+ LOG_INFO("DictionaryFactory GC old version by count")
+ .tag("dict_id", dict_id)
+ .tag("version_id", it->first)
+ .tag("dict name", it->second->dict_name());
+ versioned_map.erase(it);
+ }
+ // ttl-based: drop non-latest versions older than ttl
+ while (ttl_ms > 0 && versioned_map.size() > 1) {
+ auto it = versioned_map.begin();
+ if (it->second->commit_time_ms() >= threshold_ms) {
+ break;
+ }
+ LOG_INFO("DictionaryFactory GC old version by ttl")
+ .tag("dict_id", dict_id)
+ .tag("version_id", it->first)
+ .tag("age_sec", (now - it->second->commit_time_ms()) /
1000);
+ versioned_map.erase(it);
Review Comment:
[P1] Do not let an orphaned future version replace FE's rolled-back version
If one BE commits N+1 and another commit RPC fails, FE durably decrements
back to N. Abort cannot undo the successful BE because its staging entry is
already gone, leaving `{N, N+1}` locally. This GC then assumes numeric N+1 is
authoritative and can erase N by TTL/count, while status reports N+1 and FE's
`dataCompleted()` accepts any version `>= N`; every future plan still stamps N
and will fail on that BE. Please carry a recoverable commit/abort decision to
BE (including lost-response idempotency), preserve/report the FE-decided
version until reconciliation, and test partial success followed by DEC, abort,
and GC.
##########
be/test/exec/dictionary/dictionary_multi_version_test.cpp:
##########
@@ -0,0 +1,370 @@
+// 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.
+
+#include <gtest/gtest.h>
+
+#include <memory>
+
+#include "core/block/columns_with_type_and_name.h"
+#include "core/data_type/data_type_number.h"
+#include "exprs/function/complex_hash_map_dictionary.h"
+#include "exprs/function/dictionary.h"
+#include "exprs/function/dictionary_factory.h"
+
+namespace doris {
+
+static DictionaryPtr make_dict(const std::string& name) {
+ return create_complex_hash_map_dict_from_column(
+ name,
+ ColumnsWithTypeAndName {
+ {DataTypeInt32::ColumnType::create(),
std::make_shared<DataTypeInt32>(), ""}},
+ ColumnsWithTypeAndName {
+ ColumnWithTypeAndName {DataTypeInt32::ColumnType::create(),
+ std::make_shared<DataTypeInt32>(),
""},
+ });
+}
+
+static void commit_version(DictionaryFactory& f, int64_t dict_id, int64_t
version_id) {
+ auto dict = make_dict("dict_" + std::to_string(dict_id));
+ EXPECT_TRUE(f.refresh_dict(dict_id, version_id, dict));
+ // reset GC timer to force GC on every commit
+ f._last_gc_time_ms.store(0, std::memory_order_relaxed);
+ EXPECT_TRUE(f.commit_refresh_dict(dict_id, version_id));
+}
+
+// ============ basic multi-version ============
+
+TEST(DictionaryMultiVersionTest, GetAfterMultipleCommits) {
+ auto old = config::dictionary_max_versions;
+ config::dictionary_max_versions = 10;
+ DictionaryFactory f;
+ commit_version(f, 1, 1);
+ commit_version(f, 1, 2);
+ commit_version(f, 1, 3);
+
+ EXPECT_NE(nullptr, f.get(1, 3));
+ EXPECT_NE(nullptr, f.get(1, 2));
+ EXPECT_NE(nullptr, f.get(1, 1));
+ config::dictionary_max_versions = old;
+}
+
+TEST(DictionaryMultiVersionTest, GetMissingVersion) {
+ DictionaryFactory f;
+ commit_version(f, 1, 1);
+ EXPECT_NE(nullptr, f.get(1, 1));
+ EXPECT_EQ(nullptr, f.get(1, 99));
+ EXPECT_EQ(nullptr, f.get(999, 1));
+}
+
+// ============ count-based GC ============
+
+TEST(DictionaryMultiVersionTest, GCByCount) {
+ auto old_max = config::dictionary_max_versions;
+ config::dictionary_max_versions = 2;
+ DictionaryFactory f;
+ commit_version(f, 1, 1);
+ commit_version(f, 1, 2);
+ commit_version(f, 1, 3);
+ // max_versions=2, v=1 should be GC'd
+ EXPECT_EQ(nullptr, f.get(1, 1));
+ EXPECT_NE(nullptr, f.get(1, 2));
+ EXPECT_NE(nullptr, f.get(1, 3));
+ config::dictionary_max_versions = old_max;
+}
+
+TEST(DictionaryMultiVersionTest, GCByCountOne) {
+ auto old_max = config::dictionary_max_versions;
+ config::dictionary_max_versions = 1;
+ DictionaryFactory f;
+ commit_version(f, 1, 1);
+ commit_version(f, 1, 2);
+ EXPECT_EQ(nullptr, f.get(1, 1));
+ EXPECT_NE(nullptr, f.get(1, 2));
+ config::dictionary_max_versions = old_max;
+}
+
+TEST(DictionaryMultiVersionTest, GCKeepsLatest) {
+ auto old_max = config::dictionary_max_versions;
+ config::dictionary_max_versions = 1;
+ DictionaryFactory f;
+ for (int i = 1; i <= 5; i++) {
+ commit_version(f, 1, i);
+ }
+ EXPECT_EQ(nullptr, f.get(1, 4));
+ EXPECT_NE(nullptr, f.get(1, 5));
+ config::dictionary_max_versions = old_max;
+}
+
+// ============ TTL-based GC ============
+
+TEST(DictionaryMultiVersionTest, TTLExpired) {
+ auto old_ttl = config::dictionary_version_ttl_seconds;
+ auto old_max = config::dictionary_max_versions;
+ config::dictionary_version_ttl_seconds = 1;
+ config::dictionary_max_versions = 10;
+ DictionaryFactory f;
+ commit_version(f, 1, 1);
+ // simulate v=1 committed 2 seconds ago
+ f._dict_id_to_versioned_map[1][1]->set_commit_time_ms(UnixMillis() - 2000);
+ commit_version(f, 1, 2);
+ // GC triggered by commit should drop v=1 (expired)
+ EXPECT_EQ(nullptr, f.get(1, 1));
+ EXPECT_NE(nullptr, f.get(1, 2));
+ config::dictionary_version_ttl_seconds = old_ttl;
+ config::dictionary_max_versions = old_max;
+}
+
+TEST(DictionaryMultiVersionTest, TTLKeepsLatest) {
+ auto old_ttl = config::dictionary_version_ttl_seconds;
+ auto old_max = config::dictionary_max_versions;
+ config::dictionary_version_ttl_seconds = 1;
+ config::dictionary_max_versions = 10;
+ DictionaryFactory f;
+ commit_version(f, 1, 1);
+ f._dict_id_to_versioned_map[1][1]->set_commit_time_ms(UnixMillis() - 2000);
+ commit_version(f, 1, 2);
+ f._dict_id_to_versioned_map[1][2]->set_commit_time_ms(UnixMillis() - 2000);
+ // both expired, but latest must survive
+ EXPECT_NE(nullptr, f.get(1, 2));
+ config::dictionary_version_ttl_seconds = old_ttl;
+ config::dictionary_max_versions = old_max;
+}
+
+TEST(DictionaryMultiVersionTest, TTLZero) {
+ auto old_ttl = config::dictionary_version_ttl_seconds;
+ auto old_max = config::dictionary_max_versions;
+ config::dictionary_version_ttl_seconds = 0;
+ config::dictionary_max_versions = 10;
+ DictionaryFactory f;
+ commit_version(f, 1, 1);
+ f._dict_id_to_versioned_map[1][1]->set_commit_time_ms(UnixMillis() -
100000);
+ commit_version(f, 1, 2);
+ // ttl=0, no TTL GC; both kept
+ EXPECT_NE(nullptr, f.get(1, 1));
+ EXPECT_NE(nullptr, f.get(1, 2));
+ config::dictionary_version_ttl_seconds = old_ttl;
+ config::dictionary_max_versions = old_max;
+}
+
+// ============ extreme configs ============
+
+TEST(DictionaryMultiVersionTest, GCMaxVersionsZero) {
+ auto old = config::dictionary_max_versions;
+ config::dictionary_max_versions = 0;
+ DictionaryFactory f;
+ commit_version(f, 1, 1);
+ commit_version(f, 1, 2);
+ // 0 falls back to 1; only latest kept
+ EXPECT_EQ(nullptr, f.get(1, 1));
+ EXPECT_NE(nullptr, f.get(1, 2));
+ config::dictionary_max_versions = old;
+}
+
+TEST(DictionaryMultiVersionTest, GCMaxVersionsNegative) {
+ auto old = config::dictionary_max_versions;
+ config::dictionary_max_versions = -5;
+ DictionaryFactory f;
+ commit_version(f, 1, 1);
+ commit_version(f, 1, 2);
+ // negative falls back to 1; only latest kept
+ EXPECT_EQ(nullptr, f.get(1, 1));
+ EXPECT_NE(nullptr, f.get(1, 2));
+ config::dictionary_max_versions = old;
+}
+
+TEST(DictionaryMultiVersionTest, GCMaxVersionsLarge) {
+ auto old = config::dictionary_max_versions;
+ config::dictionary_max_versions = 1000000;
+ DictionaryFactory f;
+ for (int i = 1; i <= 5; i++) {
+ commit_version(f, 1, i);
+ }
+ // large max; all versions kept
+ for (int i = 1; i <= 5; i++) {
+ EXPECT_NE(nullptr, f.get(1, i));
+ }
+ config::dictionary_max_versions = old;
+}
+
+TEST(DictionaryMultiVersionTest, TTLNegative) {
+ auto old_ttl = config::dictionary_version_ttl_seconds;
+ auto old_max = config::dictionary_max_versions;
+ config::dictionary_version_ttl_seconds = -1;
+ config::dictionary_max_versions = 10;
+ DictionaryFactory f;
+ commit_version(f, 1, 1);
+ f._dict_id_to_versioned_map[1][1]->set_commit_time_ms(UnixMillis() -
100000);
+ commit_version(f, 1, 2);
+ // negative ttl = no TTL GC
+ EXPECT_NE(nullptr, f.get(1, 1));
+ config::dictionary_version_ttl_seconds = old_ttl;
+ config::dictionary_max_versions = old_max;
+}
+
+TEST(DictionaryMultiVersionTest, GCIntervalLarge) {
+ auto old_interval = config::dictionary_gc_interval_seconds;
+ auto old_max = config::dictionary_max_versions;
+ config::dictionary_gc_interval_seconds = 1000000;
+ config::dictionary_max_versions = 1;
+ DictionaryFactory f;
+ f._last_gc_time_ms.store(UnixMillis(), std::memory_order_relaxed);
+ // manual commit without resetting GC timer
+ auto dict1 = make_dict("dict_1");
+ EXPECT_TRUE(f.refresh_dict(1, 1, dict1));
+ EXPECT_TRUE(f.commit_refresh_dict(1, 1));
+ auto dict2 = make_dict("dict_1");
+ EXPECT_TRUE(f.refresh_dict(1, 2, dict2));
+ EXPECT_TRUE(f.commit_refresh_dict(1, 2));
+ // interval not elapsed; GC skipped; both kept
+ EXPECT_NE(nullptr, f.get(1, 1));
+ EXPECT_NE(nullptr, f.get(1, 2));
+ config::dictionary_gc_interval_seconds = old_interval;
+ config::dictionary_max_versions = old_max;
+}
+
+// ============ boundary scenarios ============
+
+TEST(DictionaryMultiVersionTest, CommitDuplicateVersion) {
+ DictionaryFactory f;
+ commit_version(f, 1, 1);
+ // commit same version again should fail
+ auto dict = make_dict("dict_1");
+ EXPECT_TRUE(f.refresh_dict(1, 1, dict));
+ auto st = f.commit_refresh_dict(1, 1);
+ EXPECT_FALSE(st.ok());
+}
+
+TEST(DictionaryMultiVersionTest, GetNonExistentDict) {
+ DictionaryFactory f;
+ EXPECT_EQ(nullptr, f.get(999, 1));
+}
+
+TEST(DictionaryMultiVersionTest, DeleteAllVersions) {
+ DictionaryFactory f;
+ config::dictionary_max_versions = 10;
Review Comment:
[P2] Restore the process-global config in every unit test
This assignment is not factory-local. `DeleteAllVersions`,
`CommitAfterDelete`, and `GetStatusReportsLatest` each set
`dictionary_max_versions=10` without restoring it, and `exec/*.cpp` tests are
linked into the same `doris_be_test` process, so later or shuffled tests run
with polluted state instead of the declared default. Please use a scoped
save/restore in all three tests so exceptional exits cannot leak the value
either.
##########
be/src/exprs/function/dictionary_factory.h:
##########
@@ -88,65 +110,121 @@ class DictionaryFactory : private boost::noncopyable {
"Version ID is not equal to the refreshing version ID. {}
: {}", version_id,
refresh_version_id);
}
- {
- // commit the dictionary
- if (_dict_id_to_version_id_map.contains(dict_id)) {
- // check version_id
- if (version_id <= _dict_id_to_version_id_map[dict_id]) {
- LOG_WARNING(
- "DictionaryFactory Failed to commit dictionary
because version ID "
- "is not greater than the existing version ID")
- .tag("dict_id", dict_id)
- .tag("version_id", version_id)
- .tag("dict name", dict->dict_name())
- .tag("existing version ID",
_dict_id_to_version_id_map[dict_id]);
- return Status::InvalidArgument(
- "Version ID is not greater than the existing
version ID for the "
- "dictionary. {} : {}",
- version_id, _dict_id_to_version_id_map[dict_id]);
- }
+ auto& versioned_map = _dict_id_to_versioned_map[dict_id];
+ if (!versioned_map.empty()) {
+ int64_t latest = versioned_map.rbegin()->first;
+ if (version_id <= latest) {
+ LOG_WARNING(
+ "DictionaryFactory Failed to commit dictionary because
version ID "
+ "is not greater than the existing version ID")
+ .tag("dict_id", dict_id)
+ .tag("version_id", version_id)
+ .tag("dict name", dict->dict_name())
+ .tag("existing version ID", latest);
+ return Status::InvalidArgument(
+ "Version ID is not greater than the existing version
ID for the "
+ "dictionary. {} : {}",
+ version_id, latest);
}
- LOG_INFO("DictionaryFactory Successfully commit dictionary")
- .tag("dict_id", dict_id)
- .tag("version_id", version_id)
- .tag("dict name", dict->dict_name());
- _dict_id_to_dict_map[dict_id] = dict;
- _dict_id_to_version_id_map[dict_id] = version_id;
- _refreshing_dict_map.erase(dict_id);
}
+ LOG_INFO("DictionaryFactory Successfully commit dictionary")
+ .tag("dict_id", dict_id)
+ .tag("version_id", version_id)
+ .tag("dict name", dict->dict_name());
+ dict->set_commit_time_ms(UnixMillis());
+ versioned_map[version_id] = dict;
+ _refreshing_dict_map.erase(dict_id);
+ lc.unlock();
+ gc_if_needed();
return Status::OK();
}
Status delete_dict(int64_t dict_id) {
VLOG_DEBUG << "DictionaryFactory delete dictionary, dict_id: " <<
dict_id;
std::unique_lock lc(_mutex);
- if (!_dict_id_to_dict_map.contains(dict_id)) {
- LOG_WARNING("DictionaryFactory Failed to delete
dictionary").tag("dict_id", dict_id);
+ auto it = _dict_id_to_versioned_map.find(dict_id);
+ if (it == _dict_id_to_versioned_map.end()) {
return Status::OK();
}
- auto dict = _dict_id_to_dict_map[dict_id];
- LOG_INFO("DictionaryFactory Successfully delete dictionary")
- .tag("dict_id", dict_id)
- .tag("dict name", dict->dict_name());
- _dict_id_to_dict_map.erase(dict_id);
- _dict_id_to_version_id_map.erase(dict_id);
+ if (it->second.empty()) {
+ LOG_WARNING("DictionaryFactory delete dictionary with empty
version map")
+ .tag("dict_id", dict_id);
+ } else {
+ auto latest_it = it->second.rbegin();
+ LOG_INFO("DictionaryFactory Successfully delete dictionary")
+ .tag("dict_id", dict_id)
+ .tag("dict name", latest_it->second->dict_name())
+ .tag("latest version_id", latest_it->first);
+ }
+ _dict_id_to_versioned_map.erase(it);
return Status::OK();
}
std::shared_ptr<MemTrackerLimiter> mem_tracker() const { return
_mem_tracker; }
+ // unified GC entry: count-based + ttl-based, with interval protection
+ void gc_if_needed() {
+ int64_t gc_interval_ms = std::max(1,
config::dictionary_gc_interval_seconds) * 1000LL;
+ int64_t now = UnixMillis();
+ if (now - _last_gc_time_ms.load(std::memory_order_relaxed) <
gc_interval_ms) {
+ return;
+ }
+ std::unique_lock<std::shared_mutex> lc(_mutex);
+ // re-check under lock to avoid duplicate GC
+ if (now - _last_gc_time_ms.load(std::memory_order_relaxed) <
gc_interval_ms) {
+ return;
+ }
+ _last_gc_time_ms.store(now, std::memory_order_relaxed);
+ _gc_all_no_lock(now);
+ }
+
void get_dictionary_status(std::vector<TDictionaryStatus>& result,
std::vector<int64_t> dict_ids);
private:
- std::map<int64_t, DictionaryPtr> _dict_id_to_dict_map;
- std::map<int64_t, int64_t> _dict_id_to_version_id_map;
+ // GC all dicts: first count-based, then ttl-based. Always keeps the
latest version.
+ void _gc_all_no_lock(int64_t now) {
+ int32_t max_versions = std::max(1, config::dictionary_max_versions);
+ int64_t ttl_ms =
static_cast<int64_t>(config::dictionary_version_ttl_seconds) * 1000;
+ int64_t threshold_ms = ttl_ms > 0 ? now - ttl_ms : 0;
+ for (auto& [dict_id, versioned_map] : _dict_id_to_versioned_map) {
+ if (versioned_map.size() <= 1) {
+ continue;
+ }
+ // count-based: drop oldest while exceeding max_versions
+ while (versioned_map.size() > static_cast<size_t>(max_versions)) {
Review Comment:
[P1] Do not evict versions still reachable by queued plans
FE embeds version N during translation, but BE does not acquire the
protecting `shared_ptr` until function `open()`, after workload-group admission
and fragment dispatch. If N+1 commits and N+2 is the first GC-eligible commit
after the 60-second interval, an N-stamped plan can still be queued under the
default 900-second query timeout; with the default cap of two this loop erases
N and the delayed `get(N)` reproduces `dictionary not found`. A fixed count
(and likewise TTL) is not a safe lifetime boundary. Please add a lease/pin that
starts before a plan can wait, or enforce a horizon covering every
planning/queue/retry timeout without count-GC overriding it, and test
interval-separated refreshes before `get()`.
##########
be/src/exprs/function/dictionary_factory.h:
##########
@@ -88,65 +110,121 @@ class DictionaryFactory : private boost::noncopyable {
"Version ID is not equal to the refreshing version ID. {}
: {}", version_id,
refresh_version_id);
}
- {
- // commit the dictionary
- if (_dict_id_to_version_id_map.contains(dict_id)) {
- // check version_id
- if (version_id <= _dict_id_to_version_id_map[dict_id]) {
- LOG_WARNING(
- "DictionaryFactory Failed to commit dictionary
because version ID "
- "is not greater than the existing version ID")
- .tag("dict_id", dict_id)
- .tag("version_id", version_id)
- .tag("dict name", dict->dict_name())
- .tag("existing version ID",
_dict_id_to_version_id_map[dict_id]);
- return Status::InvalidArgument(
- "Version ID is not greater than the existing
version ID for the "
- "dictionary. {} : {}",
- version_id, _dict_id_to_version_id_map[dict_id]);
- }
+ auto& versioned_map = _dict_id_to_versioned_map[dict_id];
+ if (!versioned_map.empty()) {
+ int64_t latest = versioned_map.rbegin()->first;
+ if (version_id <= latest) {
+ LOG_WARNING(
+ "DictionaryFactory Failed to commit dictionary because
version ID "
+ "is not greater than the existing version ID")
+ .tag("dict_id", dict_id)
+ .tag("version_id", version_id)
+ .tag("dict name", dict->dict_name())
+ .tag("existing version ID", latest);
+ return Status::InvalidArgument(
+ "Version ID is not greater than the existing version
ID for the "
+ "dictionary. {} : {}",
+ version_id, latest);
}
- LOG_INFO("DictionaryFactory Successfully commit dictionary")
- .tag("dict_id", dict_id)
- .tag("version_id", version_id)
- .tag("dict name", dict->dict_name());
- _dict_id_to_dict_map[dict_id] = dict;
- _dict_id_to_version_id_map[dict_id] = version_id;
- _refreshing_dict_map.erase(dict_id);
}
+ LOG_INFO("DictionaryFactory Successfully commit dictionary")
+ .tag("dict_id", dict_id)
+ .tag("version_id", version_id)
+ .tag("dict name", dict->dict_name());
+ dict->set_commit_time_ms(UnixMillis());
+ versioned_map[version_id] = dict;
+ _refreshing_dict_map.erase(dict_id);
+ lc.unlock();
+ gc_if_needed();
return Status::OK();
}
Status delete_dict(int64_t dict_id) {
VLOG_DEBUG << "DictionaryFactory delete dictionary, dict_id: " <<
dict_id;
std::unique_lock lc(_mutex);
- if (!_dict_id_to_dict_map.contains(dict_id)) {
- LOG_WARNING("DictionaryFactory Failed to delete
dictionary").tag("dict_id", dict_id);
+ auto it = _dict_id_to_versioned_map.find(dict_id);
+ if (it == _dict_id_to_versioned_map.end()) {
return Status::OK();
}
- auto dict = _dict_id_to_dict_map[dict_id];
- LOG_INFO("DictionaryFactory Successfully delete dictionary")
- .tag("dict_id", dict_id)
- .tag("dict name", dict->dict_name());
- _dict_id_to_dict_map.erase(dict_id);
- _dict_id_to_version_id_map.erase(dict_id);
+ if (it->second.empty()) {
+ LOG_WARNING("DictionaryFactory delete dictionary with empty
version map")
+ .tag("dict_id", dict_id);
+ } else {
+ auto latest_it = it->second.rbegin();
+ LOG_INFO("DictionaryFactory Successfully delete dictionary")
+ .tag("dict_id", dict_id)
+ .tag("dict name", latest_it->second->dict_name())
+ .tag("latest version_id", latest_it->first);
+ }
+ _dict_id_to_versioned_map.erase(it);
return Status::OK();
}
std::shared_ptr<MemTrackerLimiter> mem_tracker() const { return
_mem_tracker; }
+ // unified GC entry: count-based + ttl-based, with interval protection
+ void gc_if_needed() {
+ int64_t gc_interval_ms = std::max(1,
config::dictionary_gc_interval_seconds) * 1000LL;
+ int64_t now = UnixMillis();
+ if (now - _last_gc_time_ms.load(std::memory_order_relaxed) <
gc_interval_ms) {
+ return;
+ }
+ std::unique_lock<std::shared_mutex> lc(_mutex);
+ // re-check under lock to avoid duplicate GC
+ if (now - _last_gc_time_ms.load(std::memory_order_relaxed) <
gc_interval_ms) {
+ return;
+ }
+ _last_gc_time_ms.store(now, std::memory_order_relaxed);
+ _gc_all_no_lock(now);
+ }
+
void get_dictionary_status(std::vector<TDictionaryStatus>& result,
std::vector<int64_t> dict_ids);
private:
- std::map<int64_t, DictionaryPtr> _dict_id_to_dict_map;
- std::map<int64_t, int64_t> _dict_id_to_version_id_map;
+ // GC all dicts: first count-based, then ttl-based. Always keeps the
latest version.
+ void _gc_all_no_lock(int64_t now) {
+ int32_t max_versions = std::max(1, config::dictionary_max_versions);
+ int64_t ttl_ms =
static_cast<int64_t>(config::dictionary_version_ttl_seconds) * 1000;
+ int64_t threshold_ms = ttl_ms > 0 ? now - ttl_ms : 0;
+ for (auto& [dict_id, versioned_map] : _dict_id_to_versioned_map) {
+ if (versioned_map.size() <= 1) {
+ continue;
+ }
+ // count-based: drop oldest while exceeding max_versions
+ while (versioned_map.size() > static_cast<size_t>(max_versions)) {
+ auto it = versioned_map.begin();
+ LOG_INFO("DictionaryFactory GC old version by count")
+ .tag("dict_id", dict_id)
+ .tag("version_id", it->first)
+ .tag("dict name", it->second->dict_name());
+ versioned_map.erase(it);
Review Comment:
[P1] Retire dictionary payloads after releasing the factory lock
This erase can drop the last reference and synchronously destroy the
dictionary while the factory-wide exclusive mutex is held. A pass can retire
versions from every dictionary, and each payload may be up to the default 2
GiB, so all `get()`/status/refresh calls are blocked during hash-table and
column destruction. `commit_refresh_dict()` also waits for this sweep before
replying to FE's five-second RPC, even though it has already inserted the new
version, which can turn reclamation latency into a partial-commit rollback.
Please unlink/move retired pointers under the lock and destroy them after
unlock (and keep the synchronous commit path bounded); `delete_dict()` needs
the same treatment for its retained-version map.
##########
regression-test/suites/dictionary_p0/test_dict_version_consistency.groovy:
##########
@@ -0,0 +1,138 @@
+// 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.
+
+import java.util.concurrent.Executors
+import java.util.concurrent.TimeUnit
+
+suite("test_dict_version_consistency", "nonConcurrent") {
+ sql "drop database if exists test_dict_version_consistency"
+ sql "create database test_dict_version_consistency"
+ sql "use test_dict_version_consistency"
+
+ sql """
+ create table src_dict(
+ k varchar(100) not null,
+ v varchar(100) not null
+ )
+ DISTRIBUTED BY HASH(`k`) BUCKETS 1
+ properties("replication_num" = "1");
+ """
+ sql "insert into src_dict values ('1', 'value1'), ('2', 'value2'), ('3',
'value3')"
+ sql """
+ create dictionary dict1 using src_dict
+ (k KEY, v VALUE) LAYOUT(HASH_MAP)
+ properties('data_lifetime'='600');
+ """
+ waitAllDictionariesReady()
+
+ def queryDict = { String key ->
+ sql "select dict_get('test_dict_version_consistency.dict1', 'v',
'${key}')"
+ }
+
+ // get current dict version via show dictionaries
+ // columns: DictionaryId(0) DictionaryName(1) BaseTableName(2) Version(3)
Status(4) ...
+ def getDictVersion = {
+ def rows = sql "show dictionaries"
+ for (def row : rows) {
+ if (row[1] == "dict1") return row[3] as int
+ }
+ return -1
+ }
+
+ // wait until dict version advances past baseline (FE increaseVersion
done, commit still pending)
+ def waitVersionAdvanced = { int baseline, long timeoutMs ->
+ long deadline = System.currentTimeMillis() + timeoutMs
+ while (System.currentTimeMillis() < deadline) {
+ def v = getDictVersion()
+ if (v > baseline) return v
+ sleep(200)
+ }
+ throw new AssertionError("dict version did not advance past
${baseline} within ${timeoutMs}ms")
+ }
+
+ // baseline
+ def baselineVersion = getDictVersion()
+ def result = queryDict("1")
+ assertEquals("value1", result[0][0])
+
+ // ============ Test 1: staging fallback ============
+ // Scenario: FE increaseVersion() done (version N+1 visible) but commit
RPC to BE not yet sent
+ // (blocked at afterIncJournal debug point).
+ // Query with version N+1 should succeed via
_refreshing_dict_map fallback.
+ logger.info("=== Test 1: staging fallback ===")
+
GetDebugPoint().enableDebugPointForAllFEs("DictionaryManager.afterIncJournal")
+ try {
+ def executor = Executors.newSingleThreadExecutor()
+ def refreshFuture = executor.submit({
+ sql "use test_dict_version_consistency"
+ sql "refresh dictionary dict1"
+ })
+
+ // wait until FE has increased version (blocked at afterIncJournal,
commit not done yet)
+ def advancedVersion = waitVersionAdvanced(baselineVersion, 10000)
+ logger.info("FE version advanced to ${advancedVersion}, commit still
pending")
+
+ // query should succeed via staging fallback
+ result = queryDict("1")
+ assertEquals("value1", result[0][0])
+ logger.info("query succeeded via staging fallback")
+
+
GetDebugPoint().disableDebugPointForAllFEs("DictionaryManager.afterIncJournal")
+ refreshFuture.get(30, TimeUnit.SECONDS)
+ executor.shutdown()
+ } finally {
+
GetDebugPoint().disableDebugPointForAllFEs("DictionaryManager.afterIncJournal")
+ }
+ logger.info("=== Test 1 passed ===")
+
+ // ============ Test 2: multi-version retention ============
+ // Scenario: query holds old version N (dict_get_delay sleeps 5s at BE
get()),
+ // refresh commits new version N+1 during the sleep.
+ // Verifies _dict_id_to_versioned_map retains old version N
(max_versions=2).
+ logger.info("=== Test 2: multi-version retention ===")
+ update_all_be_config("dictionary_max_versions", "2")
+ GetDebugPoint().enableDebugPointForAllBEs("dict_get_delay", [sleep_sec: 5])
+ try {
+ def baselineVersion2 = getDictVersion()
+ // submit query in background: FE plan uses baselineVersion2, BE
sleeps 5s at get()
+ def executor = Executors.newSingleThreadExecutor()
+ def queryResult = new
java.util.concurrent.atomic.AtomicReference<List>()
+ def queryFuture = executor.submit({
+ sql "use test_dict_version_consistency"
+ queryResult.set(sql "select
dict_get('test_dict_version_consistency.dict1', 'v', '1')")
+ })
+
+ // wait for BE to enter dict_get_delay (query is now sleeping with old
version)
+ sleep(2000)
+
+ // refresh commits new version while query holds old version
+ sql "refresh dictionary dict1"
+ def newVersion = waitVersionAdvanced(baselineVersion2, 10000)
+ logger.info("refresh committed version ${newVersion}, query still
holding ${baselineVersion2}")
+
+ // query should complete successfully using old version (multi-version
retention)
+ queryFuture.get(30, TimeUnit.SECONDS)
+ executor.shutdown()
+ result = queryResult.get()
+ assertEquals("value1", result[0][0])
+ logger.info("query succeeded with old version (multi-version
retention)")
+ } finally {
+ GetDebugPoint().disableDebugPointForAllBEs("dict_get_delay")
+ update_all_be_config("dictionary_max_versions", "1")
Review Comment:
[P2] Restore the prior max-version setting
This suite changes a mutable cluster-wide BE config to 2, then always leaves
it at 1 even though this PR's new default is 2. Subsequent dictionary suites in
the same regression cluster therefore run with multi-version retention
disabled. Please capture and restore the prior value, or at least restore the
declared default of 2.
##########
regression-test/suites/dictionary_p0/test_dict_version_consistency.groovy:
##########
@@ -0,0 +1,138 @@
+// 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.
+
+import java.util.concurrent.Executors
+import java.util.concurrent.TimeUnit
+
+suite("test_dict_version_consistency", "nonConcurrent") {
+ sql "drop database if exists test_dict_version_consistency"
+ sql "create database test_dict_version_consistency"
+ sql "use test_dict_version_consistency"
+
+ sql """
+ create table src_dict(
+ k varchar(100) not null,
+ v varchar(100) not null
+ )
+ DISTRIBUTED BY HASH(`k`) BUCKETS 1
+ properties("replication_num" = "1");
+ """
+ sql "insert into src_dict values ('1', 'value1'), ('2', 'value2'), ('3',
'value3')"
+ sql """
+ create dictionary dict1 using src_dict
+ (k KEY, v VALUE) LAYOUT(HASH_MAP)
+ properties('data_lifetime'='600');
+ """
+ waitAllDictionariesReady()
+
+ def queryDict = { String key ->
+ sql "select dict_get('test_dict_version_consistency.dict1', 'v',
'${key}')"
+ }
+
+ // get current dict version via show dictionaries
+ // columns: DictionaryId(0) DictionaryName(1) BaseTableName(2) Version(3)
Status(4) ...
+ def getDictVersion = {
+ def rows = sql "show dictionaries"
+ for (def row : rows) {
+ if (row[1] == "dict1") return row[3] as int
+ }
+ return -1
+ }
+
+ // wait until dict version advances past baseline (FE increaseVersion
done, commit still pending)
+ def waitVersionAdvanced = { int baseline, long timeoutMs ->
+ long deadline = System.currentTimeMillis() + timeoutMs
+ while (System.currentTimeMillis() < deadline) {
+ def v = getDictVersion()
+ if (v > baseline) return v
+ sleep(200)
+ }
+ throw new AssertionError("dict version did not advance past
${baseline} within ${timeoutMs}ms")
+ }
+
+ // baseline
+ def baselineVersion = getDictVersion()
+ def result = queryDict("1")
+ assertEquals("value1", result[0][0])
+
+ // ============ Test 1: staging fallback ============
+ // Scenario: FE increaseVersion() done (version N+1 visible) but commit
RPC to BE not yet sent
+ // (blocked at afterIncJournal debug point).
+ // Query with version N+1 should succeed via
_refreshing_dict_map fallback.
+ logger.info("=== Test 1: staging fallback ===")
+
GetDebugPoint().enableDebugPointForAllFEs("DictionaryManager.afterIncJournal")
+ try {
+ def executor = Executors.newSingleThreadExecutor()
+ def refreshFuture = executor.submit({
+ sql "use test_dict_version_consistency"
+ sql "refresh dictionary dict1"
+ })
+
+ // wait until FE has increased version (blocked at afterIncJournal,
commit not done yet)
+ def advancedVersion = waitVersionAdvanced(baselineVersion, 10000)
+ logger.info("FE version advanced to ${advancedVersion}, commit still
pending")
+
+ // query should succeed via staging fallback
+ result = queryDict("1")
+ assertEquals("value1", result[0][0])
+ logger.info("query succeeded via staging fallback")
+
+
GetDebugPoint().disableDebugPointForAllFEs("DictionaryManager.afterIncJournal")
+ refreshFuture.get(30, TimeUnit.SECONDS)
+ executor.shutdown()
+ } finally {
+
GetDebugPoint().disableDebugPointForAllFEs("DictionaryManager.afterIncJournal")
+ }
+ logger.info("=== Test 1 passed ===")
+
+ // ============ Test 2: multi-version retention ============
+ // Scenario: query holds old version N (dict_get_delay sleeps 5s at BE
get()),
+ // refresh commits new version N+1 during the sleep.
+ // Verifies _dict_id_to_versioned_map retains old version N
(max_versions=2).
+ logger.info("=== Test 2: multi-version retention ===")
+ update_all_be_config("dictionary_max_versions", "2")
+ GetDebugPoint().enableDebugPointForAllBEs("dict_get_delay", [sleep_sec: 5])
+ try {
+ def baselineVersion2 = getDictVersion()
+ // submit query in background: FE plan uses baselineVersion2, BE
sleeps 5s at get()
+ def executor = Executors.newSingleThreadExecutor()
+ def queryResult = new
java.util.concurrent.atomic.AtomicReference<List>()
+ def queryFuture = executor.submit({
+ sql "use test_dict_version_consistency"
+ queryResult.set(sql "select
dict_get('test_dict_version_consistency.dict1', 'v', '1')")
+ })
+
+ // wait for BE to enter dict_get_delay (query is now sleeping with old
version)
+ sleep(2000)
Review Comment:
[P2] Synchronize before claiming old-version coverage
`sleep(2000)` does not prove the background query reached `dict_get_delay`
with `baselineVersion2`; if planning/admission takes longer, the refresh wins,
the query is translated with N+1, and the identical `value1` data lets this
test pass without exercising retention. Please use a barrier/hit signal before
refreshing and make N versus N+1 return distinguishable values. Also use the
suite's `thread` helper (or always cancel and shut down in `finally`) so
assertion/timeout paths do not leave non-daemon executors and thread-local JDBC
connections behind; Test 1 has the same cleanup issue.
--
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]
