be/src/format_v2/parquet/parquet_statistics.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 | | // http://www.apache.org/licenses/LICENSE-2.0 |
9 | | // Unless required by applicable law or agreed to in writing, |
10 | | // software distributed under the License is distributed on an |
11 | | // "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
12 | | // KIND, either express or implied. See the License for the |
13 | | // specific language governing permissions and limitations |
14 | | // under the License. |
15 | | |
16 | | #include "format_v2/parquet/parquet_statistics.h" |
17 | | |
18 | | #include <parquet/api/reader.h> |
19 | | #include <parquet/bloom_filter.h> |
20 | | #include <parquet/bloom_filter_reader.h> |
21 | | #include <parquet/column_page.h> |
22 | | #include <parquet/encoding.h> |
23 | | #include <parquet/page_index.h> |
24 | | #include <parquet/statistics.h> |
25 | | #include <parquet/types.h> |
26 | | |
27 | | #include <algorithm> |
28 | | #include <cstddef> |
29 | | #include <cstring> |
30 | | #include <exception> |
31 | | #include <limits> |
32 | | #include <map> |
33 | | #include <memory> |
34 | | #include <optional> |
35 | | #include <set> |
36 | | #include <string> |
37 | | #include <utility> |
38 | | #include <vector> |
39 | | |
40 | | #include "common/config.h" |
41 | | #include "core/data_type/data_type.h" |
42 | | #include "core/data_type/data_type_nullable.h" |
43 | | #include "core/data_type_serde/data_type_serde.h" |
44 | | #include "core/field.h" |
45 | | #include "exprs/expr_zonemap_filter.h" |
46 | | #include "exprs/vexpr_context.h" |
47 | | #include "format_v2/parquet/parquet_column_schema.h" |
48 | | #include "runtime/runtime_profile.h" |
49 | | #include "storage/index/zone_map/zone_map_index.h" |
50 | | #include "storage/index/zone_map/zonemap_eval_context.h" |
51 | | |
52 | | namespace doris::format::parquet { |
53 | | |
54 | | namespace { |
55 | | |
56 | | enum class ParquetRowGroupPruneReason { |
57 | | NONE, // cannot prune; must read |
58 | | STATISTICS, // excluded by ZoneMap statistics |
59 | | DICTIONARY, // excluded by dictionary |
60 | | BLOOM_FILTER, // excluded by bloom filter |
61 | | }; |
62 | | |
63 | 6 | bool bloom_logical_type_supported(const ParquetColumnSchema& column_schema) { |
64 | 6 | if (column_schema.type == nullptr) { |
65 | 0 | return false; |
66 | 0 | } |
67 | 6 | switch (remove_nullable(column_schema.type)->get_primitive_type()) { |
68 | 0 | case TYPE_BOOLEAN: |
69 | 0 | case TYPE_INT: |
70 | 3 | case TYPE_BIGINT: |
71 | 4 | case TYPE_FLOAT: |
72 | 4 | case TYPE_DOUBLE: |
73 | 6 | case TYPE_STRING: |
74 | 6 | return true; |
75 | 0 | default: |
76 | 0 | return false; |
77 | 6 | } |
78 | 6 | } |
79 | | |
80 | 424 | DecodedTimeUnit decoded_time_unit(ParquetTimeUnit time_unit) { |
81 | 424 | switch (time_unit) { |
82 | 0 | case ParquetTimeUnit::MILLIS: |
83 | 0 | return DecodedTimeUnit::MILLIS; |
84 | 2 | case ParquetTimeUnit::MICROS: |
85 | 2 | return DecodedTimeUnit::MICROS; |
86 | 0 | case ParquetTimeUnit::NANOS: |
87 | 0 | return DecodedTimeUnit::NANOS; |
88 | 422 | default: |
89 | 422 | return DecodedTimeUnit::UNKNOWN; |
90 | 424 | } |
91 | 424 | } |
92 | | |
93 | | Status read_decoded_field(const ParquetColumnSchema& column_schema, DecodedColumnView view, |
94 | 424 | Field* field, const cctz::time_zone* timezone) { |
95 | 424 | DORIS_CHECK(column_schema.type != nullptr); |
96 | 424 | DORIS_CHECK(field != nullptr); |
97 | 424 | constexpr uint8_t not_null = 0; |
98 | 424 | view.row_count = 1; |
99 | 424 | view.null_map = ¬_null; |
100 | 424 | view.time_unit = decoded_time_unit(column_schema.type_descriptor.time_unit); |
101 | 424 | view.logical_integer_bit_width = column_schema.type_descriptor.integer_bit_width; |
102 | 424 | view.logical_integer_is_signed = !column_schema.type_descriptor.is_unsigned_integer; |
103 | 424 | view.decimal_precision = column_schema.type_descriptor.decimal_precision; |
104 | 424 | view.decimal_scale = column_schema.type_descriptor.decimal_scale; |
105 | 424 | view.fixed_length = column_schema.type_descriptor.fixed_length; |
106 | 424 | view.timestamp_is_adjusted_to_utc = column_schema.type_descriptor.timestamp_is_adjusted_to_utc; |
107 | 424 | view.timezone = timezone; |
108 | 424 | return column_schema.type->get_serde()->read_field_from_decoded_value(*column_schema.type, |
109 | 424 | field, view); |
110 | 424 | } |
111 | | |
112 | | template <typename NativeType> |
113 | | bool set_decoded_field(const ParquetColumnSchema& column_schema, DecodedValueKind value_kind, |
114 | 422 | const NativeType& value, Field* field, const cctz::time_zone* timezone) { |
115 | 422 | DecodedColumnView view; |
116 | 422 | view.value_kind = value_kind; |
117 | 422 | view.values = reinterpret_cast<const uint8_t*>(&value); |
118 | 422 | return read_decoded_field(column_schema, view, field, timezone).ok(); |
119 | 422 | } Unexecuted instantiation: parquet_statistics.cpp:_ZN5doris6format7parquet12_GLOBAL__N_117set_decoded_fieldIbEEbRKNS1_19ParquetColumnSchemaENS_16DecodedValueKindERKT_PNS_5FieldEPKN4cctz9time_zoneE parquet_statistics.cpp:_ZN5doris6format7parquet12_GLOBAL__N_117set_decoded_fieldIiEEbRKNS1_19ParquetColumnSchemaENS_16DecodedValueKindERKT_PNS_5FieldEPKN4cctz9time_zoneE Line | Count | Source | 114 | 420 | const NativeType& value, Field* field, const cctz::time_zone* timezone) { | 115 | 420 | DecodedColumnView view; | 116 | 420 | view.value_kind = value_kind; | 117 | 420 | view.values = reinterpret_cast<const uint8_t*>(&value); | 118 | 420 | return read_decoded_field(column_schema, view, field, timezone).ok(); | 119 | 420 | } |
parquet_statistics.cpp:_ZN5doris6format7parquet12_GLOBAL__N_117set_decoded_fieldIlEEbRKNS1_19ParquetColumnSchemaENS_16DecodedValueKindERKT_PNS_5FieldEPKN4cctz9time_zoneE Line | Count | Source | 114 | 2 | const NativeType& value, Field* field, const cctz::time_zone* timezone) { | 115 | 2 | DecodedColumnView view; | 116 | 2 | view.value_kind = value_kind; | 117 | 2 | view.values = reinterpret_cast<const uint8_t*>(&value); | 118 | 2 | return read_decoded_field(column_schema, view, field, timezone).ok(); | 119 | 2 | } |
Unexecuted instantiation: parquet_statistics.cpp:_ZN5doris6format7parquet12_GLOBAL__N_117set_decoded_fieldIfEEbRKNS1_19ParquetColumnSchemaENS_16DecodedValueKindERKT_PNS_5FieldEPKN4cctz9time_zoneE Unexecuted instantiation: parquet_statistics.cpp:_ZN5doris6format7parquet12_GLOBAL__N_117set_decoded_fieldIdEEbRKNS1_19ParquetColumnSchemaENS_16DecodedValueKindERKT_PNS_5FieldEPKN4cctz9time_zoneE |
120 | | |
121 | | template <typename ParquetDType> |
122 | | bool set_decoded_min_max(const std::shared_ptr<::parquet::Statistics>& statistics, |
123 | | const ParquetColumnSchema& column_schema, DecodedValueKind value_kind, |
124 | | ParquetColumnStatistics* column_statistics, |
125 | 83 | const cctz::time_zone* timezone) { |
126 | 83 | auto typed_statistics = |
127 | 83 | std::static_pointer_cast<::parquet::TypedStatistics<ParquetDType>>(statistics); |
128 | 83 | if (!set_decoded_field(column_schema, value_kind, typed_statistics->min(), |
129 | 83 | &column_statistics->min_value, timezone) || |
130 | 83 | !set_decoded_field(column_schema, value_kind, typed_statistics->max(), |
131 | 83 | &column_statistics->max_value, timezone)) { |
132 | 0 | return false; |
133 | 0 | } |
134 | 83 | return true; |
135 | 83 | } Unexecuted instantiation: parquet_statistics.cpp:_ZN5doris6format7parquet12_GLOBAL__N_119set_decoded_min_maxIN7parquet12PhysicalTypeILNS4_4Type4typeE0EEEEEbRKSt10shared_ptrINS4_10StatisticsEERKNS1_19ParquetColumnSchemaENS_16DecodedValueKindEPNS1_23ParquetColumnStatisticsEPKN4cctz9time_zoneE parquet_statistics.cpp:_ZN5doris6format7parquet12_GLOBAL__N_119set_decoded_min_maxIN7parquet12PhysicalTypeILNS4_4Type4typeE1EEEEEbRKSt10shared_ptrINS4_10StatisticsEERKNS1_19ParquetColumnSchemaENS_16DecodedValueKindEPNS1_23ParquetColumnStatisticsEPKN4cctz9time_zoneE Line | Count | Source | 125 | 82 | const cctz::time_zone* timezone) { | 126 | 82 | auto typed_statistics = | 127 | 82 | std::static_pointer_cast<::parquet::TypedStatistics<ParquetDType>>(statistics); | 128 | 82 | if (!set_decoded_field(column_schema, value_kind, typed_statistics->min(), | 129 | 82 | &column_statistics->min_value, timezone) || | 130 | 82 | !set_decoded_field(column_schema, value_kind, typed_statistics->max(), | 131 | 82 | &column_statistics->max_value, timezone)) { | 132 | 0 | return false; | 133 | 0 | } | 134 | 82 | return true; | 135 | 82 | } |
parquet_statistics.cpp:_ZN5doris6format7parquet12_GLOBAL__N_119set_decoded_min_maxIN7parquet12PhysicalTypeILNS4_4Type4typeE2EEEEEbRKSt10shared_ptrINS4_10StatisticsEERKNS1_19ParquetColumnSchemaENS_16DecodedValueKindEPNS1_23ParquetColumnStatisticsEPKN4cctz9time_zoneE Line | Count | Source | 125 | 1 | const cctz::time_zone* timezone) { | 126 | 1 | auto typed_statistics = | 127 | 1 | std::static_pointer_cast<::parquet::TypedStatistics<ParquetDType>>(statistics); | 128 | 1 | if (!set_decoded_field(column_schema, value_kind, typed_statistics->min(), | 129 | 1 | &column_statistics->min_value, timezone) || | 130 | 1 | !set_decoded_field(column_schema, value_kind, typed_statistics->max(), | 131 | 1 | &column_statistics->max_value, timezone)) { | 132 | 0 | return false; | 133 | 0 | } | 134 | 1 | return true; | 135 | 1 | } |
Unexecuted instantiation: parquet_statistics.cpp:_ZN5doris6format7parquet12_GLOBAL__N_119set_decoded_min_maxIN7parquet12PhysicalTypeILNS4_4Type4typeE4EEEEEbRKSt10shared_ptrINS4_10StatisticsEERKNS1_19ParquetColumnSchemaENS_16DecodedValueKindEPNS1_23ParquetColumnStatisticsEPKN4cctz9time_zoneE Unexecuted instantiation: parquet_statistics.cpp:_ZN5doris6format7parquet12_GLOBAL__N_119set_decoded_min_maxIN7parquet12PhysicalTypeILNS4_4Type4typeE5EEEEEbRKSt10shared_ptrINS4_10StatisticsEERKNS1_19ParquetColumnSchemaENS_16DecodedValueKindEPNS1_23ParquetColumnStatisticsEPKN4cctz9time_zoneE |
136 | | |
137 | | bool set_decoded_binary_field(const ParquetColumnSchema& column_schema, DecodedValueKind value_kind, |
138 | | const StringRef& value, Field* field, |
139 | 2 | const cctz::time_zone* timezone) { |
140 | 2 | std::vector<StringRef> binary_values {value}; |
141 | 2 | DecodedColumnView view; |
142 | 2 | view.value_kind = value_kind; |
143 | 2 | view.binary_values = &binary_values; |
144 | 2 | return read_decoded_field(column_schema, view, field, timezone).ok(); |
145 | 2 | } |
146 | | |
147 | | bool set_string_min_max(const std::shared_ptr<::parquet::Statistics>& statistics, |
148 | | const ParquetColumnSchema& column_schema, |
149 | | ParquetColumnStatistics* column_statistics, |
150 | 1 | const cctz::time_zone* timezone) { |
151 | 1 | switch (statistics->physical_type()) { |
152 | 1 | case ::parquet::Type::BYTE_ARRAY: { |
153 | 1 | auto typed_statistics = |
154 | 1 | std::static_pointer_cast<::parquet::TypedStatistics<::parquet::ByteArrayType>>( |
155 | 1 | statistics); |
156 | 1 | const auto min = ::parquet::ByteArrayToString(typed_statistics->min()); |
157 | 1 | const auto max = ::parquet::ByteArrayToString(typed_statistics->max()); |
158 | 1 | if (!set_decoded_binary_field(column_schema, DecodedValueKind::BINARY, |
159 | 1 | StringRef(min.data(), min.size()), |
160 | 1 | &column_statistics->min_value, timezone) || |
161 | 1 | !set_decoded_binary_field(column_schema, DecodedValueKind::BINARY, |
162 | 1 | StringRef(max.data(), max.size()), |
163 | 1 | &column_statistics->max_value, timezone)) { |
164 | 0 | return false; |
165 | 0 | } |
166 | 1 | return true; |
167 | 1 | } |
168 | 0 | case ::parquet::Type::FIXED_LEN_BYTE_ARRAY: { |
169 | 0 | if (column_schema.descriptor == nullptr || column_schema.descriptor->type_length() <= 0) { |
170 | 0 | return false; |
171 | 0 | } |
172 | 0 | auto typed_statistics = |
173 | 0 | std::static_pointer_cast<::parquet::TypedStatistics<::parquet::FLBAType>>( |
174 | 0 | statistics); |
175 | 0 | const int type_length = column_schema.descriptor->type_length(); |
176 | 0 | const std::string min(reinterpret_cast<const char*>(typed_statistics->min().ptr), |
177 | 0 | type_length); |
178 | 0 | const std::string max(reinterpret_cast<const char*>(typed_statistics->max().ptr), |
179 | 0 | type_length); |
180 | 0 | if (!set_decoded_binary_field(column_schema, DecodedValueKind::FIXED_BINARY, |
181 | 0 | StringRef(min.data(), min.size()), |
182 | 0 | &column_statistics->min_value, timezone) || |
183 | 0 | !set_decoded_binary_field(column_schema, DecodedValueKind::FIXED_BINARY, |
184 | 0 | StringRef(max.data(), max.size()), |
185 | 0 | &column_statistics->max_value, timezone)) { |
186 | 0 | return false; |
187 | 0 | } |
188 | 0 | return true; |
189 | 0 | } |
190 | 0 | default: |
191 | 0 | return false; |
192 | 1 | } |
193 | 1 | } |
194 | | |
195 | | template <typename T> |
196 | 3 | T load_predicate_value(const char* data) { |
197 | 3 | T value; |
198 | 3 | memcpy(&value, data, sizeof(T)); |
199 | 3 | return value; |
200 | 3 | } Unexecuted instantiation: parquet_statistics.cpp:_ZN5doris6format7parquet12_GLOBAL__N_120load_predicate_valueIbEET_PKc Unexecuted instantiation: parquet_statistics.cpp:_ZN5doris6format7parquet12_GLOBAL__N_120load_predicate_valueIiEET_PKc Unexecuted instantiation: parquet_statistics.cpp:_ZN5doris6format7parquet12_GLOBAL__N_120load_predicate_valueIaEET_PKc Unexecuted instantiation: parquet_statistics.cpp:_ZN5doris6format7parquet12_GLOBAL__N_120load_predicate_valueIsEET_PKc parquet_statistics.cpp:_ZN5doris6format7parquet12_GLOBAL__N_120load_predicate_valueIlEET_PKc Line | Count | Source | 196 | 3 | T load_predicate_value(const char* data) { | 197 | 3 | T value; | 198 | 3 | memcpy(&value, data, sizeof(T)); | 199 | 3 | return value; | 200 | 3 | } |
Unexecuted instantiation: parquet_statistics.cpp:_ZN5doris6format7parquet12_GLOBAL__N_120load_predicate_valueIfEET_PKc Unexecuted instantiation: parquet_statistics.cpp:_ZN5doris6format7parquet12_GLOBAL__N_120load_predicate_valueIdEET_PKc |
201 | | |
202 | 3 | std::optional<int64_t> load_predicate_integral_value(const char* buf, size_t size) { |
203 | 3 | switch (size) { |
204 | 0 | case sizeof(int8_t): |
205 | 0 | return static_cast<int64_t>(load_predicate_value<int8_t>(buf)); |
206 | 0 | case sizeof(int16_t): |
207 | 0 | return static_cast<int64_t>(load_predicate_value<int16_t>(buf)); |
208 | 0 | case sizeof(int32_t): |
209 | 0 | return static_cast<int64_t>(load_predicate_value<int32_t>(buf)); |
210 | 3 | case sizeof(int64_t): |
211 | 3 | return load_predicate_value<int64_t>(buf); |
212 | 0 | default: |
213 | 0 | return std::nullopt; |
214 | 3 | } |
215 | 3 | } |
216 | | |
217 | | bool logical_integer_fits_physical_int32(const ParquetTypeDescriptor& type_descriptor, |
218 | 3 | int64_t value) { |
219 | 3 | const int bit_width = |
220 | 3 | type_descriptor.integer_bit_width > 0 ? type_descriptor.integer_bit_width : 32; |
221 | 3 | if (type_descriptor.is_unsigned_integer) { |
222 | 3 | const uint64_t max_value = bit_width >= 32 ? std::numeric_limits<uint32_t>::max() |
223 | 3 | : ((uint64_t {1} << bit_width) - 1); |
224 | 3 | return value >= 0 && static_cast<uint64_t>(value) <= max_value; |
225 | 3 | } |
226 | 0 | const int64_t min_value = bit_width >= 32 ? std::numeric_limits<int32_t>::min() |
227 | 0 | : -(int64_t {1} << (bit_width - 1)); |
228 | 0 | const int64_t max_value = bit_width >= 32 ? std::numeric_limits<int32_t>::max() |
229 | 0 | : ((int64_t {1} << (bit_width - 1)) - 1); |
230 | 0 | return value >= min_value && value <= max_value; |
231 | 3 | } |
232 | | |
233 | | std::optional<int32_t> convert_logical_integer_to_physical_int32( |
234 | 3 | const ParquetTypeDescriptor& type_descriptor, int64_t value) { |
235 | 3 | if (!logical_integer_fits_physical_int32(type_descriptor, value)) { |
236 | 2 | return std::nullopt; |
237 | 2 | } |
238 | 1 | if (!type_descriptor.is_unsigned_integer) { |
239 | 0 | return static_cast<int32_t>(value); |
240 | 0 | } |
241 | 1 | const auto unsigned_value = static_cast<uint32_t>(value); |
242 | 1 | int32_t physical_value; |
243 | 1 | memcpy(&physical_value, &unsigned_value, sizeof(physical_value)); |
244 | 1 | return physical_value; |
245 | 1 | } |
246 | | |
247 | | class ArrowParquetBloomFilterAdapter final : public segment_v2::BloomFilter { |
248 | | public: |
249 | | ArrowParquetBloomFilterAdapter(const ParquetColumnSchema& column_schema, |
250 | | const ::parquet::BloomFilter& bloom_filter) |
251 | 5 | : _column_schema(column_schema), _bloom_filter(bloom_filter) {} |
252 | | |
253 | 0 | void add_bytes(const char* buf, size_t size) override { DORIS_CHECK(false); } |
254 | | |
255 | 5 | bool test_bytes(const char* buf, size_t size) const override { |
256 | 5 | if (buf == nullptr) { |
257 | 0 | return true; |
258 | 0 | } |
259 | | // Parquet bloom filters are populated from the physical column carrier, while VExpr |
260 | | // literals are materialized as Doris logical values. Keep the logical type in |
261 | | // BloomFilterEvalContext for expression compatibility, and normalize to the Parquet |
262 | | // physical representation only at this adapter boundary. |
263 | 5 | switch (_column_schema.type_descriptor.physical_type) { |
264 | 0 | case ::parquet::Type::BOOLEAN: |
265 | 0 | return test_boolean(buf, size); |
266 | 3 | case ::parquet::Type::INT32: |
267 | 3 | return test_physical_int32(buf, size); |
268 | 0 | case ::parquet::Type::INT64: |
269 | 0 | return test_int64(buf, size); |
270 | 0 | case ::parquet::Type::FLOAT: |
271 | 0 | return test_float(buf, size); |
272 | 0 | case ::parquet::Type::DOUBLE: |
273 | 0 | return test_double(buf, size); |
274 | 0 | case ::parquet::Type::BYTE_ARRAY: |
275 | 0 | return test_byte_array(buf, size); |
276 | 2 | case ::parquet::Type::FIXED_LEN_BYTE_ARRAY: |
277 | 2 | return test_fixed_len_byte_array(buf, size); |
278 | 0 | default: |
279 | 0 | return true; |
280 | 5 | } |
281 | 5 | } |
282 | | |
283 | 0 | void set_has_null(bool has_null) override { DORIS_CHECK(!has_null); } |
284 | 0 | bool has_null() const override { return false; } |
285 | 0 | void add_hash(uint64_t hash) override { DORIS_CHECK(false); } |
286 | 0 | bool test_hash(uint64_t hash) const override { return _bloom_filter.FindHash(hash); } |
287 | | |
288 | | private: |
289 | 0 | bool test_boolean(const char* buf, size_t size) const { |
290 | 0 | if (size == sizeof(bool)) { |
291 | 0 | const int32_t value = load_predicate_value<bool>(buf) ? 1 : 0; |
292 | 0 | return _bloom_filter.FindHash(_bloom_filter.Hash(value)); |
293 | 0 | } |
294 | 0 | if (size == sizeof(int32_t)) { |
295 | 0 | const int32_t value = load_predicate_value<int32_t>(buf); |
296 | 0 | return _bloom_filter.FindHash(_bloom_filter.Hash(value != 0 ? 1 : 0)); |
297 | 0 | } |
298 | 0 | return true; |
299 | 0 | } |
300 | | |
301 | 3 | bool test_physical_int32(const char* buf, size_t size) const { |
302 | 3 | const auto logical_value = load_predicate_integral_value(buf, size); |
303 | 3 | if (!logical_value.has_value()) { |
304 | 0 | return true; |
305 | 0 | } |
306 | 3 | const auto physical_value = convert_logical_integer_to_physical_int32( |
307 | 3 | _column_schema.type_descriptor, *logical_value); |
308 | 3 | if (!physical_value.has_value()) { |
309 | 2 | return false; |
310 | 2 | } |
311 | 1 | return find_int32(*physical_value); |
312 | 3 | } |
313 | | |
314 | 0 | bool test_int64(const char* buf, size_t size) const { |
315 | 0 | if (size != sizeof(int64_t)) { |
316 | 0 | return true; |
317 | 0 | } |
318 | 0 | const int64_t value = load_predicate_value<int64_t>(buf); |
319 | 0 | return _bloom_filter.FindHash(_bloom_filter.Hash(value)); |
320 | 0 | } |
321 | | |
322 | 0 | bool test_float(const char* buf, size_t size) const { |
323 | 0 | if (size != sizeof(float)) { |
324 | 0 | return true; |
325 | 0 | } |
326 | 0 | const float value = load_predicate_value<float>(buf); |
327 | 0 | return _bloom_filter.FindHash(_bloom_filter.Hash(value)); |
328 | 0 | } |
329 | | |
330 | 0 | bool test_double(const char* buf, size_t size) const { |
331 | 0 | if (size != sizeof(double)) { |
332 | 0 | return true; |
333 | 0 | } |
334 | 0 | const double value = load_predicate_value<double>(buf); |
335 | 0 | return _bloom_filter.FindHash(_bloom_filter.Hash(value)); |
336 | 0 | } |
337 | | |
338 | 0 | bool test_byte_array(const char* buf, size_t size) const { |
339 | 0 | ::parquet::ByteArray value(static_cast<uint32_t>(size), |
340 | 0 | reinterpret_cast<const uint8_t*>(buf)); |
341 | 0 | return _bloom_filter.FindHash(_bloom_filter.Hash(&value)); |
342 | 0 | } |
343 | | |
344 | 2 | bool test_fixed_len_byte_array(const char* buf, size_t size) const { |
345 | 2 | if (_column_schema.type_descriptor.fixed_length <= 0) { |
346 | 0 | return true; |
347 | 0 | } |
348 | 2 | if (size != static_cast<size_t>(_column_schema.type_descriptor.fixed_length)) { |
349 | 1 | return false; |
350 | 1 | } |
351 | 1 | ::parquet::FLBA value(reinterpret_cast<const uint8_t*>(buf)); |
352 | 1 | return _bloom_filter.FindHash( |
353 | 1 | _bloom_filter.Hash(&value, _column_schema.type_descriptor.fixed_length)); |
354 | 2 | } |
355 | | |
356 | 1 | bool find_int32(int32_t value) const { |
357 | 1 | return _bloom_filter.FindHash(_bloom_filter.Hash(value)); |
358 | 1 | } |
359 | | |
360 | | const ParquetColumnSchema& _column_schema; |
361 | | const ::parquet::BloomFilter& _bloom_filter; |
362 | | }; |
363 | | |
364 | 6 | bool bloom_filter_supported(const ParquetColumnSchema& column_schema) { |
365 | 6 | if (!bloom_logical_type_supported(column_schema)) { |
366 | 0 | return false; |
367 | 0 | } |
368 | 6 | switch (column_schema.type_descriptor.physical_type) { |
369 | 0 | case ::parquet::Type::BOOLEAN: |
370 | 3 | case ::parquet::Type::INT32: |
371 | 3 | case ::parquet::Type::INT64: |
372 | 3 | case ::parquet::Type::FLOAT: |
373 | 3 | case ::parquet::Type::DOUBLE: |
374 | 3 | case ::parquet::Type::BYTE_ARRAY: |
375 | 3 | return true; |
376 | 3 | case ::parquet::Type::FIXED_LEN_BYTE_ARRAY: |
377 | 3 | return column_schema.type_descriptor.is_string_like && |
378 | 3 | column_schema.type_descriptor.fixed_length > 0; |
379 | 0 | default: |
380 | 0 | return false; |
381 | 6 | } |
382 | 6 | } |
383 | | |
384 | | bool bloom_filter_excludes(const ParquetColumnSchema& column_schema, int slot_index, |
385 | | const VExprContextSPtrs& conjuncts, |
386 | 6 | const ::parquet::BloomFilter& bloom_filter) { |
387 | 6 | if (!bloom_filter_supported(column_schema)) { |
388 | 1 | return false; |
389 | 1 | } |
390 | 5 | ArrowParquetBloomFilterAdapter adapter(column_schema, bloom_filter); |
391 | 5 | BloomFilterEvalContext ctx; |
392 | 5 | ctx.slots.emplace(slot_index, BloomFilterEvalContext::SlotBloomFilter { |
393 | 5 | .data_type = column_schema.type, |
394 | 5 | .bloom_filter = &adapter, |
395 | 5 | }); |
396 | 5 | return VExprContext::evaluate_bloom_filter(conjuncts, ctx) == ZoneMapFilterResult::kNoMatch; |
397 | 6 | } |
398 | | |
399 | | struct RowGroupBloomFilterCache { |
400 | | ::parquet::BloomFilterReader* bloom_filter_reader = nullptr; |
401 | | std::map<int, std::unique_ptr<::parquet::BloomFilter>> column_bloom_filters; |
402 | | std::set<int> loaded_columns; |
403 | | |
404 | | ::parquet::BloomFilter* get(int row_group_idx, int leaf_column_id, |
405 | 0 | ParquetPruningStats* pruning_stats) { |
406 | 0 | if (bloom_filter_reader == nullptr || leaf_column_id < 0) { |
407 | 0 | return nullptr; |
408 | 0 | } |
409 | 0 | if (loaded_columns.find(leaf_column_id) == loaded_columns.end()) { |
410 | 0 | loaded_columns.insert(leaf_column_id); |
411 | 0 | try { |
412 | 0 | std::shared_ptr<::parquet::RowGroupBloomFilterReader> row_group_reader; |
413 | 0 | if (pruning_stats != nullptr) { |
414 | 0 | SCOPED_RAW_TIMER(&pruning_stats->bloom_filter_read_time); |
415 | 0 | row_group_reader = bloom_filter_reader->RowGroup(row_group_idx); |
416 | 0 | if (row_group_reader != nullptr) { |
417 | 0 | column_bloom_filters[leaf_column_id] = |
418 | 0 | row_group_reader->GetColumnBloomFilter(leaf_column_id); |
419 | 0 | } |
420 | 0 | } else { |
421 | 0 | row_group_reader = bloom_filter_reader->RowGroup(row_group_idx); |
422 | 0 | if (row_group_reader != nullptr) { |
423 | 0 | column_bloom_filters[leaf_column_id] = |
424 | 0 | row_group_reader->GetColumnBloomFilter(leaf_column_id); |
425 | 0 | } |
426 | 0 | } |
427 | 0 | } catch (const ::parquet::ParquetException&) { |
428 | 0 | return nullptr; |
429 | 0 | } catch (const std::exception&) { |
430 | 0 | return nullptr; |
431 | 0 | } |
432 | 0 | } |
433 | 0 | auto it = column_bloom_filters.find(leaf_column_id); |
434 | 0 | return it == column_bloom_filters.end() ? nullptr : it->second.get(); |
435 | 0 | } |
436 | | }; |
437 | | |
438 | 29 | bool is_dictionary_data_encoding(::parquet::Encoding::type encoding) { |
439 | 29 | return encoding == ::parquet::Encoding::PLAIN_DICTIONARY || |
440 | 29 | encoding == ::parquet::Encoding::RLE_DICTIONARY; |
441 | 29 | } |
442 | | |
443 | 0 | bool is_level_encoding(::parquet::Encoding::type encoding) { |
444 | 0 | return encoding == ::parquet::Encoding::RLE || encoding == ::parquet::Encoding::BIT_PACKED; |
445 | 0 | } |
446 | | |
447 | 58 | bool is_data_page_type(::parquet::PageType::type page_type) { |
448 | 58 | return page_type == ::parquet::PageType::DATA_PAGE || |
449 | 58 | page_type == ::parquet::PageType::DATA_PAGE_V2; |
450 | 58 | } |
451 | | |
452 | 33 | bool is_dictionary_encoded_chunk(const ::parquet::ColumnChunkMetaData& column_metadata) { |
453 | 33 | if (!column_metadata.has_dictionary_page()) { |
454 | 4 | return false; |
455 | 4 | } |
456 | | |
457 | 29 | const auto& encoding_stats = column_metadata.encoding_stats(); |
458 | 29 | if (!encoding_stats.empty()) { |
459 | 29 | bool has_dictionary_data_page = false; |
460 | 58 | for (const auto& encoding_stat : encoding_stats) { |
461 | 58 | if (!is_data_page_type(encoding_stat.page_type) || encoding_stat.count <= 0) { |
462 | 29 | continue; |
463 | 29 | } |
464 | 29 | if (!is_dictionary_data_encoding(encoding_stat.encoding)) { |
465 | 0 | return false; |
466 | 0 | } |
467 | 29 | has_dictionary_data_page = true; |
468 | 29 | } |
469 | 29 | return has_dictionary_data_page; |
470 | 29 | } |
471 | | |
472 | 0 | bool has_dictionary_encoding = false; |
473 | 0 | for (const auto encoding : column_metadata.encodings()) { |
474 | 0 | if (is_dictionary_data_encoding(encoding)) { |
475 | 0 | has_dictionary_encoding = true; |
476 | 0 | continue; |
477 | 0 | } |
478 | 0 | if (!is_level_encoding(encoding)) { |
479 | 0 | return false; |
480 | 0 | } |
481 | 0 | } |
482 | 0 | return has_dictionary_encoding; |
483 | 0 | } |
484 | | |
485 | | bool supports_dictionary_pruning(const ParquetColumnSchema& column_schema, |
486 | 33 | const ::parquet::ColumnChunkMetaData& column_metadata) { |
487 | 33 | if (column_schema.kind != ParquetColumnSchemaKind::PRIMITIVE || |
488 | 33 | column_schema.descriptor == nullptr || column_schema.type == nullptr) { |
489 | 0 | return false; |
490 | 0 | } |
491 | 33 | if (!column_schema.type_descriptor.is_string_like) { |
492 | 0 | return false; |
493 | 0 | } |
494 | 33 | if (column_metadata.type() != ::parquet::Type::BYTE_ARRAY && |
495 | 33 | column_metadata.type() != ::parquet::Type::FIXED_LEN_BYTE_ARRAY) { |
496 | 0 | return false; |
497 | 0 | } |
498 | 33 | return true; |
499 | 33 | } |
500 | | |
501 | | struct OwnedDictionaryWords { |
502 | | std::vector<std::string> values; |
503 | | std::vector<StringRef> refs; |
504 | | |
505 | 29 | void clear() { |
506 | 29 | values.clear(); |
507 | 29 | refs.clear(); |
508 | 29 | } |
509 | | |
510 | 29 | void build_refs() { |
511 | 29 | refs.reserve(values.size()); |
512 | 39 | for (const auto& value : values) { |
513 | 39 | refs.emplace_back(value.data(), value.size()); |
514 | 39 | } |
515 | 29 | } |
516 | | }; |
517 | | |
518 | | bool read_dictionary_words(::parquet::ParquetFileReader* file_reader, int row_group_idx, |
519 | | int leaf_column_id, const ParquetColumnSchema& column_schema, |
520 | 29 | OwnedDictionaryWords* dict_words) { |
521 | 29 | DORIS_CHECK(dict_words != nullptr); |
522 | 29 | dict_words->clear(); |
523 | 29 | if (file_reader == nullptr || leaf_column_id < 0) { |
524 | 0 | return false; |
525 | 0 | } |
526 | | |
527 | 29 | auto row_group_reader = file_reader->RowGroup(row_group_idx); |
528 | 29 | if (row_group_reader == nullptr) { |
529 | 0 | return false; |
530 | 0 | } |
531 | 29 | auto page_reader = row_group_reader->GetColumnPageReader(leaf_column_id); |
532 | 29 | if (page_reader == nullptr) { |
533 | 0 | return false; |
534 | 0 | } |
535 | | |
536 | 29 | std::shared_ptr<::parquet::Page> page; |
537 | 29 | try { |
538 | 29 | page = page_reader->NextPage(); |
539 | 29 | } catch (const ::parquet::ParquetException&) { |
540 | 0 | return false; |
541 | 0 | } catch (const std::exception&) { |
542 | 0 | return false; |
543 | 0 | } |
544 | 29 | if (page == nullptr || page->type() != ::parquet::PageType::DICTIONARY_PAGE) { |
545 | 0 | return false; |
546 | 0 | } |
547 | 29 | const auto* dictionary_page = static_cast<const ::parquet::DictionaryPage*>(page.get()); |
548 | 29 | if (dictionary_page->encoding() != ::parquet::Encoding::PLAIN && |
549 | 29 | dictionary_page->encoding() != ::parquet::Encoding::PLAIN_DICTIONARY) { |
550 | 0 | return false; |
551 | 0 | } |
552 | 29 | const int32_t dictionary_length = dictionary_page->num_values(); |
553 | 29 | if (dictionary_length <= 0) { |
554 | 0 | return false; |
555 | 0 | } |
556 | 29 | const auto* dictionary_data = dictionary_page->data(); |
557 | 29 | const int dictionary_size = dictionary_page->size(); |
558 | | |
559 | 29 | dict_words->values.reserve(static_cast<size_t>(dictionary_length)); |
560 | 29 | if (column_schema.descriptor->physical_type() == ::parquet::Type::BYTE_ARRAY) { |
561 | 29 | auto decoder = ::parquet::MakeTypedDecoder<::parquet::ByteArrayType>( |
562 | 29 | ::parquet::Encoding::PLAIN, column_schema.descriptor); |
563 | 29 | decoder->SetData(dictionary_length, dictionary_data, dictionary_size); |
564 | 29 | std::vector<::parquet::ByteArray> byte_array_values(static_cast<size_t>(dictionary_length)); |
565 | 29 | if (decoder->Decode(byte_array_values.data(), dictionary_length) != dictionary_length) { |
566 | 0 | return false; |
567 | 0 | } |
568 | 68 | for (int32_t dict_idx = 0; dict_idx < dictionary_length; ++dict_idx) { |
569 | 39 | dict_words->values.emplace_back( |
570 | 39 | reinterpret_cast<const char*>(byte_array_values[dict_idx].ptr), |
571 | 39 | byte_array_values[dict_idx].len); |
572 | 39 | } |
573 | 29 | dict_words->build_refs(); |
574 | 29 | return true; |
575 | 29 | } |
576 | 0 | if (column_schema.descriptor->physical_type() == ::parquet::Type::FIXED_LEN_BYTE_ARRAY) { |
577 | 0 | const int type_length = column_schema.descriptor->type_length(); |
578 | 0 | if (type_length <= 0) { |
579 | 0 | return false; |
580 | 0 | } |
581 | 0 | auto decoder = ::parquet::MakeTypedDecoder<::parquet::FLBAType>(::parquet::Encoding::PLAIN, |
582 | 0 | column_schema.descriptor); |
583 | 0 | decoder->SetData(dictionary_length, dictionary_data, dictionary_size); |
584 | 0 | std::vector<::parquet::FixedLenByteArray> flba_values( |
585 | 0 | static_cast<size_t>(dictionary_length)); |
586 | 0 | if (decoder->Decode(flba_values.data(), dictionary_length) != dictionary_length) { |
587 | 0 | return false; |
588 | 0 | } |
589 | 0 | for (int32_t dict_idx = 0; dict_idx < dictionary_length; ++dict_idx) { |
590 | 0 | dict_words->values.emplace_back( |
591 | 0 | reinterpret_cast<const char*>(flba_values[dict_idx].ptr), type_length); |
592 | 0 | } |
593 | 0 | dict_words->build_refs(); |
594 | 0 | return true; |
595 | 0 | } |
596 | 0 | return false; |
597 | 0 | } |
598 | | |
599 | | const ParquetColumnSchema* resolve_local_leaf_schema( |
600 | | const std::vector<std::unique_ptr<ParquetColumnSchema>>& schema, |
601 | 108 | const format::LocalColumnId file_column_id) { |
602 | 108 | if (!file_column_id.is_valid() || file_column_id.value() >= static_cast<int>(schema.size())) { |
603 | 0 | return nullptr; |
604 | 0 | } |
605 | 108 | const ParquetColumnSchema* column_schema = schema[file_column_id.value()].get(); |
606 | 108 | if (column_schema == nullptr || column_schema->kind != ParquetColumnSchemaKind::PRIMITIVE || |
607 | 108 | column_schema->leaf_column_id < 0 || column_schema->max_repetition_level > 0) { |
608 | 2 | return nullptr; |
609 | 2 | } |
610 | 106 | return column_schema; |
611 | 108 | } |
612 | | |
613 | | std::optional<format::LocalColumnId> file_column_id_by_block_position( |
614 | 108 | const format::FileScanRequest& request, int block_position) { |
615 | 135 | for (const auto& [file_column_id, local_index] : request.local_positions) { |
616 | 135 | if (local_index.value() == block_position) { |
617 | 108 | return file_column_id; |
618 | 108 | } |
619 | 135 | } |
620 | 0 | return std::nullopt; |
621 | 108 | } |
622 | | |
623 | | bool has_expr_zonemap_filter(const format::FileScanRequest& request, |
624 | 342 | const RuntimeState* runtime_state) { |
625 | 342 | if (!expr_zonemap::is_expr_zonemap_filter_enabled(runtime_state)) { |
626 | 0 | return false; |
627 | 0 | } |
628 | 342 | for (const auto& conjunct : request.conjuncts) { |
629 | 148 | if (conjunct != nullptr && conjunct->root() != nullptr && |
630 | 148 | conjunct->root()->can_evaluate_zonemap_filter()) { |
631 | 101 | return true; |
632 | 101 | } |
633 | 148 | } |
634 | 241 | return false; |
635 | 342 | } |
636 | | |
637 | 65 | std::set<int> collect_expr_zonemap_slot_indexes(const VExprContextSPtrs& conjuncts) { |
638 | 65 | std::set<int> slot_indexes; |
639 | 67 | for (const auto& conjunct : conjuncts) { |
640 | 67 | if (conjunct != nullptr && conjunct->root() != nullptr && |
641 | 67 | conjunct->root()->can_evaluate_zonemap_filter()) { |
642 | 67 | conjunct->root()->collect_slot_column_ids(slot_indexes); |
643 | 67 | } |
644 | 67 | } |
645 | 65 | return slot_indexes; |
646 | 65 | } |
647 | | |
648 | | template <typename SlotIndexSelector> |
649 | | std::map<int, VExprContextSPtrs> collect_conjuncts_by_single_slot( |
650 | 324 | const VExprContextSPtrs& conjuncts, SlotIndexSelector slot_index_selector) { |
651 | 324 | std::map<int, VExprContextSPtrs> conjuncts_by_slot; |
652 | 324 | for (const auto& conjunct : conjuncts) { |
653 | 134 | const auto slot_index = slot_index_selector(conjunct); |
654 | 134 | if (slot_index >= 0) { |
655 | 33 | conjuncts_by_slot[slot_index].push_back(conjunct); |
656 | 33 | } |
657 | 134 | } |
658 | 324 | return conjuncts_by_slot; |
659 | 324 | } |
660 | | |
661 | 29 | std::vector<Field> dictionary_fields_from_words(const OwnedDictionaryWords& dict_words) { |
662 | 29 | std::vector<Field> fields; |
663 | 29 | fields.reserve(dict_words.refs.size()); |
664 | 39 | for (const auto& ref : dict_words.refs) { |
665 | 39 | fields.push_back(Field::create_field<TYPE_STRING>(String(ref.data, ref.size))); |
666 | 39 | } |
667 | 29 | return fields; |
668 | 29 | } |
669 | | |
670 | | std::shared_ptr<segment_v2::ZoneMap> make_zonemap_from_statistics( |
671 | 193 | const ParquetColumnStatistics& statistics) { |
672 | 193 | if (!statistics.has_null_count && !statistics.has_min_max) { |
673 | 0 | return nullptr; |
674 | 0 | } |
675 | 193 | segment_v2::ZoneMap zone_map; |
676 | 193 | zone_map.has_null = statistics.has_null; |
677 | 193 | zone_map.has_not_null = statistics.has_not_null; |
678 | 193 | if (!statistics.has_not_null) { |
679 | 3 | return std::make_shared<segment_v2::ZoneMap>(std::move(zone_map)); |
680 | 3 | } |
681 | 190 | if (!statistics.has_min_max) { |
682 | 0 | return nullptr; |
683 | 0 | } |
684 | 190 | zone_map.min_value = statistics.min_value; |
685 | 190 | zone_map.max_value = statistics.max_value; |
686 | 190 | return std::make_shared<segment_v2::ZoneMap>(std::move(zone_map)); |
687 | 190 | } |
688 | | |
689 | | void add_slot_zonemap(ZoneMapEvalContext* ctx, int slot_index, const DataTypePtr& data_type, |
690 | 193 | std::shared_ptr<segment_v2::ZoneMap> zone_map) { |
691 | 193 | DORIS_CHECK(ctx != nullptr); |
692 | 193 | ZoneMapEvalContext::SlotZoneMap slot_zone_map; |
693 | 193 | slot_zone_map.data_type = data_type; |
694 | 193 | slot_zone_map.zone_map = std::move(zone_map); |
695 | 193 | ctx->slots.emplace(slot_index, std::move(slot_zone_map)); |
696 | 193 | } |
697 | | |
698 | 65 | void accumulate_zonemap_stats(const ZoneMapEvalContext& ctx, ParquetPruningStats* pruning_stats) { |
699 | 65 | if (pruning_stats == nullptr) { |
700 | 0 | return; |
701 | 0 | } |
702 | 65 | pruning_stats->expr_zonemap_unusable_evals += ctx.stats.unusable_zonemap_eval_count; |
703 | 65 | pruning_stats->in_zonemap_point_check_count += ctx.stats.in_zonemap_point_check_count; |
704 | 65 | pruning_stats->in_zonemap_range_only_count += ctx.stats.in_zonemap_range_only_count; |
705 | 65 | } |
706 | | |
707 | | } // namespace |
708 | | |
709 | | ParquetColumnStatistics ParquetStatisticsUtils::TransformColumnStatistics( |
710 | | const ParquetColumnSchema& column_schema, |
711 | 90 | const std::shared_ptr<::parquet::Statistics>& statistics, const cctz::time_zone* timezone) { |
712 | 90 | ParquetColumnStatistics result; |
713 | 90 | if (statistics == nullptr) { |
714 | 2 | return result; |
715 | 2 | } |
716 | | |
717 | 88 | result.has_null = statistics->HasNullCount() && statistics->null_count() > 0; |
718 | 88 | result.has_not_null = statistics->num_values() > 0 || statistics->HasMinMax(); |
719 | 88 | result.has_null_count = statistics->HasNullCount(); |
720 | 88 | if (!result.has_not_null || !statistics->HasMinMax()) { |
721 | 4 | return result; |
722 | 4 | } |
723 | | |
724 | 84 | DORIS_CHECK(column_schema.type != nullptr); |
725 | 84 | switch (statistics->physical_type()) { |
726 | 0 | case ::parquet::Type::BOOLEAN: |
727 | 0 | result.has_min_max = set_decoded_min_max<::parquet::BooleanType>( |
728 | 0 | statistics, column_schema, DecodedValueKind::BOOL, &result, timezone); |
729 | 0 | return result; |
730 | 82 | case ::parquet::Type::INT32: |
731 | 82 | result.has_min_max = set_decoded_min_max<::parquet::Int32Type>( |
732 | 82 | statistics, column_schema, decoded_value_kind(column_schema.type_descriptor), |
733 | 82 | &result, timezone); |
734 | 82 | return result; |
735 | 1 | case ::parquet::Type::INT64: |
736 | 1 | result.has_min_max = set_decoded_min_max<::parquet::Int64Type>( |
737 | 1 | statistics, column_schema, decoded_value_kind(column_schema.type_descriptor), |
738 | 1 | &result, timezone); |
739 | 1 | return result; |
740 | 0 | case ::parquet::Type::FLOAT: |
741 | 0 | result.has_min_max = set_decoded_min_max<::parquet::FloatType>( |
742 | 0 | statistics, column_schema, DecodedValueKind::FLOAT, &result, timezone); |
743 | 0 | return result; |
744 | 0 | case ::parquet::Type::DOUBLE: |
745 | 0 | result.has_min_max = set_decoded_min_max<::parquet::DoubleType>( |
746 | 0 | statistics, column_schema, DecodedValueKind::DOUBLE, &result, timezone); |
747 | 0 | return result; |
748 | 1 | case ::parquet::Type::BYTE_ARRAY: |
749 | 1 | case ::parquet::Type::FIXED_LEN_BYTE_ARRAY: |
750 | 1 | result.has_min_max = set_string_min_max(statistics, column_schema, &result, timezone); |
751 | 1 | return result; |
752 | 0 | default: |
753 | 0 | return result; |
754 | 84 | } |
755 | 84 | } |
756 | | |
757 | | bool ParquetStatisticsUtils::BloomFilterExcludes(const ParquetColumnSchema& column_schema, |
758 | | int slot_index, const VExprContextSPtrs& conjuncts, |
759 | 6 | const ::parquet::BloomFilter& bloom_filter) { |
760 | 6 | return bloom_filter_excludes(column_schema, slot_index, conjuncts, bloom_filter); |
761 | 6 | } |
762 | | |
763 | | namespace { |
764 | | |
765 | | ParquetRowGroupPruneReason dictionary_prune_reason( |
766 | | const ::parquet::RowGroupMetaData& row_group, ::parquet::ParquetFileReader* file_reader, |
767 | | int row_group_idx, const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema, |
768 | 172 | const format::FileScanRequest& request) { |
769 | 172 | const auto conjuncts_by_slot = collect_conjuncts_by_single_slot( |
770 | 172 | request.conjuncts, expr_zonemap::single_slot_dictionary_index); |
771 | 172 | for (const auto& [slot_index, conjuncts] : conjuncts_by_slot) { |
772 | 33 | const auto file_column_id = file_column_id_by_block_position(request, slot_index); |
773 | 33 | if (!file_column_id.has_value()) { |
774 | 0 | continue; |
775 | 0 | } |
776 | 33 | const auto* column_schema = resolve_local_leaf_schema(file_schema, *file_column_id); |
777 | 33 | if (column_schema == nullptr || column_schema->type == nullptr) { |
778 | 0 | continue; |
779 | 0 | } |
780 | 33 | DCHECK_LT(column_schema->leaf_column_id, row_group.num_columns()); |
781 | 33 | auto column_chunk = row_group.ColumnChunk(column_schema->leaf_column_id); |
782 | 33 | if (column_chunk == nullptr || |
783 | 33 | !supports_dictionary_pruning(*column_schema, *column_chunk) || |
784 | 33 | !is_dictionary_encoded_chunk(*column_chunk)) { |
785 | 4 | continue; |
786 | 4 | } |
787 | | |
788 | 29 | OwnedDictionaryWords dict_words; |
789 | 29 | if (!read_dictionary_words(file_reader, row_group_idx, column_schema->leaf_column_id, |
790 | 29 | *column_schema, &dict_words)) { |
791 | 0 | continue; |
792 | 0 | } |
793 | 29 | DictionaryEvalContext ctx; |
794 | 29 | ctx.slots.emplace(slot_index, DictionaryEvalContext::SlotDictionary { |
795 | 29 | .data_type = column_schema->type, |
796 | 29 | .values = dictionary_fields_from_words(dict_words), |
797 | 29 | }); |
798 | 29 | if (VExprContext::evaluate_dictionary_filter(conjuncts, ctx) == |
799 | 29 | ZoneMapFilterResult::kNoMatch) { |
800 | 20 | return ParquetRowGroupPruneReason::DICTIONARY; |
801 | 20 | } |
802 | 29 | } |
803 | 152 | return ParquetRowGroupPruneReason::NONE; |
804 | 172 | } |
805 | | |
806 | | ParquetRowGroupPruneReason bloom_filter_prune_reason( |
807 | | int row_group_idx, const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema, |
808 | | const format::FileScanRequest& request, RowGroupBloomFilterCache* bloom_filter_cache, |
809 | 152 | ParquetPruningStats* pruning_stats) { |
810 | 152 | if (bloom_filter_cache == nullptr) { |
811 | 0 | return ParquetRowGroupPruneReason::NONE; |
812 | 0 | } |
813 | 152 | const auto conjuncts_by_slot = collect_conjuncts_by_single_slot( |
814 | 152 | request.conjuncts, expr_zonemap::single_slot_bloom_filter_index); |
815 | 152 | for (const auto& [slot_index, conjuncts] : conjuncts_by_slot) { |
816 | 0 | const auto file_column_id = file_column_id_by_block_position(request, slot_index); |
817 | 0 | if (!file_column_id.has_value()) { |
818 | 0 | continue; |
819 | 0 | } |
820 | 0 | const auto* column_schema = resolve_local_leaf_schema(file_schema, *file_column_id); |
821 | 0 | if (column_schema == nullptr || column_schema->type == nullptr || |
822 | 0 | !bloom_filter_supported(*column_schema)) { |
823 | 0 | continue; |
824 | 0 | } |
825 | 0 | auto* bloom_filter = bloom_filter_cache->get(row_group_idx, column_schema->leaf_column_id, |
826 | 0 | pruning_stats); |
827 | 0 | if (bloom_filter == nullptr) { |
828 | 0 | continue; |
829 | 0 | } |
830 | 0 | if (ParquetStatisticsUtils::BloomFilterExcludes(*column_schema, slot_index, conjuncts, |
831 | 0 | *bloom_filter)) { |
832 | 0 | return ParquetRowGroupPruneReason::BLOOM_FILTER; |
833 | 0 | } |
834 | 0 | } |
835 | 152 | return ParquetRowGroupPruneReason::NONE; |
836 | 152 | } |
837 | | |
838 | | void init_bloom_filter_cache(::parquet::ParquetFileReader* file_reader, bool enable_bloom_filter, |
839 | 124 | RowGroupBloomFilterCache* bloom_filter_cache) { |
840 | 124 | DORIS_CHECK(bloom_filter_cache != nullptr); |
841 | 124 | if (!enable_bloom_filter || file_reader == nullptr) { |
842 | 18 | return; |
843 | 18 | } |
844 | 106 | try { |
845 | 106 | bloom_filter_cache->bloom_filter_reader = &file_reader->GetBloomFilterReader(); |
846 | 106 | } catch (const ::parquet::ParquetException&) { |
847 | 0 | bloom_filter_cache->bloom_filter_reader = nullptr; |
848 | 0 | } catch (const std::exception&) { |
849 | 0 | bloom_filter_cache->bloom_filter_reader = nullptr; |
850 | 0 | } |
851 | 106 | } |
852 | | |
853 | | bool check_statistics(const ::parquet::RowGroupMetaData& row_group, |
854 | | const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema, |
855 | | const format::FileScanRequest& request, ParquetPruningStats* pruning_stats, |
856 | 65 | const cctz::time_zone* timezone) { |
857 | 65 | const auto slot_indexes = collect_expr_zonemap_slot_indexes(request.conjuncts); |
858 | 65 | if (slot_indexes.empty()) { |
859 | 0 | return false; |
860 | 0 | } |
861 | | |
862 | 65 | ZoneMapEvalContext ctx; |
863 | 66 | for (const int slot_index : slot_indexes) { |
864 | 66 | const auto file_column_id = file_column_id_by_block_position(request, slot_index); |
865 | 66 | if (!file_column_id.has_value()) { |
866 | 0 | continue; |
867 | 0 | } |
868 | 66 | const auto* column_schema = resolve_local_leaf_schema(file_schema, *file_column_id); |
869 | 66 | if (column_schema == nullptr || column_schema->type == nullptr) { |
870 | 1 | continue; |
871 | 1 | } |
872 | | |
873 | 65 | std::shared_ptr<segment_v2::ZoneMap> zone_map; |
874 | 65 | DCHECK_LT(column_schema->leaf_column_id, row_group.num_columns()); |
875 | 65 | auto column_chunk = row_group.ColumnChunk(column_schema->leaf_column_id); |
876 | 65 | if (column_chunk != nullptr) { |
877 | 65 | zone_map = |
878 | 65 | make_zonemap_from_statistics(ParquetStatisticsUtils::TransformColumnStatistics( |
879 | 65 | *column_schema, column_chunk->statistics(), timezone)); |
880 | 65 | } |
881 | 65 | add_slot_zonemap(&ctx, slot_index, column_schema->type, std::move(zone_map)); |
882 | 65 | } |
883 | | |
884 | 65 | const auto result = VExprContext::evaluate_zonemap_filter(request.conjuncts, ctx); |
885 | 65 | accumulate_zonemap_stats(ctx, pruning_stats); |
886 | 65 | return result == ZoneMapFilterResult::kNoMatch; |
887 | 65 | } |
888 | | |
889 | | Status select_row_groups_by_metadata_impl( |
890 | | const ::parquet::FileMetaData& metadata, ::parquet::ParquetFileReader* file_reader, |
891 | | const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema, |
892 | | const format::FileScanRequest& request, const std::vector<int>* candidate_row_groups, |
893 | | std::vector<int>* selected_row_groups, bool enable_bloom_filter, |
894 | | ParquetPruningStats* pruning_stats, const cctz::time_zone* timezone, |
895 | 124 | const RuntimeState* runtime_state) { |
896 | 124 | int64_t row_group_filter_time_sink = 0; |
897 | 124 | SCOPED_RAW_TIMER(pruning_stats == nullptr ? &row_group_filter_time_sink |
898 | 124 | : &pruning_stats->row_group_filter_time); |
899 | 124 | if (selected_row_groups == nullptr) { |
900 | 0 | return Status::InvalidArgument("selected_row_groups is null"); |
901 | 0 | } |
902 | 124 | selected_row_groups->clear(); |
903 | | |
904 | 124 | const int num_row_groups = metadata.num_row_groups(); |
905 | 124 | if (pruning_stats != nullptr) { |
906 | 124 | pruning_stats->total_row_groups = num_row_groups; |
907 | 124 | } |
908 | 124 | const auto candidate_size = candidate_row_groups == nullptr |
909 | 124 | ? static_cast<size_t>(num_row_groups) |
910 | 124 | : candidate_row_groups->size(); |
911 | 124 | selected_row_groups->reserve(candidate_size); |
912 | 124 | RowGroupBloomFilterCache bloom_filter_cache; |
913 | 124 | init_bloom_filter_cache(file_reader, enable_bloom_filter, &bloom_filter_cache); |
914 | 322 | for (size_t candidate_idx = 0; candidate_idx < candidate_size; ++candidate_idx) { |
915 | 198 | const int row_group_idx = candidate_row_groups == nullptr |
916 | 198 | ? static_cast<int>(candidate_idx) |
917 | 198 | : (*candidate_row_groups)[candidate_idx]; |
918 | 198 | DORIS_CHECK(row_group_idx >= 0); |
919 | 198 | DORIS_CHECK(row_group_idx < num_row_groups); |
920 | 198 | auto row_group = metadata.RowGroup(row_group_idx); |
921 | 198 | if (row_group == nullptr) { |
922 | 0 | selected_row_groups->push_back(row_group_idx); |
923 | 0 | continue; |
924 | 0 | } |
925 | 198 | ParquetRowGroupPruneReason prune_reason = ParquetRowGroupPruneReason::NONE; |
926 | 198 | if (has_expr_zonemap_filter(request, runtime_state) && |
927 | 198 | check_statistics(*row_group, file_schema, request, pruning_stats, timezone)) { |
928 | 26 | prune_reason = ParquetRowGroupPruneReason::STATISTICS; |
929 | 26 | } |
930 | | |
931 | 198 | if (prune_reason == ParquetRowGroupPruneReason::NONE) { |
932 | 172 | prune_reason = dictionary_prune_reason(*row_group, file_reader, row_group_idx, |
933 | 172 | file_schema, request); |
934 | 172 | if (prune_reason == ParquetRowGroupPruneReason::NONE) { |
935 | 152 | prune_reason = bloom_filter_prune_reason(row_group_idx, file_schema, request, |
936 | 152 | &bloom_filter_cache, pruning_stats); |
937 | 152 | } |
938 | 172 | } |
939 | | |
940 | 198 | if (prune_reason != ParquetRowGroupPruneReason::NONE) { |
941 | 46 | if (pruning_stats != nullptr) { |
942 | 46 | pruning_stats->filtered_group_rows += row_group->num_rows(); |
943 | 46 | if (prune_reason == ParquetRowGroupPruneReason::STATISTICS) { |
944 | 26 | ++pruning_stats->filtered_row_groups_by_statistics; |
945 | 26 | } else if (prune_reason == ParquetRowGroupPruneReason::DICTIONARY) { |
946 | 20 | ++pruning_stats->filtered_row_groups_by_dictionary; |
947 | 20 | } else if (prune_reason == ParquetRowGroupPruneReason::BLOOM_FILTER) { |
948 | 0 | ++pruning_stats->filtered_row_groups_by_bloom_filter; |
949 | 0 | } |
950 | 46 | } |
951 | 46 | continue; |
952 | 46 | } |
953 | 152 | selected_row_groups->push_back(row_group_idx); |
954 | 152 | } |
955 | 124 | return Status::OK(); |
956 | 124 | } |
957 | | |
958 | | } // namespace |
959 | | |
960 | | Status select_row_groups_by_metadata( |
961 | | const ::parquet::FileMetaData& metadata, ::parquet::ParquetFileReader* file_reader, |
962 | | const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema, |
963 | | const format::FileScanRequest& request, const std::vector<int>* candidate_row_groups, |
964 | | std::vector<int>* selected_row_groups, bool enable_bloom_filter, |
965 | | ParquetPruningStats* pruning_stats, const cctz::time_zone* timezone, |
966 | 124 | const RuntimeState* runtime_state) { |
967 | 124 | return select_row_groups_by_metadata_impl( |
968 | 124 | metadata, file_reader, file_schema, request, candidate_row_groups, selected_row_groups, |
969 | 124 | enable_bloom_filter, pruning_stats, timezone, runtime_state); |
970 | 124 | } |
971 | | |
972 | | namespace { |
973 | | |
974 | | template <typename ParquetDType> |
975 | | bool set_page_decoded_min_max(const std::shared_ptr<::parquet::ColumnIndex>& column_index, |
976 | | const ParquetColumnSchema& column_schema, size_t page_idx, |
977 | | DecodedValueKind value_kind, ParquetColumnStatistics* page_statistics, |
978 | 128 | const cctz::time_zone* timezone) { |
979 | 128 | const auto typed_index = |
980 | 128 | std::static_pointer_cast<::parquet::TypedColumnIndex<ParquetDType>>(column_index); |
981 | 128 | if (page_idx >= typed_index->min_values().size() || |
982 | 128 | page_idx >= typed_index->max_values().size()) { |
983 | 0 | return false; |
984 | 0 | } |
985 | 128 | if (!set_decoded_field(column_schema, value_kind, typed_index->min_values()[page_idx], |
986 | 128 | &page_statistics->min_value, timezone) || |
987 | 128 | !set_decoded_field(column_schema, value_kind, typed_index->max_values()[page_idx], |
988 | 128 | &page_statistics->max_value, timezone)) { |
989 | 0 | return false; |
990 | 0 | } |
991 | 128 | page_statistics->has_min_max = true; |
992 | 128 | return true; |
993 | 128 | } Unexecuted instantiation: parquet_statistics.cpp:_ZN5doris6format7parquet12_GLOBAL__N_124set_page_decoded_min_maxIN7parquet12PhysicalTypeILNS4_4Type4typeE0EEEEEbRKSt10shared_ptrINS4_11ColumnIndexEERKNS1_19ParquetColumnSchemaEmNS_16DecodedValueKindEPNS1_23ParquetColumnStatisticsEPKN4cctz9time_zoneE parquet_statistics.cpp:_ZN5doris6format7parquet12_GLOBAL__N_124set_page_decoded_min_maxIN7parquet12PhysicalTypeILNS4_4Type4typeE1EEEEEbRKSt10shared_ptrINS4_11ColumnIndexEERKNS1_19ParquetColumnSchemaEmNS_16DecodedValueKindEPNS1_23ParquetColumnStatisticsEPKN4cctz9time_zoneE Line | Count | Source | 978 | 128 | const cctz::time_zone* timezone) { | 979 | 128 | const auto typed_index = | 980 | 128 | std::static_pointer_cast<::parquet::TypedColumnIndex<ParquetDType>>(column_index); | 981 | 128 | if (page_idx >= typed_index->min_values().size() || | 982 | 128 | page_idx >= typed_index->max_values().size()) { | 983 | 0 | return false; | 984 | 0 | } | 985 | 128 | if (!set_decoded_field(column_schema, value_kind, typed_index->min_values()[page_idx], | 986 | 128 | &page_statistics->min_value, timezone) || | 987 | 128 | !set_decoded_field(column_schema, value_kind, typed_index->max_values()[page_idx], | 988 | 128 | &page_statistics->max_value, timezone)) { | 989 | 0 | return false; | 990 | 0 | } | 991 | 128 | page_statistics->has_min_max = true; | 992 | 128 | return true; | 993 | 128 | } |
Unexecuted instantiation: parquet_statistics.cpp:_ZN5doris6format7parquet12_GLOBAL__N_124set_page_decoded_min_maxIN7parquet12PhysicalTypeILNS4_4Type4typeE2EEEEEbRKSt10shared_ptrINS4_11ColumnIndexEERKNS1_19ParquetColumnSchemaEmNS_16DecodedValueKindEPNS1_23ParquetColumnStatisticsEPKN4cctz9time_zoneE Unexecuted instantiation: parquet_statistics.cpp:_ZN5doris6format7parquet12_GLOBAL__N_124set_page_decoded_min_maxIN7parquet12PhysicalTypeILNS4_4Type4typeE4EEEEEbRKSt10shared_ptrINS4_11ColumnIndexEERKNS1_19ParquetColumnSchemaEmNS_16DecodedValueKindEPNS1_23ParquetColumnStatisticsEPKN4cctz9time_zoneE Unexecuted instantiation: parquet_statistics.cpp:_ZN5doris6format7parquet12_GLOBAL__N_124set_page_decoded_min_maxIN7parquet12PhysicalTypeILNS4_4Type4typeE5EEEEEbRKSt10shared_ptrINS4_11ColumnIndexEERKNS1_19ParquetColumnSchemaEmNS_16DecodedValueKindEPNS1_23ParquetColumnStatisticsEPKN4cctz9time_zoneE |
994 | | |
995 | | bool set_page_string_min_max(const std::shared_ptr<::parquet::ColumnIndex>& column_index, |
996 | | const ParquetColumnSchema& column_schema, size_t page_idx, |
997 | | ParquetColumnStatistics* page_statistics, |
998 | 0 | const cctz::time_zone* timezone) { |
999 | 0 | switch (column_schema.descriptor->physical_type()) { |
1000 | 0 | case ::parquet::Type::BYTE_ARRAY: { |
1001 | 0 | const auto typed_index = |
1002 | 0 | std::static_pointer_cast<::parquet::ByteArrayColumnIndex>(column_index); |
1003 | 0 | if (page_idx >= typed_index->min_values().size() || |
1004 | 0 | page_idx >= typed_index->max_values().size()) { |
1005 | 0 | return false; |
1006 | 0 | } |
1007 | 0 | const auto min = ::parquet::ByteArrayToString(typed_index->min_values()[page_idx]); |
1008 | 0 | const auto max = ::parquet::ByteArrayToString(typed_index->max_values()[page_idx]); |
1009 | 0 | if (!set_decoded_binary_field(column_schema, DecodedValueKind::BINARY, |
1010 | 0 | StringRef(min.data(), min.size()), |
1011 | 0 | &page_statistics->min_value, timezone) || |
1012 | 0 | !set_decoded_binary_field(column_schema, DecodedValueKind::BINARY, |
1013 | 0 | StringRef(max.data(), max.size()), |
1014 | 0 | &page_statistics->max_value, timezone)) { |
1015 | 0 | return false; |
1016 | 0 | } |
1017 | 0 | page_statistics->has_min_max = true; |
1018 | 0 | return true; |
1019 | 0 | } |
1020 | 0 | case ::parquet::Type::FIXED_LEN_BYTE_ARRAY: { |
1021 | 0 | const int type_length = column_schema.descriptor->type_length(); |
1022 | 0 | if (type_length <= 0) { |
1023 | 0 | return false; |
1024 | 0 | } |
1025 | 0 | const auto typed_index = std::static_pointer_cast<::parquet::FLBAColumnIndex>(column_index); |
1026 | 0 | if (page_idx >= typed_index->min_values().size() || |
1027 | 0 | page_idx >= typed_index->max_values().size()) { |
1028 | 0 | return false; |
1029 | 0 | } |
1030 | 0 | const std::string min( |
1031 | 0 | reinterpret_cast<const char*>(typed_index->min_values()[page_idx].ptr), |
1032 | 0 | type_length); |
1033 | 0 | const std::string max( |
1034 | 0 | reinterpret_cast<const char*>(typed_index->max_values()[page_idx].ptr), |
1035 | 0 | type_length); |
1036 | 0 | if (!set_decoded_binary_field(column_schema, DecodedValueKind::FIXED_BINARY, |
1037 | 0 | StringRef(min.data(), min.size()), |
1038 | 0 | &page_statistics->min_value, timezone) || |
1039 | 0 | !set_decoded_binary_field(column_schema, DecodedValueKind::FIXED_BINARY, |
1040 | 0 | StringRef(max.data(), max.size()), |
1041 | 0 | &page_statistics->max_value, timezone)) { |
1042 | 0 | return false; |
1043 | 0 | } |
1044 | 0 | page_statistics->has_min_max = true; |
1045 | 0 | return true; |
1046 | 0 | } |
1047 | 0 | default: |
1048 | 0 | return false; |
1049 | 0 | } |
1050 | 0 | } |
1051 | | |
1052 | | bool set_page_min_max(const std::shared_ptr<::parquet::ColumnIndex>& column_index, |
1053 | | const ParquetColumnSchema& column_schema, size_t page_idx, |
1054 | 128 | ParquetColumnStatistics* page_statistics, const cctz::time_zone* timezone) { |
1055 | 128 | DORIS_CHECK(column_schema.type != nullptr); |
1056 | 128 | switch (column_schema.descriptor->physical_type()) { |
1057 | 0 | case ::parquet::Type::BOOLEAN: |
1058 | 0 | return set_page_decoded_min_max<::parquet::BooleanType>(column_index, column_schema, |
1059 | 0 | page_idx, DecodedValueKind::BOOL, |
1060 | 0 | page_statistics, timezone); |
1061 | 128 | case ::parquet::Type::INT32: |
1062 | 128 | return set_page_decoded_min_max<::parquet::Int32Type>( |
1063 | 128 | column_index, column_schema, page_idx, |
1064 | 128 | decoded_value_kind(column_schema.type_descriptor), page_statistics, timezone); |
1065 | 0 | case ::parquet::Type::INT64: |
1066 | 0 | return set_page_decoded_min_max<::parquet::Int64Type>( |
1067 | 0 | column_index, column_schema, page_idx, |
1068 | 0 | decoded_value_kind(column_schema.type_descriptor), page_statistics, timezone); |
1069 | 0 | case ::parquet::Type::FLOAT: |
1070 | 0 | return set_page_decoded_min_max<::parquet::FloatType>(column_index, column_schema, page_idx, |
1071 | 0 | DecodedValueKind::FLOAT, |
1072 | 0 | page_statistics, timezone); |
1073 | 0 | case ::parquet::Type::DOUBLE: |
1074 | 0 | return set_page_decoded_min_max<::parquet::DoubleType>(column_index, column_schema, |
1075 | 0 | page_idx, DecodedValueKind::DOUBLE, |
1076 | 0 | page_statistics, timezone); |
1077 | 0 | case ::parquet::Type::BYTE_ARRAY: |
1078 | 0 | case ::parquet::Type::FIXED_LEN_BYTE_ARRAY: |
1079 | 0 | return set_page_string_min_max(column_index, column_schema, page_idx, page_statistics, |
1080 | 0 | timezone); |
1081 | 0 | default: |
1082 | 0 | return false; |
1083 | 128 | } |
1084 | 128 | } |
1085 | | |
1086 | | bool build_page_statistics(const std::shared_ptr<::parquet::ColumnIndex>& column_index, |
1087 | | const ParquetColumnSchema& column_schema, size_t page_idx, |
1088 | | ParquetColumnStatistics* page_statistics, |
1089 | 128 | const cctz::time_zone* timezone) { |
1090 | 128 | DORIS_CHECK(page_statistics != nullptr); |
1091 | 128 | *page_statistics = ParquetColumnStatistics {}; |
1092 | | |
1093 | 128 | const auto& null_pages = column_index->null_pages(); |
1094 | 128 | if (!column_index->has_null_counts() || page_idx >= null_pages.size() || |
1095 | 128 | page_idx >= column_index->null_counts().size()) { |
1096 | 0 | return false; |
1097 | 0 | } |
1098 | | |
1099 | 128 | page_statistics->has_null_count = true; |
1100 | 128 | page_statistics->has_null = column_index->null_counts()[page_idx] > 0; |
1101 | 128 | page_statistics->has_not_null = !null_pages[page_idx]; |
1102 | 128 | if (!page_statistics->has_not_null) { |
1103 | 0 | return true; |
1104 | 0 | } |
1105 | 128 | return set_page_min_max(column_index, column_schema, page_idx, page_statistics, timezone); |
1106 | 128 | } |
1107 | | |
1108 | | std::vector<RowRange> intersect_ranges(const std::vector<RowRange>& left, |
1109 | 8 | const std::vector<RowRange>& right) { |
1110 | 8 | std::vector<RowRange> result; |
1111 | 8 | size_t left_idx = 0; |
1112 | 8 | size_t right_idx = 0; |
1113 | 15 | while (left_idx < left.size() && right_idx < right.size()) { |
1114 | 7 | const int64_t left_start = left[left_idx].start; |
1115 | 7 | const int64_t left_end = left_start + left[left_idx].length; |
1116 | 7 | const int64_t right_start = right[right_idx].start; |
1117 | 7 | const int64_t right_end = right_start + right[right_idx].length; |
1118 | 7 | const int64_t start = std::max(left_start, right_start); |
1119 | 7 | const int64_t end = std::min(left_end, right_end); |
1120 | 7 | if (start < end) { |
1121 | 7 | result.push_back(RowRange {start, end - start}); |
1122 | 7 | } |
1123 | 7 | if (left_end < right_end) { |
1124 | 0 | ++left_idx; |
1125 | 7 | } else { |
1126 | 7 | ++right_idx; |
1127 | 7 | } |
1128 | 7 | } |
1129 | 8 | return result; |
1130 | 8 | } |
1131 | | |
1132 | 7 | int64_t count_range_rows(const std::vector<RowRange>& ranges) { |
1133 | 7 | int64_t rows = 0; |
1134 | 7 | for (const auto& range : ranges) { |
1135 | 7 | rows += range.length; |
1136 | 7 | } |
1137 | 7 | return rows; |
1138 | 7 | } |
1139 | | |
1140 | | RowRange page_row_range(const ::parquet::OffsetIndex& offset_index, size_t page_idx, |
1141 | 180 | int64_t row_group_rows) { |
1142 | 180 | const auto& page_locations = offset_index.page_locations(); |
1143 | 180 | const int64_t start = page_locations[page_idx].first_row_index; |
1144 | 180 | const int64_t end = page_idx + 1 == page_locations.size() |
1145 | 180 | ? row_group_rows |
1146 | 180 | : page_locations[page_idx + 1].first_row_index; |
1147 | 180 | DORIS_CHECK(start >= 0); |
1148 | 180 | DORIS_CHECK(end >= start); |
1149 | 180 | DORIS_CHECK(end <= row_group_rows); |
1150 | 180 | return RowRange {start, end - start}; |
1151 | 180 | } |
1152 | | |
1153 | 120 | void append_row_range(const RowRange& range, std::vector<RowRange>* ranges) { |
1154 | 120 | if (range.length == 0) { |
1155 | 0 | return; |
1156 | 0 | } |
1157 | 120 | if (!ranges->empty()) { |
1158 | 106 | auto& previous = ranges->back(); |
1159 | 106 | if (previous.start + previous.length == range.start) { |
1160 | 104 | previous.length += range.length; |
1161 | 104 | return; |
1162 | 104 | } |
1163 | 106 | } |
1164 | 16 | ranges->push_back(range); |
1165 | 16 | } |
1166 | | |
1167 | | std::optional< |
1168 | | std::pair<std::shared_ptr<::parquet::ColumnIndex>, std::shared_ptr<::parquet::OffsetIndex>>> |
1169 | | load_page_indexes_for_slot(const std::shared_ptr<::parquet::RowGroupPageIndexReader>& row_group, |
1170 | | const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema, |
1171 | | const format::FileScanRequest& request, int slot_index, |
1172 | 9 | const ParquetColumnSchema** column_schema) { |
1173 | 9 | DORIS_CHECK(column_schema != nullptr); |
1174 | 9 | *column_schema = nullptr; |
1175 | 9 | const auto file_column_id = file_column_id_by_block_position(request, slot_index); |
1176 | 9 | if (!file_column_id.has_value()) { |
1177 | 0 | return std::nullopt; |
1178 | 0 | } |
1179 | 9 | *column_schema = resolve_local_leaf_schema(file_schema, *file_column_id); |
1180 | 9 | if (*column_schema == nullptr || (*column_schema)->descriptor == nullptr) { |
1181 | 1 | return std::nullopt; |
1182 | 1 | } |
1183 | | |
1184 | 8 | try { |
1185 | 8 | auto column_index = row_group->GetColumnIndex((*column_schema)->leaf_column_id); |
1186 | 8 | auto offset_index = row_group->GetOffsetIndex((*column_schema)->leaf_column_id); |
1187 | 8 | if (column_index == nullptr || offset_index == nullptr || |
1188 | 8 | column_index->null_pages().size() != offset_index->page_locations().size()) { |
1189 | 0 | return std::nullopt; |
1190 | 0 | } |
1191 | 8 | return std::make_pair(std::move(column_index), std::move(offset_index)); |
1192 | 8 | } catch (const ::parquet::ParquetException&) { |
1193 | 0 | return std::nullopt; |
1194 | 0 | } catch (const std::exception&) { |
1195 | 0 | return std::nullopt; |
1196 | 0 | } |
1197 | 8 | } |
1198 | | |
1199 | | bool select_ranges_for_expr_zonemap( |
1200 | | const std::shared_ptr<::parquet::RowGroupPageIndexReader>& row_group, |
1201 | | const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema, |
1202 | | const format::FileScanRequest& request, int slot_index, const VExprContextSPtrs& conjuncts, |
1203 | | int64_t row_group_rows, std::vector<RowRange>* ranges, ParquetPruningStats* pruning_stats, |
1204 | 9 | const cctz::time_zone* timezone) { |
1205 | 9 | DORIS_CHECK(ranges != nullptr); |
1206 | 9 | if (conjuncts.empty()) { |
1207 | 0 | return false; |
1208 | 0 | } |
1209 | 9 | const ParquetColumnSchema* column_schema = nullptr; |
1210 | 9 | const auto page_indexes = |
1211 | 9 | load_page_indexes_for_slot(row_group, file_schema, request, slot_index, &column_schema); |
1212 | 9 | if (!page_indexes.has_value()) { |
1213 | 1 | return false; |
1214 | 1 | } |
1215 | 8 | const auto& [column_index, offset_index] = *page_indexes; |
1216 | | |
1217 | 8 | ranges->clear(); |
1218 | 8 | ZoneMapEvalStats page_stats; |
1219 | 8 | const auto page_count = offset_index->page_locations().size(); |
1220 | 136 | for (size_t page_idx = 0; page_idx < page_count; ++page_idx) { |
1221 | 128 | ParquetColumnStatistics page_statistics; |
1222 | 128 | if (!build_page_statistics(column_index, *column_schema, page_idx, &page_statistics, |
1223 | 128 | timezone)) { |
1224 | 0 | ranges->clear(); |
1225 | 0 | return false; |
1226 | 0 | } |
1227 | | |
1228 | 128 | ZoneMapEvalContext ctx; |
1229 | 128 | add_slot_zonemap(&ctx, slot_index, column_schema->type, |
1230 | 128 | make_zonemap_from_statistics(page_statistics)); |
1231 | 128 | const auto result = VExprContext::evaluate_zonemap_filter(conjuncts, ctx); |
1232 | 128 | page_stats.merge_page_eval_stats(ctx.stats); |
1233 | 128 | if (result == ZoneMapFilterResult::kNoMatch) { |
1234 | 60 | continue; |
1235 | 60 | } |
1236 | 68 | append_row_range(page_row_range(*offset_index, page_idx, row_group_rows), ranges); |
1237 | 68 | } |
1238 | 8 | if (pruning_stats != nullptr) { |
1239 | 8 | pruning_stats->expr_zonemap_unusable_evals += page_stats.unusable_zonemap_eval_count; |
1240 | 8 | pruning_stats->in_zonemap_point_check_count += page_stats.in_zonemap_point_check_count; |
1241 | 8 | pruning_stats->in_zonemap_range_only_count += page_stats.in_zonemap_range_only_count; |
1242 | 8 | } |
1243 | 8 | return true; |
1244 | 8 | } |
1245 | | |
1246 | 112 | bool ranges_intersect(const std::vector<RowRange>& ranges, const RowRange& range) { |
1247 | 112 | const int64_t range_end = range.start + range.length; |
1248 | 112 | for (const auto& selected_range : ranges) { |
1249 | 112 | const int64_t selected_end = selected_range.start + selected_range.length; |
1250 | 112 | if (selected_end <= range.start) { |
1251 | 8 | continue; |
1252 | 8 | } |
1253 | 104 | if (selected_range.start >= range_end) { |
1254 | 44 | return false; |
1255 | 44 | } |
1256 | 60 | return true; |
1257 | 104 | } |
1258 | 8 | return false; |
1259 | 112 | } |
1260 | | |
1261 | | void collect_leaf_schemas(const ParquetColumnSchema& column_schema, |
1262 | | const format::LocalColumnIndex* projection, |
1263 | 7 | std::vector<const ParquetColumnSchema*>* leaf_schemas) { |
1264 | 7 | if (column_schema.kind == ParquetColumnSchemaKind::PRIMITIVE) { |
1265 | 7 | leaf_schemas->push_back(&column_schema); |
1266 | 7 | return; |
1267 | 7 | } |
1268 | 0 | for (const auto& child_schema : column_schema.children) { |
1269 | 0 | if (!format::is_child_projected(projection, child_schema->local_id)) { |
1270 | 0 | continue; |
1271 | 0 | } |
1272 | 0 | const auto* child_projection = |
1273 | 0 | format::find_child_projection(projection, child_schema->local_id); |
1274 | 0 | collect_leaf_schemas(*child_schema, child_projection, leaf_schemas); |
1275 | 0 | } |
1276 | 0 | } |
1277 | | |
1278 | | void collect_request_leaf_schemas( |
1279 | | const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema, |
1280 | | const format::FileScanRequest& request, |
1281 | 7 | std::vector<const ParquetColumnSchema*>* leaf_schemas) { |
1282 | 7 | std::set<int> seen_leaf_ids; |
1283 | 7 | auto collect_projection = [&](const format::LocalColumnIndex& projection) { |
1284 | 7 | const int32_t local_id = projection.local_id(); |
1285 | 7 | if (local_id < 0 || local_id >= static_cast<int32_t>(file_schema.size())) { |
1286 | 0 | return; |
1287 | 0 | } |
1288 | 7 | std::vector<const ParquetColumnSchema*> projection_leaf_schemas; |
1289 | 7 | collect_leaf_schemas(*file_schema[local_id], &projection, &projection_leaf_schemas); |
1290 | 7 | for (const auto* leaf_schema : projection_leaf_schemas) { |
1291 | 7 | DORIS_CHECK(leaf_schema != nullptr); |
1292 | 7 | if (seen_leaf_ids.insert(leaf_schema->leaf_column_id).second) { |
1293 | 7 | leaf_schemas->push_back(leaf_schema); |
1294 | 7 | } |
1295 | 7 | } |
1296 | 7 | }; |
1297 | 7 | for (const auto& projection : request.predicate_columns) { |
1298 | 6 | collect_projection(projection); |
1299 | 6 | } |
1300 | 7 | for (const auto& projection : request.non_predicate_columns) { |
1301 | 1 | collect_projection(projection); |
1302 | 1 | } |
1303 | 7 | } |
1304 | | |
1305 | | bool build_page_skip_plan_for_leaf( |
1306 | | const std::shared_ptr<::parquet::RowGroupPageIndexReader>& row_group, |
1307 | | const ParquetColumnSchema& column_schema, const std::vector<RowRange>& selected_ranges, |
1308 | 7 | int64_t row_group_rows, ParquetPageSkipPlan* page_skip_plan) { |
1309 | 7 | DORIS_CHECK(page_skip_plan != nullptr); |
1310 | 7 | *page_skip_plan = ParquetPageSkipPlan {}; |
1311 | 7 | if (column_schema.kind != ParquetColumnSchemaKind::PRIMITIVE || |
1312 | 7 | column_schema.descriptor == nullptr || column_schema.leaf_column_id < 0 || |
1313 | 7 | column_schema.descriptor->max_repetition_level() != 0) { |
1314 | 0 | return false; |
1315 | 0 | } |
1316 | | |
1317 | 7 | std::shared_ptr<::parquet::OffsetIndex> offset_index; |
1318 | 7 | try { |
1319 | 7 | offset_index = row_group->GetOffsetIndex(column_schema.leaf_column_id); |
1320 | 7 | } catch (const ::parquet::ParquetException&) { |
1321 | 0 | return false; |
1322 | 0 | } catch (const std::exception&) { |
1323 | 0 | return false; |
1324 | 0 | } |
1325 | 7 | if (offset_index == nullptr) { |
1326 | 0 | return false; |
1327 | 0 | } |
1328 | | |
1329 | 7 | const auto page_count = offset_index->page_locations().size(); |
1330 | 7 | page_skip_plan->leaf_column_id = column_schema.leaf_column_id; |
1331 | 7 | page_skip_plan->skipped_pages.resize(page_count); |
1332 | 7 | page_skip_plan->skipped_page_compressed_sizes.resize(page_count); |
1333 | 7 | const auto& page_locations = offset_index->page_locations(); |
1334 | 119 | for (size_t page_idx = 0; page_idx < page_count; ++page_idx) { |
1335 | 112 | const RowRange row_range = page_row_range(*offset_index, page_idx, row_group_rows); |
1336 | 112 | if (row_range.length == 0 || ranges_intersect(selected_ranges, row_range)) { |
1337 | 60 | continue; |
1338 | 60 | } |
1339 | 52 | page_skip_plan->skipped_pages[page_idx] = 1; |
1340 | 52 | page_skip_plan->skipped_page_compressed_sizes[page_idx] = |
1341 | 52 | page_locations[page_idx].compressed_page_size; |
1342 | 52 | append_row_range(row_range, &page_skip_plan->skipped_ranges); |
1343 | 52 | } |
1344 | 7 | if (page_skip_plan->empty()) { |
1345 | 0 | *page_skip_plan = ParquetPageSkipPlan {}; |
1346 | 0 | return false; |
1347 | 0 | } |
1348 | 7 | return true; |
1349 | 7 | } |
1350 | | |
1351 | | void build_page_skip_plans(const std::shared_ptr<::parquet::RowGroupPageIndexReader>& row_group, |
1352 | | const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema, |
1353 | | const format::FileScanRequest& request, |
1354 | | const std::vector<RowRange>& selected_ranges, int64_t row_group_rows, |
1355 | 7 | std::map<int, ParquetPageSkipPlan>* page_skip_plans) { |
1356 | 7 | DORIS_CHECK(page_skip_plans != nullptr); |
1357 | 7 | page_skip_plans->clear(); |
1358 | 7 | std::vector<const ParquetColumnSchema*> leaf_schemas; |
1359 | 7 | collect_request_leaf_schemas(file_schema, request, &leaf_schemas); |
1360 | 7 | for (const auto* leaf_schema : leaf_schemas) { |
1361 | 7 | DORIS_CHECK(leaf_schema != nullptr); |
1362 | 7 | ParquetPageSkipPlan page_skip_plan; |
1363 | 7 | if (build_page_skip_plan_for_leaf(row_group, *leaf_schema, selected_ranges, row_group_rows, |
1364 | 7 | &page_skip_plan)) { |
1365 | 7 | page_skip_plans->emplace(page_skip_plan.leaf_column_id, std::move(page_skip_plan)); |
1366 | 7 | } |
1367 | 7 | } |
1368 | 7 | } |
1369 | | |
1370 | | } // namespace |
1371 | | |
1372 | | Status select_row_group_ranges_by_page_index( |
1373 | | ::parquet::ParquetFileReader* file_reader, |
1374 | | const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema, |
1375 | | const format::FileScanRequest& request, int row_group_idx, int64_t row_group_rows, |
1376 | | std::vector<RowRange>* selected_ranges, std::map<int, ParquetPageSkipPlan>* page_skip_plans, |
1377 | | ParquetPruningStats* pruning_stats, const cctz::time_zone* timezone, |
1378 | 145 | const RuntimeState* runtime_state) { |
1379 | 145 | int64_t page_index_filter_time_sink = 0; |
1380 | 145 | SCOPED_RAW_TIMER(pruning_stats == nullptr ? &page_index_filter_time_sink |
1381 | 145 | : &pruning_stats->page_index_filter_time); |
1382 | 145 | DORIS_CHECK(selected_ranges != nullptr); |
1383 | 145 | selected_ranges->clear(); |
1384 | 145 | if (page_skip_plans != nullptr) { |
1385 | 145 | page_skip_plans->clear(); |
1386 | 145 | } |
1387 | 145 | if (row_group_rows <= 0) { |
1388 | 0 | return Status::OK(); |
1389 | 0 | } |
1390 | 145 | selected_ranges->push_back(RowRange {0, row_group_rows}); |
1391 | 145 | if (!config::enable_parquet_page_index || !has_expr_zonemap_filter(request, runtime_state) || |
1392 | 145 | file_reader == nullptr) { |
1393 | 109 | return Status::OK(); |
1394 | 109 | } |
1395 | | |
1396 | 36 | std::shared_ptr<::parquet::PageIndexReader> page_index_reader; |
1397 | 36 | std::shared_ptr<::parquet::RowGroupPageIndexReader> row_group_index_reader; |
1398 | 36 | try { |
1399 | 36 | if (pruning_stats != nullptr) { |
1400 | 36 | ++pruning_stats->page_index_read_calls; |
1401 | 36 | } |
1402 | 36 | { |
1403 | 36 | int64_t read_page_index_time_sink = 0; |
1404 | 36 | SCOPED_RAW_TIMER(pruning_stats == nullptr ? &read_page_index_time_sink |
1405 | 36 | : &pruning_stats->read_page_index_time); |
1406 | 36 | page_index_reader = file_reader->GetPageIndexReader(); |
1407 | 36 | if (page_index_reader == nullptr) { |
1408 | 0 | return Status::OK(); |
1409 | 0 | } |
1410 | 36 | row_group_index_reader = page_index_reader->RowGroup(row_group_idx); |
1411 | 36 | } |
1412 | 36 | } catch (const ::parquet::ParquetException&) { |
1413 | 0 | return Status::OK(); |
1414 | 0 | } catch (const std::exception&) { |
1415 | 0 | return Status::OK(); |
1416 | 0 | } |
1417 | 36 | if (row_group_index_reader == nullptr) { |
1418 | 28 | return Status::OK(); |
1419 | 28 | } |
1420 | | |
1421 | 8 | std::map<int, VExprContextSPtrs> conjuncts_by_slot; |
1422 | 10 | for (const auto& conjunct : request.conjuncts) { |
1423 | 10 | const auto slot_index = expr_zonemap::single_slot_zonemap_index(conjunct); |
1424 | 10 | if (slot_index >= 0) { |
1425 | 10 | conjuncts_by_slot[slot_index].push_back(conjunct); |
1426 | 10 | } |
1427 | 10 | } |
1428 | | |
1429 | 9 | for (const auto& [slot_index, conjuncts] : conjuncts_by_slot) { |
1430 | 9 | std::vector<RowRange> filter_ranges; |
1431 | 9 | if (!select_ranges_for_expr_zonemap(row_group_index_reader, file_schema, request, |
1432 | 9 | slot_index, conjuncts, row_group_rows, &filter_ranges, |
1433 | 9 | pruning_stats, timezone)) { |
1434 | 1 | continue; |
1435 | 1 | } |
1436 | 8 | *selected_ranges = intersect_ranges(*selected_ranges, filter_ranges); |
1437 | 8 | if (selected_ranges->empty()) { |
1438 | 1 | if (page_skip_plans != nullptr) { |
1439 | 1 | page_skip_plans->clear(); |
1440 | 1 | } |
1441 | 1 | if (pruning_stats != nullptr) { |
1442 | 1 | pruning_stats->filtered_page_rows += row_group_rows; |
1443 | 1 | ++pruning_stats->filtered_row_groups_by_page_index; |
1444 | 1 | } |
1445 | 1 | return Status::OK(); |
1446 | 1 | } |
1447 | 8 | } |
1448 | 7 | if (page_skip_plans != nullptr) { |
1449 | 7 | build_page_skip_plans(row_group_index_reader, file_schema, request, *selected_ranges, |
1450 | 7 | row_group_rows, page_skip_plans); |
1451 | 7 | } |
1452 | 7 | if (pruning_stats != nullptr) { |
1453 | 7 | const int64_t selected_rows = count_range_rows(*selected_ranges); |
1454 | 7 | DORIS_CHECK(selected_rows <= row_group_rows); |
1455 | 7 | pruning_stats->filtered_page_rows += row_group_rows - selected_rows; |
1456 | 7 | } |
1457 | 7 | return Status::OK(); |
1458 | 8 | } |
1459 | | |
1460 | | } // namespace doris::format::parquet |