be/src/storage/index/primary_key_index.cpp
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 | | #include "storage/index/primary_key_index.h" |
19 | | |
20 | | #include <butil/time.h> |
21 | | #include <gen_cpp/segment_v2.pb.h> |
22 | | |
23 | | #include <utility> |
24 | | |
25 | | #include "common/compiler_util.h" // IWYU pragma: keep |
26 | | #include "common/config.h" |
27 | | #include "io/fs/file_writer.h" |
28 | | #include "storage/index/bloom_filter/bloom_filter_index_reader.h" |
29 | | #include "storage/index/bloom_filter/bloom_filter_index_writer.h" |
30 | | #include "storage/olap_common.h" |
31 | | #include "storage/segment/encoding_info.h" |
32 | | #include "storage/types.h" |
33 | | |
34 | | namespace doris { |
35 | | |
36 | | namespace { |
37 | 65 | io::IOContext create_index_io_context(const io::IOContext* source, OlapReaderStatistics* stats) { |
38 | 65 | io::IOContext io_ctx; |
39 | 65 | if (source != nullptr) { |
40 | 3 | io_ctx = *source; |
41 | 3 | } |
42 | 65 | io_ctx.is_index_data = true; |
43 | 65 | io_ctx.is_inverted_index = false; |
44 | 65 | io_ctx.file_cache_stats = stats ? &stats->file_cache_stats : nullptr; |
45 | 65 | return io_ctx; |
46 | 65 | } |
47 | | } // namespace |
48 | | |
49 | | static bvar::Adder<size_t> g_primary_key_index_memory_bytes("doris_primary_key_index_memory_bytes"); |
50 | | |
51 | 81 | Status PrimaryKeyIndexBuilder::init() { |
52 | | // TODO(liaoxin) using the column type directly if there's only one column in unique key columns |
53 | 81 | constexpr FieldType type = FieldType::OLAP_FIELD_TYPE_VARCHAR; |
54 | 81 | segment_v2::IndexedColumnWriterOptions options; |
55 | 81 | options.write_ordinal_index = true; |
56 | 81 | options.write_value_index = true; |
57 | 81 | options.data_page_size = config::primary_key_data_page_size; |
58 | 81 | options.encoding = segment_v2::EncodingInfo::get_index_column_encoding(type); |
59 | 81 | options.compression = segment_v2::ZSTD; |
60 | 81 | _primary_key_index_builder.reset( |
61 | 81 | new segment_v2::IndexedColumnWriter(options, type, _file_writer)); |
62 | 81 | RETURN_IF_ERROR(_primary_key_index_builder->init()); |
63 | | |
64 | 81 | auto opt = segment_v2::BloomFilterOptions(); |
65 | 81 | opt.fpp = 0.01; |
66 | 81 | RETURN_IF_ERROR(segment_v2::PrimaryKeyBloomFilterIndexWriterImpl::create( |
67 | 81 | opt, type, &_bloom_filter_index_builder)); |
68 | 81 | return Status::OK(); |
69 | 81 | } |
70 | | |
71 | 134k | Status PrimaryKeyIndexBuilder::add_item(const Slice& key) { |
72 | 134k | RETURN_IF_ERROR(_primary_key_index_builder->add(&key)); |
73 | 134k | Slice key_without_seq = Slice(key.get_data(), key.get_size() - _seq_col_length - _rowid_length); |
74 | 134k | RETURN_IF_ERROR(_bloom_filter_index_builder->add_values(&key_without_seq, 1)); |
75 | | // the key is already sorted, so the first key is min_key, and |
76 | | // the last key is max_key. |
77 | 134k | if (UNLIKELY(_num_rows == 0)) { |
78 | 81 | _min_key.append(key.get_data(), key.get_size()); |
79 | 81 | } |
80 | 134k | DCHECK(key.compare(_max_key) > 0) |
81 | 0 | << "found duplicate key or key is not sorted! current key: " << key |
82 | 0 | << ", last max key: " << _max_key; |
83 | 134k | _max_key.clear(); |
84 | 134k | _max_key.append(key.get_data(), key.get_size()); |
85 | 134k | _num_rows++; |
86 | 134k | _size += key.get_size(); |
87 | 134k | return Status::OK(); |
88 | 134k | } |
89 | | |
90 | 81 | Status PrimaryKeyIndexBuilder::finalize(segment_v2::PrimaryKeyIndexMetaPB* meta) { |
91 | | // finish primary key index |
92 | 81 | RETURN_IF_ERROR(_primary_key_index_builder->finish(meta->mutable_primary_key_index())); |
93 | 81 | _disk_size += _primary_key_index_builder->disk_size(); |
94 | | |
95 | | // set min_max key, the sequence column should be removed |
96 | 81 | meta->set_min_key(min_key().to_string()); |
97 | 81 | meta->set_max_key(max_key().to_string()); |
98 | | |
99 | | // finish bloom filter index |
100 | 81 | RETURN_IF_ERROR(_bloom_filter_index_builder->flush()); |
101 | 81 | uint64_t start_size = _file_writer->bytes_appended(); |
102 | 81 | RETURN_IF_ERROR( |
103 | 81 | _bloom_filter_index_builder->finish(_file_writer, meta->mutable_bloom_filter_index())); |
104 | 81 | _disk_size += _file_writer->bytes_appended() - start_size; |
105 | 81 | _primary_key_index_builder.reset(nullptr); |
106 | 81 | _bloom_filter_index_builder.reset(nullptr); |
107 | 81 | return Status::OK(); |
108 | 81 | } |
109 | | |
110 | | Status PrimaryKeyIndexReader::parse_index(io::FileReaderSPtr file_reader, |
111 | | const segment_v2::PrimaryKeyIndexMetaPB& meta, |
112 | | OlapReaderStatistics* pk_index_load_stats, |
113 | 60 | const io::IOContext* source_io_ctx) { |
114 | | // parse primary key index |
115 | 60 | _index_reader.reset(new segment_v2::IndexedColumnReader(file_reader, meta.primary_key_index())); |
116 | 60 | _index_reader->set_is_pk_index(true); |
117 | 60 | auto io_ctx = create_index_io_context(source_io_ctx, pk_index_load_stats); |
118 | 60 | RETURN_IF_ERROR(_index_reader->load(!config::disable_pk_storage_page_cache, false, |
119 | 60 | pk_index_load_stats, &io_ctx)); |
120 | | |
121 | 60 | _index_parsed = true; |
122 | 60 | return Status::OK(); |
123 | 60 | } |
124 | | |
125 | | Status PrimaryKeyIndexReader::parse_bf(io::FileReaderSPtr file_reader, |
126 | | const segment_v2::PrimaryKeyIndexMetaPB& meta, |
127 | | OlapReaderStatistics* pk_index_load_stats, |
128 | 5 | const io::IOContext* source_io_ctx) { |
129 | | // parse bloom filter |
130 | 5 | segment_v2::ColumnIndexMetaPB column_index_meta = meta.bloom_filter_index(); |
131 | 5 | segment_v2::BloomFilterIndexReader bf_index_reader(std::move(file_reader), |
132 | 5 | column_index_meta.bloom_filter_index()); |
133 | 5 | auto io_ctx = create_index_io_context(source_io_ctx, pk_index_load_stats); |
134 | 5 | RETURN_IF_ERROR(bf_index_reader.load(!config::disable_pk_storage_page_cache, false, |
135 | 5 | pk_index_load_stats, &io_ctx)); |
136 | 5 | std::unique_ptr<segment_v2::BloomFilterIndexIterator> bf_iter; |
137 | 5 | RETURN_IF_ERROR(bf_index_reader.new_iterator(&bf_iter, pk_index_load_stats)); |
138 | 5 | RETURN_IF_ERROR(bf_iter->read_bloom_filter(0, &_bf)); |
139 | 5 | segment_v2::g_pk_total_bloom_filter_num << 1; |
140 | 5 | segment_v2::g_pk_total_bloom_filter_total_bytes << _bf->size(); |
141 | 5 | segment_v2::g_pk_read_bloom_filter_num << 1; |
142 | 5 | segment_v2::g_pk_read_bloom_filter_total_bytes << _bf->size(); |
143 | 5 | _bf_num += 1; |
144 | 5 | _bf_bytes += _bf->size(); |
145 | | |
146 | 5 | _bf_parsed = true; |
147 | | |
148 | 5 | return Status::OK(); |
149 | 5 | } |
150 | | |
151 | | } // namespace doris |