Coverage Report

Created: 2026-03-12 16:03

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 <vector>
21
22
#include "common/status.h"
23
#include "format/table/deletion_vector_reader.h"
24
#include "runtime/runtime_state.h"
25
26
namespace doris {
27
#include "common/compile_check_begin.h"
28
PaimonReader::PaimonReader(std::unique_ptr<GenericReader> file_format_reader,
29
                           RuntimeProfile* profile, RuntimeState* state,
30
                           const TFileScanRangeParams& params, const TFileRangeDesc& range,
31
                           ShardedKVCache* kv_cache, io::IOContext* io_ctx,
32
                           FileMetaCache* meta_cache)
33
0
        : TableFormatReader(std::move(file_format_reader), state, profile, params, range, io_ctx,
34
0
                            meta_cache),
35
0
          _kv_cache(kv_cache) {
36
0
    static const char* paimon_profile = "PaimonProfile";
37
0
    ADD_TIMER(_profile, paimon_profile);
38
0
    _paimon_profile.num_delete_rows =
39
0
            ADD_CHILD_COUNTER(_profile, "NumDeleteRows", TUnit::UNIT, paimon_profile);
40
0
    _paimon_profile.delete_files_read_time =
41
0
            ADD_CHILD_TIMER(_profile, "DeleteFileReadTime", paimon_profile);
42
0
    _paimon_profile.parse_deletion_vector_time =
43
0
            ADD_CHILD_TIMER(_profile, "ParseDeletionVectorTime", paimon_profile);
44
0
}
45
46
0
Status PaimonReader::init_row_filters() {
47
0
    const auto& table_desc = _range.table_format_params.paimon_params;
48
0
    if (!table_desc.__isset.deletion_file) {
49
0
        return Status::OK();
50
0
    }
51
52
    // set push down agg type to NONE because we can not do count push down opt
53
    // if there are delete files.
54
0
    if (!_range.table_format_params.paimon_params.__isset.row_count) {
55
0
        _file_format_reader->set_push_down_agg_type(TPushAggOp::NONE);
56
0
    }
57
0
    const auto& deletion_file = table_desc.deletion_file;
58
59
0
    Status create_status = Status::OK();
60
61
0
    std::string key;
62
0
    key.resize(deletion_file.path.size() + sizeof(deletion_file.offset));
63
0
    memcpy(key.data(), deletion_file.path.data(), deletion_file.path.size());
64
0
    memcpy(key.data() + deletion_file.path.size(), &deletion_file.offset,
65
0
           sizeof(deletion_file.offset));
66
67
0
    SCOPED_TIMER(_paimon_profile.delete_files_read_time);
68
0
    using DeleteRows = std::vector<int64_t>;
69
0
    _delete_rows = _kv_cache->get<DeleteRows>(key, [&]() -> DeleteRows* {
70
0
        auto* delete_rows = new DeleteRows;
71
72
0
        TFileRangeDesc delete_range;
73
        // must use __set() method to make sure __isset is true
74
0
        delete_range.__set_fs_name(_range.fs_name);
75
0
        delete_range.path = deletion_file.path;
76
0
        delete_range.start_offset = deletion_file.offset;
77
0
        delete_range.size = deletion_file.length + 4;
78
0
        delete_range.file_size = -1;
79
80
0
        DeletionVectorReader dv_reader(_state, _profile, _params, delete_range, _io_ctx);
81
0
        create_status = dv_reader.open();
82
0
        if (!create_status.ok()) [[unlikely]] {
83
0
            return nullptr;
84
0
        }
85
86
        // the reason of adding 4: https://github.com/apache/paimon/issues/3313
87
0
        size_t bytes_read = deletion_file.length + 4;
88
        // TODO: better way to alloc memeory
89
0
        std::vector<char> buffer(bytes_read);
90
0
        create_status = dv_reader.read_at(deletion_file.offset, {buffer.data(), bytes_read});
91
0
        if (!create_status.ok()) [[unlikely]] {
92
0
            return nullptr;
93
0
        }
94
95
        // parse deletion vector
96
0
        const char* buf = buffer.data();
97
0
        uint32_t actual_length;
98
0
        std::memcpy(reinterpret_cast<char*>(&actual_length), buf, 4);
99
        // change byte order to big endian
100
0
        std::reverse(reinterpret_cast<char*>(&actual_length),
101
0
                     reinterpret_cast<char*>(&actual_length) + 4);
102
0
        buf += 4;
103
0
        if (actual_length != bytes_read - 4) [[unlikely]] {
104
0
            create_status = Status::RuntimeError(
105
0
                    "DeletionVector deserialize error: length not match, "
106
0
                    "actual length: {}, expect length: {}",
107
0
                    actual_length, bytes_read - 4);
108
0
            return nullptr;
109
0
        }
110
0
        uint32_t magic_number;
111
0
        std::memcpy(reinterpret_cast<char*>(&magic_number), buf, 4);
112
        // change byte order to big endian
113
0
        std::reverse(reinterpret_cast<char*>(&magic_number),
114
0
                     reinterpret_cast<char*>(&magic_number) + 4);
115
0
        buf += 4;
116
0
        const static uint32_t MAGIC_NUMBER = 1581511376;
117
0
        if (magic_number != MAGIC_NUMBER) [[unlikely]] {
118
0
            create_status = Status::RuntimeError(
119
0
                    "DeletionVector deserialize error: invalid magic number {}", magic_number);
120
0
            return nullptr;
121
0
        }
122
123
0
        roaring::Roaring roaring_bitmap;
124
0
        SCOPED_TIMER(_paimon_profile.parse_deletion_vector_time);
125
0
        try {
126
0
            roaring_bitmap = roaring::Roaring::readSafe(buf, bytes_read - 4);
127
0
        } catch (const std::runtime_error& e) {
128
0
            create_status = Status::RuntimeError(
129
0
                    "DeletionVector deserialize error: failed to deserialize roaring bitmap, {}",
130
0
                    e.what());
131
0
            return nullptr;
132
0
        }
133
0
        delete_rows->reserve(roaring_bitmap.cardinality());
134
0
        for (auto it = roaring_bitmap.begin(); it != roaring_bitmap.end(); it++) {
135
0
            delete_rows->push_back(*it);
136
0
        }
137
0
        COUNTER_UPDATE(_paimon_profile.num_delete_rows, delete_rows->size());
138
0
        return delete_rows;
139
0
    });
140
0
    RETURN_IF_ERROR(create_status);
141
0
    if (!_delete_rows->empty()) [[likely]] {
142
0
        set_delete_rows();
143
0
    }
144
0
    return Status::OK();
145
0
}
146
147
0
Status PaimonReader::get_next_block_inner(Block* block, size_t* read_rows, bool* eof) {
148
0
    RETURN_IF_ERROR(_file_format_reader->get_next_block(block, read_rows, eof));
149
0
    return Status::OK();
150
0
}
151
#include "common/compile_check_end.h"
152
} // namespace doris