be/src/format/table/paimon_reader.cpp

Source
// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License.  You may obtain a copy of the License at
//
//   http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied.  See the License for the
// specific language governing permissions and limitations
// under the License.

#include "format/table/paimon_reader.h"

#include <fmt/format.h>

#include <cstring>
#include <vector>

#include "common/status.h"
#include "exec/common/endian.h"
#include "format/table/deletion_vector_reader.h"
#include "runtime/runtime_state.h"

namespace doris {

namespace {

constexpr static char PAIMON_BITMAP_MAGIC[] = {'\x5E', '\x43', '\xF2', '\xD0'};

} // namespace

std::string build_paimon_deletion_vector_cache_key(const TPaimonDeletionFileDesc& deletion_file) {
    return fmt::format("paimon_dv_{}#{}#{}", deletion_file.path, deletion_file.offset,
                       deletion_file.length);
}

Status decode_paimon_deletion_vector_buffer(const char* buf, size_t buffer_size,
                                            std::vector<int64_t>* delete_rows) {
    if (buffer_size < 8) [[unlikely]] {
        return Status::DataQualityError("Deletion vector file size too small: {}", buffer_size);
    }

    const uint32_t actual_length = BigEndian::Load32(buf);
    if (actual_length + 4 != buffer_size) [[unlikely]] {
        return Status::RuntimeError(
                "DeletionVector deserialize error: length not match, "
                "actual length: {}, expect length: {}",
                actual_length, buffer_size - 4);
    }

    if (memcmp(buf + sizeof(actual_length), PAIMON_BITMAP_MAGIC, 4) != 0) [[unlikely]] {
        return Status::RuntimeError("DeletionVector deserialize error: invalid magic number {}",
                                    BigEndian::Load32(buf + sizeof(actual_length)));
    }

    roaring::Roaring roaring_bitmap;
    try {
        roaring_bitmap = roaring::Roaring::readSafe(buf + 8, buffer_size - 8);
    } catch (const std::runtime_error& e) {
        return Status::RuntimeError(
                "DeletionVector deserialize error: failed to deserialize roaring bitmap, {}",
                e.what());
    }

    delete_rows->reserve(roaring_bitmap.cardinality());
    for (auto it = roaring_bitmap.begin(); it != roaring_bitmap.end(); it++) {
        delete_rows->push_back(*it);
    }
    return Status::OK();
}

// ============================================================================
// PaimonOrcReader
// ============================================================================
void PaimonOrcReader::_init_paimon_profile() {
    static const char* paimon_profile = "PaimonProfile";
    ADD_TIMER(get_profile(), paimon_profile);
    _paimon_profile.num_delete_rows =
            ADD_CHILD_COUNTER(get_profile(), "NumDeleteRows", TUnit::UNIT, paimon_profile);
    _paimon_profile.delete_files_read_time =
            ADD_CHILD_TIMER(get_profile(), "DeleteFileReadTime", paimon_profile);
    _paimon_profile.parse_deletion_vector_time =
            ADD_CHILD_TIMER(get_profile(), "ParseDeletionVectorTime", paimon_profile);
}

Status PaimonOrcReader::on_before_init_reader(ReaderInitContext* ctx) {
    _column_descs = ctx->column_descs;
    _fill_col_name_to_block_idx = ctx->col_name_to_block_idx;
    RETURN_IF_ERROR(_extract_partition_values(*ctx->range, ctx->tuple_descriptor,
                                              _fill_partition_values,
                                              &_fill_partition_value_is_null));
    const orc::Type* orc_type_ptr = nullptr;
    RETURN_IF_ERROR(get_file_type(&orc_type_ptr));

    RETURN_IF_ERROR(gen_table_info_node_by_field_id(
            get_scan_params(), get_scan_range().table_format_params.paimon_params.schema_id,
            get_tuple_descriptor(), orc_type_ptr));
    ctx->table_info_node = table_info_node_ptr;

    for (const auto& desc : *ctx->column_descs) {
        if (desc.category == ColumnCategory::REGULAR ||
            desc.category == ColumnCategory::GENERATED) {
            ctx->column_names.push_back(desc.name);
        }
    }
    return Status::OK();
}

Status PaimonOrcReader::on_after_init_reader(ReaderInitContext* /*ctx*/) {
    return _init_deletion_vector();
}

Status PaimonOrcReader::_init_deletion_vector() {
    const auto& table_desc = get_scan_range().table_format_params.paimon_params;
    if (!table_desc.__isset.deletion_file) {
        return Status::OK();
    }

    // Cannot do count push down if there are delete files
    if (!get_scan_range().table_format_params.paimon_params.__isset.row_count) {
        set_push_down_agg_type(TPushAggOp::NONE);
    }
    const auto& deletion_file = table_desc.deletion_file;

    Status create_status = Status::OK();

    SCOPED_TIMER(_paimon_profile.delete_files_read_time);
    using DeleteRows = std::vector<int64_t>;
    _delete_rows = _kv_cache->get<DeleteRows>(
            build_paimon_deletion_vector_cache_key(deletion_file), [&]() -> DeleteRows* {
                auto* delete_rows = new DeleteRows;

                TFileRangeDesc delete_range;
                delete_range.__set_fs_name(get_scan_range().fs_name);
                delete_range.path = deletion_file.path;
                delete_range.start_offset = deletion_file.offset;
                delete_range.size = deletion_file.length + 4;
                delete_range.file_size = -1;

                DeletionVectorReader dv_reader(get_state(), get_profile(), get_scan_params(),
                                               delete_range, get_io_ctx());
                create_status = dv_reader.open();
                if (!create_status.ok()) [[unlikely]] {
                    return nullptr;
                }

                size_t bytes_read = deletion_file.length + 4;
                std::vector<char> buffer(bytes_read);
                create_status =
                        dv_reader.read_at(deletion_file.offset, {buffer.data(), bytes_read});
                if (!create_status.ok()) [[unlikely]] {
                    return nullptr;
                }

                SCOPED_TIMER(_paimon_profile.parse_deletion_vector_time);
                create_status = decode_paimon_deletion_vector_buffer(buffer.data(), bytes_read,
                                                                     delete_rows);
                if (!create_status.ok()) [[unlikely]] {
                    return nullptr;
                }
                COUNTER_UPDATE(_paimon_profile.num_delete_rows, delete_rows->size());
                return delete_rows;
            });
    RETURN_IF_ERROR(create_status);
    if (!_delete_rows->empty()) [[likely]] {
        set_position_delete_rowids(_delete_rows);
    }
    return Status::OK();
}

// ============================================================================
// PaimonParquetReader
// ============================================================================
void PaimonParquetReader::_init_paimon_profile() {
    static const char* paimon_profile = "PaimonProfile";
    ADD_TIMER(get_profile(), paimon_profile);
    _paimon_profile.num_delete_rows =
            ADD_CHILD_COUNTER(get_profile(), "NumDeleteRows", TUnit::UNIT, paimon_profile);
    _paimon_profile.delete_files_read_time =
            ADD_CHILD_TIMER(get_profile(), "DeleteFileReadTime", paimon_profile);
    _paimon_profile.parse_deletion_vector_time =
            ADD_CHILD_TIMER(get_profile(), "ParseDeletionVectorTime", paimon_profile);
}

Status PaimonParquetReader::on_before_init_reader(ReaderInitContext* ctx) {
    _column_descs = ctx->column_descs;
    _fill_col_name_to_block_idx = ctx->col_name_to_block_idx;
    RETURN_IF_ERROR(_extract_partition_values(*ctx->range, ctx->tuple_descriptor,
                                              _fill_partition_values,
                                              &_fill_partition_value_is_null));
    const FieldDescriptor* field_desc = nullptr;
    RETURN_IF_ERROR(get_file_metadata_schema(&field_desc));
    DCHECK(field_desc != nullptr);

    RETURN_IF_ERROR(gen_table_info_node_by_field_id(
            get_scan_params(), get_scan_range().table_format_params.paimon_params.schema_id,
            get_tuple_descriptor(), *field_desc));
    ctx->table_info_node = table_info_node_ptr;

    for (const auto& desc : *ctx->column_descs) {
        if (desc.category == ColumnCategory::REGULAR ||
            desc.category == ColumnCategory::GENERATED) {
            ctx->column_names.push_back(desc.name);
        }
    }
    return Status::OK();
}

Status PaimonParquetReader::on_after_init_reader(ReaderInitContext* /*ctx*/) {
    return _init_deletion_vector();
}

Status PaimonParquetReader::_init_deletion_vector() {
    const auto& table_desc = get_scan_range().table_format_params.paimon_params;
    if (!table_desc.__isset.deletion_file) {
        return Status::OK();
    }

    if (!get_scan_range().table_format_params.paimon_params.__isset.row_count) {
        set_push_down_agg_type(TPushAggOp::NONE);
    }
    const auto& deletion_file = table_desc.deletion_file;

    Status create_status = Status::OK();

    SCOPED_TIMER(_paimon_profile.delete_files_read_time);
    using DeleteRows = std::vector<int64_t>;
    _delete_rows = _kv_cache->get<DeleteRows>(
            build_paimon_deletion_vector_cache_key(deletion_file), [&]() -> DeleteRows* {
                auto* delete_rows = new DeleteRows;

                TFileRangeDesc delete_range;
                delete_range.__set_fs_name(get_scan_range().fs_name);
                delete_range.path = deletion_file.path;
                delete_range.start_offset = deletion_file.offset;
                delete_range.size = deletion_file.length + 4;
                delete_range.file_size = -1;

                DeletionVectorReader dv_reader(get_state(), get_profile(), get_scan_params(),
                                               delete_range, get_io_ctx());
                create_status = dv_reader.open();
                if (!create_status.ok()) [[unlikely]] {
                    return nullptr;
                }

                size_t bytes_read = deletion_file.length + 4;
                std::vector<char> buffer(bytes_read);
                create_status =
                        dv_reader.read_at(deletion_file.offset, {buffer.data(), bytes_read});
                if (!create_status.ok()) [[unlikely]] {
                    return nullptr;
                }

                SCOPED_TIMER(_paimon_profile.parse_deletion_vector_time);
                create_status = decode_paimon_deletion_vector_buffer(buffer.data(), bytes_read,
                                                                     delete_rows);
                if (!create_status.ok()) [[unlikely]] {
                    return nullptr;
                }
                COUNTER_UPDATE(_paimon_profile.num_delete_rows, delete_rows->size());
                return delete_rows;
            });
    RETURN_IF_ERROR(create_status);
    if (!_delete_rows->empty()) [[likely]] {
        ParquetReader::set_delete_rows(_delete_rows);
    }
    return Status::OK();
}

} // namespace doris

Coverage Report

Created: 2026-07-07 18:01

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