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 <cmath> |
29 | | #include <cstddef> |
30 | | #include <cstring> |
31 | | #include <exception> |
32 | | #include <limits> |
33 | | #include <map> |
34 | | #include <memory> |
35 | | #include <optional> |
36 | | #include <set> |
37 | | #include <string> |
38 | | #include <type_traits> |
39 | | #include <utility> |
40 | | #include <vector> |
41 | | |
42 | | #include "common/cast_set.h" |
43 | | #include "common/config.h" |
44 | | #include "core/data_type/data_type.h" |
45 | | #include "core/data_type/data_type_nullable.h" |
46 | | #include "core/data_type_serde/data_type_serde.h" |
47 | | #include "core/field.h" |
48 | | #include "exprs/expr_zonemap_filter.h" |
49 | | #include "exprs/vexpr_context.h" |
50 | | #include "format_v2/parquet/parquet_column_schema.h" |
51 | | #include "format_v2/parquet/parquet_file_context.h" |
52 | | #include "format_v2/parquet/reader/native/block_split_bloom_filter.h" |
53 | | #include "format_v2/parquet/reader/native_column_reader.h" |
54 | | #include "format_v2/timestamp_statistics.h" |
55 | | #include "runtime/runtime_profile.h" |
56 | | #include "storage/index/bloom_filter/bloom_filter.h" |
57 | | #include "storage/index/zone_map/zone_map_index.h" |
58 | | #include "storage/index/zone_map/zonemap_eval_context.h" |
59 | | #include "util/thrift_util.h" |
60 | | #include "util/unaligned.h" |
61 | | |
62 | | namespace doris::format::parquet { |
63 | | |
64 | | namespace detail { |
65 | | |
66 | | Status validate_native_bloom_filter_layout(int64_t offset, uint32_t header_size, |
67 | | int64_t payload_size, int64_t declared_length, |
68 | 7 | size_t file_size) { |
69 | 7 | if (offset < 0 || header_size == 0 || payload_size < segment_v2::BloomFilter::MINIMUM_BYTES || |
70 | 7 | payload_size > segment_v2::BloomFilter::MAXIMUM_BYTES || payload_size % 32 != 0) { |
71 | 3 | return Status::Corruption( |
72 | 3 | "Invalid Parquet Bloom filter layout: offset {}, header {}, payload {}", offset, |
73 | 3 | header_size, payload_size); |
74 | 3 | } |
75 | 4 | const uint64_t unsigned_offset = static_cast<uint64_t>(offset); |
76 | 4 | const uint64_t total_size = static_cast<uint64_t>(header_size) + payload_size; |
77 | 4 | if (unsigned_offset > file_size || total_size > file_size - unsigned_offset) { |
78 | 1 | return Status::Corruption("Parquet Bloom filter range exceeds file size {}", file_size); |
79 | 1 | } |
80 | 3 | if (declared_length >= 0) { |
81 | 3 | const uint64_t unsigned_declared_length = static_cast<uint64_t>(declared_length); |
82 | 3 | if (unsigned_declared_length < total_size || |
83 | 3 | unsigned_declared_length > file_size - unsigned_offset) { |
84 | 1 | return Status::Corruption( |
85 | 1 | "Parquet Bloom filter requires {} bytes, metadata declares {}, file has {}", |
86 | 1 | total_size, declared_length, file_size - unsigned_offset); |
87 | 1 | } |
88 | 3 | } |
89 | 2 | return Status::OK(); |
90 | 3 | } |
91 | | |
92 | | bool can_use_native_footer_min_max(const ParquetTypeDescriptor& type_descriptor, |
93 | 55 | const tparquet::Statistics& statistics) { |
94 | 55 | const bool binary = type_descriptor.physical_type == ::parquet::Type::BYTE_ARRAY || |
95 | 55 | type_descriptor.physical_type == ::parquet::Type::FIXED_LEN_BYTE_ARRAY; |
96 | 55 | if (!binary) { |
97 | 50 | return true; |
98 | 50 | } |
99 | 5 | if (statistics.__isset.min_value || statistics.__isset.max_value) { |
100 | | // Do not combine a type-defined bound with a deprecated bound: the two fields can use |
101 | | // different byte ordering, so a mixed pair cannot form one valid interval. |
102 | 2 | return statistics.__isset.min_value && statistics.__isset.max_value; |
103 | 2 | } |
104 | | // Deprecated binary min/max fields were ordered with signed bytes by legacy parquet-mr, while |
105 | | // Parquet's type-defined order is unsigned. Only an equal pair is independent of that mismatch. |
106 | 3 | return statistics.__isset.min && statistics.__isset.max && statistics.min == statistics.max; |
107 | 5 | } |
108 | | |
109 | | } // namespace detail |
110 | | |
111 | | namespace { |
112 | | |
113 | | bool build_native_page_statistics(const tparquet::ColumnIndex& column_index, |
114 | | const ParquetColumnSchema& column_schema, size_t page_idx, |
115 | | ParquetColumnStatistics* page_statistics, |
116 | | const cctz::time_zone* timezone); |
117 | | |
118 | | enum class ParquetRowGroupPruneReason { |
119 | | NONE, // cannot prune; must read |
120 | | STATISTICS, // excluded by ZoneMap statistics |
121 | | DICTIONARY, // excluded by dictionary |
122 | | BLOOM_FILTER, // excluded by bloom filter |
123 | | }; |
124 | | |
125 | | Status read_native_bloom_filter(const tparquet::ColumnMetaData& metadata, |
126 | | const io::FileReaderSPtr& file, io::IOContext* io_ctx, |
127 | 1 | std::unique_ptr<native::BlockSplitBloomFilter>* result) { |
128 | 1 | if (result == nullptr || file == nullptr || !metadata.__isset.bloom_filter_offset) { |
129 | 0 | return Status::NotSupported("Parquet Bloom filter is unavailable"); |
130 | 0 | } |
131 | 1 | constexpr size_t MAX_BLOOM_HEADER_BYTES = 64; |
132 | 1 | if (metadata.bloom_filter_offset < 0 || |
133 | 1 | (metadata.__isset.bloom_filter_length && metadata.bloom_filter_length <= 0)) { |
134 | 0 | return Status::Corruption("Invalid Parquet Bloom filter offset or declared length"); |
135 | 0 | } |
136 | 1 | const uint64_t bloom_offset = static_cast<uint64_t>(metadata.bloom_filter_offset); |
137 | 1 | if (bloom_offset >= file->size()) { |
138 | 0 | return Status::Corruption("Parquet Bloom filter offset exceeds file size {}", file->size()); |
139 | 0 | } |
140 | 1 | const size_t available = file->size() - bloom_offset; |
141 | 1 | const size_t declared_available = |
142 | 1 | metadata.__isset.bloom_filter_length |
143 | 1 | ? std::min<size_t>(metadata.bloom_filter_length, available) |
144 | 1 | : available; |
145 | 1 | const size_t header_read_size = std::min(declared_available, MAX_BLOOM_HEADER_BYTES); |
146 | 1 | std::vector<uint8_t> header_buffer(header_read_size); |
147 | 1 | size_t bytes_read = 0; |
148 | 1 | RETURN_IF_ERROR(file->read_at(metadata.bloom_filter_offset, |
149 | 1 | Slice(header_buffer.data(), header_buffer.size()), &bytes_read, |
150 | 1 | io_ctx)); |
151 | 1 | tparquet::BloomFilterHeader header; |
152 | 1 | uint32_t header_size = cast_set<uint32_t>(bytes_read); |
153 | 1 | RETURN_IF_ERROR(deserialize_thrift_msg(header_buffer.data(), &header_size, true, &header)); |
154 | 1 | if (!header.algorithm.__isset.BLOCK || !header.compression.__isset.UNCOMPRESSED || |
155 | 1 | !header.hash.__isset.XXHASH || header.numBytes <= 0) { |
156 | 0 | return Status::NotSupported("Unsupported Parquet Bloom filter encoding"); |
157 | 0 | } |
158 | | |
159 | | // Validate the complete split-block layout before allocating or adding footer-controlled |
160 | | // offsets; BloomFilter::init() otherwise receives a truncated or oversized backing buffer. |
161 | 1 | RETURN_IF_ERROR(detail::validate_native_bloom_filter_layout( |
162 | 1 | metadata.bloom_filter_offset, header_size, header.numBytes, |
163 | 1 | metadata.__isset.bloom_filter_length ? metadata.bloom_filter_length : -1, |
164 | 1 | file->size())); |
165 | | |
166 | 1 | std::vector<uint8_t> data(cast_set<size_t>(header.numBytes)); |
167 | 1 | RETURN_IF_ERROR(file->read_at(static_cast<size_t>(metadata.bloom_filter_offset) + header_size, |
168 | 1 | Slice(data.data(), data.size()), &bytes_read, io_ctx)); |
169 | 1 | if (bytes_read != data.size()) { |
170 | 0 | return Status::Corruption("Truncated Parquet Bloom filter payload"); |
171 | 0 | } |
172 | 1 | auto bloom_filter = std::make_unique<native::BlockSplitBloomFilter>(); |
173 | 1 | RETURN_IF_ERROR(bloom_filter->init(reinterpret_cast<const char*>(data.data()), data.size(), |
174 | 1 | segment_v2::HashStrategyPB::XX_HASH_64)); |
175 | 1 | *result = std::move(bloom_filter); |
176 | 1 | return Status::OK(); |
177 | 1 | } |
178 | | |
179 | 16 | bool bloom_logical_type_supported(const ParquetColumnSchema& column_schema) { |
180 | 16 | if (column_schema.type == nullptr) { |
181 | 0 | return false; |
182 | 0 | } |
183 | 16 | switch (remove_nullable(column_schema.type)->get_primitive_type()) { |
184 | 0 | case TYPE_BOOLEAN: |
185 | 6 | case TYPE_INT: |
186 | 13 | case TYPE_BIGINT: |
187 | 14 | case TYPE_FLOAT: |
188 | 14 | case TYPE_DOUBLE: |
189 | 16 | case TYPE_STRING: |
190 | 16 | return true; |
191 | 0 | default: |
192 | 0 | return false; |
193 | 16 | } |
194 | 16 | } |
195 | | |
196 | 412 | DecodedTimeUnit decoded_time_unit(ParquetTimeUnit time_unit) { |
197 | 412 | switch (time_unit) { |
198 | 0 | case ParquetTimeUnit::MILLIS: |
199 | 0 | return DecodedTimeUnit::MILLIS; |
200 | 6 | case ParquetTimeUnit::MICROS: |
201 | 6 | return DecodedTimeUnit::MICROS; |
202 | 0 | case ParquetTimeUnit::NANOS: |
203 | 0 | return DecodedTimeUnit::NANOS; |
204 | 406 | default: |
205 | 406 | return DecodedTimeUnit::UNKNOWN; |
206 | 412 | } |
207 | 412 | } |
208 | | |
209 | | Status read_decoded_field(const ParquetColumnSchema& column_schema, DecodedColumnView view, |
210 | 412 | Field* field, const cctz::time_zone* timezone) { |
211 | 412 | DORIS_CHECK(column_schema.type != nullptr); |
212 | 412 | DORIS_CHECK(field != nullptr); |
213 | 412 | constexpr uint8_t not_null = 0; |
214 | 412 | view.row_count = 1; |
215 | 412 | view.null_map = ¬_null; |
216 | 412 | view.time_unit = decoded_time_unit(column_schema.type_descriptor.time_unit); |
217 | 412 | view.logical_integer_bit_width = column_schema.type_descriptor.integer_bit_width; |
218 | 412 | view.logical_integer_is_signed = !column_schema.type_descriptor.is_unsigned_integer; |
219 | 412 | view.decimal_precision = column_schema.type_descriptor.decimal_precision; |
220 | 412 | view.decimal_scale = column_schema.type_descriptor.decimal_scale; |
221 | 412 | view.fixed_length = column_schema.type_descriptor.fixed_length; |
222 | 412 | view.timestamp_is_adjusted_to_utc = column_schema.type_descriptor.timestamp_is_adjusted_to_utc; |
223 | 412 | view.timezone = timezone; |
224 | 412 | return column_schema.type->get_serde()->read_field_from_decoded_value(*column_schema.type, |
225 | 412 | field, view); |
226 | 412 | } |
227 | | |
228 | | template <typename NativeType> |
229 | | bool set_decoded_field(const ParquetColumnSchema& column_schema, DecodedValueKind value_kind, |
230 | 408 | const NativeType& value, Field* field, const cctz::time_zone* timezone) { |
231 | 408 | DecodedColumnView view; |
232 | 408 | view.value_kind = value_kind; |
233 | 408 | view.values = reinterpret_cast<const uint8_t*>(&value); |
234 | 408 | return read_decoded_field(column_schema, view, field, timezone).ok(); |
235 | 408 | } 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 | 230 | 396 | const NativeType& value, Field* field, const cctz::time_zone* timezone) { | 231 | 396 | DecodedColumnView view; | 232 | 396 | view.value_kind = value_kind; | 233 | 396 | view.values = reinterpret_cast<const uint8_t*>(&value); | 234 | 396 | return read_decoded_field(column_schema, view, field, timezone).ok(); | 235 | 396 | } |
parquet_statistics.cpp:_ZN5doris6format7parquet12_GLOBAL__N_117set_decoded_fieldIlEEbRKNS1_19ParquetColumnSchemaENS_16DecodedValueKindERKT_PNS_5FieldEPKN4cctz9time_zoneE Line | Count | Source | 230 | 6 | const NativeType& value, Field* field, const cctz::time_zone* timezone) { | 231 | 6 | DecodedColumnView view; | 232 | 6 | view.value_kind = value_kind; | 233 | 6 | view.values = reinterpret_cast<const uint8_t*>(&value); | 234 | 6 | return read_decoded_field(column_schema, view, field, timezone).ok(); | 235 | 6 | } |
Unexecuted instantiation: parquet_statistics.cpp:_ZN5doris6format7parquet12_GLOBAL__N_117set_decoded_fieldIfEEbRKNS1_19ParquetColumnSchemaENS_16DecodedValueKindERKT_PNS_5FieldEPKN4cctz9time_zoneE parquet_statistics.cpp:_ZN5doris6format7parquet12_GLOBAL__N_117set_decoded_fieldIdEEbRKNS1_19ParquetColumnSchemaENS_16DecodedValueKindERKT_PNS_5FieldEPKN4cctz9time_zoneE Line | Count | Source | 230 | 6 | const NativeType& value, Field* field, const cctz::time_zone* timezone) { | 231 | 6 | DecodedColumnView view; | 232 | 6 | view.value_kind = value_kind; | 233 | 6 | view.values = reinterpret_cast<const uint8_t*>(&value); | 234 | 6 | return read_decoded_field(column_schema, view, field, timezone).ok(); | 235 | 6 | } |
Unexecuted instantiation: parquet_statistics.cpp:_ZN5doris6format7parquet12_GLOBAL__N_117set_decoded_fieldIhEEbRKNS1_19ParquetColumnSchemaENS_16DecodedValueKindERKT_PNS_5FieldEPKN4cctz9time_zoneE |
236 | | |
237 | 6 | int64_t floor_timestamp_seconds(int64_t value, ParquetTimeUnit time_unit) { |
238 | 6 | int64_t units_per_second = 1; |
239 | 6 | switch (time_unit) { |
240 | 0 | case ParquetTimeUnit::MILLIS: |
241 | 0 | units_per_second = 1000; |
242 | 0 | break; |
243 | 6 | case ParquetTimeUnit::MICROS: |
244 | 6 | units_per_second = 1000000; |
245 | 6 | break; |
246 | 0 | case ParquetTimeUnit::NANOS: |
247 | 0 | units_per_second = 1000000000; |
248 | 0 | break; |
249 | 0 | default: |
250 | 0 | DORIS_CHECK(false); |
251 | 6 | } |
252 | 6 | return format::floor_epoch_seconds(value, units_per_second); |
253 | 6 | } |
254 | | |
255 | | bool timestamp_min_max_is_safe(const ParquetColumnSchema& column_schema, int64_t min_value, |
256 | 6 | int64_t max_value, const cctz::time_zone* timezone) { |
257 | 6 | if (min_value > max_value) { |
258 | 2 | return false; |
259 | 2 | } |
260 | 4 | if (!column_schema.type_descriptor.is_timestamp || |
261 | 4 | !column_schema.type_descriptor.timestamp_is_adjusted_to_utc || timezone == nullptr || |
262 | 4 | remove_nullable(column_schema.type)->get_primitive_type() == TYPE_TIMESTAMPTZ) { |
263 | | // TIMESTAMPTZ keeps the original UTC ordering, so local civil-time rollback does not make |
264 | | // its converted min/max non-monotonic. |
265 | 1 | return true; |
266 | 1 | } |
267 | 3 | return format::utc_timestamp_range_is_monotonic( |
268 | 3 | floor_timestamp_seconds(min_value, column_schema.type_descriptor.time_unit), |
269 | 3 | floor_timestamp_seconds(max_value, column_schema.type_descriptor.time_unit), *timezone); |
270 | 4 | } |
271 | | |
272 | | template <typename NativeType> |
273 | 211 | bool valid_min_max(const NativeType& min_value, const NativeType& max_value) { |
274 | 211 | if constexpr (std::is_floating_point_v<NativeType>) { |
275 | | // Parquet requires readers to ignore min/max statistics if either bound is NaN. |
276 | 10 | if (std::isnan(min_value) || std::isnan(max_value)) { |
277 | 7 | return false; |
278 | 7 | } |
279 | 10 | } |
280 | 3 | return true; |
281 | 211 | } Unexecuted instantiation: parquet_statistics.cpp:_ZN5doris6format7parquet12_GLOBAL__N_113valid_min_maxIbEEbRKT_S6_ parquet_statistics.cpp:_ZN5doris6format7parquet12_GLOBAL__N_113valid_min_maxIiEEbRKT_S6_ Line | Count | Source | 273 | 198 | bool valid_min_max(const NativeType& min_value, const NativeType& max_value) { | 274 | | if constexpr (std::is_floating_point_v<NativeType>) { | 275 | | // Parquet requires readers to ignore min/max statistics if either bound is NaN. | 276 | | if (std::isnan(min_value) || std::isnan(max_value)) { | 277 | | return false; | 278 | | } | 279 | | } | 280 | 198 | return true; | 281 | 198 | } |
parquet_statistics.cpp:_ZN5doris6format7parquet12_GLOBAL__N_113valid_min_maxIlEEbRKT_S6_ Line | Count | Source | 273 | 3 | bool valid_min_max(const NativeType& min_value, const NativeType& max_value) { | 274 | | if constexpr (std::is_floating_point_v<NativeType>) { | 275 | | // Parquet requires readers to ignore min/max statistics if either bound is NaN. | 276 | | if (std::isnan(min_value) || std::isnan(max_value)) { | 277 | | return false; | 278 | | } | 279 | | } | 280 | 3 | return true; | 281 | 3 | } |
parquet_statistics.cpp:_ZN5doris6format7parquet12_GLOBAL__N_113valid_min_maxIfEEbRKT_S6_ Line | Count | Source | 273 | 2 | bool valid_min_max(const NativeType& min_value, const NativeType& max_value) { | 274 | 2 | if constexpr (std::is_floating_point_v<NativeType>) { | 275 | | // Parquet requires readers to ignore min/max statistics if either bound is NaN. | 276 | 2 | if (std::isnan(min_value) || std::isnan(max_value)) { | 277 | 2 | return false; | 278 | 2 | } | 279 | 2 | } | 280 | 0 | return true; | 281 | 2 | } |
parquet_statistics.cpp:_ZN5doris6format7parquet12_GLOBAL__N_113valid_min_maxIdEEbRKT_S6_ Line | Count | Source | 273 | 8 | bool valid_min_max(const NativeType& min_value, const NativeType& max_value) { | 274 | 8 | if constexpr (std::is_floating_point_v<NativeType>) { | 275 | | // Parquet requires readers to ignore min/max statistics if either bound is NaN. | 276 | 8 | if (std::isnan(min_value) || std::isnan(max_value)) { | 277 | 5 | return false; | 278 | 5 | } | 279 | 8 | } | 280 | 3 | return true; | 281 | 8 | } |
Unexecuted instantiation: parquet_statistics.cpp:_ZN5doris6format7parquet12_GLOBAL__N_113valid_min_maxIhEEbRKT_S6_ |
282 | | |
283 | 206 | bool decoded_min_max_is_ordered(const ParquetColumnStatistics& column_statistics) { |
284 | 206 | return !(column_statistics.max_value < column_statistics.min_value); |
285 | 206 | } |
286 | | |
287 | | template <typename ParquetDType> |
288 | | bool set_decoded_min_max(const std::shared_ptr<::parquet::Statistics>& statistics, |
289 | | const ParquetColumnSchema& column_schema, DecodedValueKind value_kind, |
290 | | ParquetColumnStatistics* column_statistics, |
291 | 29 | const cctz::time_zone* timezone) { |
292 | 29 | auto typed_statistics = |
293 | 29 | std::static_pointer_cast<::parquet::TypedStatistics<ParquetDType>>(statistics); |
294 | 29 | const auto& min_value = typed_statistics->min(); |
295 | 29 | const auto& max_value = typed_statistics->max(); |
296 | 29 | if constexpr (std::is_same_v<ParquetDType, ::parquet::Int64Type>) { |
297 | 6 | if (!timestamp_min_max_is_safe(column_schema, min_value, max_value, timezone)) { |
298 | 3 | return false; |
299 | 3 | } |
300 | 6 | } |
301 | 29 | if (!valid_min_max(min_value, max_value) || |
302 | 29 | !set_decoded_field(column_schema, value_kind, min_value, &column_statistics->min_value, |
303 | 23 | timezone) || |
304 | 29 | !set_decoded_field(column_schema, value_kind, max_value, &column_statistics->max_value, |
305 | 23 | timezone)) { |
306 | 3 | return false; |
307 | 3 | } |
308 | 26 | return decoded_min_max_is_ordered(*column_statistics); |
309 | 29 | } 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 | 291 | 19 | const cctz::time_zone* timezone) { | 292 | 19 | auto typed_statistics = | 293 | 19 | std::static_pointer_cast<::parquet::TypedStatistics<ParquetDType>>(statistics); | 294 | 19 | const auto& min_value = typed_statistics->min(); | 295 | 19 | const auto& max_value = typed_statistics->max(); | 296 | | if constexpr (std::is_same_v<ParquetDType, ::parquet::Int64Type>) { | 297 | | if (!timestamp_min_max_is_safe(column_schema, min_value, max_value, timezone)) { | 298 | | return false; | 299 | | } | 300 | | } | 301 | 19 | if (!valid_min_max(min_value, max_value) || | 302 | 19 | !set_decoded_field(column_schema, value_kind, min_value, &column_statistics->min_value, | 303 | 19 | timezone) || | 304 | 19 | !set_decoded_field(column_schema, value_kind, max_value, &column_statistics->max_value, | 305 | 19 | timezone)) { | 306 | 0 | return false; | 307 | 0 | } | 308 | 19 | return decoded_min_max_is_ordered(*column_statistics); | 309 | 19 | } |
parquet_statistics.cpp:_ZN5doris6format7parquet12_GLOBAL__N_119set_decoded_min_maxIN7parquet12PhysicalTypeILNS4_4Type4typeE2EEEEEbRKSt10shared_ptrINS4_10StatisticsEERKNS1_19ParquetColumnSchemaENS_16DecodedValueKindEPNS1_23ParquetColumnStatisticsEPKN4cctz9time_zoneE Line | Count | Source | 291 | 6 | const cctz::time_zone* timezone) { | 292 | 6 | auto typed_statistics = | 293 | 6 | std::static_pointer_cast<::parquet::TypedStatistics<ParquetDType>>(statistics); | 294 | 6 | const auto& min_value = typed_statistics->min(); | 295 | 6 | const auto& max_value = typed_statistics->max(); | 296 | 6 | if constexpr (std::is_same_v<ParquetDType, ::parquet::Int64Type>) { | 297 | 6 | if (!timestamp_min_max_is_safe(column_schema, min_value, max_value, timezone)) { | 298 | 3 | return false; | 299 | 3 | } | 300 | 6 | } | 301 | 6 | if (!valid_min_max(min_value, max_value) || | 302 | 6 | !set_decoded_field(column_schema, value_kind, min_value, &column_statistics->min_value, | 303 | 3 | timezone) || | 304 | 6 | !set_decoded_field(column_schema, value_kind, max_value, &column_statistics->max_value, | 305 | 3 | timezone)) { | 306 | 0 | return false; | 307 | 0 | } | 308 | 6 | return decoded_min_max_is_ordered(*column_statistics); | 309 | 6 | } |
parquet_statistics.cpp:_ZN5doris6format7parquet12_GLOBAL__N_119set_decoded_min_maxIN7parquet12PhysicalTypeILNS4_4Type4typeE4EEEEEbRKSt10shared_ptrINS4_10StatisticsEERKNS1_19ParquetColumnSchemaENS_16DecodedValueKindEPNS1_23ParquetColumnStatisticsEPKN4cctz9time_zoneE Line | Count | Source | 291 | 1 | const cctz::time_zone* timezone) { | 292 | 1 | auto typed_statistics = | 293 | 1 | std::static_pointer_cast<::parquet::TypedStatistics<ParquetDType>>(statistics); | 294 | 1 | const auto& min_value = typed_statistics->min(); | 295 | 1 | const auto& max_value = typed_statistics->max(); | 296 | | if constexpr (std::is_same_v<ParquetDType, ::parquet::Int64Type>) { | 297 | | if (!timestamp_min_max_is_safe(column_schema, min_value, max_value, timezone)) { | 298 | | return false; | 299 | | } | 300 | | } | 301 | 1 | if (!valid_min_max(min_value, max_value) || | 302 | 1 | !set_decoded_field(column_schema, value_kind, min_value, &column_statistics->min_value, | 303 | 0 | timezone) || | 304 | 1 | !set_decoded_field(column_schema, value_kind, max_value, &column_statistics->max_value, | 305 | 1 | timezone)) { | 306 | 1 | return false; | 307 | 1 | } | 308 | 0 | return decoded_min_max_is_ordered(*column_statistics); | 309 | 1 | } |
parquet_statistics.cpp:_ZN5doris6format7parquet12_GLOBAL__N_119set_decoded_min_maxIN7parquet12PhysicalTypeILNS4_4Type4typeE5EEEEEbRKSt10shared_ptrINS4_10StatisticsEERKNS1_19ParquetColumnSchemaENS_16DecodedValueKindEPNS1_23ParquetColumnStatisticsEPKN4cctz9time_zoneE Line | Count | Source | 291 | 3 | const cctz::time_zone* timezone) { | 292 | 3 | auto typed_statistics = | 293 | 3 | std::static_pointer_cast<::parquet::TypedStatistics<ParquetDType>>(statistics); | 294 | 3 | const auto& min_value = typed_statistics->min(); | 295 | 3 | const auto& max_value = typed_statistics->max(); | 296 | | if constexpr (std::is_same_v<ParquetDType, ::parquet::Int64Type>) { | 297 | | if (!timestamp_min_max_is_safe(column_schema, min_value, max_value, timezone)) { | 298 | | return false; | 299 | | } | 300 | | } | 301 | 3 | if (!valid_min_max(min_value, max_value) || | 302 | 3 | !set_decoded_field(column_schema, value_kind, min_value, &column_statistics->min_value, | 303 | 1 | timezone) || | 304 | 3 | !set_decoded_field(column_schema, value_kind, max_value, &column_statistics->max_value, | 305 | 2 | timezone)) { | 306 | 2 | return false; | 307 | 2 | } | 308 | 1 | return decoded_min_max_is_ordered(*column_statistics); | 309 | 3 | } |
|
310 | | |
311 | | bool set_decoded_binary_field(const ParquetColumnSchema& column_schema, DecodedValueKind value_kind, |
312 | | const StringRef& value, Field* field, |
313 | 4 | const cctz::time_zone* timezone) { |
314 | 4 | std::vector<StringRef> binary_values {value}; |
315 | 4 | DecodedColumnView view; |
316 | 4 | view.value_kind = value_kind; |
317 | 4 | view.binary_values = &binary_values; |
318 | 4 | return read_decoded_field(column_schema, view, field, timezone).ok(); |
319 | 4 | } |
320 | | |
321 | | bool set_string_min_max(const std::shared_ptr<::parquet::Statistics>& statistics, |
322 | | const ParquetColumnSchema& column_schema, |
323 | | ParquetColumnStatistics* column_statistics, |
324 | 2 | const cctz::time_zone* timezone) { |
325 | 2 | switch (statistics->physical_type()) { |
326 | 2 | case ::parquet::Type::BYTE_ARRAY: { |
327 | 2 | auto typed_statistics = |
328 | 2 | std::static_pointer_cast<::parquet::TypedStatistics<::parquet::ByteArrayType>>( |
329 | 2 | statistics); |
330 | 2 | const auto min = ::parquet::ByteArrayToString(typed_statistics->min()); |
331 | 2 | const auto max = ::parquet::ByteArrayToString(typed_statistics->max()); |
332 | 2 | if (!set_decoded_binary_field(column_schema, DecodedValueKind::BINARY, |
333 | 2 | StringRef(min.data(), min.size()), |
334 | 2 | &column_statistics->min_value, timezone) || |
335 | 2 | !set_decoded_binary_field(column_schema, DecodedValueKind::BINARY, |
336 | 2 | StringRef(max.data(), max.size()), |
337 | 2 | &column_statistics->max_value, timezone)) { |
338 | 0 | return false; |
339 | 0 | } |
340 | 2 | return decoded_min_max_is_ordered(*column_statistics); |
341 | 2 | } |
342 | 0 | case ::parquet::Type::FIXED_LEN_BYTE_ARRAY: { |
343 | 0 | if (column_schema.descriptor == nullptr || column_schema.descriptor->type_length() <= 0) { |
344 | 0 | return false; |
345 | 0 | } |
346 | 0 | auto typed_statistics = |
347 | 0 | std::static_pointer_cast<::parquet::TypedStatistics<::parquet::FLBAType>>( |
348 | 0 | statistics); |
349 | 0 | const int type_length = column_schema.descriptor->type_length(); |
350 | 0 | const std::string min(reinterpret_cast<const char*>(typed_statistics->min().ptr), |
351 | 0 | type_length); |
352 | 0 | const std::string max(reinterpret_cast<const char*>(typed_statistics->max().ptr), |
353 | 0 | type_length); |
354 | 0 | if (!set_decoded_binary_field(column_schema, DecodedValueKind::FIXED_BINARY, |
355 | 0 | StringRef(min.data(), min.size()), |
356 | 0 | &column_statistics->min_value, timezone) || |
357 | 0 | !set_decoded_binary_field(column_schema, DecodedValueKind::FIXED_BINARY, |
358 | 0 | StringRef(max.data(), max.size()), |
359 | 0 | &column_statistics->max_value, timezone)) { |
360 | 0 | return false; |
361 | 0 | } |
362 | 0 | return decoded_min_max_is_ordered(*column_statistics); |
363 | 0 | } |
364 | 0 | default: |
365 | 0 | return false; |
366 | 2 | } |
367 | 2 | } |
368 | | |
369 | | template <typename T> |
370 | 8 | T load_predicate_value(const char* data) { |
371 | 8 | T value; |
372 | 8 | memcpy(&value, data, sizeof(T)); |
373 | 8 | return value; |
374 | 8 | } Unexecuted instantiation: parquet_statistics.cpp:_ZN5doris6format7parquet12_GLOBAL__N_120load_predicate_valueIbEET_PKc parquet_statistics.cpp:_ZN5doris6format7parquet12_GLOBAL__N_120load_predicate_valueIiEET_PKc Line | Count | Source | 370 | 2 | T load_predicate_value(const char* data) { | 371 | 2 | T value; | 372 | 2 | memcpy(&value, data, sizeof(T)); | 373 | 2 | return value; | 374 | 2 | } |
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 | 370 | 6 | T load_predicate_value(const char* data) { | 371 | 6 | T value; | 372 | 6 | memcpy(&value, data, sizeof(T)); | 373 | 6 | return value; | 374 | 6 | } |
Unexecuted instantiation: parquet_statistics.cpp:_ZN5doris6format7parquet12_GLOBAL__N_120load_predicate_valueIfEET_PKc Unexecuted instantiation: parquet_statistics.cpp:_ZN5doris6format7parquet12_GLOBAL__N_120load_predicate_valueIdEET_PKc |
375 | | |
376 | 8 | std::optional<int64_t> load_predicate_integral_value(const char* buf, size_t size) { |
377 | 8 | switch (size) { |
378 | 0 | case sizeof(int8_t): |
379 | 0 | return static_cast<int64_t>(load_predicate_value<int8_t>(buf)); |
380 | 0 | case sizeof(int16_t): |
381 | 0 | return static_cast<int64_t>(load_predicate_value<int16_t>(buf)); |
382 | 2 | case sizeof(int32_t): |
383 | 2 | return static_cast<int64_t>(load_predicate_value<int32_t>(buf)); |
384 | 6 | case sizeof(int64_t): |
385 | 6 | return load_predicate_value<int64_t>(buf); |
386 | 0 | default: |
387 | 0 | return std::nullopt; |
388 | 8 | } |
389 | 8 | } |
390 | | |
391 | | bool logical_integer_fits_physical_int32(const ParquetTypeDescriptor& type_descriptor, |
392 | 8 | int64_t value) { |
393 | 8 | const int bit_width = |
394 | 8 | type_descriptor.integer_bit_width > 0 ? type_descriptor.integer_bit_width : 32; |
395 | 8 | if (type_descriptor.is_unsigned_integer) { |
396 | 6 | const uint64_t max_value = bit_width >= 32 ? std::numeric_limits<uint32_t>::max() |
397 | 6 | : ((uint64_t {1} << bit_width) - 1); |
398 | 6 | return value >= 0 && static_cast<uint64_t>(value) <= max_value; |
399 | 6 | } |
400 | 2 | const int64_t min_value = bit_width >= 32 ? std::numeric_limits<int32_t>::min() |
401 | 2 | : -(int64_t {1} << (bit_width - 1)); |
402 | 2 | const int64_t max_value = bit_width >= 32 ? std::numeric_limits<int32_t>::max() |
403 | 2 | : ((int64_t {1} << (bit_width - 1)) - 1); |
404 | 2 | return value >= min_value && value <= max_value; |
405 | 8 | } |
406 | | |
407 | | std::optional<int32_t> convert_logical_integer_to_physical_int32( |
408 | 8 | const ParquetTypeDescriptor& type_descriptor, int64_t value) { |
409 | 8 | if (!logical_integer_fits_physical_int32(type_descriptor, value)) { |
410 | 3 | return std::nullopt; |
411 | 3 | } |
412 | 5 | if (!type_descriptor.is_unsigned_integer) { |
413 | 2 | return static_cast<int32_t>(value); |
414 | 2 | } |
415 | 3 | const auto unsigned_value = static_cast<uint32_t>(value); |
416 | 3 | int32_t physical_value; |
417 | 3 | memcpy(&physical_value, &unsigned_value, sizeof(physical_value)); |
418 | 3 | return physical_value; |
419 | 5 | } |
420 | | |
421 | | class ArrowParquetBloomFilterAdapter final : public segment_v2::BloomFilter { |
422 | | public: |
423 | | ArrowParquetBloomFilterAdapter(const ParquetColumnSchema& column_schema, |
424 | | const ::parquet::BloomFilter& bloom_filter) |
425 | 7 | : _column_schema(column_schema), _bloom_filter(bloom_filter) {} |
426 | | |
427 | 0 | void add_bytes(const char* buf, size_t size) override { DORIS_CHECK(false); } |
428 | | |
429 | 7 | bool test_bytes(const char* buf, size_t size) const override { |
430 | 7 | if (buf == nullptr) { |
431 | 0 | return true; |
432 | 0 | } |
433 | | // Parquet bloom filters are populated from the physical column carrier, while VExpr |
434 | | // literals are materialized as Doris logical values. Keep the logical type in |
435 | | // BloomFilterEvalContext for expression compatibility, and normalize to the Parquet |
436 | | // physical representation only at this adapter boundary. |
437 | 7 | switch (_column_schema.type_descriptor.physical_type) { |
438 | 0 | case ::parquet::Type::BOOLEAN: |
439 | 0 | return test_boolean(buf, size); |
440 | 5 | case ::parquet::Type::INT32: |
441 | 5 | return test_physical_int32(buf, size); |
442 | 0 | case ::parquet::Type::INT64: |
443 | 0 | return test_int64(buf, size); |
444 | 0 | case ::parquet::Type::FLOAT: |
445 | 0 | return test_float(buf, size); |
446 | 0 | case ::parquet::Type::DOUBLE: |
447 | 0 | return test_double(buf, size); |
448 | 0 | case ::parquet::Type::BYTE_ARRAY: |
449 | 0 | return test_byte_array(buf, size); |
450 | 2 | case ::parquet::Type::FIXED_LEN_BYTE_ARRAY: |
451 | 2 | return test_fixed_len_byte_array(buf, size); |
452 | 0 | default: |
453 | 0 | return true; |
454 | 7 | } |
455 | 7 | } |
456 | | |
457 | 0 | void set_has_null(bool has_null) override { DORIS_CHECK(!has_null); } |
458 | 0 | bool has_null() const override { return false; } |
459 | 0 | void add_hash(uint64_t hash) override { DORIS_CHECK(false); } |
460 | 0 | bool test_hash(uint64_t hash) const override { return _bloom_filter.FindHash(hash); } |
461 | | |
462 | | private: |
463 | 0 | bool test_boolean(const char* buf, size_t size) const { |
464 | 0 | if (size == sizeof(bool)) { |
465 | 0 | const int32_t value = load_predicate_value<bool>(buf) ? 1 : 0; |
466 | 0 | return _bloom_filter.FindHash(_bloom_filter.Hash(value)); |
467 | 0 | } |
468 | 0 | if (size == sizeof(int32_t)) { |
469 | 0 | const int32_t value = load_predicate_value<int32_t>(buf); |
470 | 0 | return _bloom_filter.FindHash(_bloom_filter.Hash(value != 0 ? 1 : 0)); |
471 | 0 | } |
472 | 0 | return true; |
473 | 0 | } |
474 | | |
475 | 5 | bool test_physical_int32(const char* buf, size_t size) const { |
476 | 5 | const auto logical_value = load_predicate_integral_value(buf, size); |
477 | 5 | if (!logical_value.has_value()) { |
478 | 0 | return true; |
479 | 0 | } |
480 | 5 | const auto physical_value = convert_logical_integer_to_physical_int32( |
481 | 5 | _column_schema.type_descriptor, *logical_value); |
482 | 5 | if (!physical_value.has_value()) { |
483 | 2 | return false; |
484 | 2 | } |
485 | 3 | return find_int32(*physical_value); |
486 | 5 | } |
487 | | |
488 | 0 | bool test_int64(const char* buf, size_t size) const { |
489 | 0 | if (size != sizeof(int64_t)) { |
490 | 0 | return true; |
491 | 0 | } |
492 | 0 | const int64_t value = load_predicate_value<int64_t>(buf); |
493 | 0 | return _bloom_filter.FindHash(_bloom_filter.Hash(value)); |
494 | 0 | } |
495 | | |
496 | 0 | bool test_float(const char* buf, size_t size) const { |
497 | 0 | if (size != sizeof(float)) { |
498 | 0 | return true; |
499 | 0 | } |
500 | 0 | const float value = load_predicate_value<float>(buf); |
501 | 0 | return _bloom_filter.FindHash(_bloom_filter.Hash(value)); |
502 | 0 | } |
503 | | |
504 | 0 | bool test_double(const char* buf, size_t size) const { |
505 | 0 | if (size != sizeof(double)) { |
506 | 0 | return true; |
507 | 0 | } |
508 | 0 | const double value = load_predicate_value<double>(buf); |
509 | 0 | return _bloom_filter.FindHash(_bloom_filter.Hash(value)); |
510 | 0 | } |
511 | | |
512 | 0 | bool test_byte_array(const char* buf, size_t size) const { |
513 | 0 | ::parquet::ByteArray value(static_cast<uint32_t>(size), |
514 | 0 | reinterpret_cast<const uint8_t*>(buf)); |
515 | 0 | return _bloom_filter.FindHash(_bloom_filter.Hash(&value)); |
516 | 0 | } |
517 | | |
518 | 2 | bool test_fixed_len_byte_array(const char* buf, size_t size) const { |
519 | 2 | if (_column_schema.type_descriptor.fixed_length <= 0) { |
520 | 0 | return true; |
521 | 0 | } |
522 | 2 | if (size != static_cast<size_t>(_column_schema.type_descriptor.fixed_length)) { |
523 | 1 | return false; |
524 | 1 | } |
525 | 1 | ::parquet::FLBA value(reinterpret_cast<const uint8_t*>(buf)); |
526 | 1 | return _bloom_filter.FindHash( |
527 | 1 | _bloom_filter.Hash(&value, _column_schema.type_descriptor.fixed_length)); |
528 | 2 | } |
529 | | |
530 | 3 | bool find_int32(int32_t value) const { |
531 | 3 | return _bloom_filter.FindHash(_bloom_filter.Hash(value)); |
532 | 3 | } |
533 | | |
534 | | const ParquetColumnSchema& _column_schema; |
535 | | const ::parquet::BloomFilter& _bloom_filter; |
536 | | }; |
537 | | |
538 | | class NativeParquetBloomFilterAdapter final : public segment_v2::BloomFilter { |
539 | | public: |
540 | | NativeParquetBloomFilterAdapter(const ParquetColumnSchema& column_schema, |
541 | | const segment_v2::BloomFilter& bloom_filter) |
542 | 3 | : _column_schema(column_schema), _bloom_filter(bloom_filter) {} |
543 | | |
544 | 0 | void add_bytes(const char*, size_t) override { DORIS_CHECK(false); } |
545 | | |
546 | 3 | bool test_bytes(const char* buf, size_t size) const override { |
547 | 3 | if (buf == nullptr || |
548 | 3 | _column_schema.type_descriptor.physical_type != ::parquet::Type::INT32) { |
549 | 0 | return _bloom_filter.test_bytes(buf, size); |
550 | 0 | } |
551 | 3 | const auto logical_value = load_predicate_integral_value(buf, size); |
552 | 3 | if (!logical_value.has_value()) { |
553 | 0 | return true; |
554 | 0 | } |
555 | 3 | const auto physical_value = convert_logical_integer_to_physical_int32( |
556 | 3 | _column_schema.type_descriptor, *logical_value); |
557 | 3 | if (!physical_value.has_value()) { |
558 | 1 | return false; |
559 | 1 | } |
560 | | // Native file Bloom bytes are hashed from the Parquet physical carrier, not the wider |
561 | | // Doris logical literal used by VExpr (for example UINT32 is exposed as BIGINT). |
562 | 2 | return _bloom_filter.test_bytes(reinterpret_cast<const char*>(&*physical_value), |
563 | 2 | sizeof(*physical_value)); |
564 | 3 | } |
565 | | |
566 | 0 | void set_has_null(bool has_null) override { DORIS_CHECK(!has_null); } |
567 | 0 | bool has_null() const override { return false; } |
568 | 0 | void add_hash(uint64_t) override { DORIS_CHECK(false); } |
569 | 0 | bool test_hash(uint64_t hash) const override { return _bloom_filter.test_hash(hash); } |
570 | | |
571 | | private: |
572 | | const ParquetColumnSchema& _column_schema; |
573 | | const segment_v2::BloomFilter& _bloom_filter; |
574 | | }; |
575 | | |
576 | 16 | bool bloom_filter_supported(const ParquetColumnSchema& column_schema) { |
577 | 16 | if (!bloom_logical_type_supported(column_schema)) { |
578 | 0 | return false; |
579 | 0 | } |
580 | 16 | switch (column_schema.type_descriptor.physical_type) { |
581 | 0 | case ::parquet::Type::BOOLEAN: |
582 | 13 | case ::parquet::Type::INT32: |
583 | 13 | case ::parquet::Type::INT64: |
584 | 13 | case ::parquet::Type::FLOAT: |
585 | 13 | case ::parquet::Type::DOUBLE: |
586 | 13 | case ::parquet::Type::BYTE_ARRAY: |
587 | 13 | return true; |
588 | 3 | case ::parquet::Type::FIXED_LEN_BYTE_ARRAY: |
589 | 3 | return column_schema.type_descriptor.is_string_like && |
590 | 3 | column_schema.type_descriptor.fixed_length > 0; |
591 | 0 | default: |
592 | 0 | return false; |
593 | 16 | } |
594 | 16 | } |
595 | | |
596 | | bool bloom_filter_excludes(const ParquetColumnSchema& column_schema, int slot_index, |
597 | | const VExprContextSPtrs& conjuncts, |
598 | 8 | const ::parquet::BloomFilter& bloom_filter) { |
599 | 8 | if (!bloom_filter_supported(column_schema)) { |
600 | 1 | return false; |
601 | 1 | } |
602 | 7 | ArrowParquetBloomFilterAdapter adapter(column_schema, bloom_filter); |
603 | 7 | BloomFilterEvalContext ctx; |
604 | 7 | ctx.slots.emplace(slot_index, BloomFilterEvalContext::SlotBloomFilter { |
605 | 7 | .data_type = column_schema.type, |
606 | 7 | .bloom_filter = &adapter, |
607 | 7 | }); |
608 | 7 | return VExprContext::evaluate_bloom_filter(conjuncts, ctx) == ZoneMapFilterResult::kNoMatch; |
609 | 8 | } |
610 | | |
611 | | struct RowGroupBloomFilterCache { |
612 | | using CacheKey = std::pair<int, int>; |
613 | | |
614 | | ::parquet::BloomFilterReader* bloom_filter_reader = nullptr; |
615 | | std::map<CacheKey, std::unique_ptr<::parquet::BloomFilter>> column_bloom_filters; |
616 | | std::set<CacheKey> loaded_columns; |
617 | | |
618 | | ::parquet::BloomFilter* get(int row_group_idx, int leaf_column_id, |
619 | 4 | ParquetPruningStats* pruning_stats) { |
620 | 4 | if (bloom_filter_reader == nullptr || leaf_column_id < 0) { |
621 | 2 | return nullptr; |
622 | 2 | } |
623 | 2 | const CacheKey cache_key {row_group_idx, leaf_column_id}; |
624 | 2 | if (loaded_columns.find(cache_key) == loaded_columns.end()) { |
625 | 2 | loaded_columns.insert(cache_key); |
626 | 2 | try { |
627 | 2 | std::shared_ptr<::parquet::RowGroupBloomFilterReader> row_group_reader; |
628 | 2 | if (pruning_stats != nullptr) { |
629 | 2 | SCOPED_RAW_TIMER(&pruning_stats->bloom_filter_read_time); |
630 | 2 | row_group_reader = bloom_filter_reader->RowGroup(row_group_idx); |
631 | 2 | if (row_group_reader != nullptr) { |
632 | 2 | column_bloom_filters[cache_key] = |
633 | 2 | row_group_reader->GetColumnBloomFilter(leaf_column_id); |
634 | 2 | } |
635 | 2 | } else { |
636 | 0 | row_group_reader = bloom_filter_reader->RowGroup(row_group_idx); |
637 | 0 | if (row_group_reader != nullptr) { |
638 | 0 | column_bloom_filters[cache_key] = |
639 | 0 | row_group_reader->GetColumnBloomFilter(leaf_column_id); |
640 | 0 | } |
641 | 0 | } |
642 | 2 | } catch (const ::parquet::ParquetException&) { |
643 | 0 | return nullptr; |
644 | 0 | } catch (const std::exception&) { |
645 | 0 | return nullptr; |
646 | 0 | } |
647 | 2 | } |
648 | 2 | auto it = column_bloom_filters.find(cache_key); |
649 | 2 | return it == column_bloom_filters.end() ? nullptr : it->second.get(); |
650 | 2 | } |
651 | | }; |
652 | | |
653 | 13 | bool is_dictionary_data_encoding(::parquet::Encoding::type encoding) { |
654 | 13 | return encoding == ::parquet::Encoding::PLAIN_DICTIONARY || |
655 | 13 | encoding == ::parquet::Encoding::RLE_DICTIONARY; |
656 | 13 | } |
657 | | |
658 | 0 | bool is_level_encoding(::parquet::Encoding::type encoding) { |
659 | 0 | return encoding == ::parquet::Encoding::RLE || encoding == ::parquet::Encoding::BIT_PACKED; |
660 | 0 | } |
661 | | |
662 | 26 | bool is_data_page_type(::parquet::PageType::type page_type) { |
663 | 26 | return page_type == ::parquet::PageType::DATA_PAGE || |
664 | 26 | page_type == ::parquet::PageType::DATA_PAGE_V2; |
665 | 26 | } |
666 | | |
667 | 17 | bool is_dictionary_encoded_chunk(const ::parquet::ColumnChunkMetaData& column_metadata) { |
668 | 17 | if (!column_metadata.has_dictionary_page()) { |
669 | 4 | return false; |
670 | 4 | } |
671 | | |
672 | 13 | const auto& encoding_stats = column_metadata.encoding_stats(); |
673 | 13 | if (!encoding_stats.empty()) { |
674 | 13 | bool has_dictionary_data_page = false; |
675 | 26 | for (const auto& encoding_stat : encoding_stats) { |
676 | 26 | if (!is_data_page_type(encoding_stat.page_type) || encoding_stat.count <= 0) { |
677 | 13 | continue; |
678 | 13 | } |
679 | 13 | if (!is_dictionary_data_encoding(encoding_stat.encoding)) { |
680 | 0 | return false; |
681 | 0 | } |
682 | 13 | has_dictionary_data_page = true; |
683 | 13 | } |
684 | 13 | return has_dictionary_data_page; |
685 | 13 | } |
686 | | |
687 | 0 | bool has_dictionary_encoding = false; |
688 | 0 | for (const auto encoding : column_metadata.encodings()) { |
689 | 0 | if (is_dictionary_data_encoding(encoding)) { |
690 | 0 | has_dictionary_encoding = true; |
691 | 0 | continue; |
692 | 0 | } |
693 | 0 | if (!is_level_encoding(encoding)) { |
694 | 0 | return false; |
695 | 0 | } |
696 | 0 | } |
697 | 0 | return has_dictionary_encoding; |
698 | 0 | } |
699 | | |
700 | | bool supports_dictionary_pruning(const ParquetColumnSchema& column_schema, |
701 | 17 | const ::parquet::ColumnChunkMetaData& column_metadata) { |
702 | 17 | if (column_schema.kind != ParquetColumnSchemaKind::PRIMITIVE || |
703 | 17 | column_schema.descriptor == nullptr || column_schema.type == nullptr) { |
704 | 0 | return false; |
705 | 0 | } |
706 | 17 | if (!column_schema.type_descriptor.is_string_like) { |
707 | 0 | return false; |
708 | 0 | } |
709 | 17 | if (column_metadata.type() != ::parquet::Type::BYTE_ARRAY && |
710 | 17 | column_metadata.type() != ::parquet::Type::FIXED_LEN_BYTE_ARRAY) { |
711 | 0 | return false; |
712 | 0 | } |
713 | 17 | return true; |
714 | 17 | } |
715 | | |
716 | | } // namespace |
717 | | |
718 | | bool read_dictionary_words(::parquet::ParquetFileReader* file_reader, int row_group_idx, |
719 | | int leaf_column_id, const ParquetColumnSchema& column_schema, |
720 | 13 | ParquetDictionaryWords* dict_words) { |
721 | 13 | DORIS_CHECK(dict_words != nullptr); |
722 | 13 | dict_words->clear(); |
723 | 13 | if (file_reader == nullptr || leaf_column_id < 0) { |
724 | 0 | return false; |
725 | 0 | } |
726 | | |
727 | 13 | auto row_group_reader = file_reader->RowGroup(row_group_idx); |
728 | 13 | if (row_group_reader == nullptr) { |
729 | 0 | return false; |
730 | 0 | } |
731 | 13 | auto page_reader = row_group_reader->GetColumnPageReader(leaf_column_id); |
732 | 13 | if (page_reader == nullptr) { |
733 | 0 | return false; |
734 | 0 | } |
735 | | |
736 | 13 | std::shared_ptr<::parquet::Page> page; |
737 | 13 | try { |
738 | 13 | page = page_reader->NextPage(); |
739 | 13 | } catch (const ::parquet::ParquetException&) { |
740 | 0 | return false; |
741 | 0 | } catch (const std::exception&) { |
742 | 0 | return false; |
743 | 0 | } |
744 | 13 | if (page == nullptr || page->type() != ::parquet::PageType::DICTIONARY_PAGE) { |
745 | 0 | return false; |
746 | 0 | } |
747 | 13 | const auto* dictionary_page = static_cast<const ::parquet::DictionaryPage*>(page.get()); |
748 | 13 | if (dictionary_page->encoding() != ::parquet::Encoding::PLAIN && |
749 | 13 | dictionary_page->encoding() != ::parquet::Encoding::PLAIN_DICTIONARY) { |
750 | 0 | return false; |
751 | 0 | } |
752 | 13 | const int32_t dictionary_length = dictionary_page->num_values(); |
753 | 13 | if (dictionary_length <= 0) { |
754 | 0 | return false; |
755 | 0 | } |
756 | 13 | const auto* dictionary_data = dictionary_page->data(); |
757 | 13 | const int dictionary_size = dictionary_page->size(); |
758 | | |
759 | 13 | dict_words->values.reserve(static_cast<size_t>(dictionary_length)); |
760 | 13 | if (column_schema.descriptor->physical_type() == ::parquet::Type::BYTE_ARRAY) { |
761 | 13 | auto decoder = ::parquet::MakeTypedDecoder<::parquet::ByteArrayType>( |
762 | 13 | ::parquet::Encoding::PLAIN, column_schema.descriptor); |
763 | 13 | decoder->SetData(dictionary_length, dictionary_data, dictionary_size); |
764 | 13 | std::vector<::parquet::ByteArray> byte_array_values(static_cast<size_t>(dictionary_length)); |
765 | 13 | if (decoder->Decode(byte_array_values.data(), dictionary_length) != dictionary_length) { |
766 | 0 | return false; |
767 | 0 | } |
768 | 32 | for (int32_t dict_idx = 0; dict_idx < dictionary_length; ++dict_idx) { |
769 | 19 | dict_words->values.emplace_back( |
770 | 19 | reinterpret_cast<const char*>(byte_array_values[dict_idx].ptr), |
771 | 19 | byte_array_values[dict_idx].len); |
772 | 19 | } |
773 | 13 | dict_words->build_refs(); |
774 | 13 | return true; |
775 | 13 | } |
776 | 0 | if (column_schema.descriptor->physical_type() == ::parquet::Type::FIXED_LEN_BYTE_ARRAY) { |
777 | 0 | const int type_length = column_schema.descriptor->type_length(); |
778 | 0 | if (type_length <= 0) { |
779 | 0 | return false; |
780 | 0 | } |
781 | 0 | auto decoder = ::parquet::MakeTypedDecoder<::parquet::FLBAType>(::parquet::Encoding::PLAIN, |
782 | 0 | column_schema.descriptor); |
783 | 0 | decoder->SetData(dictionary_length, dictionary_data, dictionary_size); |
784 | 0 | std::vector<::parquet::FixedLenByteArray> flba_values( |
785 | 0 | static_cast<size_t>(dictionary_length)); |
786 | 0 | if (decoder->Decode(flba_values.data(), dictionary_length) != dictionary_length) { |
787 | 0 | return false; |
788 | 0 | } |
789 | 0 | for (int32_t dict_idx = 0; dict_idx < dictionary_length; ++dict_idx) { |
790 | 0 | dict_words->values.emplace_back( |
791 | 0 | reinterpret_cast<const char*>(flba_values[dict_idx].ptr), type_length); |
792 | 0 | } |
793 | 0 | dict_words->build_refs(); |
794 | 0 | return true; |
795 | 0 | } |
796 | 0 | return false; |
797 | 0 | } |
798 | | |
799 | 13 | std::vector<Field> dictionary_fields_from_words(const ParquetDictionaryWords& dict_words) { |
800 | 13 | std::vector<Field> fields; |
801 | 13 | fields.reserve(dict_words.refs.size()); |
802 | 19 | for (const auto& ref : dict_words.refs) { |
803 | 19 | fields.push_back(Field::create_field<TYPE_STRING>(String(ref.data, ref.size))); |
804 | 19 | } |
805 | 13 | return fields; |
806 | 13 | } |
807 | | |
808 | | namespace { |
809 | | |
810 | | const ParquetColumnSchema* resolve_local_leaf_schema( |
811 | | const std::vector<std::unique_ptr<ParquetColumnSchema>>& schema, |
812 | 136 | const format::LocalColumnId file_column_id) { |
813 | 136 | if (!file_column_id.is_valid() || file_column_id.value() >= static_cast<int>(schema.size())) { |
814 | 0 | return nullptr; |
815 | 0 | } |
816 | 136 | const ParquetColumnSchema* column_schema = schema[file_column_id.value()].get(); |
817 | 136 | if (column_schema == nullptr || column_schema->kind != ParquetColumnSchemaKind::PRIMITIVE || |
818 | 136 | column_schema->leaf_column_id < 0 || column_schema->max_repetition_level > 0) { |
819 | 2 | return nullptr; |
820 | 2 | } |
821 | 134 | return column_schema; |
822 | 136 | } |
823 | | |
824 | | std::optional<format::LocalColumnId> file_column_id_by_block_position( |
825 | 136 | const format::FileScanRequest& request, int block_position) { |
826 | 171 | for (const auto& [file_column_id, local_index] : request.local_positions) { |
827 | 171 | if (local_index.value() == block_position) { |
828 | 136 | return file_column_id; |
829 | 136 | } |
830 | 171 | } |
831 | 0 | return std::nullopt; |
832 | 136 | } |
833 | | |
834 | | bool has_expr_zonemap_filter(const format::FileScanRequest& request, |
835 | 716 | const RuntimeState* runtime_state) { |
836 | 716 | if (!expr_zonemap::is_expr_zonemap_filter_enabled(runtime_state)) { |
837 | 0 | return false; |
838 | 0 | } |
839 | 716 | for (const auto& conjunct : request.conjuncts) { |
840 | 256 | if (conjunct != nullptr && conjunct->root() != nullptr && |
841 | 256 | conjunct->root()->can_evaluate_zonemap_filter()) { |
842 | 171 | return true; |
843 | 171 | } |
844 | 256 | } |
845 | 545 | return false; |
846 | 716 | } |
847 | | |
848 | 78 | std::set<int> collect_expr_zonemap_slot_indexes(const VExprContextSPtrs& conjuncts) { |
849 | 78 | std::set<int> slot_indexes; |
850 | 85 | for (const auto& conjunct : conjuncts) { |
851 | 85 | if (conjunct != nullptr && conjunct->root() != nullptr && |
852 | 85 | conjunct->root()->can_evaluate_zonemap_filter()) { |
853 | 82 | conjunct->root()->collect_slot_column_ids(slot_indexes); |
854 | 82 | } |
855 | 85 | } |
856 | 78 | return slot_indexes; |
857 | 78 | } |
858 | | |
859 | | template <typename SlotIndexSelector> |
860 | | std::map<int, VExprContextSPtrs> collect_conjuncts_by_single_slot( |
861 | 495 | const VExprContextSPtrs& conjuncts, SlotIndexSelector slot_index_selector) { |
862 | 495 | std::map<int, VExprContextSPtrs> conjuncts_by_slot; |
863 | 495 | for (const auto& conjunct : conjuncts) { |
864 | 193 | const auto slot_index = slot_index_selector(conjunct); |
865 | 193 | if (slot_index >= 0) { |
866 | 45 | conjuncts_by_slot[slot_index].push_back(conjunct); |
867 | 45 | } |
868 | 193 | } |
869 | 495 | return conjuncts_by_slot; |
870 | 495 | } |
871 | | |
872 | | std::shared_ptr<segment_v2::ZoneMap> make_zonemap_from_statistics( |
873 | 194 | const ParquetColumnStatistics& statistics) { |
874 | 194 | if (!statistics.has_null_count && !statistics.has_min_max) { |
875 | 12 | return nullptr; |
876 | 12 | } |
877 | 182 | segment_v2::ZoneMap zone_map; |
878 | 182 | zone_map.has_null = statistics.has_null; |
879 | 182 | zone_map.has_not_null = statistics.has_not_null; |
880 | 182 | if (!statistics.has_not_null) { |
881 | 3 | return std::make_shared<segment_v2::ZoneMap>(std::move(zone_map)); |
882 | 3 | } |
883 | 179 | if (!statistics.has_min_max) { |
884 | | // Null counts remain trustworthy when min/max decoding fails (for example, because a |
885 | | // floating-point bound is NaN). pass_all prevents range pruning without discarding the |
886 | | // has_null/has_not_null flags needed by IS NULL and IS NOT NULL predicates. |
887 | 2 | zone_map.pass_all = true; |
888 | 2 | return std::make_shared<segment_v2::ZoneMap>(std::move(zone_map)); |
889 | 2 | } |
890 | 177 | zone_map.min_value = statistics.min_value; |
891 | 177 | zone_map.max_value = statistics.max_value; |
892 | 177 | return std::make_shared<segment_v2::ZoneMap>(std::move(zone_map)); |
893 | 179 | } |
894 | | |
895 | | void add_slot_zonemap(ZoneMapEvalContext* ctx, int slot_index, const DataTypePtr& data_type, |
896 | 192 | std::shared_ptr<segment_v2::ZoneMap> zone_map) { |
897 | 192 | DORIS_CHECK(ctx != nullptr); |
898 | 192 | ZoneMapEvalContext::SlotZoneMap slot_zone_map; |
899 | 192 | slot_zone_map.data_type = data_type; |
900 | 192 | slot_zone_map.zone_map = std::move(zone_map); |
901 | 192 | ctx->slots.emplace(slot_index, std::move(slot_zone_map)); |
902 | 192 | } |
903 | | |
904 | 78 | void accumulate_zonemap_stats(const ZoneMapEvalContext& ctx, ParquetPruningStats* pruning_stats) { |
905 | 78 | if (pruning_stats == nullptr) { |
906 | 0 | return; |
907 | 0 | } |
908 | 78 | pruning_stats->expr_zonemap_unusable_evals += ctx.stats.unusable_zonemap_eval_count; |
909 | 78 | pruning_stats->in_zonemap_point_check_count += ctx.stats.in_zonemap_point_check_count; |
910 | 78 | pruning_stats->in_zonemap_range_only_count += ctx.stats.in_zonemap_range_only_count; |
911 | 78 | } |
912 | | |
913 | | } // namespace |
914 | | |
915 | | bool can_use_parquet_page_index(const format::FileScanRequest& request, |
916 | 205 | const RuntimeState* runtime_state) { |
917 | 205 | return config::enable_parquet_page_index && has_expr_zonemap_filter(request, runtime_state); |
918 | 205 | } |
919 | | |
920 | | std::shared_ptr<segment_v2::ZoneMap> ParquetStatisticsUtils::MakeZoneMap( |
921 | 194 | const ParquetColumnStatistics& statistics) { |
922 | 194 | return make_zonemap_from_statistics(statistics); |
923 | 194 | } |
924 | | |
925 | | ParquetColumnStatistics ParquetStatisticsUtils::TransformColumnStatistics( |
926 | | const ParquetColumnSchema& column_schema, |
927 | 36 | const std::shared_ptr<::parquet::Statistics>& statistics, const cctz::time_zone* timezone) { |
928 | 36 | ParquetColumnStatistics result; |
929 | 36 | if (statistics == nullptr) { |
930 | 1 | return result; |
931 | 1 | } |
932 | | |
933 | 35 | result.has_null = !statistics->HasNullCount() || statistics->null_count() > 0; |
934 | 35 | result.has_not_null = statistics->num_values() > 0 || statistics->HasMinMax(); |
935 | 35 | result.has_null_count = statistics->HasNullCount(); |
936 | 35 | if (!result.has_not_null || !statistics->HasMinMax()) { |
937 | 4 | return result; |
938 | 4 | } |
939 | | |
940 | 31 | DORIS_CHECK(column_schema.type != nullptr); |
941 | 31 | switch (statistics->physical_type()) { |
942 | 0 | case ::parquet::Type::BOOLEAN: |
943 | 0 | result.has_min_max = set_decoded_min_max<::parquet::BooleanType>( |
944 | 0 | statistics, column_schema, DecodedValueKind::BOOL, &result, timezone); |
945 | 0 | return result; |
946 | 19 | case ::parquet::Type::INT32: |
947 | 19 | result.has_min_max = set_decoded_min_max<::parquet::Int32Type>( |
948 | 19 | statistics, column_schema, decoded_value_kind(column_schema.type_descriptor), |
949 | 19 | &result, timezone); |
950 | 19 | return result; |
951 | 6 | case ::parquet::Type::INT64: |
952 | 6 | result.has_min_max = set_decoded_min_max<::parquet::Int64Type>( |
953 | 6 | statistics, column_schema, decoded_value_kind(column_schema.type_descriptor), |
954 | 6 | &result, timezone); |
955 | 6 | return result; |
956 | 1 | case ::parquet::Type::FLOAT: |
957 | 1 | result.has_min_max = set_decoded_min_max<::parquet::FloatType>( |
958 | 1 | statistics, column_schema, DecodedValueKind::FLOAT, &result, timezone); |
959 | 1 | return result; |
960 | 3 | case ::parquet::Type::DOUBLE: |
961 | 3 | result.has_min_max = set_decoded_min_max<::parquet::DoubleType>( |
962 | 3 | statistics, column_schema, DecodedValueKind::DOUBLE, &result, timezone); |
963 | 3 | return result; |
964 | 2 | case ::parquet::Type::BYTE_ARRAY: |
965 | 2 | case ::parquet::Type::FIXED_LEN_BYTE_ARRAY: |
966 | 2 | result.has_min_max = set_string_min_max(statistics, column_schema, &result, timezone); |
967 | 2 | return result; |
968 | 0 | default: |
969 | 0 | return result; |
970 | 31 | } |
971 | 31 | } |
972 | | |
973 | | ParquetColumnStatistics ParquetStatisticsUtils::TransformColumnStatistics( |
974 | | const ParquetColumnSchema& column_schema, const tparquet::Statistics* statistics, |
975 | 80 | int64_t column_value_count, const cctz::time_zone* timezone) { |
976 | 80 | ParquetColumnStatistics result; |
977 | 80 | if (statistics == nullptr || column_value_count < 0) { |
978 | 13 | return result; |
979 | 13 | } |
980 | | |
981 | 67 | if (statistics->__isset.null_count && statistics->null_count > column_value_count) { |
982 | | // An impossible null count makes all derived min/max and all-null flags untrustworthy; |
983 | | // disable pruning instead of turning corrupt footer metadata into false negatives. |
984 | 1 | return result; |
985 | 1 | } |
986 | | |
987 | 66 | const bool has_null_count = statistics->__isset.null_count && statistics->null_count >= 0; |
988 | 66 | const int64_t null_count = has_null_count ? statistics->null_count : 0; |
989 | 66 | const bool has_not_null = has_null_count ? column_value_count > null_count : true; |
990 | 66 | const std::string* min_value = statistics->__isset.min_value |
991 | 66 | ? &statistics->min_value |
992 | 66 | : (statistics->__isset.min ? &statistics->min : nullptr); |
993 | 66 | const std::string* max_value = statistics->__isset.max_value |
994 | 66 | ? &statistics->max_value |
995 | 66 | : (statistics->__isset.max ? &statistics->max : nullptr); |
996 | | |
997 | 66 | tparquet::ColumnIndex index; |
998 | 66 | index.__set_null_pages({!has_not_null}); |
999 | 66 | index.__set_null_counts({null_count}); |
1000 | 66 | if (min_value != nullptr && max_value != nullptr) { |
1001 | 66 | index.__set_min_values({*min_value}); |
1002 | 66 | index.__set_max_values({*max_value}); |
1003 | 66 | } |
1004 | | // Footer statistics and page indexes share the same little-endian physical encoding. Reusing |
1005 | | // one decoder keeps native row-group and page pruning identical for logical types and NaNs. |
1006 | 66 | if (!build_native_page_statistics(index, column_schema, 0, &result, timezone)) { |
1007 | 0 | return {}; |
1008 | 0 | } |
1009 | 66 | if (!has_null_count) { |
1010 | 0 | result.has_null_count = false; |
1011 | 0 | result.has_null = true; |
1012 | 0 | } |
1013 | 66 | return result; |
1014 | 66 | } |
1015 | | |
1016 | | bool ParquetStatisticsUtils::BloomFilterExcludes(const ParquetColumnSchema& column_schema, |
1017 | | int slot_index, const VExprContextSPtrs& conjuncts, |
1018 | 8 | const ::parquet::BloomFilter& bloom_filter) { |
1019 | 8 | return bloom_filter_excludes(column_schema, slot_index, conjuncts, bloom_filter); |
1020 | 8 | } |
1021 | | |
1022 | | bool ParquetStatisticsUtils::NativeBloomFilterExcludes( |
1023 | | const ParquetColumnSchema& column_schema, int slot_index, |
1024 | 3 | const VExprContextSPtrs& conjuncts, const segment_v2::BloomFilter& bloom_filter) { |
1025 | 3 | if (!bloom_filter_supported(column_schema)) { |
1026 | 0 | return false; |
1027 | 0 | } |
1028 | 3 | NativeParquetBloomFilterAdapter adapter(column_schema, bloom_filter); |
1029 | 3 | BloomFilterEvalContext ctx; |
1030 | 3 | ctx.slots.emplace(slot_index, BloomFilterEvalContext::SlotBloomFilter { |
1031 | 3 | .data_type = column_schema.type, |
1032 | 3 | .bloom_filter = &adapter, |
1033 | 3 | }); |
1034 | 3 | return VExprContext::evaluate_bloom_filter(conjuncts, ctx) == ZoneMapFilterResult::kNoMatch; |
1035 | 3 | } |
1036 | | |
1037 | | namespace { |
1038 | | |
1039 | | ParquetRowGroupPruneReason dictionary_prune_reason( |
1040 | | const ::parquet::RowGroupMetaData& row_group, ::parquet::ParquetFileReader* file_reader, |
1041 | | int row_group_idx, const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema, |
1042 | 42 | const format::FileScanRequest& request) { |
1043 | 42 | const auto conjuncts_by_slot = collect_conjuncts_by_single_slot( |
1044 | 42 | request.conjuncts, expr_zonemap::single_slot_dictionary_index); |
1045 | 42 | for (const auto& [slot_index, conjuncts] : conjuncts_by_slot) { |
1046 | 17 | const auto file_column_id = file_column_id_by_block_position(request, slot_index); |
1047 | 17 | if (!file_column_id.has_value()) { |
1048 | 0 | continue; |
1049 | 0 | } |
1050 | 17 | const auto* column_schema = resolve_local_leaf_schema(file_schema, *file_column_id); |
1051 | 17 | if (column_schema == nullptr || column_schema->type == nullptr) { |
1052 | 0 | continue; |
1053 | 0 | } |
1054 | 17 | DCHECK_LT(column_schema->leaf_column_id, row_group.num_columns()); |
1055 | 17 | auto column_chunk = row_group.ColumnChunk(column_schema->leaf_column_id); |
1056 | 17 | if (column_chunk == nullptr || |
1057 | 17 | !supports_dictionary_pruning(*column_schema, *column_chunk) || |
1058 | 17 | !is_dictionary_encoded_chunk(*column_chunk)) { |
1059 | 4 | continue; |
1060 | 4 | } |
1061 | | |
1062 | 13 | ParquetDictionaryWords dict_words; |
1063 | 13 | if (!read_dictionary_words(file_reader, row_group_idx, column_schema->leaf_column_id, |
1064 | 13 | *column_schema, &dict_words)) { |
1065 | 0 | continue; |
1066 | 0 | } |
1067 | 13 | DictionaryEvalContext ctx; |
1068 | 13 | ctx.slots.emplace(slot_index, DictionaryEvalContext::SlotDictionary { |
1069 | 13 | .data_type = column_schema->type, |
1070 | 13 | .values = dictionary_fields_from_words(dict_words), |
1071 | 13 | }); |
1072 | 13 | if (VExprContext::evaluate_dictionary_filter(conjuncts, ctx) == |
1073 | 13 | ZoneMapFilterResult::kNoMatch) { |
1074 | 10 | return ParquetRowGroupPruneReason::DICTIONARY; |
1075 | 10 | } |
1076 | 13 | } |
1077 | 32 | return ParquetRowGroupPruneReason::NONE; |
1078 | 42 | } |
1079 | | |
1080 | | ParquetRowGroupPruneReason bloom_filter_prune_reason( |
1081 | | int row_group_idx, const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema, |
1082 | | const format::FileScanRequest& request, RowGroupBloomFilterCache* bloom_filter_cache, |
1083 | 32 | ParquetPruningStats* pruning_stats) { |
1084 | 32 | if (bloom_filter_cache == nullptr) { |
1085 | 0 | return ParquetRowGroupPruneReason::NONE; |
1086 | 0 | } |
1087 | 32 | const auto conjuncts_by_slot = collect_conjuncts_by_single_slot( |
1088 | 32 | request.conjuncts, expr_zonemap::single_slot_bloom_filter_index); |
1089 | 32 | for (const auto& [slot_index, conjuncts] : conjuncts_by_slot) { |
1090 | 4 | const auto file_column_id = file_column_id_by_block_position(request, slot_index); |
1091 | 4 | if (!file_column_id.has_value()) { |
1092 | 0 | continue; |
1093 | 0 | } |
1094 | 4 | const auto* column_schema = resolve_local_leaf_schema(file_schema, *file_column_id); |
1095 | 4 | if (column_schema == nullptr || column_schema->type == nullptr || |
1096 | 4 | !bloom_filter_supported(*column_schema)) { |
1097 | 0 | continue; |
1098 | 0 | } |
1099 | 4 | auto* bloom_filter = bloom_filter_cache->get(row_group_idx, column_schema->leaf_column_id, |
1100 | 4 | pruning_stats); |
1101 | 4 | if (bloom_filter == nullptr) { |
1102 | 2 | continue; |
1103 | 2 | } |
1104 | 2 | if (ParquetStatisticsUtils::BloomFilterExcludes(*column_schema, slot_index, conjuncts, |
1105 | 2 | *bloom_filter)) { |
1106 | 1 | return ParquetRowGroupPruneReason::BLOOM_FILTER; |
1107 | 1 | } |
1108 | 2 | } |
1109 | 31 | return ParquetRowGroupPruneReason::NONE; |
1110 | 32 | } |
1111 | | |
1112 | | void init_bloom_filter_cache(::parquet::ParquetFileReader* file_reader, bool enable_bloom_filter, |
1113 | 22 | RowGroupBloomFilterCache* bloom_filter_cache) { |
1114 | 22 | DORIS_CHECK(bloom_filter_cache != nullptr); |
1115 | 22 | if (!enable_bloom_filter || file_reader == nullptr) { |
1116 | 19 | return; |
1117 | 19 | } |
1118 | 3 | try { |
1119 | 3 | bloom_filter_cache->bloom_filter_reader = &file_reader->GetBloomFilterReader(); |
1120 | 3 | } catch (const ::parquet::ParquetException&) { |
1121 | 0 | bloom_filter_cache->bloom_filter_reader = nullptr; |
1122 | 0 | } catch (const std::exception&) { |
1123 | 0 | bloom_filter_cache->bloom_filter_reader = nullptr; |
1124 | 0 | } |
1125 | 3 | } |
1126 | | |
1127 | | bool check_statistics(const ::parquet::RowGroupMetaData& row_group, |
1128 | | const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema, |
1129 | | const format::FileScanRequest& request, ParquetPruningStats* pruning_stats, |
1130 | 18 | const cctz::time_zone* timezone) { |
1131 | 18 | const auto slot_indexes = collect_expr_zonemap_slot_indexes(request.conjuncts); |
1132 | 18 | if (slot_indexes.empty()) { |
1133 | 0 | return false; |
1134 | 0 | } |
1135 | | |
1136 | 18 | ZoneMapEvalContext ctx; |
1137 | 19 | for (const int slot_index : slot_indexes) { |
1138 | 19 | const auto file_column_id = file_column_id_by_block_position(request, slot_index); |
1139 | 19 | if (!file_column_id.has_value()) { |
1140 | 0 | continue; |
1141 | 0 | } |
1142 | 19 | const auto* column_schema = resolve_local_leaf_schema(file_schema, *file_column_id); |
1143 | 19 | if (column_schema == nullptr || column_schema->type == nullptr) { |
1144 | 1 | continue; |
1145 | 1 | } |
1146 | | |
1147 | 18 | std::shared_ptr<segment_v2::ZoneMap> zone_map; |
1148 | 18 | DCHECK_LT(column_schema->leaf_column_id, row_group.num_columns()); |
1149 | 18 | auto column_chunk = row_group.ColumnChunk(column_schema->leaf_column_id); |
1150 | 18 | if (column_chunk != nullptr) { |
1151 | 18 | zone_map = ParquetStatisticsUtils::MakeZoneMap( |
1152 | 18 | ParquetStatisticsUtils::TransformColumnStatistics( |
1153 | 18 | *column_schema, column_chunk->statistics(), timezone)); |
1154 | 18 | } |
1155 | 18 | add_slot_zonemap(&ctx, slot_index, column_schema->type, std::move(zone_map)); |
1156 | 18 | } |
1157 | | |
1158 | 18 | const auto result = VExprContext::evaluate_zonemap_filter(request.conjuncts, ctx); |
1159 | 18 | accumulate_zonemap_stats(ctx, pruning_stats); |
1160 | 18 | return result == ZoneMapFilterResult::kNoMatch; |
1161 | 18 | } |
1162 | | |
1163 | | void collect_filtered_leaf_ids(const ParquetColumnSchema& column_schema, |
1164 | | const format::LocalColumnIndex* projection, |
1165 | 45 | std::set<int>* leaf_column_ids) { |
1166 | 45 | if (column_schema.kind == ParquetColumnSchemaKind::PRIMITIVE) { |
1167 | 45 | if (column_schema.leaf_column_id >= 0) { |
1168 | 45 | leaf_column_ids->insert(column_schema.leaf_column_id); |
1169 | 45 | } |
1170 | 45 | return; |
1171 | 45 | } |
1172 | 0 | for (const auto& child_schema : column_schema.children) { |
1173 | 0 | if (!format::is_child_projected(projection, child_schema->local_id)) { |
1174 | 0 | continue; |
1175 | 0 | } |
1176 | 0 | collect_filtered_leaf_ids(*child_schema, |
1177 | 0 | format::find_child_projection(projection, child_schema->local_id), |
1178 | 0 | leaf_column_ids); |
1179 | 0 | } |
1180 | 0 | } |
1181 | | |
1182 | | int64_t requested_compressed_bytes( |
1183 | | const ::parquet::RowGroupMetaData& row_group, |
1184 | | const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema, |
1185 | 19 | const format::FileScanRequest& request) { |
1186 | 19 | std::set<int> leaf_column_ids; |
1187 | 19 | auto collect_projection = [&](const format::LocalColumnIndex& projection) { |
1188 | 5 | const int32_t local_id = projection.local_id(); |
1189 | 5 | if (local_id < 0 || local_id >= static_cast<int32_t>(file_schema.size()) || |
1190 | 5 | file_schema[local_id] == nullptr) { |
1191 | 0 | return; |
1192 | 0 | } |
1193 | 5 | collect_filtered_leaf_ids(*file_schema[local_id], &projection, &leaf_column_ids); |
1194 | 5 | }; |
1195 | 19 | for (const auto& projection : request.predicate_columns) { |
1196 | 5 | collect_projection(projection); |
1197 | 5 | } |
1198 | 19 | for (const auto& projection : request.non_predicate_columns) { |
1199 | 0 | collect_projection(projection); |
1200 | 0 | } |
1201 | | |
1202 | 19 | int64_t bytes = 0; |
1203 | 19 | for (const int leaf_column_id : leaf_column_ids) { |
1204 | 5 | if (leaf_column_id < 0 || leaf_column_id >= row_group.num_columns()) { |
1205 | 0 | continue; |
1206 | 0 | } |
1207 | 5 | const auto column_chunk = row_group.ColumnChunk(leaf_column_id); |
1208 | 5 | if (column_chunk != nullptr && column_chunk->total_compressed_size() > 0) { |
1209 | 5 | bytes += column_chunk->total_compressed_size(); |
1210 | 5 | } |
1211 | 5 | } |
1212 | 19 | return bytes; |
1213 | 19 | } |
1214 | | |
1215 | | Status select_row_groups_by_metadata_impl( |
1216 | | const ::parquet::FileMetaData& metadata, ::parquet::ParquetFileReader* file_reader, |
1217 | | const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema, |
1218 | | const format::FileScanRequest& request, const std::vector<int>* candidate_row_groups, |
1219 | | std::vector<int>* selected_row_groups, bool enable_bloom_filter, |
1220 | | ParquetPruningStats* pruning_stats, const cctz::time_zone* timezone, |
1221 | 22 | const RuntimeState* runtime_state) { |
1222 | 22 | int64_t row_group_filter_time_sink = 0; |
1223 | 22 | SCOPED_RAW_TIMER(pruning_stats == nullptr ? &row_group_filter_time_sink |
1224 | 22 | : &pruning_stats->row_group_filter_time); |
1225 | 22 | if (selected_row_groups == nullptr) { |
1226 | 0 | return Status::InvalidArgument("selected_row_groups is null"); |
1227 | 0 | } |
1228 | 22 | selected_row_groups->clear(); |
1229 | | |
1230 | 22 | const int num_row_groups = metadata.num_row_groups(); |
1231 | 22 | const auto candidate_size = candidate_row_groups == nullptr |
1232 | 22 | ? static_cast<size_t>(num_row_groups) |
1233 | 22 | : candidate_row_groups->size(); |
1234 | 22 | if (pruning_stats != nullptr) { |
1235 | | // Scan-range ownership is decided before metadata pruning. Count only row groups owned by |
1236 | | // this split so a file divided into multiple splits does not report the full-file total and |
1237 | | // out-of-split groups once per split. |
1238 | 22 | pruning_stats->total_row_groups = cast_set<int64_t>(candidate_size); |
1239 | 22 | } |
1240 | 22 | selected_row_groups->reserve(candidate_size); |
1241 | 22 | RowGroupBloomFilterCache bloom_filter_cache; |
1242 | 22 | init_bloom_filter_cache(file_reader, enable_bloom_filter, &bloom_filter_cache); |
1243 | 72 | for (size_t candidate_idx = 0; candidate_idx < candidate_size; ++candidate_idx) { |
1244 | 50 | const int row_group_idx = candidate_row_groups == nullptr |
1245 | 50 | ? static_cast<int>(candidate_idx) |
1246 | 50 | : (*candidate_row_groups)[candidate_idx]; |
1247 | 50 | DORIS_CHECK(row_group_idx >= 0); |
1248 | 50 | DORIS_CHECK(row_group_idx < num_row_groups); |
1249 | 50 | auto row_group = metadata.RowGroup(row_group_idx); |
1250 | 50 | if (row_group == nullptr) { |
1251 | 0 | selected_row_groups->push_back(row_group_idx); |
1252 | 0 | continue; |
1253 | 0 | } |
1254 | 50 | ParquetRowGroupPruneReason prune_reason = ParquetRowGroupPruneReason::NONE; |
1255 | 50 | if (has_expr_zonemap_filter(request, runtime_state) && |
1256 | 50 | check_statistics(*row_group, file_schema, request, pruning_stats, timezone)) { |
1257 | 8 | prune_reason = ParquetRowGroupPruneReason::STATISTICS; |
1258 | 8 | } |
1259 | | |
1260 | 50 | if (prune_reason == ParquetRowGroupPruneReason::NONE) { |
1261 | 42 | prune_reason = dictionary_prune_reason(*row_group, file_reader, row_group_idx, |
1262 | 42 | file_schema, request); |
1263 | 42 | if (prune_reason == ParquetRowGroupPruneReason::NONE) { |
1264 | 32 | prune_reason = bloom_filter_prune_reason(row_group_idx, file_schema, request, |
1265 | 32 | &bloom_filter_cache, pruning_stats); |
1266 | 32 | } |
1267 | 42 | } |
1268 | | |
1269 | 50 | if (prune_reason != ParquetRowGroupPruneReason::NONE) { |
1270 | 19 | if (pruning_stats != nullptr) { |
1271 | 19 | pruning_stats->filtered_group_rows += row_group->num_rows(); |
1272 | | // FilteredBytes must describe the IO actually avoided by this scan projection; |
1273 | | // counting every physical child overstates savings for nested-column projection. |
1274 | 19 | pruning_stats->filtered_bytes += |
1275 | 19 | requested_compressed_bytes(*row_group, file_schema, request); |
1276 | 19 | if (prune_reason == ParquetRowGroupPruneReason::STATISTICS) { |
1277 | 8 | ++pruning_stats->filtered_row_groups_by_statistics; |
1278 | 11 | } else if (prune_reason == ParquetRowGroupPruneReason::DICTIONARY) { |
1279 | 10 | ++pruning_stats->filtered_row_groups_by_dictionary; |
1280 | 10 | } else if (prune_reason == ParquetRowGroupPruneReason::BLOOM_FILTER) { |
1281 | 1 | ++pruning_stats->filtered_row_groups_by_bloom_filter; |
1282 | 1 | } |
1283 | 19 | } |
1284 | 19 | continue; |
1285 | 19 | } |
1286 | 31 | selected_row_groups->push_back(row_group_idx); |
1287 | 31 | } |
1288 | 22 | return Status::OK(); |
1289 | 22 | } |
1290 | | |
1291 | | } // namespace |
1292 | | |
1293 | | Status select_row_groups_by_metadata( |
1294 | | const ::parquet::FileMetaData& metadata, ::parquet::ParquetFileReader* file_reader, |
1295 | | const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema, |
1296 | | const format::FileScanRequest& request, const std::vector<int>* candidate_row_groups, |
1297 | | std::vector<int>* selected_row_groups, bool enable_bloom_filter, |
1298 | | ParquetPruningStats* pruning_stats, const cctz::time_zone* timezone, |
1299 | 22 | const RuntimeState* runtime_state) { |
1300 | 22 | return select_row_groups_by_metadata_impl( |
1301 | 22 | metadata, file_reader, file_schema, request, candidate_row_groups, selected_row_groups, |
1302 | 22 | enable_bloom_filter, pruning_stats, timezone, runtime_state); |
1303 | 22 | } |
1304 | | |
1305 | | namespace { |
1306 | | |
1307 | 90 | bool native_metadata_predicate_is_type_safe(const ParquetColumnSchema& column_schema) { |
1308 | 90 | DORIS_CHECK(column_schema.type != nullptr); |
1309 | | // Raw VARBINARY file slots may feed table-side STRING casts. Footer/page metadata is still in |
1310 | | // the pre-cast domain, so using it for a rewritten table predicate can cause false negatives. |
1311 | 90 | return remove_nullable(column_schema.type)->get_primitive_type() != TYPE_VARBINARY; |
1312 | 90 | } |
1313 | | |
1314 | | bool check_native_statistics(const tparquet::RowGroup& row_group, |
1315 | | const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema, |
1316 | | const format::FileScanRequest& request, |
1317 | 60 | ParquetPruningStats* pruning_stats, const cctz::time_zone* timezone) { |
1318 | 60 | const auto slot_indexes = collect_expr_zonemap_slot_indexes(request.conjuncts); |
1319 | 60 | if (slot_indexes.empty()) { |
1320 | 0 | return false; |
1321 | 0 | } |
1322 | 60 | ZoneMapEvalContext ctx; |
1323 | 62 | for (const int slot_index : slot_indexes) { |
1324 | 62 | const auto file_column_id = file_column_id_by_block_position(request, slot_index); |
1325 | 62 | if (!file_column_id.has_value()) { |
1326 | 0 | continue; |
1327 | 0 | } |
1328 | 62 | const auto* column_schema = resolve_local_leaf_schema(file_schema, *file_column_id); |
1329 | 62 | if (column_schema == nullptr || column_schema->type == nullptr || |
1330 | 62 | !native_metadata_predicate_is_type_safe(*column_schema) || |
1331 | 62 | column_schema->leaf_column_id >= static_cast<int>(row_group.columns.size())) { |
1332 | 0 | continue; |
1333 | 0 | } |
1334 | 62 | const auto& chunk = row_group.columns[column_schema->leaf_column_id]; |
1335 | 62 | std::shared_ptr<segment_v2::ZoneMap> zone_map; |
1336 | 62 | if (chunk.__isset.meta_data) { |
1337 | 62 | const auto& column_metadata = chunk.meta_data; |
1338 | 62 | const auto* statistics = |
1339 | 62 | column_metadata.__isset.statistics ? &column_metadata.statistics : nullptr; |
1340 | 62 | if (statistics != nullptr && !detail::can_use_native_footer_min_max( |
1341 | 50 | column_schema->type_descriptor, *statistics)) { |
1342 | 0 | statistics = nullptr; |
1343 | 0 | } |
1344 | 62 | zone_map = ParquetStatisticsUtils::MakeZoneMap( |
1345 | 62 | ParquetStatisticsUtils::TransformColumnStatistics( |
1346 | 62 | *column_schema, statistics, column_metadata.num_values, timezone)); |
1347 | 62 | } |
1348 | 62 | add_slot_zonemap(&ctx, slot_index, column_schema->type, std::move(zone_map)); |
1349 | 62 | } |
1350 | 60 | const auto result = VExprContext::evaluate_zonemap_filter(request.conjuncts, ctx); |
1351 | 60 | accumulate_zonemap_stats(ctx, pruning_stats); |
1352 | 60 | return result == ZoneMapFilterResult::kNoMatch; |
1353 | 60 | } |
1354 | | |
1355 | 22 | bool is_native_dictionary_data_encoding(tparquet::Encoding::type encoding) { |
1356 | 22 | return encoding == tparquet::Encoding::PLAIN_DICTIONARY || |
1357 | 22 | encoding == tparquet::Encoding::RLE_DICTIONARY; |
1358 | 22 | } |
1359 | | |
1360 | 0 | bool is_native_level_encoding(tparquet::Encoding::type encoding) { |
1361 | 0 | return encoding == tparquet::Encoding::RLE || encoding == tparquet::Encoding::BIT_PACKED; |
1362 | 0 | } |
1363 | | |
1364 | 22 | bool is_native_dictionary_encoded_chunk(const tparquet::ColumnMetaData& metadata) { |
1365 | 22 | if (!metadata.__isset.dictionary_page_offset || metadata.dictionary_page_offset < 0) { |
1366 | 0 | return false; |
1367 | 0 | } |
1368 | 22 | if (metadata.__isset.encoding_stats && !metadata.encoding_stats.empty()) { |
1369 | 22 | bool has_dictionary_data_page = false; |
1370 | 44 | for (const auto& encoding_stat : metadata.encoding_stats) { |
1371 | 44 | if ((encoding_stat.page_type != tparquet::PageType::DATA_PAGE && |
1372 | 44 | encoding_stat.page_type != tparquet::PageType::DATA_PAGE_V2) || |
1373 | 44 | encoding_stat.count <= 0) { |
1374 | 22 | continue; |
1375 | 22 | } |
1376 | 22 | if (!is_native_dictionary_data_encoding(encoding_stat.encoding)) { |
1377 | 0 | return false; |
1378 | 0 | } |
1379 | 22 | has_dictionary_data_page = true; |
1380 | 22 | } |
1381 | 22 | return has_dictionary_data_page; |
1382 | 22 | } |
1383 | 0 | bool has_dictionary_encoding = false; |
1384 | 0 | for (const auto encoding : metadata.encodings) { |
1385 | 0 | if (is_native_dictionary_data_encoding(encoding)) { |
1386 | 0 | has_dictionary_encoding = true; |
1387 | 0 | } else if (!is_native_level_encoding(encoding)) { |
1388 | 0 | return false; |
1389 | 0 | } |
1390 | 0 | } |
1391 | 0 | return has_dictionary_encoding; |
1392 | 0 | } |
1393 | | |
1394 | | const format::LocalColumnIndex* find_request_projection(const format::FileScanRequest& request, |
1395 | 23 | format::LocalColumnId file_column_id) { |
1396 | 23 | for (const auto& projection : request.predicate_columns) { |
1397 | 23 | if (projection.local_id() == file_column_id.value()) { |
1398 | 23 | return &projection; |
1399 | 23 | } |
1400 | 23 | } |
1401 | 0 | for (const auto& projection : request.non_predicate_columns) { |
1402 | 0 | if (projection.local_id() == file_column_id.value()) { |
1403 | 0 | return &projection; |
1404 | 0 | } |
1405 | 0 | } |
1406 | 0 | return nullptr; |
1407 | 0 | } |
1408 | | |
1409 | | ParquetRowGroupPruneReason native_dictionary_prune_reason( |
1410 | | const tparquet::RowGroup& row_group, int row_group_idx, |
1411 | | const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema, |
1412 | | const format::FileScanRequest& request, const cctz::time_zone* timezone, |
1413 | 216 | ParquetFileContext* file_context) { |
1414 | 216 | if (file_context == nullptr || file_context->native_metadata == nullptr) { |
1415 | 1 | return ParquetRowGroupPruneReason::NONE; |
1416 | 1 | } |
1417 | 215 | const auto conjuncts_by_slot = collect_conjuncts_by_single_slot( |
1418 | 215 | request.conjuncts, expr_zonemap::single_slot_dictionary_index); |
1419 | 215 | for (const auto& [slot_index, conjuncts] : conjuncts_by_slot) { |
1420 | 23 | const auto file_column_id = file_column_id_by_block_position(request, slot_index); |
1421 | 23 | if (!file_column_id.has_value()) { |
1422 | 0 | continue; |
1423 | 0 | } |
1424 | 23 | const auto* column_schema = resolve_local_leaf_schema(file_schema, *file_column_id); |
1425 | 23 | const auto* projection = find_request_projection(request, *file_column_id); |
1426 | 23 | if (column_schema == nullptr || projection == nullptr || column_schema->type == nullptr || |
1427 | 23 | !column_schema->type_descriptor.is_string_like || |
1428 | 23 | column_schema->leaf_column_id >= static_cast<int>(row_group.columns.size())) { |
1429 | 0 | continue; |
1430 | 0 | } |
1431 | 23 | if (!native_metadata_predicate_is_type_safe(*column_schema)) { |
1432 | | // The file-local VARBINARY may feed a table-side STRING cast. Pruning before that cast |
1433 | | // can compare different Field kinds and incorrectly discard a matching row group. |
1434 | 1 | continue; |
1435 | 1 | } |
1436 | 22 | const auto& chunk = row_group.columns[column_schema->leaf_column_id]; |
1437 | 22 | if (!chunk.__isset.meta_data || |
1438 | 22 | (chunk.meta_data.type != tparquet::Type::BYTE_ARRAY && |
1439 | 22 | chunk.meta_data.type != tparquet::Type::FIXED_LEN_BYTE_ARRAY) || |
1440 | 22 | !is_native_dictionary_encoded_chunk(chunk.meta_data)) { |
1441 | 0 | continue; |
1442 | 0 | } |
1443 | 22 | std::unique_ptr<ParquetColumnReader> reader; |
1444 | 22 | const std::vector<RowRange> ranges {{0, row_group.num_rows}}; |
1445 | 22 | const std::unordered_map<int, tparquet::OffsetIndex> offset_indexes; |
1446 | 22 | const auto status = NativeColumnReader::create( |
1447 | 22 | *column_schema, projection, file_context->native_file, |
1448 | 22 | file_context->native_metadata, row_group_idx, ranges, offset_indexes, timezone, |
1449 | 22 | file_context->native_io_ctx, nullptr, file_context->native_page_cache_enabled, |
1450 | 22 | file_context->native_page_cache_file_key, true, {}, &reader); |
1451 | 22 | if (!status.ok() || reader == nullptr) { |
1452 | 0 | continue; |
1453 | 0 | } |
1454 | 22 | auto dictionary_result = reader->dictionary_values(); |
1455 | 22 | if (!dictionary_result.has_value()) { |
1456 | 0 | continue; |
1457 | 0 | } |
1458 | 22 | auto dictionary = std::move(dictionary_result).value(); |
1459 | 22 | std::vector<Field> values(dictionary->size()); |
1460 | 78 | for (size_t value_idx = 0; value_idx < dictionary->size(); ++value_idx) { |
1461 | 56 | dictionary->get(value_idx, values[value_idx]); |
1462 | 56 | } |
1463 | 22 | DictionaryEvalContext ctx; |
1464 | 22 | ctx.slots.emplace(slot_index, DictionaryEvalContext::SlotDictionary { |
1465 | 22 | .data_type = column_schema->type, |
1466 | 22 | .values = std::move(values), |
1467 | 22 | }); |
1468 | 22 | if (VExprContext::evaluate_dictionary_filter(conjuncts, ctx) == |
1469 | 22 | ZoneMapFilterResult::kNoMatch) { |
1470 | 10 | return ParquetRowGroupPruneReason::DICTIONARY; |
1471 | 10 | } |
1472 | 22 | } |
1473 | 205 | return ParquetRowGroupPruneReason::NONE; |
1474 | 215 | } |
1475 | | |
1476 | | ParquetRowGroupPruneReason native_bloom_filter_prune_reason( |
1477 | | const tparquet::RowGroup& row_group, |
1478 | | const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema, |
1479 | | const format::FileScanRequest& request, ParquetFileContext* file_context, |
1480 | 206 | ParquetPruningStats* pruning_stats) { |
1481 | 206 | if (file_context == nullptr || file_context->native_file == nullptr) { |
1482 | 0 | return ParquetRowGroupPruneReason::NONE; |
1483 | 0 | } |
1484 | 206 | const auto conjuncts_by_slot = collect_conjuncts_by_single_slot( |
1485 | 206 | request.conjuncts, expr_zonemap::single_slot_bloom_filter_index); |
1486 | 206 | for (const auto& [slot_index, conjuncts] : conjuncts_by_slot) { |
1487 | 1 | const auto file_column_id = file_column_id_by_block_position(request, slot_index); |
1488 | 1 | if (!file_column_id.has_value()) { |
1489 | 0 | continue; |
1490 | 0 | } |
1491 | 1 | const auto* column_schema = resolve_local_leaf_schema(file_schema, *file_column_id); |
1492 | 1 | if (column_schema == nullptr || column_schema->type == nullptr || |
1493 | 1 | !native_metadata_predicate_is_type_safe(*column_schema) || |
1494 | 1 | !bloom_filter_supported(*column_schema) || |
1495 | 1 | column_schema->leaf_column_id >= static_cast<int>(row_group.columns.size())) { |
1496 | 0 | continue; |
1497 | 0 | } |
1498 | 1 | const auto& chunk = row_group.columns[column_schema->leaf_column_id]; |
1499 | 1 | if (!chunk.__isset.meta_data) { |
1500 | 0 | continue; |
1501 | 0 | } |
1502 | 1 | std::unique_ptr<native::BlockSplitBloomFilter> bloom_filter; |
1503 | 1 | Status status; |
1504 | 1 | { |
1505 | 1 | int64_t timer_sink = 0; |
1506 | 1 | SCOPED_RAW_TIMER(pruning_stats == nullptr ? &timer_sink |
1507 | 1 | : &pruning_stats->bloom_filter_read_time); |
1508 | 1 | status = read_native_bloom_filter(chunk.meta_data, file_context->native_file, |
1509 | 1 | file_context->native_io_ctx, &bloom_filter); |
1510 | 1 | } |
1511 | 1 | if (!status.ok() || bloom_filter == nullptr) { |
1512 | 0 | continue; |
1513 | 0 | } |
1514 | 1 | if (ParquetStatisticsUtils::NativeBloomFilterExcludes(*column_schema, slot_index, conjuncts, |
1515 | 1 | *bloom_filter)) { |
1516 | 0 | return ParquetRowGroupPruneReason::BLOOM_FILTER; |
1517 | 0 | } |
1518 | 1 | } |
1519 | 206 | return ParquetRowGroupPruneReason::NONE; |
1520 | 206 | } |
1521 | | |
1522 | | int64_t native_requested_compressed_bytes( |
1523 | | const tparquet::RowGroup& row_group, |
1524 | | const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema, |
1525 | 28 | const format::FileScanRequest& request) { |
1526 | 28 | std::set<int> leaf_column_ids; |
1527 | 42 | auto collect_projection = [&](const format::LocalColumnIndex& projection) { |
1528 | 42 | const int32_t local_id = projection.local_id(); |
1529 | 42 | if (local_id < 0 || local_id >= static_cast<int32_t>(file_schema.size()) || |
1530 | 42 | file_schema[local_id] == nullptr) { |
1531 | 2 | return; |
1532 | 2 | } |
1533 | 40 | collect_filtered_leaf_ids(*file_schema[local_id], &projection, &leaf_column_ids); |
1534 | 40 | }; |
1535 | 28 | for (const auto& projection : request.predicate_columns) { |
1536 | 23 | collect_projection(projection); |
1537 | 23 | } |
1538 | 28 | for (const auto& projection : request.non_predicate_columns) { |
1539 | 19 | collect_projection(projection); |
1540 | 19 | } |
1541 | 28 | int64_t bytes = 0; |
1542 | 40 | for (const int leaf_column_id : leaf_column_ids) { |
1543 | 40 | if (leaf_column_id < 0 || leaf_column_id >= static_cast<int>(row_group.columns.size())) { |
1544 | 0 | continue; |
1545 | 0 | } |
1546 | 40 | const auto& chunk = row_group.columns[leaf_column_id]; |
1547 | 40 | if (chunk.__isset.meta_data && chunk.meta_data.total_compressed_size > 0) { |
1548 | 40 | bytes += chunk.meta_data.total_compressed_size; |
1549 | 40 | } |
1550 | 40 | } |
1551 | 28 | return bytes; |
1552 | 28 | } |
1553 | | |
1554 | | } // namespace |
1555 | | |
1556 | | Status select_row_groups_by_metadata( |
1557 | | const tparquet::FileMetaData& metadata, |
1558 | | const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema, |
1559 | | const format::FileScanRequest& request, const std::vector<int>* candidate_row_groups, |
1560 | | std::vector<int>* selected_row_groups, bool enable_bloom_filter, |
1561 | | ParquetPruningStats* pruning_stats, const cctz::time_zone* timezone, |
1562 | 176 | const RuntimeState* runtime_state, ParquetFileContext* file_context) { |
1563 | 176 | int64_t timer_sink = 0; |
1564 | 176 | SCOPED_RAW_TIMER(pruning_stats == nullptr ? &timer_sink |
1565 | 176 | : &pruning_stats->row_group_filter_time); |
1566 | 176 | if (selected_row_groups == nullptr) { |
1567 | 0 | return Status::InvalidArgument("selected_row_groups is null"); |
1568 | 0 | } |
1569 | 176 | selected_row_groups->clear(); |
1570 | 176 | const size_t candidate_size = candidate_row_groups == nullptr ? metadata.row_groups.size() |
1571 | 176 | : candidate_row_groups->size(); |
1572 | 176 | if (pruning_stats != nullptr) { |
1573 | 176 | pruning_stats->total_row_groups = cast_set<int64_t>(candidate_size); |
1574 | 176 | } |
1575 | 176 | selected_row_groups->reserve(candidate_size); |
1576 | 410 | for (size_t candidate_idx = 0; candidate_idx < candidate_size; ++candidate_idx) { |
1577 | 234 | const int row_group_idx = candidate_row_groups == nullptr |
1578 | 234 | ? static_cast<int>(candidate_idx) |
1579 | 234 | : (*candidate_row_groups)[candidate_idx]; |
1580 | 234 | DORIS_CHECK(row_group_idx >= 0 && |
1581 | 234 | row_group_idx < static_cast<int>(metadata.row_groups.size())); |
1582 | 234 | const auto& row_group = metadata.row_groups[row_group_idx]; |
1583 | 234 | ParquetRowGroupPruneReason prune_reason = ParquetRowGroupPruneReason::NONE; |
1584 | 234 | if (has_expr_zonemap_filter(request, runtime_state) && |
1585 | 234 | check_native_statistics(row_group, file_schema, request, pruning_stats, timezone)) { |
1586 | 18 | prune_reason = ParquetRowGroupPruneReason::STATISTICS; |
1587 | 18 | } |
1588 | 234 | if (prune_reason == ParquetRowGroupPruneReason::NONE) { |
1589 | 216 | prune_reason = native_dictionary_prune_reason(row_group, row_group_idx, file_schema, |
1590 | 216 | request, timezone, file_context); |
1591 | 216 | } |
1592 | 234 | if (prune_reason == ParquetRowGroupPruneReason::NONE && enable_bloom_filter) { |
1593 | 206 | prune_reason = native_bloom_filter_prune_reason(row_group, file_schema, request, |
1594 | 206 | file_context, pruning_stats); |
1595 | 206 | } |
1596 | 234 | if (prune_reason == ParquetRowGroupPruneReason::NONE) { |
1597 | 206 | selected_row_groups->push_back(row_group_idx); |
1598 | 206 | continue; |
1599 | 206 | } |
1600 | 28 | if (pruning_stats != nullptr) { |
1601 | 28 | pruning_stats->filtered_group_rows += row_group.num_rows; |
1602 | 28 | pruning_stats->filtered_bytes += |
1603 | 28 | native_requested_compressed_bytes(row_group, file_schema, request); |
1604 | 28 | if (prune_reason == ParquetRowGroupPruneReason::STATISTICS) { |
1605 | 18 | ++pruning_stats->filtered_row_groups_by_statistics; |
1606 | 18 | } else if (prune_reason == ParquetRowGroupPruneReason::DICTIONARY) { |
1607 | 10 | ++pruning_stats->filtered_row_groups_by_dictionary; |
1608 | 10 | } else { |
1609 | 0 | ++pruning_stats->filtered_row_groups_by_bloom_filter; |
1610 | 0 | } |
1611 | 28 | } |
1612 | 28 | } |
1613 | 176 | return Status::OK(); |
1614 | 176 | } |
1615 | | |
1616 | | namespace { |
1617 | | |
1618 | | template <typename ParquetDType> |
1619 | | bool set_page_decoded_min_max(const std::shared_ptr<::parquet::ColumnIndex>& column_index, |
1620 | | const ParquetColumnSchema& column_schema, size_t page_idx, |
1621 | | DecodedValueKind value_kind, ParquetColumnStatistics* page_statistics, |
1622 | 87 | const cctz::time_zone* timezone) { |
1623 | 87 | const auto typed_index = |
1624 | 87 | std::static_pointer_cast<::parquet::TypedColumnIndex<ParquetDType>>(column_index); |
1625 | 87 | if (page_idx >= typed_index->min_values().size() || |
1626 | 87 | page_idx >= typed_index->max_values().size()) { |
1627 | 0 | return false; |
1628 | 0 | } |
1629 | 87 | const auto& min_value = typed_index->min_values()[page_idx]; |
1630 | 87 | const auto& max_value = typed_index->max_values()[page_idx]; |
1631 | 87 | if constexpr (std::is_same_v<ParquetDType, ::parquet::Int64Type>) { |
1632 | 0 | if (!timestamp_min_max_is_safe(column_schema, min_value, max_value, timezone)) { |
1633 | 0 | return false; |
1634 | 0 | } |
1635 | 0 | } |
1636 | 87 | if (!valid_min_max(min_value, max_value)) { |
1637 | | // A NaN invalidates only this page's bounds, not the ColumnIndex itself. Keep the page |
1638 | | // conservatively by returning usable null-count statistics with has_min_max=false, while |
1639 | | // allowing later pages with finite bounds to remain eligible for pruning. |
1640 | 4 | return true; |
1641 | 4 | } |
1642 | 83 | if (!set_decoded_field(column_schema, value_kind, min_value, &page_statistics->min_value, |
1643 | 83 | timezone) || |
1644 | 83 | !set_decoded_field(column_schema, value_kind, max_value, &page_statistics->max_value, |
1645 | 83 | timezone)) { |
1646 | 0 | return false; |
1647 | 0 | } |
1648 | 83 | if (!decoded_min_max_is_ordered(*page_statistics)) { |
1649 | 1 | return true; |
1650 | 1 | } |
1651 | 82 | page_statistics->has_min_max = true; |
1652 | 82 | return true; |
1653 | 83 | } 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 | 1622 | 81 | const cctz::time_zone* timezone) { | 1623 | 81 | const auto typed_index = | 1624 | 81 | std::static_pointer_cast<::parquet::TypedColumnIndex<ParquetDType>>(column_index); | 1625 | 81 | if (page_idx >= typed_index->min_values().size() || | 1626 | 81 | page_idx >= typed_index->max_values().size()) { | 1627 | 0 | return false; | 1628 | 0 | } | 1629 | 81 | const auto& min_value = typed_index->min_values()[page_idx]; | 1630 | 81 | const auto& max_value = typed_index->max_values()[page_idx]; | 1631 | | if constexpr (std::is_same_v<ParquetDType, ::parquet::Int64Type>) { | 1632 | | if (!timestamp_min_max_is_safe(column_schema, min_value, max_value, timezone)) { | 1633 | | return false; | 1634 | | } | 1635 | | } | 1636 | 81 | if (!valid_min_max(min_value, max_value)) { | 1637 | | // A NaN invalidates only this page's bounds, not the ColumnIndex itself. Keep the page | 1638 | | // conservatively by returning usable null-count statistics with has_min_max=false, while | 1639 | | // allowing later pages with finite bounds to remain eligible for pruning. | 1640 | 0 | return true; | 1641 | 0 | } | 1642 | 81 | if (!set_decoded_field(column_schema, value_kind, min_value, &page_statistics->min_value, | 1643 | 81 | timezone) || | 1644 | 81 | !set_decoded_field(column_schema, value_kind, max_value, &page_statistics->max_value, | 1645 | 81 | timezone)) { | 1646 | 0 | return false; | 1647 | 0 | } | 1648 | 81 | if (!decoded_min_max_is_ordered(*page_statistics)) { | 1649 | 1 | return true; | 1650 | 1 | } | 1651 | 80 | page_statistics->has_min_max = true; | 1652 | 80 | return true; | 1653 | 81 | } |
Unexecuted instantiation: parquet_statistics.cpp:_ZN5doris6format7parquet12_GLOBAL__N_124set_page_decoded_min_maxIN7parquet12PhysicalTypeILNS4_4Type4typeE2EEEEEbRKSt10shared_ptrINS4_11ColumnIndexEERKNS1_19ParquetColumnSchemaEmNS_16DecodedValueKindEPNS1_23ParquetColumnStatisticsEPKN4cctz9time_zoneE parquet_statistics.cpp:_ZN5doris6format7parquet12_GLOBAL__N_124set_page_decoded_min_maxIN7parquet12PhysicalTypeILNS4_4Type4typeE4EEEEEbRKSt10shared_ptrINS4_11ColumnIndexEERKNS1_19ParquetColumnSchemaEmNS_16DecodedValueKindEPNS1_23ParquetColumnStatisticsEPKN4cctz9time_zoneE Line | Count | Source | 1622 | 1 | const cctz::time_zone* timezone) { | 1623 | 1 | const auto typed_index = | 1624 | 1 | std::static_pointer_cast<::parquet::TypedColumnIndex<ParquetDType>>(column_index); | 1625 | 1 | if (page_idx >= typed_index->min_values().size() || | 1626 | 1 | page_idx >= typed_index->max_values().size()) { | 1627 | 0 | return false; | 1628 | 0 | } | 1629 | 1 | const auto& min_value = typed_index->min_values()[page_idx]; | 1630 | 1 | const auto& max_value = typed_index->max_values()[page_idx]; | 1631 | | if constexpr (std::is_same_v<ParquetDType, ::parquet::Int64Type>) { | 1632 | | if (!timestamp_min_max_is_safe(column_schema, min_value, max_value, timezone)) { | 1633 | | return false; | 1634 | | } | 1635 | | } | 1636 | 1 | if (!valid_min_max(min_value, max_value)) { | 1637 | | // A NaN invalidates only this page's bounds, not the ColumnIndex itself. Keep the page | 1638 | | // conservatively by returning usable null-count statistics with has_min_max=false, while | 1639 | | // allowing later pages with finite bounds to remain eligible for pruning. | 1640 | 1 | return true; | 1641 | 1 | } | 1642 | 0 | if (!set_decoded_field(column_schema, value_kind, min_value, &page_statistics->min_value, | 1643 | 0 | timezone) || | 1644 | 0 | !set_decoded_field(column_schema, value_kind, max_value, &page_statistics->max_value, | 1645 | 0 | timezone)) { | 1646 | 0 | return false; | 1647 | 0 | } | 1648 | 0 | if (!decoded_min_max_is_ordered(*page_statistics)) { | 1649 | 0 | return true; | 1650 | 0 | } | 1651 | 0 | page_statistics->has_min_max = true; | 1652 | 0 | return true; | 1653 | 0 | } |
parquet_statistics.cpp:_ZN5doris6format7parquet12_GLOBAL__N_124set_page_decoded_min_maxIN7parquet12PhysicalTypeILNS4_4Type4typeE5EEEEEbRKSt10shared_ptrINS4_11ColumnIndexEERKNS1_19ParquetColumnSchemaEmNS_16DecodedValueKindEPNS1_23ParquetColumnStatisticsEPKN4cctz9time_zoneE Line | Count | Source | 1622 | 5 | const cctz::time_zone* timezone) { | 1623 | 5 | const auto typed_index = | 1624 | 5 | std::static_pointer_cast<::parquet::TypedColumnIndex<ParquetDType>>(column_index); | 1625 | 5 | if (page_idx >= typed_index->min_values().size() || | 1626 | 5 | page_idx >= typed_index->max_values().size()) { | 1627 | 0 | return false; | 1628 | 0 | } | 1629 | 5 | const auto& min_value = typed_index->min_values()[page_idx]; | 1630 | 5 | const auto& max_value = typed_index->max_values()[page_idx]; | 1631 | | if constexpr (std::is_same_v<ParquetDType, ::parquet::Int64Type>) { | 1632 | | if (!timestamp_min_max_is_safe(column_schema, min_value, max_value, timezone)) { | 1633 | | return false; | 1634 | | } | 1635 | | } | 1636 | 5 | if (!valid_min_max(min_value, max_value)) { | 1637 | | // A NaN invalidates only this page's bounds, not the ColumnIndex itself. Keep the page | 1638 | | // conservatively by returning usable null-count statistics with has_min_max=false, while | 1639 | | // allowing later pages with finite bounds to remain eligible for pruning. | 1640 | 3 | return true; | 1641 | 3 | } | 1642 | 2 | if (!set_decoded_field(column_schema, value_kind, min_value, &page_statistics->min_value, | 1643 | 2 | timezone) || | 1644 | 2 | !set_decoded_field(column_schema, value_kind, max_value, &page_statistics->max_value, | 1645 | 2 | timezone)) { | 1646 | 0 | return false; | 1647 | 0 | } | 1648 | 2 | if (!decoded_min_max_is_ordered(*page_statistics)) { | 1649 | 0 | return true; | 1650 | 0 | } | 1651 | 2 | page_statistics->has_min_max = true; | 1652 | 2 | return true; | 1653 | 2 | } |
|
1654 | | |
1655 | | bool set_page_string_min_max(const std::shared_ptr<::parquet::ColumnIndex>& column_index, |
1656 | | const ParquetColumnSchema& column_schema, size_t page_idx, |
1657 | | ParquetColumnStatistics* page_statistics, |
1658 | 0 | const cctz::time_zone* timezone) { |
1659 | 0 | switch (column_schema.type_descriptor.physical_type) { |
1660 | 0 | case ::parquet::Type::BYTE_ARRAY: { |
1661 | 0 | const auto typed_index = |
1662 | 0 | std::static_pointer_cast<::parquet::ByteArrayColumnIndex>(column_index); |
1663 | 0 | if (page_idx >= typed_index->min_values().size() || |
1664 | 0 | page_idx >= typed_index->max_values().size()) { |
1665 | 0 | return false; |
1666 | 0 | } |
1667 | 0 | const auto min = ::parquet::ByteArrayToString(typed_index->min_values()[page_idx]); |
1668 | 0 | const auto max = ::parquet::ByteArrayToString(typed_index->max_values()[page_idx]); |
1669 | 0 | if (!set_decoded_binary_field(column_schema, DecodedValueKind::BINARY, |
1670 | 0 | StringRef(min.data(), min.size()), |
1671 | 0 | &page_statistics->min_value, timezone) || |
1672 | 0 | !set_decoded_binary_field(column_schema, DecodedValueKind::BINARY, |
1673 | 0 | StringRef(max.data(), max.size()), |
1674 | 0 | &page_statistics->max_value, timezone)) { |
1675 | 0 | return false; |
1676 | 0 | } |
1677 | 0 | if (!decoded_min_max_is_ordered(*page_statistics)) { |
1678 | 0 | return true; |
1679 | 0 | } |
1680 | 0 | page_statistics->has_min_max = true; |
1681 | 0 | return true; |
1682 | 0 | } |
1683 | 0 | case ::parquet::Type::FIXED_LEN_BYTE_ARRAY: { |
1684 | 0 | const int type_length = column_schema.descriptor->type_length(); |
1685 | 0 | if (type_length <= 0) { |
1686 | 0 | return false; |
1687 | 0 | } |
1688 | 0 | const auto typed_index = std::static_pointer_cast<::parquet::FLBAColumnIndex>(column_index); |
1689 | 0 | if (page_idx >= typed_index->min_values().size() || |
1690 | 0 | page_idx >= typed_index->max_values().size()) { |
1691 | 0 | return false; |
1692 | 0 | } |
1693 | 0 | const std::string min( |
1694 | 0 | reinterpret_cast<const char*>(typed_index->min_values()[page_idx].ptr), |
1695 | 0 | type_length); |
1696 | 0 | const std::string max( |
1697 | 0 | reinterpret_cast<const char*>(typed_index->max_values()[page_idx].ptr), |
1698 | 0 | type_length); |
1699 | 0 | if (!set_decoded_binary_field(column_schema, DecodedValueKind::FIXED_BINARY, |
1700 | 0 | StringRef(min.data(), min.size()), |
1701 | 0 | &page_statistics->min_value, timezone) || |
1702 | 0 | !set_decoded_binary_field(column_schema, DecodedValueKind::FIXED_BINARY, |
1703 | 0 | StringRef(max.data(), max.size()), |
1704 | 0 | &page_statistics->max_value, timezone)) { |
1705 | 0 | return false; |
1706 | 0 | } |
1707 | 0 | if (!decoded_min_max_is_ordered(*page_statistics)) { |
1708 | 0 | return true; |
1709 | 0 | } |
1710 | 0 | page_statistics->has_min_max = true; |
1711 | 0 | return true; |
1712 | 0 | } |
1713 | 0 | default: |
1714 | 0 | return false; |
1715 | 0 | } |
1716 | 0 | } |
1717 | | |
1718 | | bool set_page_min_max(const std::shared_ptr<::parquet::ColumnIndex>& column_index, |
1719 | | const ParquetColumnSchema& column_schema, size_t page_idx, |
1720 | 87 | ParquetColumnStatistics* page_statistics, const cctz::time_zone* timezone) { |
1721 | 87 | DORIS_CHECK(column_schema.type != nullptr); |
1722 | 87 | switch (column_schema.type_descriptor.physical_type) { |
1723 | 0 | case ::parquet::Type::BOOLEAN: |
1724 | 0 | return set_page_decoded_min_max<::parquet::BooleanType>(column_index, column_schema, |
1725 | 0 | page_idx, DecodedValueKind::BOOL, |
1726 | 0 | page_statistics, timezone); |
1727 | 81 | case ::parquet::Type::INT32: |
1728 | 81 | return set_page_decoded_min_max<::parquet::Int32Type>( |
1729 | 81 | column_index, column_schema, page_idx, |
1730 | 81 | decoded_value_kind(column_schema.type_descriptor), page_statistics, timezone); |
1731 | 0 | case ::parquet::Type::INT64: |
1732 | 0 | return set_page_decoded_min_max<::parquet::Int64Type>( |
1733 | 0 | column_index, column_schema, page_idx, |
1734 | 0 | decoded_value_kind(column_schema.type_descriptor), page_statistics, timezone); |
1735 | 1 | case ::parquet::Type::FLOAT: |
1736 | 1 | return set_page_decoded_min_max<::parquet::FloatType>(column_index, column_schema, page_idx, |
1737 | 1 | DecodedValueKind::FLOAT, |
1738 | 1 | page_statistics, timezone); |
1739 | 5 | case ::parquet::Type::DOUBLE: |
1740 | 5 | return set_page_decoded_min_max<::parquet::DoubleType>(column_index, column_schema, |
1741 | 5 | page_idx, DecodedValueKind::DOUBLE, |
1742 | 5 | page_statistics, timezone); |
1743 | 0 | case ::parquet::Type::BYTE_ARRAY: |
1744 | 0 | case ::parquet::Type::FIXED_LEN_BYTE_ARRAY: |
1745 | 0 | return set_page_string_min_max(column_index, column_schema, page_idx, page_statistics, |
1746 | 0 | timezone); |
1747 | 0 | default: |
1748 | 0 | return false; |
1749 | 87 | } |
1750 | 87 | } |
1751 | | |
1752 | | bool build_page_statistics(const std::shared_ptr<::parquet::ColumnIndex>& column_index, |
1753 | | const ParquetColumnSchema& column_schema, size_t page_idx, |
1754 | | ParquetColumnStatistics* page_statistics, |
1755 | 87 | const cctz::time_zone* timezone) { |
1756 | 87 | DORIS_CHECK(page_statistics != nullptr); |
1757 | 87 | *page_statistics = ParquetColumnStatistics {}; |
1758 | | |
1759 | 87 | const auto& null_pages = column_index->null_pages(); |
1760 | 87 | if (!column_index->has_null_counts() || page_idx >= null_pages.size() || |
1761 | 87 | page_idx >= column_index->null_counts().size()) { |
1762 | 0 | return false; |
1763 | 0 | } |
1764 | | |
1765 | 87 | page_statistics->has_null_count = true; |
1766 | 87 | page_statistics->has_null = column_index->null_counts()[page_idx] > 0; |
1767 | 87 | page_statistics->has_not_null = !null_pages[page_idx]; |
1768 | 87 | if (!page_statistics->has_not_null) { |
1769 | 0 | return true; |
1770 | 0 | } |
1771 | 87 | return set_page_min_max(column_index, column_schema, page_idx, page_statistics, timezone); |
1772 | 87 | } |
1773 | | |
1774 | | std::vector<RowRange> intersect_ranges(const std::vector<RowRange>& left, |
1775 | 7 | const std::vector<RowRange>& right) { |
1776 | 7 | std::vector<RowRange> result; |
1777 | 7 | size_t left_idx = 0; |
1778 | 7 | size_t right_idx = 0; |
1779 | 13 | while (left_idx < left.size() && right_idx < right.size()) { |
1780 | 6 | const int64_t left_start = left[left_idx].start; |
1781 | 6 | const int64_t left_end = left_start + left[left_idx].length; |
1782 | 6 | const int64_t right_start = right[right_idx].start; |
1783 | 6 | const int64_t right_end = right_start + right[right_idx].length; |
1784 | 6 | const int64_t start = std::max(left_start, right_start); |
1785 | 6 | const int64_t end = std::min(left_end, right_end); |
1786 | 6 | if (start < end) { |
1787 | 6 | result.push_back(RowRange {start, end - start}); |
1788 | 6 | } |
1789 | 6 | if (left_end < right_end) { |
1790 | 0 | ++left_idx; |
1791 | 6 | } else { |
1792 | 6 | ++right_idx; |
1793 | 6 | } |
1794 | 6 | } |
1795 | 7 | return result; |
1796 | 7 | } |
1797 | | |
1798 | 6 | int64_t count_range_rows(const std::vector<RowRange>& ranges) { |
1799 | 6 | int64_t rows = 0; |
1800 | 6 | for (const auto& range : ranges) { |
1801 | 6 | rows += range.length; |
1802 | 6 | } |
1803 | 6 | return rows; |
1804 | 6 | } |
1805 | | |
1806 | | RowRange page_row_range(const ::parquet::OffsetIndex& offset_index, size_t page_idx, |
1807 | 108 | int64_t row_group_rows) { |
1808 | 108 | const auto& page_locations = offset_index.page_locations(); |
1809 | 108 | const int64_t start = page_locations[page_idx].first_row_index; |
1810 | 108 | const int64_t end = page_idx + 1 == page_locations.size() |
1811 | 108 | ? row_group_rows |
1812 | 108 | : page_locations[page_idx + 1].first_row_index; |
1813 | 108 | DORIS_CHECK(start >= 0); |
1814 | 108 | DORIS_CHECK(end >= start); |
1815 | 108 | DORIS_CHECK(end <= row_group_rows); |
1816 | 108 | return RowRange {start, end - start}; |
1817 | 108 | } |
1818 | | |
1819 | 112 | void append_row_range(const RowRange& range, std::vector<RowRange>* ranges) { |
1820 | 112 | if (range.length == 0) { |
1821 | 0 | return; |
1822 | 0 | } |
1823 | 112 | if (!ranges->empty()) { |
1824 | 99 | auto& previous = ranges->back(); |
1825 | 99 | if (previous.start + previous.length == range.start) { |
1826 | 97 | previous.length += range.length; |
1827 | 97 | return; |
1828 | 97 | } |
1829 | 99 | } |
1830 | 15 | ranges->push_back(range); |
1831 | 15 | } |
1832 | | |
1833 | | std::optional< |
1834 | | std::pair<std::shared_ptr<::parquet::ColumnIndex>, std::shared_ptr<::parquet::OffsetIndex>>> |
1835 | | load_page_indexes_for_slot(const std::shared_ptr<::parquet::RowGroupPageIndexReader>& row_group, |
1836 | | const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema, |
1837 | | const format::FileScanRequest& request, int slot_index, |
1838 | 6 | const ParquetColumnSchema** column_schema) { |
1839 | 6 | DORIS_CHECK(column_schema != nullptr); |
1840 | 6 | *column_schema = nullptr; |
1841 | 6 | const auto file_column_id = file_column_id_by_block_position(request, slot_index); |
1842 | 6 | if (!file_column_id.has_value()) { |
1843 | 0 | return std::nullopt; |
1844 | 0 | } |
1845 | 6 | *column_schema = resolve_local_leaf_schema(file_schema, *file_column_id); |
1846 | 6 | if (*column_schema == nullptr || (*column_schema)->descriptor == nullptr) { |
1847 | 1 | return std::nullopt; |
1848 | 1 | } |
1849 | | |
1850 | 5 | try { |
1851 | 5 | auto column_index = row_group->GetColumnIndex((*column_schema)->leaf_column_id); |
1852 | 5 | auto offset_index = row_group->GetOffsetIndex((*column_schema)->leaf_column_id); |
1853 | 5 | if (column_index == nullptr || offset_index == nullptr || |
1854 | 5 | column_index->null_pages().size() != offset_index->page_locations().size()) { |
1855 | 0 | return std::nullopt; |
1856 | 0 | } |
1857 | 5 | return std::make_pair(std::move(column_index), std::move(offset_index)); |
1858 | 5 | } catch (const ::parquet::ParquetException&) { |
1859 | 0 | return std::nullopt; |
1860 | 0 | } catch (const std::exception&) { |
1861 | 0 | return std::nullopt; |
1862 | 0 | } |
1863 | 5 | } |
1864 | | |
1865 | | bool select_ranges_for_expr_zonemap( |
1866 | | const std::shared_ptr<::parquet::RowGroupPageIndexReader>& row_group, |
1867 | | const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema, |
1868 | | const format::FileScanRequest& request, int slot_index, const VExprContextSPtrs& conjuncts, |
1869 | | int64_t row_group_rows, std::vector<RowRange>* ranges, ParquetPruningStats* pruning_stats, |
1870 | 6 | const cctz::time_zone* timezone) { |
1871 | 6 | DORIS_CHECK(ranges != nullptr); |
1872 | 6 | if (conjuncts.empty()) { |
1873 | 0 | return false; |
1874 | 0 | } |
1875 | 6 | const ParquetColumnSchema* column_schema = nullptr; |
1876 | 6 | int64_t parse_page_index_time_sink = 0; |
1877 | 6 | std::optional<std::pair<std::shared_ptr<::parquet::ColumnIndex>, |
1878 | 6 | std::shared_ptr<::parquet::OffsetIndex>>> |
1879 | 6 | page_indexes; |
1880 | 6 | { |
1881 | | // Arrow materializes the serialized page-index objects lazily in these getters, so keep |
1882 | | // that cost separate from predicate evaluation when diagnosing a slow page-index scan. |
1883 | 6 | SCOPED_RAW_TIMER(pruning_stats == nullptr ? &parse_page_index_time_sink |
1884 | 6 | : &pruning_stats->parse_page_index_time); |
1885 | 6 | page_indexes = load_page_indexes_for_slot(row_group, file_schema, request, slot_index, |
1886 | 6 | &column_schema); |
1887 | 6 | } |
1888 | 6 | if (!page_indexes.has_value()) { |
1889 | 1 | return false; |
1890 | 1 | } |
1891 | 5 | const auto& [column_index, offset_index] = *page_indexes; |
1892 | | |
1893 | 5 | ranges->clear(); |
1894 | 5 | ZoneMapEvalStats page_stats; |
1895 | 5 | const auto page_count = offset_index->page_locations().size(); |
1896 | 85 | for (size_t page_idx = 0; page_idx < page_count; ++page_idx) { |
1897 | 80 | ParquetColumnStatistics page_statistics; |
1898 | 80 | if (!ParquetStatisticsUtils::TransformColumnIndexStatistics( |
1899 | 80 | column_index, *column_schema, page_idx, &page_statistics, timezone)) { |
1900 | 0 | ranges->clear(); |
1901 | 0 | return false; |
1902 | 0 | } |
1903 | | |
1904 | 80 | ZoneMapEvalContext ctx; |
1905 | 80 | add_slot_zonemap(&ctx, slot_index, column_schema->type, |
1906 | 80 | ParquetStatisticsUtils::MakeZoneMap(page_statistics)); |
1907 | 80 | const auto result = VExprContext::evaluate_zonemap_filter(conjuncts, ctx); |
1908 | 80 | page_stats.merge_page_eval_stats(ctx.stats); |
1909 | 80 | if (result == ZoneMapFilterResult::kNoMatch) { |
1910 | 36 | continue; |
1911 | 36 | } |
1912 | 44 | append_row_range(page_row_range(*offset_index, page_idx, row_group_rows), ranges); |
1913 | 44 | } |
1914 | 5 | if (pruning_stats != nullptr) { |
1915 | 5 | pruning_stats->expr_zonemap_unusable_evals += page_stats.unusable_zonemap_eval_count; |
1916 | 5 | pruning_stats->in_zonemap_point_check_count += page_stats.in_zonemap_point_check_count; |
1917 | 5 | pruning_stats->in_zonemap_range_only_count += page_stats.in_zonemap_range_only_count; |
1918 | 5 | } |
1919 | 5 | return true; |
1920 | 5 | } |
1921 | | |
1922 | 112 | bool ranges_intersect(const std::vector<RowRange>& ranges, const RowRange& range) { |
1923 | 112 | const int64_t range_end = range.start + range.length; |
1924 | 112 | for (const auto& selected_range : ranges) { |
1925 | 112 | const int64_t selected_end = selected_range.start + selected_range.length; |
1926 | 112 | if (selected_end <= range.start) { |
1927 | 8 | continue; |
1928 | 8 | } |
1929 | 104 | if (selected_range.start >= range_end) { |
1930 | 44 | return false; |
1931 | 44 | } |
1932 | 60 | return true; |
1933 | 104 | } |
1934 | 8 | return false; |
1935 | 112 | } |
1936 | | |
1937 | | void collect_leaf_schemas(const ParquetColumnSchema& column_schema, |
1938 | | const format::LocalColumnIndex* projection, |
1939 | 7 | std::vector<const ParquetColumnSchema*>* leaf_schemas) { |
1940 | 7 | if (column_schema.kind == ParquetColumnSchemaKind::PRIMITIVE) { |
1941 | 7 | leaf_schemas->push_back(&column_schema); |
1942 | 7 | return; |
1943 | 7 | } |
1944 | 0 | for (const auto& child_schema : column_schema.children) { |
1945 | 0 | if (!format::is_child_projected(projection, child_schema->local_id)) { |
1946 | 0 | continue; |
1947 | 0 | } |
1948 | 0 | const auto* child_projection = |
1949 | 0 | format::find_child_projection(projection, child_schema->local_id); |
1950 | 0 | collect_leaf_schemas(*child_schema, child_projection, leaf_schemas); |
1951 | 0 | } |
1952 | 0 | } |
1953 | | |
1954 | | void collect_request_leaf_schemas( |
1955 | | const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema, |
1956 | | const format::FileScanRequest& request, |
1957 | 6 | std::vector<const ParquetColumnSchema*>* leaf_schemas) { |
1958 | 6 | std::set<int> seen_leaf_ids; |
1959 | 7 | auto collect_projection = [&](const format::LocalColumnIndex& projection) { |
1960 | 7 | const int32_t local_id = projection.local_id(); |
1961 | 7 | if (local_id < 0 || local_id >= static_cast<int32_t>(file_schema.size())) { |
1962 | 0 | return; |
1963 | 0 | } |
1964 | 7 | std::vector<const ParquetColumnSchema*> projection_leaf_schemas; |
1965 | 7 | collect_leaf_schemas(*file_schema[local_id], &projection, &projection_leaf_schemas); |
1966 | 7 | for (const auto* leaf_schema : projection_leaf_schemas) { |
1967 | 7 | DORIS_CHECK(leaf_schema != nullptr); |
1968 | 7 | if (seen_leaf_ids.insert(leaf_schema->leaf_column_id).second) { |
1969 | 7 | leaf_schemas->push_back(leaf_schema); |
1970 | 7 | } |
1971 | 7 | } |
1972 | 7 | }; |
1973 | 6 | for (const auto& projection : request.predicate_columns) { |
1974 | 6 | collect_projection(projection); |
1975 | 6 | } |
1976 | 6 | for (const auto& projection : request.non_predicate_columns) { |
1977 | 1 | collect_projection(projection); |
1978 | 1 | } |
1979 | 6 | } |
1980 | | |
1981 | | bool build_page_skip_plan_for_leaf( |
1982 | | const std::shared_ptr<::parquet::RowGroupPageIndexReader>& row_group, |
1983 | | const ParquetColumnSchema& column_schema, const std::vector<RowRange>& selected_ranges, |
1984 | 4 | int64_t row_group_rows, ParquetPageSkipPlan* page_skip_plan) { |
1985 | 4 | DORIS_CHECK(page_skip_plan != nullptr); |
1986 | 4 | *page_skip_plan = ParquetPageSkipPlan {}; |
1987 | 4 | if (column_schema.kind != ParquetColumnSchemaKind::PRIMITIVE || |
1988 | 4 | column_schema.descriptor == nullptr || column_schema.leaf_column_id < 0 || |
1989 | 4 | column_schema.descriptor->max_repetition_level() != 0) { |
1990 | 0 | return false; |
1991 | 0 | } |
1992 | | |
1993 | 4 | std::shared_ptr<::parquet::OffsetIndex> offset_index; |
1994 | 4 | try { |
1995 | 4 | offset_index = row_group->GetOffsetIndex(column_schema.leaf_column_id); |
1996 | 4 | } catch (const ::parquet::ParquetException&) { |
1997 | 0 | return false; |
1998 | 0 | } catch (const std::exception&) { |
1999 | 0 | return false; |
2000 | 0 | } |
2001 | 4 | if (offset_index == nullptr) { |
2002 | 0 | return false; |
2003 | 0 | } |
2004 | | |
2005 | 4 | const auto page_count = offset_index->page_locations().size(); |
2006 | 4 | page_skip_plan->leaf_column_id = column_schema.leaf_column_id; |
2007 | 4 | page_skip_plan->skipped_pages.resize(page_count); |
2008 | 4 | page_skip_plan->skipped_page_compressed_sizes.resize(page_count); |
2009 | 4 | const auto& page_locations = offset_index->page_locations(); |
2010 | 68 | for (size_t page_idx = 0; page_idx < page_count; ++page_idx) { |
2011 | 64 | const RowRange row_range = page_row_range(*offset_index, page_idx, row_group_rows); |
2012 | 64 | if (row_range.length == 0 || ranges_intersect(selected_ranges, row_range)) { |
2013 | 36 | continue; |
2014 | 36 | } |
2015 | 28 | page_skip_plan->skipped_pages[page_idx] = 1; |
2016 | 28 | page_skip_plan->skipped_page_compressed_sizes[page_idx] = |
2017 | 28 | page_locations[page_idx].compressed_page_size; |
2018 | 28 | append_row_range(row_range, &page_skip_plan->skipped_ranges); |
2019 | 28 | } |
2020 | 4 | if (page_skip_plan->empty()) { |
2021 | 0 | *page_skip_plan = ParquetPageSkipPlan {}; |
2022 | 0 | return false; |
2023 | 0 | } |
2024 | 4 | return true; |
2025 | 4 | } |
2026 | | |
2027 | | void build_page_skip_plans(const std::shared_ptr<::parquet::RowGroupPageIndexReader>& row_group, |
2028 | | const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema, |
2029 | | const format::FileScanRequest& request, |
2030 | | const std::vector<RowRange>& selected_ranges, int64_t row_group_rows, |
2031 | | std::map<int, ParquetPageSkipPlan>* page_skip_plans, |
2032 | 4 | ParquetPruningStats* pruning_stats) { |
2033 | 4 | DORIS_CHECK(page_skip_plans != nullptr); |
2034 | 4 | page_skip_plans->clear(); |
2035 | 4 | std::vector<const ParquetColumnSchema*> leaf_schemas; |
2036 | 4 | collect_request_leaf_schemas(file_schema, request, &leaf_schemas); |
2037 | 4 | for (const auto* leaf_schema : leaf_schemas) { |
2038 | 4 | DORIS_CHECK(leaf_schema != nullptr); |
2039 | 4 | ParquetPageSkipPlan page_skip_plan; |
2040 | 4 | int64_t parse_page_index_time_sink = 0; |
2041 | 4 | bool has_skip_plan = false; |
2042 | 4 | { |
2043 | | // Offset indexes for output-only columns may not have been touched by ZoneMap |
2044 | | // filtering; include their lazy materialization in the same parse timer. |
2045 | 4 | SCOPED_RAW_TIMER(pruning_stats == nullptr ? &parse_page_index_time_sink |
2046 | 4 | : &pruning_stats->parse_page_index_time); |
2047 | 4 | has_skip_plan = build_page_skip_plan_for_leaf(row_group, *leaf_schema, selected_ranges, |
2048 | 4 | row_group_rows, &page_skip_plan); |
2049 | 4 | } |
2050 | 4 | if (has_skip_plan) { |
2051 | 4 | page_skip_plans->emplace(page_skip_plan.leaf_column_id, std::move(page_skip_plan)); |
2052 | 4 | } |
2053 | 4 | } |
2054 | 4 | } |
2055 | | |
2056 | | } // namespace |
2057 | | |
2058 | | bool ParquetStatisticsUtils::TransformColumnIndexStatistics( |
2059 | | const std::shared_ptr<::parquet::ColumnIndex>& column_index, |
2060 | | const ParquetColumnSchema& column_schema, size_t page_idx, |
2061 | 87 | ParquetColumnStatistics* page_statistics, const cctz::time_zone* timezone) { |
2062 | 87 | return build_page_statistics(column_index, column_schema, page_idx, page_statistics, timezone); |
2063 | 87 | } |
2064 | | |
2065 | | Status select_row_group_ranges_by_page_index( |
2066 | | ::parquet::ParquetFileReader* file_reader, |
2067 | | const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema, |
2068 | | const format::FileScanRequest& request, int row_group_idx, int64_t row_group_rows, |
2069 | | std::vector<RowRange>* selected_ranges, std::map<int, ParquetPageSkipPlan>* page_skip_plans, |
2070 | | ParquetPruningStats* pruning_stats, const cctz::time_zone* timezone, |
2071 | 21 | const RuntimeState* runtime_state) { |
2072 | 21 | int64_t page_index_filter_time_sink = 0; |
2073 | 21 | SCOPED_RAW_TIMER(pruning_stats == nullptr ? &page_index_filter_time_sink |
2074 | 21 | : &pruning_stats->page_index_filter_time); |
2075 | 21 | DORIS_CHECK(selected_ranges != nullptr); |
2076 | 21 | selected_ranges->clear(); |
2077 | 21 | if (page_skip_plans != nullptr) { |
2078 | 21 | page_skip_plans->clear(); |
2079 | 21 | } |
2080 | 21 | if (row_group_rows <= 0) { |
2081 | 0 | return Status::OK(); |
2082 | 0 | } |
2083 | 21 | selected_ranges->push_back(RowRange {0, row_group_rows}); |
2084 | 21 | if (!config::enable_parquet_page_index || !has_expr_zonemap_filter(request, runtime_state) || |
2085 | 21 | file_reader == nullptr) { |
2086 | 14 | return Status::OK(); |
2087 | 14 | } |
2088 | | |
2089 | 7 | std::shared_ptr<::parquet::PageIndexReader> page_index_reader; |
2090 | 7 | std::shared_ptr<::parquet::RowGroupPageIndexReader> row_group_index_reader; |
2091 | 7 | try { |
2092 | 7 | if (pruning_stats != nullptr) { |
2093 | 7 | ++pruning_stats->page_index_read_calls; |
2094 | 7 | } |
2095 | 7 | { |
2096 | 7 | int64_t read_page_index_time_sink = 0; |
2097 | 7 | SCOPED_RAW_TIMER(pruning_stats == nullptr ? &read_page_index_time_sink |
2098 | 7 | : &pruning_stats->read_page_index_time); |
2099 | 7 | page_index_reader = file_reader->GetPageIndexReader(); |
2100 | 7 | if (page_index_reader == nullptr) { |
2101 | 0 | return Status::OK(); |
2102 | 0 | } |
2103 | 7 | row_group_index_reader = page_index_reader->RowGroup(row_group_idx); |
2104 | 7 | } |
2105 | 7 | } catch (const ::parquet::ParquetException&) { |
2106 | 0 | return Status::OK(); |
2107 | 0 | } catch (const std::exception&) { |
2108 | 0 | return Status::OK(); |
2109 | 0 | } |
2110 | 7 | if (row_group_index_reader == nullptr) { |
2111 | 2 | return Status::OK(); |
2112 | 2 | } |
2113 | | |
2114 | 5 | std::map<int, VExprContextSPtrs> conjuncts_by_slot; |
2115 | 7 | for (const auto& conjunct : request.conjuncts) { |
2116 | 7 | const auto slot_index = expr_zonemap::single_slot_zonemap_index(conjunct); |
2117 | 7 | if (slot_index >= 0) { |
2118 | 7 | conjuncts_by_slot[slot_index].push_back(conjunct); |
2119 | 7 | } |
2120 | 7 | } |
2121 | | |
2122 | 6 | for (const auto& [slot_index, conjuncts] : conjuncts_by_slot) { |
2123 | 6 | std::vector<RowRange> filter_ranges; |
2124 | 6 | if (!select_ranges_for_expr_zonemap(row_group_index_reader, file_schema, request, |
2125 | 6 | slot_index, conjuncts, row_group_rows, &filter_ranges, |
2126 | 6 | pruning_stats, timezone)) { |
2127 | 1 | continue; |
2128 | 1 | } |
2129 | 5 | *selected_ranges = intersect_ranges(*selected_ranges, filter_ranges); |
2130 | 5 | if (selected_ranges->empty()) { |
2131 | 1 | if (page_skip_plans != nullptr) { |
2132 | 1 | page_skip_plans->clear(); |
2133 | 1 | } |
2134 | 1 | if (pruning_stats != nullptr) { |
2135 | 1 | pruning_stats->filtered_page_rows += row_group_rows; |
2136 | 1 | ++pruning_stats->filtered_row_groups_by_page_index; |
2137 | 1 | } |
2138 | 1 | return Status::OK(); |
2139 | 1 | } |
2140 | 5 | } |
2141 | 4 | if (page_skip_plans != nullptr) { |
2142 | 4 | build_page_skip_plans(row_group_index_reader, file_schema, request, *selected_ranges, |
2143 | 4 | row_group_rows, page_skip_plans, pruning_stats); |
2144 | 4 | } |
2145 | 4 | if (pruning_stats != nullptr) { |
2146 | 4 | const int64_t selected_rows = count_range_rows(*selected_ranges); |
2147 | 4 | DORIS_CHECK(selected_rows <= row_group_rows); |
2148 | 4 | pruning_stats->filtered_page_rows += row_group_rows - selected_rows; |
2149 | 4 | } |
2150 | 4 | return Status::OK(); |
2151 | 5 | } |
2152 | | |
2153 | | namespace { |
2154 | | |
2155 | | template <typename ValueType> |
2156 | | bool set_native_page_scalar_min_max(const tparquet::ColumnIndex& column_index, |
2157 | | const ParquetColumnSchema& column_schema, size_t page_idx, |
2158 | | DecodedValueKind kind, ParquetColumnStatistics* page_statistics, |
2159 | 98 | const cctz::time_zone* timezone) { |
2160 | 98 | if (page_idx >= column_index.min_values.size() || page_idx >= column_index.max_values.size() || |
2161 | 98 | column_index.min_values[page_idx].size() != sizeof(ValueType) || |
2162 | 98 | column_index.max_values[page_idx].size() != sizeof(ValueType)) { |
2163 | 0 | return false; |
2164 | 0 | } |
2165 | 98 | const auto min_value = unaligned_load<ValueType>(column_index.min_values[page_idx].data()); |
2166 | 98 | const auto max_value = unaligned_load<ValueType>(column_index.max_values[page_idx].data()); |
2167 | 98 | if constexpr (std::is_same_v<ValueType, int64_t>) { |
2168 | 0 | if (!timestamp_min_max_is_safe(column_schema, min_value, max_value, timezone)) { |
2169 | 0 | return false; |
2170 | 0 | } |
2171 | 0 | } |
2172 | 98 | if (!valid_min_max(min_value, max_value)) { |
2173 | 0 | return true; |
2174 | 0 | } |
2175 | 98 | if (!set_decoded_field(column_schema, kind, min_value, &page_statistics->min_value, timezone) || |
2176 | 98 | !set_decoded_field(column_schema, kind, max_value, &page_statistics->max_value, timezone)) { |
2177 | 0 | return false; |
2178 | 0 | } |
2179 | 98 | if (decoded_min_max_is_ordered(*page_statistics)) { |
2180 | 98 | page_statistics->has_min_max = true; |
2181 | 98 | } |
2182 | 98 | return true; |
2183 | 98 | } Unexecuted instantiation: parquet_statistics.cpp:_ZN5doris6format7parquet12_GLOBAL__N_130set_native_page_scalar_min_maxIhEEbRKN8tparquet11ColumnIndexERKNS1_19ParquetColumnSchemaEmNS_16DecodedValueKindEPNS1_23ParquetColumnStatisticsEPKN4cctz9time_zoneE parquet_statistics.cpp:_ZN5doris6format7parquet12_GLOBAL__N_130set_native_page_scalar_min_maxIiEEbRKN8tparquet11ColumnIndexERKNS1_19ParquetColumnSchemaEmNS_16DecodedValueKindEPNS1_23ParquetColumnStatisticsEPKN4cctz9time_zoneE Line | Count | Source | 2159 | 98 | const cctz::time_zone* timezone) { | 2160 | 98 | if (page_idx >= column_index.min_values.size() || page_idx >= column_index.max_values.size() || | 2161 | 98 | column_index.min_values[page_idx].size() != sizeof(ValueType) || | 2162 | 98 | column_index.max_values[page_idx].size() != sizeof(ValueType)) { | 2163 | 0 | return false; | 2164 | 0 | } | 2165 | 98 | const auto min_value = unaligned_load<ValueType>(column_index.min_values[page_idx].data()); | 2166 | 98 | const auto max_value = unaligned_load<ValueType>(column_index.max_values[page_idx].data()); | 2167 | | if constexpr (std::is_same_v<ValueType, int64_t>) { | 2168 | | if (!timestamp_min_max_is_safe(column_schema, min_value, max_value, timezone)) { | 2169 | | return false; | 2170 | | } | 2171 | | } | 2172 | 98 | if (!valid_min_max(min_value, max_value)) { | 2173 | 0 | return true; | 2174 | 0 | } | 2175 | 98 | if (!set_decoded_field(column_schema, kind, min_value, &page_statistics->min_value, timezone) || | 2176 | 98 | !set_decoded_field(column_schema, kind, max_value, &page_statistics->max_value, timezone)) { | 2177 | 0 | return false; | 2178 | 0 | } | 2179 | 98 | if (decoded_min_max_is_ordered(*page_statistics)) { | 2180 | 98 | page_statistics->has_min_max = true; | 2181 | 98 | } | 2182 | 98 | return true; | 2183 | 98 | } |
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 |
2184 | | |
2185 | | bool build_native_page_statistics(const tparquet::ColumnIndex& column_index, |
2186 | | const ParquetColumnSchema& column_schema, size_t page_idx, |
2187 | | ParquetColumnStatistics* page_statistics, |
2188 | 100 | const cctz::time_zone* timezone) { |
2189 | 100 | DORIS_CHECK(page_statistics != nullptr); |
2190 | 100 | *page_statistics = {}; |
2191 | 100 | if (!column_index.__isset.null_counts || page_idx >= column_index.null_pages.size() || |
2192 | 100 | page_idx >= column_index.null_counts.size()) { |
2193 | 0 | return false; |
2194 | 0 | } |
2195 | 100 | const int64_t null_count = column_index.null_counts[page_idx]; |
2196 | 100 | if (null_count < 0 || (column_index.null_pages[page_idx] && null_count == 0)) { |
2197 | | // Contradictory optional index metadata must disable pruning; treating it as an empty |
2198 | | // null set can make IS NULL/IS NOT NULL discard rows without reading the data page. |
2199 | 2 | return false; |
2200 | 2 | } |
2201 | 98 | page_statistics->has_null_count = true; |
2202 | 98 | page_statistics->has_null = null_count > 0; |
2203 | 98 | page_statistics->has_not_null = !column_index.null_pages[page_idx]; |
2204 | 98 | if (!page_statistics->has_not_null) { |
2205 | 0 | return true; |
2206 | 0 | } |
2207 | 98 | switch (column_schema.type_descriptor.physical_type) { |
2208 | 0 | case ::parquet::Type::BOOLEAN: |
2209 | 0 | return set_native_page_scalar_min_max<uint8_t>(column_index, column_schema, page_idx, |
2210 | 0 | DecodedValueKind::BOOL, page_statistics, |
2211 | 0 | timezone); |
2212 | 98 | case ::parquet::Type::INT32: |
2213 | 98 | return set_native_page_scalar_min_max<int32_t>( |
2214 | 98 | column_index, column_schema, page_idx, |
2215 | 98 | decoded_value_kind(column_schema.type_descriptor), page_statistics, timezone); |
2216 | 0 | case ::parquet::Type::INT64: |
2217 | 0 | return set_native_page_scalar_min_max<int64_t>( |
2218 | 0 | column_index, column_schema, page_idx, |
2219 | 0 | decoded_value_kind(column_schema.type_descriptor), page_statistics, timezone); |
2220 | 0 | case ::parquet::Type::FLOAT: |
2221 | 0 | return set_native_page_scalar_min_max<float>(column_index, column_schema, page_idx, |
2222 | 0 | DecodedValueKind::FLOAT, page_statistics, |
2223 | 0 | timezone); |
2224 | 0 | case ::parquet::Type::DOUBLE: |
2225 | 0 | return set_native_page_scalar_min_max<double>(column_index, column_schema, page_idx, |
2226 | 0 | DecodedValueKind::DOUBLE, page_statistics, |
2227 | 0 | timezone); |
2228 | 0 | case ::parquet::Type::BYTE_ARRAY: |
2229 | 0 | case ::parquet::Type::FIXED_LEN_BYTE_ARRAY: { |
2230 | 0 | if (page_idx >= column_index.min_values.size() || |
2231 | 0 | page_idx >= column_index.max_values.size()) { |
2232 | 0 | return false; |
2233 | 0 | } |
2234 | 0 | const auto& min_value = column_index.min_values[page_idx]; |
2235 | 0 | const auto& max_value = column_index.max_values[page_idx]; |
2236 | 0 | const bool fixed = column_schema.type_descriptor.physical_type == |
2237 | 0 | ::parquet::Type::FIXED_LEN_BYTE_ARRAY; |
2238 | 0 | if (fixed && |
2239 | 0 | (column_schema.type_descriptor.fixed_length <= 0 || |
2240 | 0 | min_value.size() != static_cast<size_t>(column_schema.type_descriptor.fixed_length) || |
2241 | 0 | max_value.size() != static_cast<size_t>(column_schema.type_descriptor.fixed_length))) { |
2242 | 0 | return false; |
2243 | 0 | } |
2244 | 0 | const auto kind = fixed ? DecodedValueKind::FIXED_BINARY : DecodedValueKind::BINARY; |
2245 | 0 | if (!set_decoded_binary_field(column_schema, kind, |
2246 | 0 | StringRef(min_value.data(), min_value.size()), |
2247 | 0 | &page_statistics->min_value, timezone) || |
2248 | 0 | !set_decoded_binary_field(column_schema, kind, |
2249 | 0 | StringRef(max_value.data(), max_value.size()), |
2250 | 0 | &page_statistics->max_value, timezone)) { |
2251 | 0 | return false; |
2252 | 0 | } |
2253 | 0 | if (decoded_min_max_is_ordered(*page_statistics)) { |
2254 | 0 | page_statistics->has_min_max = true; |
2255 | 0 | } |
2256 | 0 | return true; |
2257 | 0 | } |
2258 | 0 | default: |
2259 | 0 | return false; |
2260 | 98 | } |
2261 | 98 | } |
2262 | | |
2263 | | RowRange native_page_row_range(const tparquet::OffsetIndex& offset_index, size_t page_idx, |
2264 | 64 | int64_t row_group_rows) { |
2265 | 64 | const auto& locations = offset_index.page_locations; |
2266 | 64 | const int64_t start = locations[page_idx].first_row_index; |
2267 | 64 | const int64_t end = page_idx + 1 == locations.size() ? row_group_rows |
2268 | 64 | : locations[page_idx + 1].first_row_index; |
2269 | 64 | return {.start = start, .length = end - start}; |
2270 | 64 | } |
2271 | | |
2272 | | } // namespace |
2273 | | |
2274 | | Status select_row_group_ranges_by_native_page_index( |
2275 | | const std::unordered_map<int, NativeParquetPageIndex>& page_indexes, |
2276 | | const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema, |
2277 | | const format::FileScanRequest& request, int64_t row_group_rows, |
2278 | | std::vector<RowRange>* selected_ranges, std::map<int, ParquetPageSkipPlan>* page_skip_plans, |
2279 | | ParquetPruningStats* pruning_stats, const cctz::time_zone* timezone, |
2280 | 207 | const RuntimeState* runtime_state) { |
2281 | 207 | int64_t filter_time_sink = 0; |
2282 | 207 | SCOPED_RAW_TIMER(pruning_stats == nullptr ? &filter_time_sink |
2283 | 207 | : &pruning_stats->page_index_filter_time); |
2284 | 207 | DORIS_CHECK(selected_ranges != nullptr); |
2285 | 207 | selected_ranges->clear(); |
2286 | 207 | selected_ranges->push_back({.start = 0, .length = row_group_rows}); |
2287 | 207 | if (page_skip_plans != nullptr) { |
2288 | 205 | page_skip_plans->clear(); |
2289 | 205 | } |
2290 | 207 | if (row_group_rows <= 0 || !config::enable_parquet_page_index || |
2291 | 207 | !has_expr_zonemap_filter(request, runtime_state) || page_indexes.empty()) { |
2292 | 203 | return Status::OK(); |
2293 | 203 | } |
2294 | 4 | if (pruning_stats != nullptr) { |
2295 | 2 | ++pruning_stats->page_index_read_calls; |
2296 | 2 | } |
2297 | | |
2298 | 4 | std::map<int, VExprContextSPtrs> conjuncts_by_slot; |
2299 | 4 | for (const auto& conjunct : request.conjuncts) { |
2300 | 4 | const auto slot_index = expr_zonemap::single_slot_zonemap_index(conjunct); |
2301 | 4 | if (slot_index >= 0) { |
2302 | 4 | conjuncts_by_slot[slot_index].push_back(conjunct); |
2303 | 4 | } |
2304 | 4 | } |
2305 | 4 | for (const auto& [slot_index, conjuncts] : conjuncts_by_slot) { |
2306 | 4 | const auto file_column_id = file_column_id_by_block_position(request, slot_index); |
2307 | 4 | if (!file_column_id.has_value()) { |
2308 | 0 | continue; |
2309 | 0 | } |
2310 | 4 | const auto* column_schema = resolve_local_leaf_schema(file_schema, *file_column_id); |
2311 | 4 | if (column_schema == nullptr || column_schema->type == nullptr || |
2312 | 4 | !native_metadata_predicate_is_type_safe(*column_schema)) { |
2313 | 0 | continue; |
2314 | 0 | } |
2315 | 4 | const auto index_it = page_indexes.find(column_schema->leaf_column_id); |
2316 | 4 | if (index_it == page_indexes.end()) { |
2317 | 0 | continue; |
2318 | 0 | } |
2319 | 4 | const auto& indexes = index_it->second; |
2320 | 4 | std::vector<RowRange> filter_ranges; |
2321 | 4 | bool usable = true; |
2322 | 36 | for (size_t page_idx = 0; page_idx < indexes.offset_index.page_locations.size(); |
2323 | 34 | ++page_idx) { |
2324 | 34 | ParquetColumnStatistics statistics; |
2325 | 34 | if (!build_native_page_statistics(indexes.column_index, *column_schema, page_idx, |
2326 | 34 | &statistics, timezone)) { |
2327 | 2 | usable = false; |
2328 | 2 | break; |
2329 | 2 | } |
2330 | 32 | ZoneMapEvalContext ctx; |
2331 | 32 | add_slot_zonemap(&ctx, slot_index, column_schema->type, |
2332 | 32 | ParquetStatisticsUtils::MakeZoneMap(statistics)); |
2333 | 32 | if (VExprContext::evaluate_zonemap_filter(conjuncts, ctx) != |
2334 | 32 | ZoneMapFilterResult::kNoMatch) { |
2335 | 16 | append_row_range( |
2336 | 16 | native_page_row_range(indexes.offset_index, page_idx, row_group_rows), |
2337 | 16 | &filter_ranges); |
2338 | 16 | } |
2339 | 32 | if (pruning_stats != nullptr) { |
2340 | 32 | pruning_stats->expr_zonemap_unusable_evals += ctx.stats.unusable_zonemap_eval_count; |
2341 | 32 | pruning_stats->in_zonemap_point_check_count += |
2342 | 32 | ctx.stats.in_zonemap_point_check_count; |
2343 | 32 | pruning_stats->in_zonemap_range_only_count += ctx.stats.in_zonemap_range_only_count; |
2344 | 32 | } |
2345 | 32 | } |
2346 | 4 | if (!usable) { |
2347 | 2 | continue; |
2348 | 2 | } |
2349 | 2 | *selected_ranges = intersect_ranges(*selected_ranges, filter_ranges); |
2350 | 2 | if (selected_ranges->empty()) { |
2351 | 0 | if (pruning_stats != nullptr) { |
2352 | 0 | pruning_stats->filtered_page_rows += row_group_rows; |
2353 | 0 | ++pruning_stats->filtered_row_groups_by_page_index; |
2354 | 0 | } |
2355 | 0 | return Status::OK(); |
2356 | 0 | } |
2357 | 2 | } |
2358 | | |
2359 | 4 | if (page_skip_plans != nullptr) { |
2360 | 2 | std::vector<const ParquetColumnSchema*> leaves; |
2361 | 2 | collect_request_leaf_schemas(file_schema, request, &leaves); |
2362 | 3 | for (const auto* leaf : leaves) { |
2363 | 3 | const auto index_it = page_indexes.find(leaf->leaf_column_id); |
2364 | 3 | if (index_it == page_indexes.end() || leaf->max_repetition_level != 0) { |
2365 | 0 | continue; |
2366 | 0 | } |
2367 | 3 | const auto& offset_index = index_it->second.offset_index; |
2368 | 3 | ParquetPageSkipPlan skip_plan; |
2369 | 3 | skip_plan.leaf_column_id = leaf->leaf_column_id; |
2370 | 3 | skip_plan.skipped_pages.resize(offset_index.page_locations.size()); |
2371 | 3 | skip_plan.skipped_page_compressed_sizes.resize(offset_index.page_locations.size()); |
2372 | 51 | for (size_t page_idx = 0; page_idx < offset_index.page_locations.size(); ++page_idx) { |
2373 | 48 | const auto range = native_page_row_range(offset_index, page_idx, row_group_rows); |
2374 | 48 | if (range.length == 0 || ranges_intersect(*selected_ranges, range)) { |
2375 | 24 | continue; |
2376 | 24 | } |
2377 | 24 | skip_plan.skipped_pages[page_idx] = 1; |
2378 | 24 | skip_plan.skipped_page_compressed_sizes[page_idx] = |
2379 | 24 | offset_index.page_locations[page_idx].compressed_page_size; |
2380 | 24 | append_row_range(range, &skip_plan.skipped_ranges); |
2381 | 24 | } |
2382 | 3 | if (!skip_plan.empty()) { |
2383 | 3 | page_skip_plans->emplace(skip_plan.leaf_column_id, std::move(skip_plan)); |
2384 | 3 | } |
2385 | 3 | } |
2386 | 2 | } |
2387 | 4 | if (pruning_stats != nullptr) { |
2388 | 2 | pruning_stats->filtered_page_rows += row_group_rows - count_range_rows(*selected_ranges); |
2389 | 2 | } |
2390 | 4 | return Status::OK(); |
2391 | 4 | } |
2392 | | |
2393 | | } // namespace doris::format::parquet |