Coverage Report

Created: 2026-07-09 12:42

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
be/src/format/table/paimon_reader.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
//
9
//   http://www.apache.org/licenses/LICENSE-2.0
10
//
11
// Unless required by applicable law or agreed to in writing,
12
// software distributed under the License is distributed on an
13
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14
// KIND, either express or implied.  See the License for the
15
// specific language governing permissions and limitations
16
// under the License.
17
18
#include "format/table/paimon_reader.h"
19
20
#include <fmt/format.h>
21
22
#include <cstring>
23
#include <memory>
24
#include <vector>
25
26
#include "common/status.h"
27
#include "exec/common/endian.h"
28
#include "format/table/deletion_vector_reader.h"
29
#include "runtime/runtime_state.h"
30
31
namespace doris {
32
33
namespace {
34
35
constexpr static char PAIMON_BITMAP_MAGIC[] = {'\x5E', '\x43', '\xF2', '\xD0'};
36
37
} // namespace
38
39
21.6k
std::string build_paimon_deletion_vector_cache_key(const TPaimonDeletionFileDesc& deletion_file) {
40
21.6k
    return fmt::format("paimon_dv_{}#{}#{}", deletion_file.path, deletion_file.offset,
41
21.6k
                       deletion_file.length);
42
21.6k
}
43
44
Status decode_paimon_deletion_vector_buffer(const char* buf, size_t buffer_size,
45
237
                                            std::vector<int64_t>* delete_rows) {
46
237
    if (buffer_size < 8) [[unlikely]] {
47
2
        return Status::DataQualityError("Deletion vector file size too small: {}", buffer_size);
48
2
    }
49
50
235
    const uint32_t actual_length = BigEndian::Load32(buf);
51
235
    if (actual_length + 4 != buffer_size) [[unlikely]] {
52
2
        return Status::RuntimeError(
53
2
                "DeletionVector deserialize error: length not match, "
54
2
                "actual length: {}, expect length: {}",
55
2
                actual_length, buffer_size - 4);
56
2
    }
57
58
233
    if (memcmp(buf + sizeof(actual_length), PAIMON_BITMAP_MAGIC, 4) != 0) [[unlikely]] {
59
2
        return Status::RuntimeError("DeletionVector deserialize error: invalid magic number {}",
60
2
                                    BigEndian::Load32(buf + sizeof(actual_length)));
61
2
    }
62
63
231
    roaring::Roaring roaring_bitmap;
64
231
    try {
65
231
        roaring_bitmap = roaring::Roaring::readSafe(buf + 8, buffer_size - 8);
66
231
    } catch (const std::runtime_error& e) {
67
2
        return Status::RuntimeError(
68
2
                "DeletionVector deserialize error: failed to deserialize roaring bitmap, {}",
69
2
                e.what());
70
2
    }
71
72
231
    delete_rows->reserve(roaring_bitmap.cardinality());
73
465
    for (auto it = roaring_bitmap.begin(); it != roaring_bitmap.end(); it++) {
74
234
        delete_rows->push_back(*it);
75
234
    }
76
231
    return Status::OK();
77
231
}
78
79
// ============================================================================
80
// PaimonOrcReader
81
// ============================================================================
82
7.67k
void PaimonOrcReader::_init_paimon_profile() {
83
7.67k
    static const char* paimon_profile = "PaimonProfile";
84
7.67k
    ADD_TIMER(get_profile(), paimon_profile);
85
7.67k
    _paimon_profile.num_delete_rows =
86
7.67k
            ADD_CHILD_COUNTER(get_profile(), "NumDeleteRows", TUnit::UNIT, paimon_profile);
87
7.67k
    _paimon_profile.delete_files_read_time =
88
7.67k
            ADD_CHILD_TIMER(get_profile(), "DeleteFileReadTime", paimon_profile);
89
7.67k
    _paimon_profile.parse_deletion_vector_time =
90
7.67k
            ADD_CHILD_TIMER(get_profile(), "ParseDeletionVectorTime", paimon_profile);
91
7.67k
}
92
93
7.68k
Status PaimonOrcReader::on_before_init_reader(ReaderInitContext* ctx) {
94
7.68k
    _column_descs = ctx->column_descs;
95
7.68k
    _fill_col_name_to_block_idx = ctx->col_name_to_block_idx;
96
7.68k
    RETURN_IF_ERROR(_extract_partition_values(*ctx->range, ctx->tuple_descriptor,
97
7.68k
                                              _fill_partition_values,
98
7.68k
                                              &_fill_partition_value_is_null));
99
7.68k
    const orc::Type* orc_type_ptr = nullptr;
100
7.68k
    RETURN_IF_ERROR(get_file_type(&orc_type_ptr));
101
102
7.68k
    RETURN_IF_ERROR(gen_table_info_node_by_field_id(
103
7.68k
            get_scan_params(), get_scan_range().table_format_params.paimon_params.schema_id,
104
7.68k
            get_tuple_descriptor(), orc_type_ptr));
105
7.68k
    ctx->table_info_node = table_info_node_ptr;
106
107
20.2k
    for (const auto& desc : *ctx->column_descs) {
108
20.2k
        if (desc.category == ColumnCategory::REGULAR ||
109
20.2k
            desc.category == ColumnCategory::GENERATED) {
110
20.0k
            ctx->column_names.push_back(desc.name);
111
20.0k
        }
112
20.2k
    }
113
7.68k
    return Status::OK();
114
7.68k
}
115
116
7.69k
Status PaimonOrcReader::on_after_init_reader(ReaderInitContext* /*ctx*/) {
117
7.69k
    return _init_deletion_vector();
118
7.69k
}
119
120
7.69k
Status PaimonOrcReader::_init_deletion_vector() {
121
7.69k
    const auto& table_desc = get_scan_range().table_format_params.paimon_params;
122
7.69k
    if (!table_desc.__isset.deletion_file) {
123
2.09k
        return Status::OK();
124
2.09k
    }
125
126
    // Cannot do count push down if there are delete files
127
5.60k
    if (!get_scan_range().table_format_params.paimon_params.__isset.row_count) {
128
5.60k
        set_push_down_agg_type(TPushAggOp::NONE);
129
5.60k
    }
130
5.60k
    const auto& deletion_file = table_desc.deletion_file;
131
132
5.60k
    Status create_status = Status::OK();
133
134
5.60k
    SCOPED_TIMER(_paimon_profile.delete_files_read_time);
135
5.60k
    using DeleteRows = std::vector<int64_t>;
136
5.60k
    _delete_rows = _kv_cache->get<DeleteRows>(
137
5.60k
            build_paimon_deletion_vector_cache_key(deletion_file), [&]() -> DeleteRows* {
138
97
                auto delete_rows = std::make_unique<DeleteRows>();
139
140
97
                TFileRangeDesc delete_range;
141
97
                delete_range.__set_fs_name(get_scan_range().fs_name);
142
97
                delete_range.path = deletion_file.path;
143
97
                delete_range.start_offset = deletion_file.offset;
144
97
                delete_range.size = deletion_file.length + 4;
145
97
                delete_range.file_size = -1;
146
147
97
                DeletionVectorReader dv_reader(get_state(), get_profile(), get_scan_params(),
148
97
                                               delete_range, get_io_ctx());
149
97
                create_status = dv_reader.open();
150
97
                if (!create_status.ok()) [[unlikely]] {
151
1
                    return nullptr;
152
1
                }
153
154
96
                size_t bytes_read = deletion_file.length + 4;
155
96
                std::vector<char> buffer(bytes_read);
156
96
                create_status =
157
96
                        dv_reader.read_at(deletion_file.offset, {buffer.data(), bytes_read});
158
96
                if (!create_status.ok()) [[unlikely]] {
159
0
                    return nullptr;
160
0
                }
161
162
96
                SCOPED_TIMER(_paimon_profile.parse_deletion_vector_time);
163
96
                create_status = decode_paimon_deletion_vector_buffer(buffer.data(), bytes_read,
164
96
                                                                     delete_rows.get());
165
96
                if (!create_status.ok()) [[unlikely]] {
166
0
                    return nullptr;
167
0
                }
168
96
                COUNTER_UPDATE(_paimon_profile.num_delete_rows, delete_rows->size());
169
96
                return delete_rows.release();
170
96
            });
171
5.60k
    RETURN_IF_ERROR(create_status);
172
5.60k
    if (!_delete_rows->empty()) [[likely]] {
173
5.60k
        set_position_delete_rowids(_delete_rows);
174
5.60k
    }
175
5.60k
    return Status::OK();
176
5.60k
}
177
178
// ============================================================================
179
// PaimonParquetReader
180
// ============================================================================
181
87
void PaimonParquetReader::_init_paimon_profile() {
182
87
    static const char* paimon_profile = "PaimonProfile";
183
87
    ADD_TIMER(get_profile(), paimon_profile);
184
87
    _paimon_profile.num_delete_rows =
185
87
            ADD_CHILD_COUNTER(get_profile(), "NumDeleteRows", TUnit::UNIT, paimon_profile);
186
87
    _paimon_profile.delete_files_read_time =
187
87
            ADD_CHILD_TIMER(get_profile(), "DeleteFileReadTime", paimon_profile);
188
87
    _paimon_profile.parse_deletion_vector_time =
189
87
            ADD_CHILD_TIMER(get_profile(), "ParseDeletionVectorTime", paimon_profile);
190
87
}
191
192
86
Status PaimonParquetReader::on_before_init_reader(ReaderInitContext* ctx) {
193
86
    _column_descs = ctx->column_descs;
194
86
    _fill_col_name_to_block_idx = ctx->col_name_to_block_idx;
195
86
    RETURN_IF_ERROR(_extract_partition_values(*ctx->range, ctx->tuple_descriptor,
196
86
                                              _fill_partition_values,
197
86
                                              &_fill_partition_value_is_null));
198
86
    const FieldDescriptor* field_desc = nullptr;
199
86
    RETURN_IF_ERROR(get_file_metadata_schema(&field_desc));
200
86
    DCHECK(field_desc != nullptr);
201
202
86
    RETURN_IF_ERROR(gen_table_info_node_by_field_id(
203
86
            get_scan_params(), get_scan_range().table_format_params.paimon_params.schema_id,
204
86
            get_tuple_descriptor(), *field_desc));
205
86
    ctx->table_info_node = table_info_node_ptr;
206
207
138
    for (const auto& desc : *ctx->column_descs) {
208
138
        if (desc.category == ColumnCategory::REGULAR ||
209
138
            desc.category == ColumnCategory::GENERATED) {
210
92
            ctx->column_names.push_back(desc.name);
211
92
        }
212
138
    }
213
86
    return Status::OK();
214
86
}
215
216
86
Status PaimonParquetReader::on_after_init_reader(ReaderInitContext* /*ctx*/) {
217
86
    return _init_deletion_vector();
218
86
}
219
220
87
Status PaimonParquetReader::_init_deletion_vector() {
221
87
    const auto& table_desc = get_scan_range().table_format_params.paimon_params;
222
87
    if (!table_desc.__isset.deletion_file) {
223
86
        return Status::OK();
224
86
    }
225
226
1
    if (!get_scan_range().table_format_params.paimon_params.__isset.row_count) {
227
1
        set_push_down_agg_type(TPushAggOp::NONE);
228
1
    }
229
1
    const auto& deletion_file = table_desc.deletion_file;
230
231
1
    Status create_status = Status::OK();
232
233
1
    SCOPED_TIMER(_paimon_profile.delete_files_read_time);
234
1
    using DeleteRows = std::vector<int64_t>;
235
1
    _delete_rows = _kv_cache->get<DeleteRows>(
236
1
            build_paimon_deletion_vector_cache_key(deletion_file), [&]() -> DeleteRows* {
237
1
                auto delete_rows = std::make_unique<DeleteRows>();
238
239
1
                TFileRangeDesc delete_range;
240
1
                delete_range.__set_fs_name(get_scan_range().fs_name);
241
1
                delete_range.path = deletion_file.path;
242
1
                delete_range.start_offset = deletion_file.offset;
243
1
                delete_range.size = deletion_file.length + 4;
244
1
                delete_range.file_size = -1;
245
246
1
                DeletionVectorReader dv_reader(get_state(), get_profile(), get_scan_params(),
247
1
                                               delete_range, get_io_ctx());
248
1
                create_status = dv_reader.open();
249
1
                if (!create_status.ok()) [[unlikely]] {
250
1
                    return nullptr;
251
1
                }
252
253
0
                size_t bytes_read = deletion_file.length + 4;
254
0
                std::vector<char> buffer(bytes_read);
255
0
                create_status =
256
0
                        dv_reader.read_at(deletion_file.offset, {buffer.data(), bytes_read});
257
0
                if (!create_status.ok()) [[unlikely]] {
258
0
                    return nullptr;
259
0
                }
260
261
0
                SCOPED_TIMER(_paimon_profile.parse_deletion_vector_time);
262
0
                create_status = decode_paimon_deletion_vector_buffer(buffer.data(), bytes_read,
263
0
                                                                     delete_rows.get());
264
0
                if (!create_status.ok()) [[unlikely]] {
265
0
                    return nullptr;
266
0
                }
267
0
                COUNTER_UPDATE(_paimon_profile.num_delete_rows, delete_rows->size());
268
0
                return delete_rows.release();
269
0
            });
270
1
    RETURN_IF_ERROR(create_status);
271
0
    if (!_delete_rows->empty()) [[likely]] {
272
0
        ParquetReader::set_delete_rows(_delete_rows);
273
0
    }
274
0
    return Status::OK();
275
1
}
276
277
} // namespace doris