Coverage Report

Created: 2026-08-14 23:14

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
be/src/exprs/function/dictionary_factory.h
Line
Count
Source
1
// Licensed to the Apache Software Foundation (ASF) under one
2
// or more contributor license agreements.  See the NOTICE file
3
// distributed with this work for additional information
4
// regarding copyright ownership.  The ASF licenses this file
5
// to you under the Apache License, Version 2.0 (the
6
// "License"); you may not use this file except in compliance
7
// with the License.  You may obtain a copy of the License at
8
//
9
//   http://www.apache.org/licenses/LICENSE-2.0
10
//
11
// Unless required by applicable law or agreed to in writing,
12
// software distributed under the License is distributed on an
13
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14
// KIND, either express or implied.  See the License for the
15
// specific language governing permissions and limitations
16
// under the License.
17
18
#pragma once
19
20
#include <gen_cpp/BackendService_types.h>
21
22
#include <mutex>
23
24
#include "common/config.h"
25
#include "common/logging.h"
26
#include "exprs/function/dictionary.h"
27
28
namespace doris {
29
class MemTrackerLimiter;
30
}
31
namespace doris {
32
33
class DictionaryFactory : private boost::noncopyable {
34
public:
35
    DictionaryFactory();
36
    ~DictionaryFactory();
37
38
    // Returns nullptr if failed
39
2
    std::shared_ptr<const IDictionary> get(int64_t dict_id, int64_t version_id) {
40
2
        std::unique_lock lc(_mutex);
41
        // dict_id and version_id must match
42
2
        if (_dict_id_to_dict_map.contains(dict_id) &&
43
2
            _dict_id_to_version_id_map[dict_id] == version_id) {
44
0
            return _dict_id_to_dict_map[dict_id];
45
0
        }
46
2
        return nullptr;
47
2
    }
48
49
9
    Status refresh_dict(int64_t dict_id, int64_t version_id, DictionaryPtr dict) {
50
9
        VLOG_DEBUG << "DictionaryFactory refresh dictionary"
51
0
                   << " dict_id: " << dict_id << " version_id: " << version_id
52
0
                   << " dict name: " << dict->dict_name();
53
9
        std::unique_lock lc(_mutex);
54
9
        dict->_mem_tracker = _mem_tracker;
55
9
        _refreshing_dict_map[dict_id] = std::make_pair(version_id, dict);
56
        // Set the mem tracker for the dictionary
57
9
        return Status::OK();
58
9
    }
59
60
3
    Status abort_refresh_dict(int64_t dict_id, int64_t version_id) {
61
3
        VLOG_DEBUG << "DictionaryFactory abort refresh dictionary"
62
0
                   << " dict_id: " << dict_id << " version_id: " << version_id;
63
3
        std::unique_lock lc(_mutex);
64
3
        if (!_refreshing_dict_map.contains(dict_id)) {
65
            // FE will abort all, including succeed and failed.
66
1
            return Status::OK();
67
1
        }
68
2
        auto [refresh_version_id, dict] = _refreshing_dict_map[dict_id];
69
2
        if (version_id != refresh_version_id) {
70
1
            return Status::InvalidArgument(
71
1
                    "Version ID is not equal to the refreshing version ID. {} : {}", version_id,
72
1
                    refresh_version_id);
73
1
        }
74
1
        _refreshing_dict_map.erase(dict_id);
75
1
        return Status::OK();
76
2
    }
77
78
8
    Status commit_refresh_dict(int64_t dict_id, int64_t version_id) {
79
8
        VLOG_DEBUG << "DictionaryFactory commit refresh dictionary"
80
0
                   << " dict_id: " << dict_id << " version_id: " << version_id;
81
8
        std::unique_lock lc(_mutex);
82
8
        if (!_refreshing_dict_map.contains(dict_id)) {
83
1
            return Status::InvalidArgument("Dictionary is not refreshing dict_id: {}", dict_id);
84
1
        }
85
7
        auto [refresh_version_id, dict] = _refreshing_dict_map[dict_id];
86
7
        if (version_id != refresh_version_id) {
87
2
            return Status::InvalidArgument(
88
2
                    "Version ID is not equal to the refreshing version ID. {} : {}", version_id,
89
2
                    refresh_version_id);
90
2
        }
91
5
        {
92
            // commit the dictionary
93
5
            if (_dict_id_to_version_id_map.contains(dict_id)) {
94
                // check version_id
95
2
                if (version_id <= _dict_id_to_version_id_map[dict_id]) {
96
1
                    LOG_WARNING(
97
1
                            "DictionaryFactory Failed to commit dictionary because version ID "
98
1
                            "is not greater than the existing version ID")
99
1
                            .tag("dict_id", dict_id)
100
1
                            .tag("version_id", version_id)
101
1
                            .tag("dict name", dict->dict_name())
102
1
                            .tag("existing version ID", _dict_id_to_version_id_map[dict_id]);
103
1
                    return Status::InvalidArgument(
104
1
                            "Version ID is not greater than the existing version ID for the "
105
1
                            "dictionary. {} : {}",
106
1
                            version_id, _dict_id_to_version_id_map[dict_id]);
107
1
                }
108
2
            }
109
4
            LOG_INFO("DictionaryFactory Successfully commit dictionary")
110
4
                    .tag("dict_id", dict_id)
111
4
                    .tag("version_id", version_id)
112
4
                    .tag("dict name", dict->dict_name());
113
4
            _dict_id_to_dict_map[dict_id] = dict;
114
4
            _dict_id_to_version_id_map[dict_id] = version_id;
115
4
            _refreshing_dict_map.erase(dict_id);
116
4
        }
117
0
        return Status::OK();
118
5
    }
119
120
0
    Status delete_dict(int64_t dict_id) {
121
0
        VLOG_DEBUG << "DictionaryFactory delete dictionary, dict_id: " << dict_id;
122
0
        std::unique_lock lc(_mutex);
123
0
        if (!_dict_id_to_dict_map.contains(dict_id)) {
124
0
            LOG_WARNING("DictionaryFactory Failed to delete dictionary").tag("dict_id", dict_id);
125
0
            return Status::OK();
126
0
        }
127
0
        auto dict = _dict_id_to_dict_map[dict_id];
128
0
        LOG_INFO("DictionaryFactory Successfully delete dictionary")
129
0
                .tag("dict_id", dict_id)
130
0
                .tag("dict name", dict->dict_name());
131
0
        _dict_id_to_dict_map.erase(dict_id);
132
0
        _dict_id_to_version_id_map.erase(dict_id);
133
0
        return Status::OK();
134
0
    }
135
136
0
    std::shared_ptr<MemTrackerLimiter> mem_tracker() const { return _mem_tracker; }
137
138
    void get_dictionary_status(std::vector<TDictionaryStatus>& result,
139
                               std::vector<int64_t> dict_ids);
140
141
private:
142
    std::map<int64_t, DictionaryPtr> _dict_id_to_dict_map;
143
    std::map<int64_t, int64_t> _dict_id_to_version_id_map;
144
145
    std::map<int64_t, std::pair<int64_t, DictionaryPtr>>
146
            _refreshing_dict_map; // dict_id -> (version_id, dict)
147
148
    std::shared_mutex _mutex;
149
150
    std::shared_ptr<MemTrackerLimiter> _mem_tracker;
151
};
152
153
} // namespace doris