Coverage Report

Created: 2026-08-07 19:43

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