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 <memory>
#include <utility>
#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 validate_paimon_deletion_vector_descriptor(const TPaimonDeletionFileDesc& deletion_file,
                                                  size_t& bytes_read) {
    if (!deletion_file.__isset.path || !deletion_file.__isset.offset ||
        !deletion_file.__isset.length) {
        return Status::DataQualityError(
                "Paimon deletion file descriptor misses path/offset/length");
    }
    return validate_paimon_deletion_vector_read_range(deletion_file.offset, deletion_file.length,
                                                      bytes_read);
}

Status decode_paimon_deletion_vector_buffer(const char* buf, size_t buffer_size,
                                            DeletionVector* deletion_vector) {
    if (deletion_vector == nullptr) {
        return Status::InvalidArgument("deletion vector output must not be null");
    }
    if (buf == nullptr) {
        return Status::DataQualityError("Paimon deletion vector blob is null");
    }
    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 (static_cast<uint64_t>(actual_length) + 4 != buffer_size) [[unlikely]] {
        return Status::DataQualityError(
                "Paimon deletion vector length mismatch, expected: {}, actual: {}",
                static_cast<uint64_t>(actual_length) + 4, buffer_size);
    }

    if (memcmp(buf + sizeof(actual_length), PAIMON_BITMAP_MAGIC, 4) != 0) [[unlikely]] {
        return Status::DataQualityError(
                "Paimon deletion vector magic number mismatch, expected: {}, actual: {}",
                BigEndian::Load32(PAIMON_BITMAP_MAGIC),
                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());
    }

    *deletion_vector |= DeletionVector(std::move(roaring_bitmap));
    return Status::OK();
}

namespace {

template <typename Profile>
void init_deletion_vector_cache_profile(RuntimeProfile* profile, const char* parent,
                                        Profile* counters) {
    counters->decoded_cache_hit_count =
            ADD_CHILD_COUNTER(profile, "DeletionVectorDecodedCacheHitCount", TUnit::UNIT, parent);
    counters->decoded_cache_miss_count =
            ADD_CHILD_COUNTER(profile, "DeletionVectorDecodedCacheMissCount", TUnit::UNIT, parent);
    counters->file_cache_hit_count =
            ADD_CHILD_COUNTER(profile, "DeletionVectorFileCacheHitCount", TUnit::UNIT, parent);
    counters->file_cache_miss_count =
            ADD_CHILD_COUNTER(profile, "DeletionVectorFileCacheMissCount", TUnit::UNIT, parent);
    counters->file_cache_peer_read_count =
            ADD_CHILD_COUNTER(profile, "DeletionVectorFileCachePeerReadCount", TUnit::UNIT, parent);
}

template <typename Profile>
void update_deletion_vector_file_cache_profile(const DeletionVectorReader& reader,
                                               Profile* counters) {
    const auto& stats = reader.file_cache_statistics();
    COUNTER_UPDATE(counters->file_cache_hit_count, stats.num_local_io_total);
    COUNTER_UPDATE(counters->file_cache_miss_count, stats.num_remote_io_total);
    COUNTER_UPDATE(counters->file_cache_peer_read_count, stats.num_peer_io_total);
}

} // namespace

// ============================================================================
// 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);
    init_deletion_vector_cache_profile(get_profile(), paimon_profile, &_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;
    size_t bytes_read = 0;
    RETURN_IF_ERROR(validate_paimon_deletion_vector_descriptor(deletion_file, bytes_read));

    Status create_status = Status::OK();

    SCOPED_TIMER(_paimon_profile.delete_files_read_time);
    bool decoded_cache_hit = false;
    _deletion_vector = _kv_cache->get<DeletionVector>(
            build_paimon_deletion_vector_cache_key(deletion_file),
            [&]() -> DeletionVector* {
                auto deletion_vector = std::make_unique<DeletionVector>();

                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 = static_cast<int64_t>(bytes_read);
                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;
                }

                std::vector<char> buffer(bytes_read);
                create_status =
                        dv_reader.read_at(deletion_file.offset, {buffer.data(), bytes_read});
                update_deletion_vector_file_cache_profile(dv_reader, &_paimon_profile);
                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,
                                                                     deletion_vector.get());
                if (!create_status.ok()) [[unlikely]] {
                    return nullptr;
                }
                COUNTER_UPDATE(_paimon_profile.num_delete_rows, deletion_vector->cardinality());
                return deletion_vector.release();
            },
            &decoded_cache_hit);
    RETURN_IF_ERROR(create_status);
    COUNTER_UPDATE(decoded_cache_hit ? _paimon_profile.decoded_cache_hit_count
                                     : _paimon_profile.decoded_cache_miss_count,
                   1);
    if (!_deletion_vector->isEmpty()) [[likely]] {
        set_deletion_vector(_deletion_vector);
    }
    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);
    init_deletion_vector_cache_profile(get_profile(), paimon_profile, &_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;
    size_t bytes_read = 0;
    RETURN_IF_ERROR(validate_paimon_deletion_vector_descriptor(deletion_file, bytes_read));

    Status create_status = Status::OK();

    SCOPED_TIMER(_paimon_profile.delete_files_read_time);
    bool decoded_cache_hit = false;
    _deletion_vector = _kv_cache->get<DeletionVector>(
            build_paimon_deletion_vector_cache_key(deletion_file),
            [&]() -> DeletionVector* {
                auto deletion_vector = std::make_unique<DeletionVector>();

                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 = static_cast<int64_t>(bytes_read);
                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;
                }

                std::vector<char> buffer(bytes_read);
                create_status =
                        dv_reader.read_at(deletion_file.offset, {buffer.data(), bytes_read});
                update_deletion_vector_file_cache_profile(dv_reader, &_paimon_profile);
                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,
                                                                     deletion_vector.get());
                if (!create_status.ok()) [[unlikely]] {
                    return nullptr;
                }
                COUNTER_UPDATE(_paimon_profile.num_delete_rows, deletion_vector->cardinality());
                return deletion_vector.release();
            },
            &decoded_cache_hit);
    RETURN_IF_ERROR(create_status);
    COUNTER_UPDATE(decoded_cache_hit ? _paimon_profile.decoded_cache_hit_count
                                     : _paimon_profile.decoded_cache_miss_count,
                   1);
    if (!_deletion_vector->isEmpty()) [[likely]] {
        ParquetReader::set_deletion_vector(_deletion_vector);
    }
    return Status::OK();
}

} // namespace doris

Coverage Report

Created: 2026-07-20 15:57