Coverage Report

Created: 2026-07-05 00:48

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