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 <algorithm> |
19 | | #include <cmath> |
20 | | #include <cstddef> |
21 | | #include <cstring> |
22 | | #include <exception> |
23 | | #include <limits> |
24 | | #include <map> |
25 | | #include <memory> |
26 | | #include <optional> |
27 | | #include <set> |
28 | | #include <string> |
29 | | #include <type_traits> |
30 | | #include <utility> |
31 | | #include <vector> |
32 | | |
33 | | #include "common/cast_set.h" |
34 | | #include "common/config.h" |
35 | | #include "core/data_type/data_type.h" |
36 | | #include "core/data_type/data_type_nullable.h" |
37 | | #include "core/data_type_serde/data_type_serde.h" |
38 | | #include "core/field.h" |
39 | | #include "exprs/expr_zonemap_filter.h" |
40 | | #include "exprs/vexpr_context.h" |
41 | | #include "format_v2/parquet/parquet_column_schema.h" |
42 | | #include "format_v2/parquet/parquet_file_context.h" |
43 | | #include "format_v2/parquet/reader/native/block_split_bloom_filter.h" |
44 | | #include "format_v2/parquet/reader/native_column_reader.h" |
45 | | #include "format_v2/timestamp_statistics.h" |
46 | | #include "runtime/runtime_profile.h" |
47 | | #include "storage/index/bloom_filter/bloom_filter.h" |
48 | | #include "storage/index/zone_map/zone_map_index.h" |
49 | | #include "storage/index/zone_map/zonemap_eval_context.h" |
50 | | #include "util/thrift_util.h" |
51 | | #include "util/unaligned.h" |
52 | | |
53 | | namespace doris::format::parquet { |
54 | | |
55 | | namespace detail { |
56 | | |
57 | | Status validate_native_bloom_filter_layout(int64_t offset, uint32_t header_size, |
58 | | int64_t payload_size, int64_t declared_length, |
59 | 1 | size_t file_size) { |
60 | 1 | if (offset < 0 || header_size == 0 || payload_size < segment_v2::BloomFilter::MINIMUM_BYTES || |
61 | 1 | payload_size > segment_v2::BloomFilter::MAXIMUM_BYTES || payload_size % 32 != 0) { |
62 | 0 | return Status::Corruption( |
63 | 0 | "Invalid Parquet Bloom filter layout: offset {}, header {}, payload {}", offset, |
64 | 0 | header_size, payload_size); |
65 | 0 | } |
66 | 1 | const uint64_t unsigned_offset = static_cast<uint64_t>(offset); |
67 | 1 | const uint64_t total_size = static_cast<uint64_t>(header_size) + payload_size; |
68 | 1 | if (unsigned_offset > file_size || total_size > file_size - unsigned_offset) { |
69 | 0 | return Status::Corruption("Parquet Bloom filter range exceeds file size {}", file_size); |
70 | 0 | } |
71 | 1 | if (declared_length >= 0) { |
72 | 1 | const uint64_t unsigned_declared_length = static_cast<uint64_t>(declared_length); |
73 | 1 | if (unsigned_declared_length < total_size || |
74 | 1 | unsigned_declared_length > file_size - unsigned_offset) { |
75 | 0 | return Status::Corruption( |
76 | 0 | "Parquet Bloom filter requires {} bytes, metadata declares {}, file has {}", |
77 | 0 | total_size, declared_length, file_size - unsigned_offset); |
78 | 0 | } |
79 | 1 | } |
80 | 1 | return Status::OK(); |
81 | 1 | } |
82 | | |
83 | 96 | bool has_supported_type_defined_order(const tparquet::FileMetaData& metadata, int leaf_column_id) { |
84 | 96 | return leaf_column_id >= 0 && metadata.__isset.column_orders && |
85 | 96 | leaf_column_id < static_cast<int>(metadata.column_orders.size()) && |
86 | 96 | metadata.column_orders[leaf_column_id].__isset.TYPE_ORDER; |
87 | 96 | } |
88 | | |
89 | | tparquet::Statistics sanitize_native_footer_statistics(const ParquetTypeDescriptor& type_descriptor, |
90 | | const tparquet::Statistics& statistics, |
91 | 94 | bool has_type_defined_order) { |
92 | 94 | auto sanitized = statistics; |
93 | 94 | if (!has_type_defined_order || !sanitized.__isset.min_value || !sanitized.__isset.max_value) { |
94 | 6 | sanitized.__isset.min_value = false; |
95 | 6 | sanitized.__isset.max_value = false; |
96 | 6 | sanitized.min_value.clear(); |
97 | 6 | sanitized.max_value.clear(); |
98 | 6 | } |
99 | 94 | const bool binary = type_descriptor.physical_type == tparquet::Type::BYTE_ARRAY || |
100 | 94 | type_descriptor.physical_type == tparquet::Type::FIXED_LEN_BYTE_ARRAY; |
101 | 94 | if (!sanitized.__isset.min || !sanitized.__isset.max || |
102 | 94 | (binary && sanitized.min != sanitized.max)) { |
103 | 8 | sanitized.__isset.min = false; |
104 | 8 | sanitized.__isset.max = false; |
105 | 8 | sanitized.min.clear(); |
106 | 8 | sanitized.max.clear(); |
107 | 8 | } |
108 | 94 | return sanitized; |
109 | 94 | } |
110 | | |
111 | | bool can_use_native_footer_min_max(const ParquetTypeDescriptor& type_descriptor, |
112 | | const tparquet::Statistics& statistics, |
113 | 25 | bool has_type_defined_order) { |
114 | | // Inexact bounds remain useful for pruning, but returning them as aggregate values changes the |
115 | | // query result. Missing exactness fields are legacy-compatible; only an explicit false rejects. |
116 | 25 | if ((statistics.__isset.is_min_value_exact && !statistics.is_min_value_exact) || |
117 | 25 | (statistics.__isset.is_max_value_exact && !statistics.is_max_value_exact)) { |
118 | 2 | return false; |
119 | 2 | } |
120 | 23 | const auto sanitized = |
121 | 23 | sanitize_native_footer_statistics(type_descriptor, statistics, has_type_defined_order); |
122 | 23 | return (sanitized.__isset.min_value && sanitized.__isset.max_value) || |
123 | 23 | (sanitized.__isset.min && sanitized.__isset.max); |
124 | 25 | } |
125 | | |
126 | | } // namespace detail |
127 | | |
128 | | namespace { |
129 | | |
130 | | bool build_native_page_statistics(const tparquet::ColumnIndex& column_index, |
131 | | const ParquetColumnSchema& column_schema, size_t page_idx, |
132 | | int64_t page_rows, ParquetColumnStatistics* page_statistics, |
133 | | const cctz::time_zone* timezone); |
134 | | |
135 | | enum class ParquetRowGroupPruneReason { |
136 | | NONE, // cannot prune; must read |
137 | | STATISTICS, // excluded by ZoneMap statistics |
138 | | DICTIONARY, // excluded by dictionary |
139 | | BLOOM_FILTER, // excluded by bloom filter |
140 | | }; |
141 | | |
142 | | Status read_native_bloom_filter(const tparquet::ColumnMetaData& metadata, |
143 | | const io::FileReaderSPtr& file, io::IOContext* io_ctx, |
144 | 1 | std::unique_ptr<native::BlockSplitBloomFilter>* result) { |
145 | 1 | if (result == nullptr || file == nullptr || !metadata.__isset.bloom_filter_offset) { |
146 | 0 | return Status::NotSupported("Parquet Bloom filter is unavailable"); |
147 | 0 | } |
148 | 1 | constexpr size_t MAX_BLOOM_HEADER_BYTES = 64; |
149 | 1 | if (metadata.bloom_filter_offset < 0 || |
150 | 1 | (metadata.__isset.bloom_filter_length && metadata.bloom_filter_length <= 0)) { |
151 | 0 | return Status::Corruption("Invalid Parquet Bloom filter offset or declared length"); |
152 | 0 | } |
153 | 1 | const uint64_t bloom_offset = static_cast<uint64_t>(metadata.bloom_filter_offset); |
154 | 1 | if (bloom_offset >= file->size()) { |
155 | 0 | return Status::Corruption("Parquet Bloom filter offset exceeds file size {}", file->size()); |
156 | 0 | } |
157 | 1 | const size_t available = file->size() - bloom_offset; |
158 | 1 | const size_t declared_available = |
159 | 1 | metadata.__isset.bloom_filter_length |
160 | 1 | ? std::min<size_t>(metadata.bloom_filter_length, available) |
161 | 1 | : available; |
162 | 1 | const size_t header_read_size = std::min(declared_available, MAX_BLOOM_HEADER_BYTES); |
163 | 1 | std::vector<uint8_t> header_buffer(header_read_size); |
164 | 1 | size_t bytes_read = 0; |
165 | 1 | RETURN_IF_ERROR(file->read_at(metadata.bloom_filter_offset, |
166 | 1 | Slice(header_buffer.data(), header_buffer.size()), &bytes_read, |
167 | 1 | io_ctx)); |
168 | 1 | tparquet::BloomFilterHeader header; |
169 | 1 | uint32_t header_size = cast_set<uint32_t>(bytes_read); |
170 | 1 | RETURN_IF_ERROR(deserialize_thrift_msg(header_buffer.data(), &header_size, true, &header)); |
171 | 1 | if (!header.algorithm.__isset.BLOCK || !header.compression.__isset.UNCOMPRESSED || |
172 | 1 | !header.hash.__isset.XXHASH || header.numBytes <= 0) { |
173 | 0 | return Status::NotSupported("Unsupported Parquet Bloom filter encoding"); |
174 | 0 | } |
175 | | |
176 | | // Validate the complete split-block layout before allocating or adding footer-controlled |
177 | | // offsets; BloomFilter::init() otherwise receives a truncated or oversized backing buffer. |
178 | 1 | RETURN_IF_ERROR(detail::validate_native_bloom_filter_layout( |
179 | 1 | metadata.bloom_filter_offset, header_size, header.numBytes, |
180 | 1 | metadata.__isset.bloom_filter_length ? metadata.bloom_filter_length : -1, |
181 | 1 | file->size())); |
182 | | |
183 | 1 | std::vector<uint8_t> data(cast_set<size_t>(header.numBytes)); |
184 | 1 | RETURN_IF_ERROR(file->read_at(static_cast<size_t>(metadata.bloom_filter_offset) + header_size, |
185 | 1 | Slice(data.data(), data.size()), &bytes_read, io_ctx)); |
186 | 1 | if (bytes_read != data.size()) { |
187 | 0 | return Status::Corruption("Truncated Parquet Bloom filter payload"); |
188 | 0 | } |
189 | 1 | auto bloom_filter = std::make_unique<native::BlockSplitBloomFilter>(); |
190 | 1 | RETURN_IF_ERROR(bloom_filter->init(reinterpret_cast<const char*>(data.data()), data.size(), |
191 | 1 | segment_v2::HashStrategyPB::XX_HASH_64)); |
192 | 1 | *result = std::move(bloom_filter); |
193 | 1 | return Status::OK(); |
194 | 1 | } |
195 | | |
196 | 4 | bool bloom_logical_type_supported(const ParquetColumnSchema& column_schema) { |
197 | 4 | if (column_schema.type == nullptr) { |
198 | 0 | return false; |
199 | 0 | } |
200 | 4 | switch (remove_nullable(column_schema.type)->get_primitive_type()) { |
201 | 0 | case TYPE_BOOLEAN: |
202 | 0 | case TYPE_INT: |
203 | 4 | case TYPE_BIGINT: |
204 | 4 | case TYPE_FLOAT: |
205 | 4 | case TYPE_DOUBLE: |
206 | 4 | case TYPE_STRING: |
207 | 4 | return true; |
208 | 0 | default: |
209 | 0 | return false; |
210 | 4 | } |
211 | 4 | } |
212 | | |
213 | 244 | DecodedTimeUnit decoded_time_unit(ParquetTimeUnit time_unit) { |
214 | 244 | switch (time_unit) { |
215 | 0 | case ParquetTimeUnit::MILLIS: |
216 | 0 | return DecodedTimeUnit::MILLIS; |
217 | 0 | case ParquetTimeUnit::MICROS: |
218 | 0 | return DecodedTimeUnit::MICROS; |
219 | 0 | case ParquetTimeUnit::NANOS: |
220 | 0 | return DecodedTimeUnit::NANOS; |
221 | 244 | default: |
222 | 244 | return DecodedTimeUnit::UNKNOWN; |
223 | 244 | } |
224 | 244 | } |
225 | | |
226 | | Status read_decoded_field(const ParquetColumnSchema& column_schema, DecodedColumnView view, |
227 | 244 | Field* field, const cctz::time_zone* timezone) { |
228 | 244 | DORIS_CHECK(column_schema.type != nullptr); |
229 | 244 | DORIS_CHECK(field != nullptr); |
230 | 244 | constexpr uint8_t not_null = 0; |
231 | 244 | view.row_count = 1; |
232 | 244 | view.null_map = ¬_null; |
233 | 244 | view.time_unit = decoded_time_unit(column_schema.type_descriptor.time_unit); |
234 | 244 | view.logical_integer_bit_width = column_schema.type_descriptor.integer_bit_width; |
235 | 244 | view.logical_integer_is_signed = !column_schema.type_descriptor.is_unsigned_integer; |
236 | 244 | view.decimal_precision = column_schema.type_descriptor.decimal_precision; |
237 | 244 | view.decimal_scale = column_schema.type_descriptor.decimal_scale; |
238 | 244 | view.fixed_length = column_schema.type_descriptor.fixed_length; |
239 | 244 | view.timestamp_is_adjusted_to_utc = column_schema.type_descriptor.timestamp_is_adjusted_to_utc; |
240 | 244 | view.timezone = timezone; |
241 | | // Statistics are pruning proofs, not row materialization. A malformed non-NULL bound must |
242 | | // disable pruning instead of being converted to NULL under permissive scan semantics. |
243 | 244 | view.enable_strict_mode = true; |
244 | 244 | RETURN_IF_ERROR(column_schema.type->get_serde()->read_field_from_decoded_value( |
245 | 244 | *column_schema.type, field, view)); |
246 | 242 | if (field->is_null()) { |
247 | 0 | return Status::DataQualityError("Non-NULL Parquet statistic decoded as NULL"); |
248 | 0 | } |
249 | 242 | return Status::OK(); |
250 | 242 | } |
251 | | |
252 | | template <typename NativeType> |
253 | | bool set_decoded_field(const ParquetColumnSchema& column_schema, DecodedValueKind value_kind, |
254 | 244 | const NativeType& value, Field* field, const cctz::time_zone* timezone) { |
255 | 244 | DecodedColumnView view; |
256 | 244 | view.value_kind = value_kind; |
257 | 244 | view.values = reinterpret_cast<const uint8_t*>(&value); |
258 | 244 | return read_decoded_field(column_schema, view, field, timezone).ok(); |
259 | 244 | } parquet_statistics.cpp:_ZN5doris6format7parquet12_GLOBAL__N_117set_decoded_fieldIhEEbRKNS1_19ParquetColumnSchemaENS_16DecodedValueKindERKT_PNS_5FieldEPKN4cctz9time_zoneE Line | Count | Source | 254 | 4 | const NativeType& value, Field* field, const cctz::time_zone* timezone) { | 255 | 4 | DecodedColumnView view; | 256 | 4 | view.value_kind = value_kind; | 257 | 4 | view.values = reinterpret_cast<const uint8_t*>(&value); | 258 | 4 | return read_decoded_field(column_schema, view, field, timezone).ok(); | 259 | 4 | } |
parquet_statistics.cpp:_ZN5doris6format7parquet12_GLOBAL__N_117set_decoded_fieldIiEEbRKNS1_19ParquetColumnSchemaENS_16DecodedValueKindERKT_PNS_5FieldEPKN4cctz9time_zoneE Line | Count | Source | 254 | 240 | const NativeType& value, Field* field, const cctz::time_zone* timezone) { | 255 | 240 | DecodedColumnView view; | 256 | 240 | view.value_kind = value_kind; | 257 | 240 | view.values = reinterpret_cast<const uint8_t*>(&value); | 258 | 240 | return read_decoded_field(column_schema, view, field, timezone).ok(); | 259 | 240 | } |
Unexecuted instantiation: parquet_statistics.cpp:_ZN5doris6format7parquet12_GLOBAL__N_117set_decoded_fieldIlEEbRKNS1_19ParquetColumnSchemaENS_16DecodedValueKindERKT_PNS_5FieldEPKN4cctz9time_zoneE 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 |
260 | | |
261 | 0 | int64_t floor_timestamp_seconds(int64_t value, ParquetTimeUnit time_unit) { |
262 | 0 | int64_t units_per_second = 1; |
263 | 0 | switch (time_unit) { |
264 | 0 | case ParquetTimeUnit::MILLIS: |
265 | 0 | units_per_second = 1000; |
266 | 0 | break; |
267 | 0 | case ParquetTimeUnit::MICROS: |
268 | 0 | units_per_second = 1000000; |
269 | 0 | break; |
270 | 0 | case ParquetTimeUnit::NANOS: |
271 | 0 | units_per_second = 1000000000; |
272 | 0 | break; |
273 | 0 | default: |
274 | 0 | DORIS_CHECK(false); |
275 | 0 | } |
276 | 0 | return format::floor_epoch_seconds(value, units_per_second); |
277 | 0 | } |
278 | | |
279 | | bool timestamp_min_max_is_safe(const ParquetColumnSchema& column_schema, int64_t min_value, |
280 | 0 | int64_t max_value, const cctz::time_zone* timezone) { |
281 | 0 | if (min_value > max_value) { |
282 | 0 | return false; |
283 | 0 | } |
284 | 0 | if (!column_schema.type_descriptor.is_timestamp || |
285 | 0 | !column_schema.type_descriptor.timestamp_is_adjusted_to_utc || timezone == nullptr || |
286 | 0 | remove_nullable(column_schema.type)->get_primitive_type() == TYPE_TIMESTAMPTZ) { |
287 | | // TIMESTAMPTZ keeps the original UTC ordering, so local civil-time rollback does not make |
288 | | // its converted min/max non-monotonic. |
289 | 0 | return true; |
290 | 0 | } |
291 | 0 | return format::utc_timestamp_range_is_monotonic( |
292 | 0 | floor_timestamp_seconds(min_value, column_schema.type_descriptor.time_unit), |
293 | 0 | floor_timestamp_seconds(max_value, column_schema.type_descriptor.time_unit), *timezone); |
294 | 0 | } |
295 | | |
296 | | template <typename NativeType> |
297 | 123 | bool valid_min_max(const NativeType& min_value, const NativeType& max_value) { |
298 | 123 | if constexpr (std::is_floating_point_v<NativeType>) { |
299 | | // Parquet requires readers to ignore min/max statistics if either bound is NaN. |
300 | 0 | if (std::isnan(min_value) || std::isnan(max_value)) { |
301 | 0 | return false; |
302 | 0 | } |
303 | 0 | } |
304 | 0 | return true; |
305 | 123 | } parquet_statistics.cpp:_ZN5doris6format7parquet12_GLOBAL__N_113valid_min_maxIhEEbRKT_S6_ Line | Count | Source | 297 | 2 | bool valid_min_max(const NativeType& min_value, const NativeType& max_value) { | 298 | | if constexpr (std::is_floating_point_v<NativeType>) { | 299 | | // Parquet requires readers to ignore min/max statistics if either bound is NaN. | 300 | | if (std::isnan(min_value) || std::isnan(max_value)) { | 301 | | return false; | 302 | | } | 303 | | } | 304 | 2 | return true; | 305 | 2 | } |
parquet_statistics.cpp:_ZN5doris6format7parquet12_GLOBAL__N_113valid_min_maxIiEEbRKT_S6_ Line | Count | Source | 297 | 121 | bool valid_min_max(const NativeType& min_value, const NativeType& max_value) { | 298 | | if constexpr (std::is_floating_point_v<NativeType>) { | 299 | | // Parquet requires readers to ignore min/max statistics if either bound is NaN. | 300 | | if (std::isnan(min_value) || std::isnan(max_value)) { | 301 | | return false; | 302 | | } | 303 | | } | 304 | 121 | return true; | 305 | 121 | } |
Unexecuted instantiation: parquet_statistics.cpp:_ZN5doris6format7parquet12_GLOBAL__N_113valid_min_maxIlEEbRKT_S6_ Unexecuted instantiation: parquet_statistics.cpp:_ZN5doris6format7parquet12_GLOBAL__N_113valid_min_maxIfEEbRKT_S6_ Unexecuted instantiation: parquet_statistics.cpp:_ZN5doris6format7parquet12_GLOBAL__N_113valid_min_maxIdEEbRKT_S6_ |
306 | | |
307 | 121 | bool decoded_min_max_is_ordered(const ParquetColumnStatistics& column_statistics) { |
308 | 121 | return !(column_statistics.max_value < column_statistics.min_value); |
309 | 121 | } |
310 | | |
311 | | bool set_decoded_binary_field(const ParquetColumnSchema& column_schema, DecodedValueKind value_kind, |
312 | | const StringRef& value, Field* field, |
313 | 0 | const cctz::time_zone* timezone) { |
314 | 0 | std::vector<StringRef> binary_values {value}; |
315 | 0 | DecodedColumnView view; |
316 | 0 | view.value_kind = value_kind; |
317 | 0 | view.binary_values = &binary_values; |
318 | 0 | return read_decoded_field(column_schema, view, field, timezone).ok(); |
319 | 0 | } |
320 | | |
321 | | template <typename T> |
322 | 3 | T load_predicate_value(const char* data) { |
323 | 3 | T value; |
324 | 3 | memcpy(&value, data, sizeof(T)); |
325 | 3 | return value; |
326 | 3 | } Unexecuted instantiation: parquet_statistics.cpp:_ZN5doris6format7parquet12_GLOBAL__N_120load_predicate_valueIaEET_PKc Unexecuted instantiation: parquet_statistics.cpp:_ZN5doris6format7parquet12_GLOBAL__N_120load_predicate_valueIsEET_PKc Unexecuted instantiation: parquet_statistics.cpp:_ZN5doris6format7parquet12_GLOBAL__N_120load_predicate_valueIiEET_PKc parquet_statistics.cpp:_ZN5doris6format7parquet12_GLOBAL__N_120load_predicate_valueIlEET_PKc Line | Count | Source | 322 | 3 | T load_predicate_value(const char* data) { | 323 | 3 | T value; | 324 | 3 | memcpy(&value, data, sizeof(T)); | 325 | 3 | return value; | 326 | 3 | } |
|
327 | | |
328 | 3 | std::optional<int64_t> load_predicate_integral_value(const char* buf, size_t size) { |
329 | 3 | switch (size) { |
330 | 0 | case sizeof(int8_t): |
331 | 0 | return static_cast<int64_t>(load_predicate_value<int8_t>(buf)); |
332 | 0 | case sizeof(int16_t): |
333 | 0 | return static_cast<int64_t>(load_predicate_value<int16_t>(buf)); |
334 | 0 | case sizeof(int32_t): |
335 | 0 | return static_cast<int64_t>(load_predicate_value<int32_t>(buf)); |
336 | 3 | case sizeof(int64_t): |
337 | 3 | return load_predicate_value<int64_t>(buf); |
338 | 0 | default: |
339 | 0 | return std::nullopt; |
340 | 3 | } |
341 | 3 | } |
342 | | |
343 | | bool logical_integer_fits_physical_int32(const ParquetTypeDescriptor& type_descriptor, |
344 | 3 | int64_t value) { |
345 | 3 | const int bit_width = |
346 | 3 | type_descriptor.integer_bit_width > 0 ? type_descriptor.integer_bit_width : 32; |
347 | 3 | if (type_descriptor.is_unsigned_integer) { |
348 | 3 | const uint64_t max_value = bit_width >= 32 ? std::numeric_limits<uint32_t>::max() |
349 | 3 | : ((uint64_t {1} << bit_width) - 1); |
350 | 3 | return value >= 0 && static_cast<uint64_t>(value) <= max_value; |
351 | 3 | } |
352 | 0 | const int64_t min_value = bit_width >= 32 ? std::numeric_limits<int32_t>::min() |
353 | 0 | : -(int64_t {1} << (bit_width - 1)); |
354 | 0 | const int64_t max_value = bit_width >= 32 ? std::numeric_limits<int32_t>::max() |
355 | 0 | : ((int64_t {1} << (bit_width - 1)) - 1); |
356 | 0 | return value >= min_value && value <= max_value; |
357 | 3 | } |
358 | | |
359 | | std::optional<int32_t> convert_logical_integer_to_physical_int32( |
360 | 3 | const ParquetTypeDescriptor& type_descriptor, int64_t value) { |
361 | 3 | if (!logical_integer_fits_physical_int32(type_descriptor, value)) { |
362 | 1 | return std::nullopt; |
363 | 1 | } |
364 | 2 | if (!type_descriptor.is_unsigned_integer) { |
365 | 0 | return static_cast<int32_t>(value); |
366 | 0 | } |
367 | 2 | const auto unsigned_value = static_cast<uint32_t>(value); |
368 | 2 | int32_t physical_value; |
369 | 2 | memcpy(&physical_value, &unsigned_value, sizeof(physical_value)); |
370 | 2 | return physical_value; |
371 | 2 | } |
372 | | |
373 | | class NativeParquetBloomFilterAdapter final : public segment_v2::BloomFilter { |
374 | | public: |
375 | | NativeParquetBloomFilterAdapter(const ParquetColumnSchema& column_schema, |
376 | | const segment_v2::BloomFilter& bloom_filter) |
377 | 3 | : _column_schema(column_schema), _bloom_filter(bloom_filter) {} |
378 | | |
379 | 0 | void add_bytes(const char*, size_t) override { DORIS_CHECK(false); } |
380 | | |
381 | 3 | bool test_bytes(const char* buf, size_t size) const override { |
382 | 3 | if (buf == nullptr || |
383 | 3 | _column_schema.type_descriptor.physical_type != tparquet::Type::INT32) { |
384 | 0 | return _bloom_filter.test_bytes(buf, size); |
385 | 0 | } |
386 | 3 | const auto logical_value = load_predicate_integral_value(buf, size); |
387 | 3 | if (!logical_value.has_value()) { |
388 | 0 | return true; |
389 | 0 | } |
390 | 3 | const auto physical_value = convert_logical_integer_to_physical_int32( |
391 | 3 | _column_schema.type_descriptor, *logical_value); |
392 | 3 | if (!physical_value.has_value()) { |
393 | 1 | return false; |
394 | 1 | } |
395 | | // Native file Bloom bytes are hashed from the Parquet physical carrier, not the wider |
396 | | // Doris logical literal used by VExpr (for example UINT32 is exposed as BIGINT). |
397 | 2 | return _bloom_filter.test_bytes(reinterpret_cast<const char*>(&*physical_value), |
398 | 2 | sizeof(*physical_value)); |
399 | 3 | } |
400 | | |
401 | 0 | void set_has_null(bool has_null) override { DORIS_CHECK(!has_null); } |
402 | 0 | bool has_null() const override { return false; } |
403 | 0 | void add_hash(uint64_t) override { DORIS_CHECK(false); } |
404 | 0 | bool test_hash(uint64_t hash) const override { return _bloom_filter.test_hash(hash); } |
405 | | |
406 | | private: |
407 | | const ParquetColumnSchema& _column_schema; |
408 | | const segment_v2::BloomFilter& _bloom_filter; |
409 | | }; |
410 | | |
411 | 4 | bool bloom_filter_supported(const ParquetColumnSchema& column_schema) { |
412 | 4 | if (!bloom_logical_type_supported(column_schema)) { |
413 | 0 | return false; |
414 | 0 | } |
415 | 4 | switch (column_schema.type_descriptor.physical_type) { |
416 | 0 | case tparquet::Type::BOOLEAN: |
417 | 4 | case tparquet::Type::INT32: |
418 | 4 | case tparquet::Type::INT64: |
419 | 4 | case tparquet::Type::FLOAT: |
420 | 4 | case tparquet::Type::DOUBLE: |
421 | 4 | case tparquet::Type::BYTE_ARRAY: |
422 | 4 | return true; |
423 | 0 | case tparquet::Type::FIXED_LEN_BYTE_ARRAY: |
424 | 0 | return column_schema.type_descriptor.is_string_like && |
425 | 0 | column_schema.type_descriptor.fixed_length > 0; |
426 | 0 | default: |
427 | 0 | return false; |
428 | 4 | } |
429 | 4 | } |
430 | | |
431 | | const ParquetColumnSchema* resolve_local_leaf_schema( |
432 | | const std::vector<std::unique_ptr<ParquetColumnSchema>>& schema, |
433 | 119 | const format::LocalColumnId file_column_id) { |
434 | 119 | if (!file_column_id.is_valid() || file_column_id.value() >= static_cast<int>(schema.size())) { |
435 | 0 | return nullptr; |
436 | 0 | } |
437 | 119 | const ParquetColumnSchema* column_schema = schema[file_column_id.value()].get(); |
438 | 119 | if (column_schema == nullptr || column_schema->kind != ParquetColumnSchemaKind::PRIMITIVE || |
439 | 119 | column_schema->leaf_column_id < 0 || column_schema->max_repetition_level > 0) { |
440 | 0 | return nullptr; |
441 | 0 | } |
442 | 119 | return column_schema; |
443 | 119 | } |
444 | | |
445 | | std::optional<format::LocalColumnId> file_column_id_by_block_position( |
446 | 119 | const format::FileScanRequest& request, int block_position) { |
447 | 180 | for (const auto& [file_column_id, local_index] : request.local_positions) { |
448 | 180 | if (local_index.value() == block_position) { |
449 | 119 | return file_column_id; |
450 | 119 | } |
451 | 180 | } |
452 | 0 | return std::nullopt; |
453 | 119 | } |
454 | | |
455 | | bool has_expr_zonemap_filter(const format::FileScanRequest& request, |
456 | 752 | const RuntimeState* runtime_state) { |
457 | 752 | if (!expr_zonemap::is_expr_zonemap_filter_enabled(runtime_state)) { |
458 | 0 | return false; |
459 | 0 | } |
460 | 752 | for (const auto& conjunct : request.conjuncts) { |
461 | 271 | if (conjunct != nullptr && conjunct->root() != nullptr && |
462 | 271 | conjunct->root()->can_evaluate_zonemap_filter()) { |
463 | 163 | return true; |
464 | 163 | } |
465 | 271 | } |
466 | 589 | return false; |
467 | 752 | } |
468 | | |
469 | 67 | std::set<int> collect_expr_zonemap_slot_indexes(const VExprContextSPtrs& conjuncts) { |
470 | 67 | std::set<int> slot_indexes; |
471 | 72 | for (const auto& conjunct : conjuncts) { |
472 | 72 | if (conjunct != nullptr && conjunct->root() != nullptr && |
473 | 72 | conjunct->root()->can_evaluate_zonemap_filter()) { |
474 | 69 | conjunct->root()->collect_slot_column_ids(slot_indexes); |
475 | 69 | } |
476 | 72 | } |
477 | 67 | return slot_indexes; |
478 | 67 | } |
479 | | |
480 | | template <typename SlotIndexSelector> |
481 | | std::map<int, VExprContextSPtrs> collect_conjuncts_by_single_slot( |
482 | 489 | const VExprContextSPtrs& conjuncts, SlotIndexSelector slot_index_selector) { |
483 | 489 | std::map<int, VExprContextSPtrs> conjuncts_by_slot; |
484 | 489 | for (const auto& conjunct : conjuncts) { |
485 | 177 | const auto slot_index = slot_index_selector(conjunct); |
486 | 177 | if (slot_index >= 0) { |
487 | 41 | conjuncts_by_slot[slot_index].push_back(conjunct); |
488 | 41 | } |
489 | 177 | } |
490 | 489 | return conjuncts_by_slot; |
491 | 489 | } |
492 | | |
493 | | std::shared_ptr<segment_v2::ZoneMap> make_zonemap_from_statistics( |
494 | 119 | const ParquetColumnStatistics& statistics) { |
495 | 119 | if (!statistics.has_null_count && !statistics.has_min_max) { |
496 | 15 | return nullptr; |
497 | 15 | } |
498 | 104 | segment_v2::ZoneMap zone_map; |
499 | 104 | zone_map.has_null = statistics.has_null; |
500 | 104 | zone_map.has_not_null = statistics.has_not_null; |
501 | 104 | if (!statistics.has_not_null) { |
502 | 0 | return std::make_shared<segment_v2::ZoneMap>(std::move(zone_map)); |
503 | 0 | } |
504 | 104 | if (!statistics.has_min_max) { |
505 | | // Null counts remain trustworthy when min/max decoding fails (for example, because a |
506 | | // floating-point bound is NaN). pass_all prevents range pruning without discarding the |
507 | | // has_null/has_not_null flags needed by IS NULL and IS NOT NULL predicates. |
508 | 0 | zone_map.pass_all = true; |
509 | 0 | return std::make_shared<segment_v2::ZoneMap>(std::move(zone_map)); |
510 | 0 | } |
511 | 104 | zone_map.min_value = statistics.min_value; |
512 | 104 | zone_map.max_value = statistics.max_value; |
513 | 104 | return std::make_shared<segment_v2::ZoneMap>(std::move(zone_map)); |
514 | 104 | } |
515 | | |
516 | | void add_slot_zonemap(ZoneMapEvalContext* ctx, int slot_index, const DataTypePtr& data_type, |
517 | 119 | std::shared_ptr<segment_v2::ZoneMap> zone_map) { |
518 | 119 | DORIS_CHECK(ctx != nullptr); |
519 | 119 | ZoneMapEvalContext::SlotZoneMap slot_zone_map; |
520 | 119 | slot_zone_map.data_type = data_type; |
521 | 119 | slot_zone_map.zone_map = std::move(zone_map); |
522 | 119 | ctx->slots.emplace(slot_index, std::move(slot_zone_map)); |
523 | 119 | } |
524 | | |
525 | 67 | void accumulate_zonemap_stats(const ZoneMapEvalContext& ctx, ParquetPruningStats* pruning_stats) { |
526 | 67 | if (pruning_stats == nullptr) { |
527 | 2 | return; |
528 | 2 | } |
529 | 65 | pruning_stats->expr_zonemap_unusable_evals += ctx.stats.unusable_zonemap_eval_count; |
530 | 65 | pruning_stats->in_zonemap_point_check_count += ctx.stats.in_zonemap_point_check_count; |
531 | 65 | pruning_stats->in_zonemap_range_only_count += ctx.stats.in_zonemap_range_only_count; |
532 | 65 | } |
533 | | |
534 | | } // namespace |
535 | | |
536 | | bool can_use_parquet_page_index(const format::FileScanRequest& request, |
537 | 233 | const RuntimeState* runtime_state) { |
538 | 233 | return config::enable_parquet_page_index && has_expr_zonemap_filter(request, runtime_state); |
539 | 233 | } |
540 | | |
541 | | std::shared_ptr<segment_v2::ZoneMap> ParquetStatisticsUtils::MakeZoneMap( |
542 | 119 | const ParquetColumnStatistics& statistics) { |
543 | 119 | return make_zonemap_from_statistics(statistics); |
544 | 119 | } |
545 | | |
546 | | ParquetColumnStatistics ParquetStatisticsUtils::TransformColumnStatistics( |
547 | | const ParquetColumnSchema& column_schema, const tparquet::Statistics* statistics, |
548 | 90 | int64_t column_value_count, const cctz::time_zone* timezone) { |
549 | 90 | ParquetColumnStatistics result; |
550 | 90 | if (statistics == nullptr || column_value_count < 0) { |
551 | 15 | return result; |
552 | 15 | } |
553 | | |
554 | 75 | if (statistics->__isset.null_count && statistics->null_count > column_value_count) { |
555 | | // An impossible null count makes all derived min/max and all-null flags untrustworthy; |
556 | | // disable pruning instead of turning corrupt footer metadata into false negatives. |
557 | 0 | return result; |
558 | 0 | } |
559 | | |
560 | 75 | const bool has_null_count = statistics->__isset.null_count && statistics->null_count >= 0; |
561 | 75 | const int64_t null_count = has_null_count ? statistics->null_count : 0; |
562 | 75 | const bool has_not_null = has_null_count ? column_value_count > null_count : true; |
563 | 75 | const std::string* min_value = statistics->__isset.min_value |
564 | 75 | ? &statistics->min_value |
565 | 75 | : (statistics->__isset.min ? &statistics->min : nullptr); |
566 | 75 | const std::string* max_value = statistics->__isset.max_value |
567 | 75 | ? &statistics->max_value |
568 | 75 | : (statistics->__isset.max ? &statistics->max : nullptr); |
569 | | |
570 | 75 | tparquet::ColumnIndex index; |
571 | 75 | index.__set_null_pages({!has_not_null}); |
572 | 75 | index.__set_null_counts({null_count}); |
573 | 75 | if (min_value != nullptr && max_value != nullptr) { |
574 | 74 | index.__set_min_values({*min_value}); |
575 | 74 | index.__set_max_values({*max_value}); |
576 | 74 | } |
577 | | // Footer statistics and page indexes share the same little-endian physical encoding. Reusing |
578 | | // one decoder keeps native row-group and page pruning identical for logical types and NaNs. |
579 | 75 | if (!build_native_page_statistics(index, column_schema, 0, column_value_count, &result, |
580 | 75 | timezone)) { |
581 | 4 | return {}; |
582 | 4 | } |
583 | 71 | if (!has_null_count) { |
584 | 0 | result.has_null_count = false; |
585 | 0 | result.has_null = true; |
586 | 0 | } |
587 | 71 | return result; |
588 | 75 | } |
589 | | |
590 | | bool ParquetStatisticsUtils::NativeBloomFilterExcludes( |
591 | | const ParquetColumnSchema& column_schema, int slot_index, |
592 | 3 | const VExprContextSPtrs& conjuncts, const segment_v2::BloomFilter& bloom_filter) { |
593 | 3 | if (!bloom_filter_supported(column_schema)) { |
594 | 0 | return false; |
595 | 0 | } |
596 | 3 | NativeParquetBloomFilterAdapter adapter(column_schema, bloom_filter); |
597 | 3 | BloomFilterEvalContext ctx; |
598 | 3 | ctx.slots.emplace(slot_index, BloomFilterEvalContext::SlotBloomFilter { |
599 | 3 | .data_type = column_schema.type, |
600 | 3 | .bloom_filter = &adapter, |
601 | 3 | }); |
602 | 3 | return VExprContext::evaluate_bloom_filter(conjuncts, ctx) == ZoneMapFilterResult::kNoMatch; |
603 | 3 | } |
604 | | |
605 | | namespace { |
606 | | |
607 | | void collect_filtered_leaf_ids(const ParquetColumnSchema& column_schema, |
608 | | const format::LocalColumnIndex* projection, |
609 | 64 | std::set<int>* leaf_column_ids) { |
610 | 64 | if (column_schema.kind == ParquetColumnSchemaKind::PRIMITIVE) { |
611 | 64 | if (column_schema.leaf_column_id >= 0) { |
612 | 64 | leaf_column_ids->insert(column_schema.leaf_column_id); |
613 | 64 | } |
614 | 64 | return; |
615 | 64 | } |
616 | 0 | for (const auto& child_schema : column_schema.children) { |
617 | 0 | if (!format::is_child_projected(projection, child_schema->local_id)) { |
618 | 0 | continue; |
619 | 0 | } |
620 | 0 | collect_filtered_leaf_ids(*child_schema, |
621 | 0 | format::find_child_projection(projection, child_schema->local_id), |
622 | 0 | leaf_column_ids); |
623 | 0 | } |
624 | 0 | } |
625 | | |
626 | 114 | bool native_metadata_predicate_is_type_safe(const ParquetColumnSchema& column_schema) { |
627 | 114 | DORIS_CHECK(column_schema.type != nullptr); |
628 | | // Raw VARBINARY file slots may feed table-side STRING casts. Footer/page metadata is still in |
629 | | // the pre-cast domain, so using it for a rewritten table predicate can cause false negatives. |
630 | 114 | return remove_nullable(column_schema.type)->get_primitive_type() != TYPE_VARBINARY; |
631 | 114 | } |
632 | | |
633 | | bool check_native_statistics(const tparquet::FileMetaData& metadata, |
634 | | const tparquet::RowGroup& row_group, |
635 | | const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema, |
636 | | const format::FileScanRequest& request, |
637 | 67 | ParquetPruningStats* pruning_stats, const cctz::time_zone* timezone) { |
638 | 67 | const auto slot_indexes = collect_expr_zonemap_slot_indexes(request.conjuncts); |
639 | 67 | if (slot_indexes.empty()) { |
640 | 0 | return false; |
641 | 0 | } |
642 | 67 | ZoneMapEvalContext ctx; |
643 | 69 | for (const int slot_index : slot_indexes) { |
644 | 69 | const auto file_column_id = file_column_id_by_block_position(request, slot_index); |
645 | 69 | if (!file_column_id.has_value()) { |
646 | 0 | continue; |
647 | 0 | } |
648 | 69 | const auto* column_schema = resolve_local_leaf_schema(file_schema, *file_column_id); |
649 | 69 | if (column_schema == nullptr || column_schema->type == nullptr || |
650 | 69 | !native_metadata_predicate_is_type_safe(*column_schema) || |
651 | 69 | column_schema->leaf_column_id >= static_cast<int>(row_group.columns.size())) { |
652 | 0 | continue; |
653 | 0 | } |
654 | 69 | const auto& chunk = row_group.columns[column_schema->leaf_column_id]; |
655 | 69 | std::shared_ptr<segment_v2::ZoneMap> zone_map; |
656 | 69 | if (chunk.__isset.meta_data) { |
657 | 69 | const auto& column_metadata = chunk.meta_data; |
658 | 69 | std::optional<tparquet::Statistics> safe_statistics; |
659 | 69 | if (column_metadata.__isset.statistics) { |
660 | 55 | safe_statistics = detail::sanitize_native_footer_statistics( |
661 | 55 | column_schema->type_descriptor, column_metadata.statistics, |
662 | 55 | detail::has_supported_type_defined_order(metadata, |
663 | 55 | column_schema->leaf_column_id)); |
664 | 55 | } |
665 | 69 | zone_map = ParquetStatisticsUtils::MakeZoneMap( |
666 | 69 | ParquetStatisticsUtils::TransformColumnStatistics( |
667 | 69 | *column_schema, |
668 | 69 | safe_statistics.has_value() ? &*safe_statistics : nullptr, |
669 | 69 | column_metadata.num_values, timezone)); |
670 | 69 | } |
671 | 69 | add_slot_zonemap(&ctx, slot_index, column_schema->type, std::move(zone_map)); |
672 | 69 | } |
673 | 67 | const auto result = VExprContext::evaluate_zonemap_filter(request.conjuncts, ctx); |
674 | 67 | accumulate_zonemap_stats(ctx, pruning_stats); |
675 | 67 | return result == ZoneMapFilterResult::kNoMatch; |
676 | 67 | } |
677 | | |
678 | 35 | bool is_native_dictionary_data_encoding(tparquet::Encoding::type encoding) { |
679 | 35 | return encoding == tparquet::Encoding::PLAIN_DICTIONARY || |
680 | 35 | encoding == tparquet::Encoding::RLE_DICTIONARY; |
681 | 35 | } |
682 | | |
683 | 0 | bool is_native_level_encoding(tparquet::Encoding::type encoding) { |
684 | 0 | return encoding == tparquet::Encoding::RLE || encoding == tparquet::Encoding::BIT_PACKED; |
685 | 0 | } |
686 | | |
687 | 35 | bool is_native_dictionary_encoded_chunk(const tparquet::ColumnMetaData& metadata) { |
688 | 35 | if (!metadata.__isset.dictionary_page_offset || metadata.dictionary_page_offset < 0) { |
689 | 0 | return false; |
690 | 0 | } |
691 | 35 | if (metadata.__isset.encoding_stats && !metadata.encoding_stats.empty()) { |
692 | 35 | bool has_dictionary_data_page = false; |
693 | 70 | for (const auto& encoding_stat : metadata.encoding_stats) { |
694 | 70 | if ((encoding_stat.page_type != tparquet::PageType::DATA_PAGE && |
695 | 70 | encoding_stat.page_type != tparquet::PageType::DATA_PAGE_V2) || |
696 | 70 | encoding_stat.count <= 0) { |
697 | 35 | continue; |
698 | 35 | } |
699 | 35 | if (!is_native_dictionary_data_encoding(encoding_stat.encoding)) { |
700 | 0 | return false; |
701 | 0 | } |
702 | 35 | has_dictionary_data_page = true; |
703 | 35 | } |
704 | 35 | return has_dictionary_data_page; |
705 | 35 | } |
706 | 0 | bool has_dictionary_encoding = false; |
707 | 0 | for (const auto encoding : metadata.encodings) { |
708 | 0 | if (is_native_dictionary_data_encoding(encoding)) { |
709 | 0 | has_dictionary_encoding = true; |
710 | 0 | } else if (!is_native_level_encoding(encoding)) { |
711 | 0 | return false; |
712 | 0 | } |
713 | 0 | } |
714 | 0 | return has_dictionary_encoding; |
715 | 0 | } |
716 | | |
717 | | const format::LocalColumnIndex* find_request_projection(const format::FileScanRequest& request, |
718 | 40 | format::LocalColumnId file_column_id) { |
719 | 50 | for (const auto& projection : request.predicate_columns) { |
720 | 50 | if (projection.local_id() == file_column_id.value()) { |
721 | 40 | return &projection; |
722 | 40 | } |
723 | 50 | } |
724 | 0 | for (const auto& projection : request.non_predicate_columns) { |
725 | 0 | if (projection.local_id() == file_column_id.value()) { |
726 | 0 | return &projection; |
727 | 0 | } |
728 | 0 | } |
729 | 0 | return nullptr; |
730 | 0 | } |
731 | | |
732 | | ParquetRowGroupPruneReason native_dictionary_prune_reason( |
733 | | const tparquet::RowGroup& row_group, int row_group_idx, |
734 | | const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema, |
735 | | const format::FileScanRequest& request, const cctz::time_zone* timezone, |
736 | 257 | ParquetFileContext* file_context, const ParquetColumnReaderProfile& column_reader_profile) { |
737 | 257 | if (file_context == nullptr || file_context->native_metadata == nullptr) { |
738 | 2 | return ParquetRowGroupPruneReason::NONE; |
739 | 2 | } |
740 | 255 | const auto conjuncts_by_slot = collect_conjuncts_by_single_slot( |
741 | 255 | request.conjuncts, expr_zonemap::single_slot_dictionary_index); |
742 | 255 | for (const auto& [slot_index, conjuncts] : conjuncts_by_slot) { |
743 | 40 | const auto file_column_id = file_column_id_by_block_position(request, slot_index); |
744 | 40 | if (!file_column_id.has_value()) { |
745 | 0 | continue; |
746 | 0 | } |
747 | 40 | const auto* column_schema = resolve_local_leaf_schema(file_schema, *file_column_id); |
748 | 40 | const auto* projection = find_request_projection(request, *file_column_id); |
749 | 40 | if (column_schema == nullptr || projection == nullptr || column_schema->type == nullptr || |
750 | 40 | !column_schema->type_descriptor.is_string_like || |
751 | 40 | column_schema->leaf_column_id >= static_cast<int>(row_group.columns.size())) { |
752 | 5 | continue; |
753 | 5 | } |
754 | 35 | if (!native_metadata_predicate_is_type_safe(*column_schema)) { |
755 | | // The file-local VARBINARY may feed a table-side STRING cast. Pruning before that cast |
756 | | // can compare different Field kinds and incorrectly discard a matching row group. |
757 | 0 | continue; |
758 | 0 | } |
759 | 35 | const auto& chunk = row_group.columns[column_schema->leaf_column_id]; |
760 | 35 | if (!chunk.__isset.meta_data || |
761 | 35 | (chunk.meta_data.type != tparquet::Type::BYTE_ARRAY && |
762 | 35 | chunk.meta_data.type != tparquet::Type::FIXED_LEN_BYTE_ARRAY) || |
763 | 35 | !is_native_dictionary_encoded_chunk(chunk.meta_data)) { |
764 | 0 | continue; |
765 | 0 | } |
766 | 35 | std::unique_ptr<ParquetColumnReader> reader; |
767 | 35 | const std::vector<RowRange> ranges {{0, row_group.num_rows}}; |
768 | 35 | const std::unordered_map<int, tparquet::OffsetIndex> offset_indexes; |
769 | | // Metadata pruning uses the real native reader, so its page work must be attributed to the |
770 | | // scan profile even when the row group is eliminated before execution readers are built. |
771 | 35 | const auto status = NativeColumnReader::create( |
772 | 35 | *column_schema, projection, file_context->native_file, |
773 | 35 | file_context->native_metadata, row_group_idx, ranges, offset_indexes, timezone, |
774 | 35 | file_context->native_io_ctx, nullptr, file_context->native_page_cache_enabled, |
775 | 35 | file_context->native_page_cache_file_key, true, column_reader_profile, &reader); |
776 | 35 | if (!status.ok() || reader == nullptr) { |
777 | 0 | continue; |
778 | 0 | } |
779 | 35 | auto dictionary_result = reader->dictionary_values(); |
780 | 35 | if (!dictionary_result.has_value()) { |
781 | 0 | continue; |
782 | 0 | } |
783 | 35 | auto dictionary = std::move(dictionary_result).value(); |
784 | 35 | std::vector<Field> values(dictionary->size()); |
785 | 106 | for (size_t value_idx = 0; value_idx < dictionary->size(); ++value_idx) { |
786 | 71 | dictionary->get(value_idx, values[value_idx]); |
787 | 71 | } |
788 | 35 | DictionaryEvalContext ctx; |
789 | 35 | ctx.slots.emplace(slot_index, DictionaryEvalContext::SlotDictionary { |
790 | 35 | .data_type = column_schema->type, |
791 | 35 | .values = std::move(values), |
792 | 35 | }); |
793 | 35 | if (VExprContext::evaluate_dictionary_filter(conjuncts, ctx) == |
794 | 35 | ZoneMapFilterResult::kNoMatch) { |
795 | 22 | return ParquetRowGroupPruneReason::DICTIONARY; |
796 | 22 | } |
797 | 35 | } |
798 | 233 | return ParquetRowGroupPruneReason::NONE; |
799 | 255 | } |
800 | | |
801 | | ParquetRowGroupPruneReason native_bloom_filter_prune_reason( |
802 | | const tparquet::RowGroup& row_group, |
803 | | const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema, |
804 | | const format::FileScanRequest& request, ParquetFileContext* file_context, |
805 | 234 | ParquetPruningStats* pruning_stats) { |
806 | 234 | if (file_context == nullptr || file_context->native_file == nullptr) { |
807 | 0 | return ParquetRowGroupPruneReason::NONE; |
808 | 0 | } |
809 | 234 | const auto conjuncts_by_slot = collect_conjuncts_by_single_slot( |
810 | 234 | request.conjuncts, expr_zonemap::single_slot_bloom_filter_index); |
811 | 234 | for (const auto& [slot_index, conjuncts] : conjuncts_by_slot) { |
812 | 1 | const auto file_column_id = file_column_id_by_block_position(request, slot_index); |
813 | 1 | if (!file_column_id.has_value()) { |
814 | 0 | continue; |
815 | 0 | } |
816 | 1 | const auto* column_schema = resolve_local_leaf_schema(file_schema, *file_column_id); |
817 | 1 | if (column_schema == nullptr || column_schema->type == nullptr || |
818 | 1 | !native_metadata_predicate_is_type_safe(*column_schema) || |
819 | 1 | !bloom_filter_supported(*column_schema) || |
820 | 1 | column_schema->leaf_column_id >= static_cast<int>(row_group.columns.size())) { |
821 | 0 | continue; |
822 | 0 | } |
823 | 1 | const auto& chunk = row_group.columns[column_schema->leaf_column_id]; |
824 | 1 | if (!chunk.__isset.meta_data) { |
825 | 0 | continue; |
826 | 0 | } |
827 | 1 | std::unique_ptr<native::BlockSplitBloomFilter> bloom_filter; |
828 | 1 | Status status; |
829 | 1 | { |
830 | 1 | int64_t timer_sink = 0; |
831 | 1 | SCOPED_RAW_TIMER(pruning_stats == nullptr ? &timer_sink |
832 | 1 | : &pruning_stats->bloom_filter_read_time); |
833 | 1 | status = read_native_bloom_filter(chunk.meta_data, file_context->native_file, |
834 | 1 | file_context->native_io_ctx, &bloom_filter); |
835 | 1 | } |
836 | 1 | if (!status.ok() || bloom_filter == nullptr) { |
837 | 0 | continue; |
838 | 0 | } |
839 | 1 | if (ParquetStatisticsUtils::NativeBloomFilterExcludes(*column_schema, slot_index, conjuncts, |
840 | 1 | *bloom_filter)) { |
841 | 0 | return ParquetRowGroupPruneReason::BLOOM_FILTER; |
842 | 0 | } |
843 | 1 | } |
844 | 234 | return ParquetRowGroupPruneReason::NONE; |
845 | 234 | } |
846 | | |
847 | | int64_t native_requested_compressed_bytes( |
848 | | const tparquet::RowGroup& row_group, |
849 | | const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema, |
850 | 40 | const format::FileScanRequest& request) { |
851 | 40 | std::set<int> leaf_column_ids; |
852 | 66 | auto collect_projection = [&](const format::LocalColumnIndex& projection) { |
853 | 66 | const int32_t local_id = projection.local_id(); |
854 | 66 | if (local_id < 0 || local_id >= static_cast<int32_t>(file_schema.size()) || |
855 | 66 | file_schema[local_id] == nullptr) { |
856 | 2 | return; |
857 | 2 | } |
858 | 64 | collect_filtered_leaf_ids(*file_schema[local_id], &projection, &leaf_column_ids); |
859 | 64 | }; |
860 | 40 | for (const auto& projection : request.predicate_columns) { |
861 | 35 | collect_projection(projection); |
862 | 35 | } |
863 | 40 | for (const auto& projection : request.non_predicate_columns) { |
864 | 31 | collect_projection(projection); |
865 | 31 | } |
866 | 40 | int64_t bytes = 0; |
867 | 64 | for (const int leaf_column_id : leaf_column_ids) { |
868 | 64 | if (leaf_column_id < 0 || leaf_column_id >= static_cast<int>(row_group.columns.size())) { |
869 | 0 | continue; |
870 | 0 | } |
871 | 64 | const auto& chunk = row_group.columns[leaf_column_id]; |
872 | 64 | if (chunk.__isset.meta_data && chunk.meta_data.total_compressed_size > 0) { |
873 | 64 | bytes += chunk.meta_data.total_compressed_size; |
874 | 64 | } |
875 | 64 | } |
876 | 40 | return bytes; |
877 | 40 | } |
878 | | |
879 | | } // namespace |
880 | | |
881 | | Status select_row_groups_by_metadata( |
882 | | const tparquet::FileMetaData& metadata, |
883 | | const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema, |
884 | | const format::FileScanRequest& request, const std::vector<int>* candidate_row_groups, |
885 | | std::vector<int>* selected_row_groups, bool enable_bloom_filter, |
886 | | ParquetPruningStats* pruning_stats, const cctz::time_zone* timezone, |
887 | | const RuntimeState* runtime_state, ParquetFileContext* file_context, |
888 | | const ParquetColumnReaderProfile& column_reader_profile, |
889 | 467 | ParquetMetadataProbeMode probe_mode) { |
890 | 467 | int64_t timer_sink = 0; |
891 | 467 | SCOPED_RAW_TIMER(pruning_stats == nullptr ? &timer_sink |
892 | 467 | : &pruning_stats->row_group_filter_time); |
893 | 467 | if (selected_row_groups == nullptr) { |
894 | 0 | return Status::InvalidArgument("selected_row_groups is null"); |
895 | 0 | } |
896 | 467 | selected_row_groups->clear(); |
897 | 467 | const size_t candidate_size = candidate_row_groups == nullptr ? metadata.row_groups.size() |
898 | 467 | : candidate_row_groups->size(); |
899 | 467 | if (pruning_stats != nullptr) { |
900 | 464 | pruning_stats->total_row_groups = cast_set<int64_t>(candidate_size); |
901 | 464 | } |
902 | 467 | selected_row_groups->reserve(candidate_size); |
903 | 1.00k | for (size_t candidate_idx = 0; candidate_idx < candidate_size; ++candidate_idx) { |
904 | 537 | const int row_group_idx = candidate_row_groups == nullptr |
905 | 537 | ? static_cast<int>(candidate_idx) |
906 | 537 | : (*candidate_row_groups)[candidate_idx]; |
907 | 537 | if (row_group_idx < 0 || row_group_idx >= static_cast<int>(metadata.row_groups.size())) { |
908 | | // Candidate ids originate in external split metadata; a corrupt id must not terminate |
909 | | // the BE while planning an otherwise recoverable file scan. |
910 | 1 | return Status::Corruption("Invalid Parquet row group candidate {} for {} row groups", |
911 | 1 | row_group_idx, metadata.row_groups.size()); |
912 | 1 | } |
913 | 536 | const auto& row_group = metadata.row_groups[row_group_idx]; |
914 | 536 | if (row_group.num_rows < 0) { |
915 | 0 | return Status::Corruption("Parquet row group {} has negative row count {}", |
916 | 0 | row_group_idx, row_group.num_rows); |
917 | 0 | } |
918 | 536 | if (row_group.num_rows == 0) { |
919 | | // Native metadata probes construct positive row ranges; empty groups contribute no |
920 | | // rows and must be discarded before dictionary, statistics, or Bloom reader setup. |
921 | 1 | continue; |
922 | 1 | } |
923 | 535 | ParquetRowGroupPruneReason prune_reason = ParquetRowGroupPruneReason::NONE; |
924 | 535 | if (probe_mode != ParquetMetadataProbeMode::EXPENSIVE_ONLY && |
925 | 535 | has_expr_zonemap_filter(request, runtime_state) && |
926 | 535 | check_native_statistics(metadata, row_group, file_schema, request, pruning_stats, |
927 | 67 | timezone)) { |
928 | 19 | prune_reason = ParquetRowGroupPruneReason::STATISTICS; |
929 | 19 | } |
930 | 535 | if (probe_mode != ParquetMetadataProbeMode::FOOTER_ONLY && |
931 | 535 | prune_reason == ParquetRowGroupPruneReason::NONE) { |
932 | 257 | prune_reason = |
933 | 257 | native_dictionary_prune_reason(row_group, row_group_idx, file_schema, request, |
934 | 257 | timezone, file_context, column_reader_profile); |
935 | 257 | } |
936 | 535 | if (probe_mode != ParquetMetadataProbeMode::FOOTER_ONLY && |
937 | 535 | prune_reason == ParquetRowGroupPruneReason::NONE && enable_bloom_filter) { |
938 | 234 | prune_reason = native_bloom_filter_prune_reason(row_group, file_schema, request, |
939 | 234 | file_context, pruning_stats); |
940 | 234 | } |
941 | 535 | if (prune_reason == ParquetRowGroupPruneReason::NONE) { |
942 | 494 | selected_row_groups->push_back(row_group_idx); |
943 | 494 | continue; |
944 | 494 | } |
945 | 41 | if (pruning_stats != nullptr) { |
946 | 40 | pruning_stats->filtered_group_rows += row_group.num_rows; |
947 | 40 | pruning_stats->filtered_bytes += |
948 | 40 | native_requested_compressed_bytes(row_group, file_schema, request); |
949 | 40 | if (prune_reason == ParquetRowGroupPruneReason::STATISTICS) { |
950 | 18 | ++pruning_stats->filtered_row_groups_by_statistics; |
951 | 22 | } else if (prune_reason == ParquetRowGroupPruneReason::DICTIONARY) { |
952 | 22 | ++pruning_stats->filtered_row_groups_by_dictionary; |
953 | 22 | } else { |
954 | 0 | ++pruning_stats->filtered_row_groups_by_bloom_filter; |
955 | 0 | } |
956 | 40 | } |
957 | 41 | } |
958 | 466 | return Status::OK(); |
959 | 467 | } |
960 | | |
961 | | namespace { |
962 | | |
963 | | std::vector<RowRange> intersect_ranges(const std::vector<RowRange>& left, |
964 | 5 | const std::vector<RowRange>& right) { |
965 | 5 | std::vector<RowRange> result; |
966 | 5 | size_t left_idx = 0; |
967 | 5 | size_t right_idx = 0; |
968 | 9 | while (left_idx < left.size() && right_idx < right.size()) { |
969 | 4 | const int64_t left_start = left[left_idx].start; |
970 | 4 | const int64_t left_end = left_start + left[left_idx].length; |
971 | 4 | const int64_t right_start = right[right_idx].start; |
972 | 4 | const int64_t right_end = right_start + right[right_idx].length; |
973 | 4 | const int64_t start = std::max(left_start, right_start); |
974 | 4 | const int64_t end = std::min(left_end, right_end); |
975 | 4 | if (start < end) { |
976 | 4 | result.push_back(RowRange {start, end - start}); |
977 | 4 | } |
978 | 4 | if (left_end < right_end) { |
979 | 0 | ++left_idx; |
980 | 4 | } else { |
981 | 4 | ++right_idx; |
982 | 4 | } |
983 | 4 | } |
984 | 5 | return result; |
985 | 5 | } |
986 | | |
987 | 3 | int64_t count_range_rows(const std::vector<RowRange>& ranges) { |
988 | 3 | int64_t rows = 0; |
989 | 3 | for (const auto& range : ranges) { |
990 | 3 | rows += range.length; |
991 | 3 | } |
992 | 3 | return rows; |
993 | 3 | } |
994 | | |
995 | 57 | void append_row_range(const RowRange& range, std::vector<RowRange>* ranges) { |
996 | 57 | if (range.length == 0) { |
997 | 0 | return; |
998 | 0 | } |
999 | 57 | if (!ranges->empty()) { |
1000 | 50 | auto& previous = ranges->back(); |
1001 | 50 | if (previous.start + previous.length == range.start) { |
1002 | 50 | previous.length += range.length; |
1003 | 50 | return; |
1004 | 50 | } |
1005 | 50 | } |
1006 | 7 | ranges->push_back(range); |
1007 | 7 | } |
1008 | | |
1009 | 69 | bool ranges_intersect(const std::vector<RowRange>& ranges, const RowRange& range) { |
1010 | 69 | const int64_t range_end = range.start + range.length; |
1011 | 69 | for (const auto& selected_range : ranges) { |
1012 | 69 | const int64_t selected_end = selected_range.start + selected_range.length; |
1013 | 69 | if (selected_end <= range.start) { |
1014 | 0 | continue; |
1015 | 0 | } |
1016 | 69 | if (selected_range.start >= range_end) { |
1017 | 24 | return false; |
1018 | 24 | } |
1019 | 45 | return true; |
1020 | 69 | } |
1021 | 0 | return false; |
1022 | 69 | } |
1023 | | |
1024 | | void collect_leaf_schemas(const ParquetColumnSchema& column_schema, |
1025 | | const format::LocalColumnIndex* projection, |
1026 | 9 | std::vector<const ParquetColumnSchema*>* leaf_schemas) { |
1027 | 9 | if (column_schema.kind == ParquetColumnSchemaKind::PRIMITIVE) { |
1028 | 9 | leaf_schemas->push_back(&column_schema); |
1029 | 9 | return; |
1030 | 9 | } |
1031 | 0 | for (const auto& child_schema : column_schema.children) { |
1032 | 0 | if (!format::is_child_projected(projection, child_schema->local_id)) { |
1033 | 0 | continue; |
1034 | 0 | } |
1035 | 0 | const auto* child_projection = |
1036 | 0 | format::find_child_projection(projection, child_schema->local_id); |
1037 | 0 | collect_leaf_schemas(*child_schema, child_projection, leaf_schemas); |
1038 | 0 | } |
1039 | 0 | } |
1040 | | |
1041 | | void collect_request_leaf_schemas( |
1042 | | const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema, |
1043 | | const format::FileScanRequest& request, |
1044 | 8 | std::vector<const ParquetColumnSchema*>* leaf_schemas) { |
1045 | 8 | std::set<int> seen_leaf_ids; |
1046 | 9 | auto collect_projection = [&](const format::LocalColumnIndex& projection) { |
1047 | 9 | const int32_t local_id = projection.local_id(); |
1048 | 9 | if (local_id < 0 || local_id >= static_cast<int32_t>(file_schema.size())) { |
1049 | 0 | return; |
1050 | 0 | } |
1051 | 9 | std::vector<const ParquetColumnSchema*> projection_leaf_schemas; |
1052 | 9 | collect_leaf_schemas(*file_schema[local_id], &projection, &projection_leaf_schemas); |
1053 | 9 | for (const auto* leaf_schema : projection_leaf_schemas) { |
1054 | 9 | DORIS_CHECK(leaf_schema != nullptr); |
1055 | 9 | if (seen_leaf_ids.insert(leaf_schema->leaf_column_id).second) { |
1056 | 9 | leaf_schemas->push_back(leaf_schema); |
1057 | 9 | } |
1058 | 9 | } |
1059 | 9 | }; |
1060 | 8 | for (const auto& projection : request.predicate_columns) { |
1061 | 8 | collect_projection(projection); |
1062 | 8 | } |
1063 | 8 | for (const auto& projection : request.non_predicate_columns) { |
1064 | 1 | collect_projection(projection); |
1065 | 1 | } |
1066 | 8 | } |
1067 | | |
1068 | | template <typename ValueType> |
1069 | | bool set_native_page_scalar_min_max(const tparquet::ColumnIndex& column_index, |
1070 | | const ParquetColumnSchema& column_schema, size_t page_idx, |
1071 | | DecodedValueKind kind, ParquetColumnStatistics* page_statistics, |
1072 | 124 | const cctz::time_zone* timezone) { |
1073 | 124 | if (page_idx >= column_index.min_values.size() || page_idx >= column_index.max_values.size() || |
1074 | 124 | column_index.min_values[page_idx].size() != sizeof(ValueType) || |
1075 | 124 | column_index.max_values[page_idx].size() != sizeof(ValueType)) { |
1076 | 1 | return false; |
1077 | 1 | } |
1078 | 123 | const auto min_value = unaligned_load<ValueType>(column_index.min_values[page_idx].data()); |
1079 | 123 | const auto max_value = unaligned_load<ValueType>(column_index.max_values[page_idx].data()); |
1080 | 123 | if constexpr (std::is_integral_v<ValueType>) { |
1081 | 123 | if (remove_nullable(column_schema.type)->get_primitive_type() == TYPE_TIMEV2) { |
1082 | 2 | int64_t units_per_day = 0; |
1083 | 2 | switch (column_schema.type_descriptor.time_unit) { |
1084 | 2 | case ParquetTimeUnit::MILLIS: |
1085 | 2 | units_per_day = 86400000; |
1086 | 2 | break; |
1087 | 0 | case ParquetTimeUnit::MICROS: |
1088 | 0 | units_per_day = 86400000000; |
1089 | 0 | break; |
1090 | 0 | case ParquetTimeUnit::NANOS: |
1091 | 0 | units_per_day = 86400000000000; |
1092 | 0 | break; |
1093 | 0 | default: |
1094 | 0 | return false; |
1095 | 2 | } |
1096 | | // TIME statistics are pruning proofs. Validate the raw carrier before rescaling so an |
1097 | | // invalid bound at or beyond 24:00 cannot publish a misleading ZoneMap. |
1098 | 2 | if (min_value < 0 || max_value < 0 || min_value >= units_per_day || |
1099 | 2 | max_value >= units_per_day) { |
1100 | 2 | return false; |
1101 | 2 | } |
1102 | 2 | } |
1103 | 123 | } |
1104 | 121 | if constexpr (std::is_same_v<ValueType, int64_t>) { |
1105 | 0 | if (!timestamp_min_max_is_safe(column_schema, min_value, max_value, timezone)) { |
1106 | 0 | return false; |
1107 | 0 | } |
1108 | 0 | } |
1109 | 123 | if (!valid_min_max(min_value, max_value)) { |
1110 | 0 | return true; |
1111 | 0 | } |
1112 | 123 | if (!set_decoded_field(column_schema, kind, min_value, &page_statistics->min_value, timezone) || |
1113 | 123 | !set_decoded_field(column_schema, kind, max_value, &page_statistics->max_value, timezone)) { |
1114 | 2 | return false; |
1115 | 2 | } |
1116 | 121 | if (decoded_min_max_is_ordered(*page_statistics)) { |
1117 | 119 | page_statistics->has_min_max = true; |
1118 | 119 | } |
1119 | 121 | return true; |
1120 | 123 | } parquet_statistics.cpp:_ZN5doris6format7parquet12_GLOBAL__N_130set_native_page_scalar_min_maxIiEEbRKN8tparquet11ColumnIndexERKNS1_19ParquetColumnSchemaEmNS_16DecodedValueKindEPNS1_23ParquetColumnStatisticsEPKN4cctz9time_zoneE Line | Count | Source | 1072 | 124 | const cctz::time_zone* timezone) { | 1073 | 124 | if (page_idx >= column_index.min_values.size() || page_idx >= column_index.max_values.size() || | 1074 | 124 | column_index.min_values[page_idx].size() != sizeof(ValueType) || | 1075 | 124 | column_index.max_values[page_idx].size() != sizeof(ValueType)) { | 1076 | 1 | return false; | 1077 | 1 | } | 1078 | 123 | const auto min_value = unaligned_load<ValueType>(column_index.min_values[page_idx].data()); | 1079 | 123 | const auto max_value = unaligned_load<ValueType>(column_index.max_values[page_idx].data()); | 1080 | 123 | if constexpr (std::is_integral_v<ValueType>) { | 1081 | 123 | if (remove_nullable(column_schema.type)->get_primitive_type() == TYPE_TIMEV2) { | 1082 | 2 | int64_t units_per_day = 0; | 1083 | 2 | switch (column_schema.type_descriptor.time_unit) { | 1084 | 2 | case ParquetTimeUnit::MILLIS: | 1085 | 2 | units_per_day = 86400000; | 1086 | 2 | break; | 1087 | 0 | case ParquetTimeUnit::MICROS: | 1088 | 0 | units_per_day = 86400000000; | 1089 | 0 | break; | 1090 | 0 | case ParquetTimeUnit::NANOS: | 1091 | 0 | units_per_day = 86400000000000; | 1092 | 0 | break; | 1093 | 0 | default: | 1094 | 0 | return false; | 1095 | 2 | } | 1096 | | // TIME statistics are pruning proofs. Validate the raw carrier before rescaling so an | 1097 | | // invalid bound at or beyond 24:00 cannot publish a misleading ZoneMap. | 1098 | 2 | if (min_value < 0 || max_value < 0 || min_value >= units_per_day || | 1099 | 2 | max_value >= units_per_day) { | 1100 | 2 | return false; | 1101 | 2 | } | 1102 | 2 | } | 1103 | 123 | } | 1104 | | if constexpr (std::is_same_v<ValueType, int64_t>) { | 1105 | | if (!timestamp_min_max_is_safe(column_schema, min_value, max_value, timezone)) { | 1106 | | return false; | 1107 | | } | 1108 | | } | 1109 | 123 | if (!valid_min_max(min_value, max_value)) { | 1110 | 0 | return true; | 1111 | 0 | } | 1112 | 123 | if (!set_decoded_field(column_schema, kind, min_value, &page_statistics->min_value, timezone) || | 1113 | 123 | !set_decoded_field(column_schema, kind, max_value, &page_statistics->max_value, timezone)) { | 1114 | 2 | return false; | 1115 | 2 | } | 1116 | 121 | if (decoded_min_max_is_ordered(*page_statistics)) { | 1117 | 119 | page_statistics->has_min_max = true; | 1118 | 119 | } | 1119 | 121 | return true; | 1120 | 123 | } |
Unexecuted instantiation: parquet_statistics.cpp:_ZN5doris6format7parquet12_GLOBAL__N_130set_native_page_scalar_min_maxIlEEbRKN8tparquet11ColumnIndexERKNS1_19ParquetColumnSchemaEmNS_16DecodedValueKindEPNS1_23ParquetColumnStatisticsEPKN4cctz9time_zoneE Unexecuted instantiation: parquet_statistics.cpp:_ZN5doris6format7parquet12_GLOBAL__N_130set_native_page_scalar_min_maxIfEEbRKN8tparquet11ColumnIndexERKNS1_19ParquetColumnSchemaEmNS_16DecodedValueKindEPNS1_23ParquetColumnStatisticsEPKN4cctz9time_zoneE Unexecuted instantiation: parquet_statistics.cpp:_ZN5doris6format7parquet12_GLOBAL__N_130set_native_page_scalar_min_maxIdEEbRKN8tparquet11ColumnIndexERKNS1_19ParquetColumnSchemaEmNS_16DecodedValueKindEPNS1_23ParquetColumnStatisticsEPKN4cctz9time_zoneE |
1121 | | |
1122 | | bool set_native_page_boolean_min_max(const tparquet::ColumnIndex& column_index, |
1123 | | const ParquetColumnSchema& column_schema, size_t page_idx, |
1124 | | ParquetColumnStatistics* page_statistics, |
1125 | 2 | const cctz::time_zone* timezone) { |
1126 | 2 | if (page_idx >= column_index.min_values.size() || page_idx >= column_index.max_values.size() || |
1127 | 2 | column_index.min_values[page_idx].size() != 1 || |
1128 | 2 | column_index.max_values[page_idx].size() != 1) { |
1129 | 0 | return false; |
1130 | 0 | } |
1131 | | // Parquet BOOLEAN statistics use the same one-bit value representation as PLAIN pages; bits |
1132 | | // outside the value bit are padding and must not change false into true. |
1133 | 2 | const uint8_t min_value = static_cast<uint8_t>(column_index.min_values[page_idx][0]) & 1; |
1134 | 2 | const uint8_t max_value = static_cast<uint8_t>(column_index.max_values[page_idx][0]) & 1; |
1135 | 2 | if (!valid_min_max(min_value, max_value)) { |
1136 | 0 | return true; |
1137 | 0 | } |
1138 | 2 | if (!set_decoded_field(column_schema, DecodedValueKind::BOOL, min_value, |
1139 | 2 | &page_statistics->min_value, timezone) || |
1140 | 2 | !set_decoded_field(column_schema, DecodedValueKind::BOOL, max_value, |
1141 | 2 | &page_statistics->max_value, timezone)) { |
1142 | 0 | return false; |
1143 | 0 | } |
1144 | 2 | if (decoded_min_max_is_ordered(*page_statistics)) { |
1145 | 2 | page_statistics->has_min_max = true; |
1146 | 2 | } |
1147 | 2 | return true; |
1148 | 2 | } |
1149 | | |
1150 | | bool build_native_page_statistics(const tparquet::ColumnIndex& column_index, |
1151 | | const ParquetColumnSchema& column_schema, size_t page_idx, |
1152 | | int64_t page_rows, ParquetColumnStatistics* page_statistics, |
1153 | 128 | const cctz::time_zone* timezone) { |
1154 | 128 | DORIS_CHECK(page_statistics != nullptr); |
1155 | 128 | *page_statistics = {}; |
1156 | 128 | if (!column_index.__isset.null_counts || page_idx >= column_index.null_pages.size() || |
1157 | 128 | page_idx >= column_index.null_counts.size()) { |
1158 | 0 | return false; |
1159 | 0 | } |
1160 | 128 | const int64_t null_count = column_index.null_counts[page_idx]; |
1161 | 128 | const bool all_null = column_index.null_pages[page_idx]; |
1162 | 128 | if (page_rows < 0 || null_count < 0 || null_count > page_rows || |
1163 | 128 | all_null != (null_count == page_rows)) { |
1164 | | // The caller supplies the exact flat page or row-group span. Contradictory optional null |
1165 | | // metadata must disable pruning instead of turning a partial span into an all-null proof. |
1166 | 2 | return false; |
1167 | 2 | } |
1168 | 126 | page_statistics->has_null_count = true; |
1169 | 126 | page_statistics->has_null = null_count > 0; |
1170 | 126 | page_statistics->has_not_null = !all_null; |
1171 | 126 | if (!page_statistics->has_not_null) { |
1172 | 0 | return true; |
1173 | 0 | } |
1174 | 126 | switch (column_schema.type_descriptor.physical_type) { |
1175 | 2 | case tparquet::Type::BOOLEAN: |
1176 | 2 | return set_native_page_boolean_min_max(column_index, column_schema, page_idx, |
1177 | 2 | page_statistics, timezone); |
1178 | 124 | case tparquet::Type::INT32: |
1179 | 124 | return set_native_page_scalar_min_max<int32_t>( |
1180 | 124 | column_index, column_schema, page_idx, |
1181 | 124 | decoded_value_kind(column_schema.type_descriptor), page_statistics, timezone); |
1182 | 0 | case tparquet::Type::INT64: |
1183 | 0 | return set_native_page_scalar_min_max<int64_t>( |
1184 | 0 | column_index, column_schema, page_idx, |
1185 | 0 | decoded_value_kind(column_schema.type_descriptor), page_statistics, timezone); |
1186 | 0 | case tparquet::Type::FLOAT: |
1187 | 0 | return set_native_page_scalar_min_max<float>(column_index, column_schema, page_idx, |
1188 | 0 | DecodedValueKind::FLOAT, page_statistics, |
1189 | 0 | timezone); |
1190 | 0 | case tparquet::Type::DOUBLE: |
1191 | 0 | return set_native_page_scalar_min_max<double>(column_index, column_schema, page_idx, |
1192 | 0 | DecodedValueKind::DOUBLE, page_statistics, |
1193 | 0 | timezone); |
1194 | 0 | case tparquet::Type::BYTE_ARRAY: |
1195 | 0 | case tparquet::Type::FIXED_LEN_BYTE_ARRAY: { |
1196 | 0 | if (page_idx >= column_index.min_values.size() || |
1197 | 0 | page_idx >= column_index.max_values.size()) { |
1198 | 0 | return false; |
1199 | 0 | } |
1200 | 0 | const auto& min_value = column_index.min_values[page_idx]; |
1201 | 0 | const auto& max_value = column_index.max_values[page_idx]; |
1202 | 0 | const bool fixed = |
1203 | 0 | column_schema.type_descriptor.physical_type == tparquet::Type::FIXED_LEN_BYTE_ARRAY; |
1204 | 0 | if (fixed && |
1205 | 0 | (column_schema.type_descriptor.fixed_length <= 0 || |
1206 | 0 | min_value.size() != static_cast<size_t>(column_schema.type_descriptor.fixed_length) || |
1207 | 0 | max_value.size() != static_cast<size_t>(column_schema.type_descriptor.fixed_length))) { |
1208 | 0 | return false; |
1209 | 0 | } |
1210 | 0 | const auto kind = fixed ? DecodedValueKind::FIXED_BINARY : DecodedValueKind::BINARY; |
1211 | 0 | if (!set_decoded_binary_field(column_schema, kind, |
1212 | 0 | StringRef(min_value.data(), min_value.size()), |
1213 | 0 | &page_statistics->min_value, timezone) || |
1214 | 0 | !set_decoded_binary_field(column_schema, kind, |
1215 | 0 | StringRef(max_value.data(), max_value.size()), |
1216 | 0 | &page_statistics->max_value, timezone)) { |
1217 | 0 | return false; |
1218 | 0 | } |
1219 | 0 | if (decoded_min_max_is_ordered(*page_statistics)) { |
1220 | 0 | page_statistics->has_min_max = true; |
1221 | 0 | } |
1222 | 0 | return true; |
1223 | 0 | } |
1224 | 0 | default: |
1225 | 0 | return false; |
1226 | 126 | } |
1227 | 126 | } |
1228 | | |
1229 | | RowRange native_page_row_range(const tparquet::OffsetIndex& offset_index, size_t page_idx, |
1230 | 122 | int64_t row_group_rows) { |
1231 | 122 | const auto& locations = offset_index.page_locations; |
1232 | 122 | const int64_t start = locations[page_idx].first_row_index; |
1233 | 122 | const int64_t end = page_idx + 1 == locations.size() ? row_group_rows |
1234 | 122 | : locations[page_idx + 1].first_row_index; |
1235 | 122 | return {.start = start, .length = end - start}; |
1236 | 122 | } |
1237 | | |
1238 | | } // namespace |
1239 | | |
1240 | | Status select_row_group_ranges_by_native_page_index( |
1241 | | const tparquet::FileMetaData& metadata, |
1242 | | const std::unordered_map<int, NativeParquetPageIndex>& page_indexes, |
1243 | | const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema, |
1244 | | const format::FileScanRequest& request, int64_t row_group_rows, |
1245 | | std::vector<RowRange>* selected_ranges, std::map<int, ParquetPageSkipPlan>* page_skip_plans, |
1246 | | ParquetPruningStats* pruning_stats, const cctz::time_zone* timezone, |
1247 | 239 | const RuntimeState* runtime_state) { |
1248 | 239 | int64_t filter_time_sink = 0; |
1249 | 239 | SCOPED_RAW_TIMER(pruning_stats == nullptr ? &filter_time_sink |
1250 | 239 | : &pruning_stats->page_index_filter_time); |
1251 | 239 | DORIS_CHECK(selected_ranges != nullptr); |
1252 | 239 | selected_ranges->clear(); |
1253 | 239 | selected_ranges->push_back({.start = 0, .length = row_group_rows}); |
1254 | 239 | if (page_skip_plans != nullptr) { |
1255 | 239 | page_skip_plans->clear(); |
1256 | 239 | } |
1257 | 239 | if (row_group_rows <= 0 || !config::enable_parquet_page_index || |
1258 | 239 | !has_expr_zonemap_filter(request, runtime_state) || page_indexes.empty()) { |
1259 | 230 | return Status::OK(); |
1260 | 230 | } |
1261 | 9 | if (pruning_stats != nullptr) { |
1262 | 3 | ++pruning_stats->page_index_read_calls; |
1263 | 3 | } |
1264 | | |
1265 | 9 | std::map<int, VExprContextSPtrs> conjuncts_by_slot; |
1266 | 9 | for (const auto& conjunct : request.conjuncts) { |
1267 | 9 | const auto slot_index = expr_zonemap::single_slot_zonemap_index(conjunct); |
1268 | 9 | if (slot_index >= 0) { |
1269 | 9 | conjuncts_by_slot[slot_index].push_back(conjunct); |
1270 | 9 | } |
1271 | 9 | } |
1272 | 9 | for (const auto& [slot_index, conjuncts] : conjuncts_by_slot) { |
1273 | 9 | const auto file_column_id = file_column_id_by_block_position(request, slot_index); |
1274 | 9 | if (!file_column_id.has_value()) { |
1275 | 0 | continue; |
1276 | 0 | } |
1277 | 9 | const auto* column_schema = resolve_local_leaf_schema(file_schema, *file_column_id); |
1278 | 9 | if (column_schema == nullptr || column_schema->type == nullptr || |
1279 | 9 | !native_metadata_predicate_is_type_safe(*column_schema) || |
1280 | 9 | !detail::has_supported_type_defined_order(metadata, column_schema->leaf_column_id)) { |
1281 | 1 | continue; |
1282 | 1 | } |
1283 | 8 | const auto index_it = page_indexes.find(column_schema->leaf_column_id); |
1284 | 8 | if (index_it == page_indexes.end()) { |
1285 | 0 | continue; |
1286 | 0 | } |
1287 | 8 | const auto& indexes = index_it->second; |
1288 | 8 | std::vector<RowRange> filter_ranges; |
1289 | 8 | bool usable = true; |
1290 | 58 | for (size_t page_idx = 0; page_idx < indexes.offset_index.page_locations.size(); |
1291 | 53 | ++page_idx) { |
1292 | 53 | const auto page_range = |
1293 | 53 | native_page_row_range(indexes.offset_index, page_idx, row_group_rows); |
1294 | 53 | ParquetColumnStatistics statistics; |
1295 | 53 | if (!build_native_page_statistics(indexes.column_index, *column_schema, page_idx, |
1296 | 53 | page_range.length, &statistics, timezone)) { |
1297 | 3 | usable = false; |
1298 | 3 | break; |
1299 | 3 | } |
1300 | 50 | ZoneMapEvalContext ctx; |
1301 | 50 | add_slot_zonemap(&ctx, slot_index, column_schema->type, |
1302 | 50 | ParquetStatisticsUtils::MakeZoneMap(statistics)); |
1303 | 50 | if (VExprContext::evaluate_zonemap_filter(conjuncts, ctx) != |
1304 | 50 | ZoneMapFilterResult::kNoMatch) { |
1305 | 33 | append_row_range(page_range, &filter_ranges); |
1306 | 33 | } |
1307 | 50 | if (pruning_stats != nullptr) { |
1308 | 48 | pruning_stats->expr_zonemap_unusable_evals += ctx.stats.unusable_zonemap_eval_count; |
1309 | 48 | pruning_stats->in_zonemap_point_check_count += |
1310 | 48 | ctx.stats.in_zonemap_point_check_count; |
1311 | 48 | pruning_stats->in_zonemap_range_only_count += ctx.stats.in_zonemap_range_only_count; |
1312 | 48 | } |
1313 | 50 | } |
1314 | 8 | if (!usable) { |
1315 | 3 | continue; |
1316 | 3 | } |
1317 | 5 | *selected_ranges = intersect_ranges(*selected_ranges, filter_ranges); |
1318 | 5 | if (selected_ranges->empty()) { |
1319 | 1 | if (pruning_stats != nullptr) { |
1320 | 0 | pruning_stats->filtered_page_rows += row_group_rows; |
1321 | 0 | ++pruning_stats->filtered_row_groups_by_page_index; |
1322 | 0 | } |
1323 | 1 | return Status::OK(); |
1324 | 1 | } |
1325 | 5 | } |
1326 | | |
1327 | 8 | if (page_skip_plans != nullptr) { |
1328 | 8 | std::vector<const ParquetColumnSchema*> leaves; |
1329 | 8 | collect_request_leaf_schemas(file_schema, request, &leaves); |
1330 | 9 | for (const auto* leaf : leaves) { |
1331 | 9 | const auto index_it = page_indexes.find(leaf->leaf_column_id); |
1332 | 9 | if (index_it == page_indexes.end() || leaf->max_repetition_level != 0) { |
1333 | 0 | continue; |
1334 | 0 | } |
1335 | 9 | const auto& offset_index = index_it->second.offset_index; |
1336 | 9 | ParquetPageSkipPlan skip_plan; |
1337 | 9 | skip_plan.leaf_column_id = leaf->leaf_column_id; |
1338 | 9 | skip_plan.skipped_pages.resize(offset_index.page_locations.size()); |
1339 | 9 | skip_plan.skipped_page_compressed_sizes.resize(offset_index.page_locations.size()); |
1340 | 78 | for (size_t page_idx = 0; page_idx < offset_index.page_locations.size(); ++page_idx) { |
1341 | 69 | const auto range = native_page_row_range(offset_index, page_idx, row_group_rows); |
1342 | 69 | if (range.length == 0 || ranges_intersect(*selected_ranges, range)) { |
1343 | 45 | continue; |
1344 | 45 | } |
1345 | 24 | skip_plan.skipped_pages[page_idx] = 1; |
1346 | 24 | skip_plan.skipped_page_compressed_sizes[page_idx] = |
1347 | 24 | offset_index.page_locations[page_idx].compressed_page_size; |
1348 | 24 | append_row_range(range, &skip_plan.skipped_ranges); |
1349 | 24 | } |
1350 | 9 | if (!skip_plan.empty()) { |
1351 | 3 | page_skip_plans->emplace(skip_plan.leaf_column_id, std::move(skip_plan)); |
1352 | 3 | } |
1353 | 9 | } |
1354 | 8 | } |
1355 | 8 | if (pruning_stats != nullptr) { |
1356 | 3 | pruning_stats->filtered_page_rows += row_group_rows - count_range_rows(*selected_ranges); |
1357 | 3 | } |
1358 | 8 | return Status::OK(); |
1359 | 9 | } |
1360 | | |
1361 | | } // namespace doris::format::parquet |