cambyzju commented on code in PR #66812:
URL: https://github.com/apache/doris/pull/66812#discussion_r3829644062
##########
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:
这是个设计权衡的问题：保留版本数量控制 + TTL控制，足以覆盖大部分需求。
如果按照 Flying Query 携带的字典版本来 GC，可以解决找不到字典的问题，但是系统资源不可控，可能会导致 OOM 等问题。
--
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]
