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 <stddef.h> |
21 | | #include <stdint.h> |
22 | | |
23 | | #include <algorithm> |
24 | | #include <cstdint> |
25 | | #include <memory> |
26 | | #include <string> |
27 | | #include <unordered_map> |
28 | | #include <utility> |
29 | | #include <vector> |
30 | | |
31 | | #include "common/consts.h" |
32 | | #include "common/status.h" |
33 | | #include "core/column/column.h" |
34 | | #include "exprs/aggregate/aggregate_function.h" |
35 | | #include "io/io_common.h" |
36 | | #include "runtime/thread_context.h" |
37 | | #include "storage/olap_common.h" |
38 | | #include "storage/tablet/tablet_schema.h" |
39 | | #include "storage/utils.h" |
40 | | |
41 | | namespace doris { |
42 | | |
43 | | class ReadSchema; |
44 | | class Block; |
45 | | class SlotDescriptor; |
46 | | using ReadSchemaSPtr = std::shared_ptr<ReadSchema>; |
47 | | |
48 | | // Select columns by their ordinal in `columns`, preserving the requested order and duplicates. |
49 | | // Keeping this source-layout operation outside ReadSchema prevents those ordinals from being |
50 | | // confused with the dense ordinals of the resulting ReadSchema. |
51 | | std::vector<TabletColumnPtr> project_columns_by_ordinal( |
52 | | const std::vector<TabletColumnPtr>& columns, |
53 | | const std::vector<ColumnId>& source_column_ordinals); |
54 | | |
55 | | // The dense, ordered column layout consumed by one storage reader. For example, if the caller's |
56 | | // Block is [k1, v1] and a historical delete predicate needs dropped_v2, the layout is |
57 | | // [k1, v1, dropped_v2]: num_block_columns() is 2, num_read_columns() is 3, and every reader-side |
58 | | // ColumnId is an ordinal in this list. TabletColumn keeps physical storage metadata, while |
59 | | // data_type() records the expected materialized type, such as a pruned STRUCT type. |
60 | | class ReadSchema { |
61 | | public: |
62 | | using SequenceMap = std::unordered_map<ColumnId, std::vector<ColumnId>>; |
63 | | using ChangelogValueColumnPairs = std::vector<std::pair<ColumnId, ColumnId>>; |
64 | | |
65 | | explicit ReadSchema(std::vector<TabletColumnPtr> columns); |
66 | | |
67 | | // Every column is initially caller-visible. Historical delete-predicate columns may be |
68 | | // appended later without changing num_block_columns(). |
69 | | explicit ReadSchema(std::vector<TabletColumnPtr> columns, std::vector<DataTypePtr> read_types); |
70 | | |
71 | | static IColumn::MutablePtr get_predicate_column_ptr(const DataTypePtr& data_type, |
72 | | const ReaderType reader_type); |
73 | | |
74 | | // Builds the column `slot` reads from the slot's name, unique id, type and default value. |
75 | | // For a column the tablet schema does not have yet, e.g. one a light schema change just added. |
76 | | static TabletColumn tablet_column_from_slot(const SlotDescriptor& slot); |
77 | | |
78 | 11.0k | const std::vector<TabletColumnPtr>& columns() const { return _read_columns; } |
79 | | |
80 | 92.2k | DataTypePtr data_type(size_t ordinal) const { |
81 | 92.2k | DCHECK_LT(ordinal, _read_types.size()); |
82 | 92.2k | return _read_types[ordinal]; |
83 | 92.2k | } |
84 | | |
85 | | // Append private historical delete-predicate columns without extending the Block prefix. |
86 | | void append_dropped_columns(std::vector<TabletColumn> columns); |
87 | | |
88 | | // Append COMPACTION_ROWID_COL to the Block prefix before any historical delete columns. |
89 | | // It carries each source row's physical location through compaction filtering and merge. |
90 | | void append_compaction_rowid_column(); |
91 | | |
92 | | // Create caller-visible Blocks from the FE-slot prefix. |
93 | | Block create_read_block() const; |
94 | | |
95 | | std::string read_columns_to_string() const; |
96 | | |
97 | | // Always sets the two facts below; the flags add the sequence mapping and the changelog |
98 | | // before-image mapping. |
99 | | Status init_from_tablet_schema(const TabletSchema& tablet_schema, |
100 | | bool merge_by_sequence_mapping, bool map_changelog_columns); |
101 | | |
102 | | // Whether the tablet schema defines a sequence mapping -- not whether this read merges by |
103 | | // one. A MoW query does not merge, yet a value predicate still must not be pushed down. |
104 | 560 | bool tablet_has_sequence_map() const { return _tablet_has_sequence_map; } |
105 | | |
106 | | // Whether the tablet schema materializes variant subcolumns as extracted columns. A |
107 | | // compaction read then takes them as flat leaves. |
108 | 1.57k | bool tablet_has_extracted_variant_columns() const { |
109 | 1.57k | return _tablet_has_extracted_variant_columns; |
110 | 1.57k | } |
111 | | |
112 | 3 | const SequenceMap& sequence_map() const { return _sequence_map; } |
113 | | |
114 | | // Return the matching before-image ordinal for a changelog value column. For example, in |
115 | | // [v1, v2, __DORIS_CHANGELOG_BEFORE_v1__, __DORIS_CHANGELOG_BEFORE_v2__], 0 maps to 2 and 1 |
116 | | // maps to 3. Columns without a before image, including TSO/LSN/OP, map to themselves. |
117 | 104 | ColumnId before_column_ordinal(ColumnId ordinal) const { |
118 | 104 | DCHECK_LT(ordinal, _before_column_ordinals.size()); |
119 | 104 | return _before_column_ordinals[ordinal]; |
120 | 104 | } |
121 | | |
122 | 19 | const ChangelogValueColumnPairs& changelog_value_column_pairs() const { |
123 | 19 | return _changelog_value_column_pairs; |
124 | 19 | } |
125 | | |
126 | 19 | bool changelog_value_pairs_complete() const { return _changelog_value_pairs_complete; } |
127 | | |
128 | 1.30M | const TabletColumn* column(size_t ordinal) const { return _read_columns[ordinal].get(); } |
129 | | |
130 | | // Total columns used inside storage, including appended storage-only columns. |
131 | | // Use this for per-column state and iteration over the complete ReadSchema. |
132 | 169k | size_t num_read_columns() const { return _read_columns.size(); } |
133 | | |
134 | | // Columns materialized in caller Blocks. They are the ReadSchema prefix before |
135 | | // appended storage-only columns; use this for Block layout and position bounds. |
136 | 1.05M | size_t num_block_columns() const { return _num_block_columns; } |
137 | | |
138 | | // Number of key columns present in the caller-visible ReadSchema. This is deliberately not the |
139 | | // full TabletSchema key count. Merge readers additionally require these columns to be the full, |
140 | | // leading storage-key prefix; direct/projected readers may contain fewer key columns. |
141 | 1.06M | size_t num_key_columns() const { return _num_key_columns; } |
142 | | |
143 | | // All special-column ordinals below address the caller-visible Block prefix and are -1 when |
144 | | // absent. A changelog layout may be [k1, v1, __DORIS_CHANGELOG_BEFORE_v1__, TSO, LSN, OP]; a |
145 | | // snapshot layout may instead contain COMMIT_TSO. |
146 | | // Logical-delete marker used by unique-key reads. |
147 | 3.01k | int32_t delete_sign_ordinal() const { return _delete_sign_ordinal; } |
148 | | // Sequence column used to choose the winning row during merge. |
149 | 499 | int32_t sequence_ordinal() const { return _sequence_ordinal; } |
150 | | // Synthetic row identifier returned by queries or carried by compaction reads. |
151 | 900 | int32_t rowid_ordinal() const { return _rowid_ordinal; } |
152 | | // Rowset version synthesized for single-version reads. |
153 | 0 | int32_t version_ordinal() const { return _version_ordinal; } |
154 | | // Changelog transaction timestamp. |
155 | 1.69k | int32_t tso_ordinal() const { return _tso_ordinal; } |
156 | | // Changelog log-sequence number. |
157 | 28 | int32_t lsn_ordinal() const { return _lsn_ordinal; } |
158 | | // Changelog operation kind, such as INSERT, UPDATE, or DELETE. |
159 | 76 | int32_t op_ordinal() const { return _op_ordinal; } |
160 | | // Snapshot commit timestamp. |
161 | 1 | int32_t commit_tso_ordinal() const { return _commit_tso_ordinal; } |
162 | | |
163 | | // -1 if no column with this unique id is present. |
164 | 369 | int32_t ordinal_by_uid(int32_t unique_id) const { |
165 | 369 | auto it = _uid_to_ordinal.find(unique_id); |
166 | 369 | return it == _uid_to_ordinal.end() ? -1 : it->second; |
167 | 369 | } |
168 | | |
169 | | private: |
170 | | void _init_read_types(); |
171 | | void _init_before_column_ordinals(); |
172 | | Status _init_sequence_map(const TabletSchema& tablet_schema); |
173 | | void _init_changelog_column_mappings(const TabletSchema& tablet_schema); |
174 | | |
175 | 1.44k | void _init_descriptors() { |
176 | 1.44k | DORIS_CHECK_LE(_num_block_columns, _read_columns.size()); |
177 | 1.44k | DORIS_CHECK_EQ(_read_columns.size(), _read_types.size()); |
178 | 1.44k | _num_key_columns = 0; |
179 | 1.44k | _delete_sign_ordinal = -1; |
180 | 1.44k | _sequence_ordinal = -1; |
181 | 1.44k | _rowid_ordinal = -1; |
182 | 1.44k | _version_ordinal = -1; |
183 | 1.44k | _tso_ordinal = -1; |
184 | 1.44k | _lsn_ordinal = -1; |
185 | 1.44k | _op_ordinal = -1; |
186 | 1.44k | _commit_tso_ordinal = -1; |
187 | 1.44k | _before_column_ordinals.clear(); |
188 | 1.44k | _uid_to_ordinal.clear(); |
189 | 7.69k | for (uint32_t i = 0; i < _read_columns.size(); ++i) { |
190 | 6.24k | const auto& col = *_read_columns[i]; |
191 | 6.24k | if (col.unique_id() >= 0) { |
192 | 6.03k | _uid_to_ordinal.emplace(col.unique_id(), i); |
193 | 6.03k | } |
194 | 6.24k | } |
195 | 7.69k | for (uint32_t i = 0; i < _num_block_columns; ++i) { |
196 | 6.24k | const auto& col = *_read_columns[i]; |
197 | 6.24k | if (col.is_key()) { |
198 | 1.60k | ++_num_key_columns; |
199 | 1.60k | } |
200 | 6.24k | if (col.name() == DELETE_SIGN) { |
201 | 201 | _delete_sign_ordinal = i; |
202 | 201 | } |
203 | 6.24k | if (col.name() == SEQUENCE_COL) { |
204 | 61 | _sequence_ordinal = i; |
205 | 61 | } |
206 | 6.24k | if (col.name().starts_with(BeConsts::GLOBAL_ROWID_COL) || |
207 | 6.24k | col.name() == BeConsts::COMPACTION_ROWID_COL) { |
208 | 153 | _rowid_ordinal = i; |
209 | 153 | } |
210 | 6.24k | if (col.name() == VERSION_COL) { |
211 | 26 | _version_ordinal = i; |
212 | 26 | } |
213 | 6.24k | if (col.name() == CHANGELOG_TSO_COL) { |
214 | 64 | _tso_ordinal = i; |
215 | 64 | } |
216 | 6.24k | if (col.name() == CHANGELOG_LSN_COL) { |
217 | 56 | _lsn_ordinal = i; |
218 | 56 | } |
219 | 6.24k | if (col.name() == CHANGELOG_OP_COL) { |
220 | 58 | _op_ordinal = i; |
221 | 58 | } |
222 | 6.24k | if (col.name() == COMMIT_TSO_COL) { |
223 | 11 | _commit_tso_ordinal = i; |
224 | 11 | } |
225 | 6.24k | } |
226 | 1.44k | if (_op_ordinal >= 0) { |
227 | 58 | _init_before_column_ordinals(); |
228 | 58 | } |
229 | 1.44k | } |
230 | | |
231 | | // Example: storage has k(uid=1, INT) and |
232 | | // s(uid=2, STRUCT<a:INT,b:STRING,c:BIGINT>). The scan needs k and only s.a/s.c, while |
233 | | // a historical delete predicate `old_v = 0` needs the dropped column old_v(uid=3, INT): |
234 | | // [0] _read_columns: k:INT (uid=1) |
235 | | // _read_types: INT |
236 | | // [1] _read_columns: s:STRUCT<a:INT,b:STRING,c:BIGINT> (uid=2) |
237 | | // _read_types: STRUCT<a:INT,c:BIGINT> |
238 | | // [2] _read_columns: old_v:INT (uid=3, dropped) |
239 | | // _read_types: INT |
240 | | // `_num_block_columns` is 2, so create_read_block() materializes ordinals [0, 2). Ordinal 2 |
241 | | // is read only for delete filtering and never appears in the caller Block. |
242 | | std::vector<TabletColumnPtr> _read_columns; |
243 | | // Types aligned by ordinal with `_read_columns`. |
244 | | std::vector<DataTypePtr> _read_types; |
245 | | // Boundary between the columns materialized in Blocks and the appended dropped columns. |
246 | | size_t _num_block_columns = 0; |
247 | | |
248 | | size_t _num_key_columns = 0; |
249 | | int32_t _delete_sign_ordinal = -1; |
250 | | int32_t _sequence_ordinal = -1; |
251 | | int32_t _rowid_ordinal = -1; |
252 | | int32_t _version_ordinal = -1; |
253 | | int32_t _tso_ordinal = -1; |
254 | | int32_t _lsn_ordinal = -1; |
255 | | int32_t _op_ordinal = -1; |
256 | | int32_t _commit_tso_ordinal = -1; |
257 | | std::unordered_map<int32_t, int32_t> _uid_to_ordinal; |
258 | | SequenceMap _sequence_map; |
259 | | std::vector<ColumnId> _before_column_ordinals; |
260 | | ChangelogValueColumnPairs _changelog_value_column_pairs; |
261 | | bool _changelog_value_pairs_complete = false; |
262 | | bool _tablet_has_sequence_map = false; |
263 | | bool _tablet_has_extracted_variant_columns = false; |
264 | | }; |
265 | | |
266 | | } // namespace doris |