be/src/exec/sink/viceberg_delete_sink.cpp

// 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 "exec/sink/viceberg_delete_sink.h"

#include <fmt/format.h>
#include <rapidjson/stringbuffer.h>
#include <rapidjson/writer.h>
#include <zlib.h>

#include "common/logging.h"
#include "core/block/column_with_type_and_name.h"
#include "core/column/column_nullable.h"
#include "core/column/column_string.h"
#include "core/column/column_struct.h"
#include "core/column/column_vector.h"
#include "core/data_type/data_type_factory.hpp"
#include "core/data_type/data_type_nullable.h"
#include "core/data_type/data_type_number.h"
#include "core/data_type/data_type_string.h"
#include "core/data_type/data_type_struct.h"
#include "exec/common/endian.h"
#include "exec/sink/writer/external_file_writer_compatibility.h"
#include "exprs/vexpr.h"
#include "format/table/deletion_vector.h"
#include "format/table/iceberg_delete_file_reader_helper.h"
#include "format/transformer/vfile_format_transformer.h"
#include "io/file_factory.h"
#include "runtime/runtime_state.h"
#include "util/debug_points.h"
#include "util/slice.h"
#include "util/string_util.h"
#include "util/uid_util.h"

namespace doris {

namespace {

class RewriteBitmapVisitor final : public IcebergPositionDeleteVisitor {
public:
    RewriteBitmapVisitor(const std::string& referenced_data_file_path,
                         roaring::Roaring64Map* rows_to_delete)
            : _referenced_data_file_path(referenced_data_file_path),
              _rows_to_delete(rows_to_delete) {}

    Status visit(const std::string& file_path, int64_t pos) override {
        if (_rows_to_delete == nullptr) {
            return Status::InvalidArgument("rows_to_delete is null");
        }
        if (file_path == _referenced_data_file_path) {
            _rows_to_delete->add(static_cast<uint64_t>(pos));
        }
        return Status::OK();
    }

private:
    const std::string& _referenced_data_file_path;
    roaring::Roaring64Map* _rows_to_delete;
};

Status load_rewritable_delete_rows(RuntimeState* state, RuntimeProfile* profile,
                                   const std::string& referenced_data_file_path,
                                   const std::vector<TIcebergDeleteFileDesc>& delete_files,
                                   const std::map<std::string, std::string>& hadoop_conf,
                                   TFileType::type file_type,
                                   const std::vector<TNetworkAddress>& broker_addresses,
                                   roaring::Roaring64Map* rows_to_delete) {
    if (rows_to_delete == nullptr) {
        return Status::InvalidArgument("rows_to_delete is null");
    }
    if (state == nullptr || profile == nullptr || delete_files.empty()) {
        return Status::OK();
    }

    TFileScanRangeParams params =
            build_iceberg_delete_scan_range_params(hadoop_conf, file_type, broker_addresses);
    IcebergDeleteFileIOContext delete_file_io_ctx(state);
    IcebergDeleteFileReaderOptions options;
    options.state = state;
    options.profile = profile;
    options.scan_params = &params;
    options.io_ctx = &delete_file_io_ctx.io_ctx;
    options.batch_size = 102400;

    for (const auto& delete_file : delete_files) {
        if (is_iceberg_deletion_vector(delete_file)) {
            RETURN_IF_ERROR(read_iceberg_deletion_vector(delete_file, options, rows_to_delete));
            continue;
        }
        RewriteBitmapVisitor visitor(referenced_data_file_path, rows_to_delete);
        RETURN_IF_ERROR(read_iceberg_position_delete_file(delete_file, options, &visitor));
    }
    return Status::OK();
}

} // namespace

Status calculate_iceberg_deletion_vector_content_size(size_t bitmap_size, int64_t* content_size) {
    DORIS_CHECK(content_size != nullptr);
    constexpr size_t max_bitmap_size = static_cast<size_t>(MAX_ICEBERG_DELETION_VECTOR_BYTES) -
                                       ICEBERG_DELETION_VECTOR_BLOB_OVERHEAD_BYTES;
    if (bitmap_size > max_bitmap_size) {
        return Status::NotSupported(
                "Iceberg deletion vector bitmap size exceeds Doris supported limit: {}, "
                "maximum bitmap size: {}, content size limit: {}",
                bitmap_size, max_bitmap_size, MAX_ICEBERG_DELETION_VECTOR_BYTES);
    }
    *content_size = static_cast<int64_t>(bitmap_size + ICEBERG_DELETION_VECTOR_BLOB_OVERHEAD_BYTES);
    return Status::OK();
}

VIcebergDeleteSink::VIcebergDeleteSink(const TDataSink& t_sink,
                                       const VExprContextSPtrs& output_exprs,
                                       std::shared_ptr<Dependency> dep,
                                       std::shared_ptr<Dependency> fin_dep)
        : AsyncResultWriter(output_exprs, dep, fin_dep), _t_sink(t_sink) {
    DCHECK(_t_sink.__isset.iceberg_delete_sink);
}

Status VIcebergDeleteSink::init_properties(ObjectPool* pool) {
    const auto& delete_sink = _t_sink.iceberg_delete_sink;

    _delete_type = delete_sink.delete_type;
    if (_delete_type != TFileContent::POSITION_DELETES) {
        return Status::NotSupported("Iceberg delete only supports position delete files");
    }

    // Get file format settings
    if (delete_sink.__isset.file_format) {
        _file_format_type = delete_sink.file_format;
    }

    if (delete_sink.__isset.compress_type) {
        _compress_type = delete_sink.compress_type;
    }

    // Get output path and table location
    if (delete_sink.__isset.output_path) {
        _output_path = delete_sink.output_path;
    }

    if (delete_sink.__isset.table_location) {
        _table_location = delete_sink.table_location;
    }

    // Get Hadoop configuration
    if (delete_sink.__isset.hadoop_config) {
        _hadoop_conf.insert(delete_sink.hadoop_config.begin(), delete_sink.hadoop_config.end());
    }

    if (delete_sink.__isset.file_type) {
        _file_type = delete_sink.file_type;
    }

    if (delete_sink.__isset.broker_addresses) {
        _broker_addresses.assign(delete_sink.broker_addresses.begin(),
                                 delete_sink.broker_addresses.end());
    }

    // Get partition information
    if (delete_sink.__isset.partition_spec_id) {
        _partition_spec_id = delete_sink.partition_spec_id;
    }

    if (delete_sink.__isset.partition_data_json) {
        _partition_data_json = delete_sink.partition_data_json;
    }

    if (delete_sink.__isset.format_version) {
        _format_version = delete_sink.format_version;
    }

    // for merge old deletion vector and old position delete to a new deletion vector.
    if (_format_version >= 3 && delete_sink.__isset.rewritable_delete_file_sets) {
        for (const auto& delete_file_set : delete_sink.rewritable_delete_file_sets) {
            if (!delete_file_set.__isset.referenced_data_file_path ||
                !delete_file_set.__isset.delete_files ||
                delete_file_set.referenced_data_file_path.empty() ||
                delete_file_set.delete_files.empty()) {
                continue;
            }
            _rewritable_delete_files.emplace(delete_file_set.referenced_data_file_path,
                                             delete_file_set.delete_files);
        }
    }

    return Status::OK();
}

Status VIcebergDeleteSink::open(RuntimeState* state, RuntimeProfile* profile) {
    _state = state;

    RETURN_IF_ERROR(validate_external_file_report_ack(state->query_options()));

    // Initialize counters
    _written_rows_counter = ADD_COUNTER(profile, "RowsWritten", TUnit::UNIT);
    _send_data_timer = ADD_TIMER(profile, "SendDataTime");
    _write_delete_files_timer = ADD_TIMER(profile, "WriteDeleteFilesTime");
    _delete_file_count_counter = ADD_COUNTER(profile, "DeleteFileCount", TUnit::UNIT);
    _open_timer = ADD_TIMER(profile, "OpenTime");
    _close_timer = ADD_TIMER(profile, "CloseTime");

    SCOPED_TIMER(_open_timer);

    if (_format_version < 3) {
        RETURN_IF_ERROR(_init_position_delete_output_exprs());
    }

    LOG(INFO) << fmt::format(
            "VIcebergDeleteSink opened: delete_type={}, output_path={}, format_version={}",
            to_string(_delete_type), _output_path, _format_version);

    return Status::OK();
}

Status VIcebergDeleteSink::write(RuntimeState* state, Block& block) {
    SCOPED_TIMER(_send_data_timer);

    if (block.rows() == 0) {
        return Status::OK();
    }

    _row_count += block.rows();

    if (_delete_type != TFileContent::POSITION_DELETES) {
        return Status::NotSupported("Iceberg delete only supports position delete files");
    }

    // Extract $row_id column and group by file_path
    RETURN_IF_ERROR(_collect_position_deletes(block, _file_deletions));

    if (_written_rows_counter) {
        COUNTER_UPDATE(_written_rows_counter, block.rows());
    }

    return Status::OK();
}

Status VIcebergDeleteSink::close(Status close_status) {
    SCOPED_TIMER(_close_timer);

    if (!close_status.ok()) {
        LOG(WARNING) << fmt::format("VIcebergDeleteSink close with error: {}",
                                    close_status.to_string());
        _cleanup_created_files();
        return close_status;
    }

    Status write_status = Status::OK();
    DBUG_EXECUTE_IF("VIcebergDeleteSink.close.inject_failure", {
        write_status = Status::InternalError("injected Iceberg delete close failure");
    });

    if (_delete_type == TFileContent::POSITION_DELETES && !_file_deletions.empty()) {
        SCOPED_TIMER(_write_delete_files_timer);
        if (write_status.ok()) {
            write_status = _format_version >= 3 ? _write_deletion_vector_files(_file_deletions)
                                                : _write_position_delete_files(_file_deletions);
        }
    }
    if (!write_status.ok()) {
        _cleanup_created_files();
        return write_status;
    }

    // Update counters
    if (_delete_file_count_counter) {
        COUNTER_UPDATE(_delete_file_count_counter, _delete_file_count);
    }

    LOG(INFO) << fmt::format("VIcebergDeleteSink closed: rows={}, delete_files={}", _row_count,
                             _delete_file_count);

    if (_state != nullptr) {
        for (auto& commit_data : _commit_data_list) {
            Status report_status = _state->add_iceberg_commit_datas(std::move(commit_data));
            if (!report_status.ok()) {
                _cleanup_created_files();
                return report_status;
            }
        }
    }

    if (!_defer_file_cleanup_until_outer_close) {
        _transfer_created_files_to_report_cleanup();
    }

    return Status::OK();
}

void VIcebergDeleteSink::finish_deferred_file_cleanup(Status outer_status) {
    if (!outer_status.ok()) {
        _cleanup_created_files();
    } else {
        _transfer_created_files_to_report_cleanup();
    }
    _defer_file_cleanup_until_outer_close = false;
}

void VIcebergDeleteSink::_transfer_created_files_to_report_cleanup() {
    DCHECK(_state != nullptr);
    for (auto& created_file : _created_files) {
        _state->add_rejected_external_file_report_cleanup(
                [cleanup_fs = std::move(created_file.first),
                 cleanup_path = std::move(created_file.second)] {
                    WARN_IF_ERROR(cleanup_fs->delete_file(cleanup_path),
                                  "failed to delete an Iceberg delete file after report failure");
                });
    }
    _created_files.clear();
}

void VIcebergDeleteSink::_cleanup_created_files() {
    for (const auto& [fs, path] : _created_files) {
        Status delete_status = fs->delete_file(path);
        if (!delete_status.ok() && !delete_status.is<ErrorCode::NOT_FOUND>()) {
            LOG(WARNING) << fmt::format("Failed to delete Iceberg delete file {}: {}", path,
                                        delete_status.to_string());
        }
    }
    _created_files.clear();
}

int VIcebergDeleteSink::_get_row_id_column_index(const Block& block) {
    // Find __DORIS_ICEBERG_ROWID_COL__ column in block
    for (size_t i = 0; i < block.columns(); ++i) {
        const auto& col_name = block.get_by_position(i).name;
        if (col_name == doris::BeConsts::ICEBERG_ROWID_COL) {
            return static_cast<int>(i);
        }
    }
    return -1;
}

Status VIcebergDeleteSink::_collect_position_deletes(
        const Block& block, std::map<std::string, IcebergFileDeletion>& file_deletions) {
    // Find row id column
    int row_id_col_idx = _get_row_id_column_index(block);
    if (row_id_col_idx < 0) {
        return Status::InternalError(
                "__DORIS_ICEBERG_ROWID_COL__ column not found in block for position delete");
    }

    const auto& row_id_col = block.get_by_position(row_id_col_idx);
    const IColumn* row_id_data = row_id_col.column.get();
    const IDataType* row_id_type = row_id_col.type.get();
    const auto* nullable_col = check_and_get_column<ColumnNullable>(row_id_data);
    if (nullable_col != nullptr) {
        row_id_data = nullable_col->get_nested_column_ptr().get();
    }
    const auto* nullable_type = check_and_get_data_type<DataTypeNullable>(row_id_type);
    if (nullable_type != nullptr) {
        row_id_type = nullable_type->get_nested_type().get();
    }
    const auto* struct_col = check_and_get_column<ColumnStruct>(row_id_data);
    const auto* struct_type = check_and_get_data_type<DataTypeStruct>(row_id_type);
    if (!struct_col || !struct_type) {
        return Status::InternalError("__DORIS_ICEBERG_ROWID_COL__ column is not a struct column");
    }

    // __DORIS_ICEBERG_ROWID_COL__ struct:
    // (file_path: STRING, row_position: BIGINT, partition_spec_id: INT, partition_data: STRING)
    size_t field_count = struct_col->tuple_size();
    if (field_count < 2) {
        return Status::InternalError(
                "__DORIS_ICEBERG_ROWID_COL__ struct must have at least 2 fields "
                "(file_path, row_position)");
    }

    auto normalize = [](const std::string& name) { return doris::to_lower(name); };

    int file_path_idx = -1;
    int row_position_idx = -1;
    int spec_id_idx = -1;
    int partition_data_idx = -1;
    const auto& field_names = struct_type->get_element_names();
    for (size_t i = 0; i < field_names.size(); ++i) {
        std::string name = normalize(field_names[i]);
        if (file_path_idx < 0 && name == "file_path") {
            file_path_idx = static_cast<int>(i);
        } else if (row_position_idx < 0 && name == "row_position") {
            row_position_idx = static_cast<int>(i);
        } else if (spec_id_idx < 0 && name == "partition_spec_id") {
            spec_id_idx = static_cast<int>(i);
        } else if (partition_data_idx < 0 && name == "partition_data") {
            partition_data_idx = static_cast<int>(i);
        }
    }

    if (file_path_idx < 0 || row_position_idx < 0) {
        return Status::InternalError(
                "__DORIS_ICEBERG_ROWID_COL__ must contain standard fields file_path and "
                "row_position");
    }
    if (field_count >= 3 && spec_id_idx < 0) {
        return Status::InternalError(
                "__DORIS_ICEBERG_ROWID_COL__ must use standard field name partition_spec_id");
    }
    if (field_count >= 4 && partition_data_idx < 0) {
        return Status::InternalError(
                "__DORIS_ICEBERG_ROWID_COL__ must use standard field name partition_data");
    }

    const auto* file_path_col = check_and_get_column<ColumnString>(
            remove_nullable(struct_col->get_column_ptr(file_path_idx)).get());
    const auto* row_position_col = check_and_get_column<ColumnVector<TYPE_BIGINT>>(
            remove_nullable(struct_col->get_column_ptr(row_position_idx)).get());

    if (!file_path_col || !row_position_col) {
        return Status::InternalError(
                "__DORIS_ICEBERG_ROWID_COL__ struct fields have incorrect types");
    }

    const ColumnVector<TYPE_INT>* spec_id_col = nullptr;
    const ColumnString* partition_data_col = nullptr;
    if (spec_id_idx >= 0 && spec_id_idx < static_cast<int>(field_count)) {
        spec_id_col = check_and_get_column<ColumnVector<TYPE_INT>>(
                remove_nullable(struct_col->get_column_ptr(spec_id_idx)).get());
        if (!spec_id_col) {
            return Status::InternalError(
                    "__DORIS_ICEBERG_ROWID_COL__ partition_spec_id has incorrect type");
        }
    }
    if (partition_data_idx >= 0 && partition_data_idx < static_cast<int>(field_count)) {
        partition_data_col = check_and_get_column<ColumnString>(
                remove_nullable(struct_col->get_column_ptr(partition_data_idx)).get());
        if (!partition_data_col) {
            return Status::InternalError(
                    "__DORIS_ICEBERG_ROWID_COL__ partition_data has incorrect type");
        }
    }

    // Group by file_path using roaring bitmap
    for (size_t i = 0; i < block.rows(); ++i) {
        std::string file_path = file_path_col->get_data_at(i).to_string();
        int64_t row_position = row_position_col->get_element(i);
        if (row_position < 0) {
            return Status::InternalError("Invalid row_position {} in row_id column", row_position);
        }

        int32_t partition_spec_id = _partition_spec_id;
        std::string partition_data_json = _partition_data_json;
        if (spec_id_col != nullptr) {
            partition_spec_id = spec_id_col->get_element(i);
        }
        if (partition_data_col != nullptr) {
            partition_data_json = partition_data_col->get_data_at(i).to_string();
        }

        auto [iter, inserted] = file_deletions.emplace(
                file_path, IcebergFileDeletion(partition_spec_id, partition_data_json));
        if (!inserted) {
            if (iter->second.partition_spec_id != partition_spec_id ||
                iter->second.partition_data_json != partition_data_json) {
                LOG(WARNING) << fmt::format(
                        "Mismatched partition info for file {}, existing spec_id={}, data={}, "
                        "new spec_id={}, data={}",
                        file_path, iter->second.partition_spec_id, iter->second.partition_data_json,
                        partition_spec_id, partition_data_json);
            }
        }
        iter->second.rows_to_delete.add(static_cast<uint64_t>(row_position));
    }

    return Status::OK();
}

Status VIcebergDeleteSink::_write_position_delete_files(
        const std::map<std::string, IcebergFileDeletion>& file_deletions) {
    constexpr size_t kBatchSize = 4096;
    for (const auto& [data_file_path, deletion] : file_deletions) {
        if (deletion.rows_to_delete.isEmpty()) {
            continue;
        }
        // Generate unique delete file path
        std::string delete_file_path = _generate_delete_file_path(data_file_path);

        // Create delete file writer
        auto writer = VIcebergDeleteFileWriterFactory::create_writer(
                TFileContent::POSITION_DELETES, delete_file_path, _file_format_type,
                _compress_type);

        // Build column names for position delete
        std::vector<std::string> column_names = {"file_path", "pos"};

        if (_position_delete_output_expr_ctxs.empty()) {
            RETURN_IF_ERROR(_init_position_delete_output_exprs());
        }

        // Open writer
        Status open_status =
                writer->open(_state, _state->runtime_profile(), _position_delete_output_expr_ctxs,
                             column_names, _hadoop_conf, _file_type, _broker_addresses);
        if (writer->file_system() != nullptr) {
            _created_files.emplace_back(writer->file_system(), delete_file_path);
        }
        RETURN_IF_ERROR(open_status);

        // Build block with (file_path, pos) columns
        std::vector<int64_t> positions;
        positions.reserve(kBatchSize);
        for (auto it = deletion.rows_to_delete.begin(); it != deletion.rows_to_delete.end(); ++it) {
            positions.push_back(static_cast<int64_t>(*it));
            if (positions.size() >= kBatchSize) {
                Block delete_block;
                RETURN_IF_ERROR(
                        _build_position_delete_block(data_file_path, positions, delete_block));
                RETURN_IF_ERROR(writer->write(delete_block));
                positions.clear();
            }
        }
        if (!positions.empty()) {
            Block delete_block;
            RETURN_IF_ERROR(_build_position_delete_block(data_file_path, positions, delete_block));
            RETURN_IF_ERROR(writer->write(delete_block));
        }

        // Set partition info on writer before close
        writer->set_partition_info(deletion.partition_spec_id, deletion.partition_data_json);

        // Close writer and collect commit data
        TIcebergCommitData commit_data;
        RETURN_IF_ERROR(writer->close(commit_data));

        // Set referenced data file path
        commit_data.__set_referenced_data_file_path(data_file_path);

        _commit_data_list.push_back(commit_data);
        _delete_file_count++;

        VLOG(1) << fmt::format("Written position delete file: path={}, rows={}, referenced_file={}",
                               delete_file_path, commit_data.row_count, data_file_path);
    }

    return Status::OK();
}

Status VIcebergDeleteSink::_init_position_delete_output_exprs() {
    if (!_position_delete_output_expr_ctxs.empty()) {
        return Status::OK();
    }

    std::vector<TExpr> texprs;
    texprs.reserve(2);

    std::string empty_string;
    TExprNode file_path_node =
            create_texpr_node_from(&empty_string, PrimitiveType::TYPE_STRING, 0, 0);
    file_path_node.__set_num_children(0);
    file_path_node.__set_output_scale(0);
    file_path_node.__set_is_nullable(false);
    TExpr file_path_expr;
    file_path_expr.nodes.emplace_back(std::move(file_path_node));
    texprs.emplace_back(std::move(file_path_expr));

    int64_t zero = 0;
    TExprNode pos_node = create_texpr_node_from(&zero, PrimitiveType::TYPE_BIGINT, 0, 0);
    pos_node.__set_num_children(0);
    pos_node.__set_output_scale(0);
    pos_node.__set_is_nullable(false);
    TExpr pos_expr;
    pos_expr.nodes.emplace_back(std::move(pos_node));
    texprs.emplace_back(std::move(pos_expr));

    RETURN_IF_ERROR(VExpr::create_expr_trees(texprs, _position_delete_output_expr_ctxs));
    return Status::OK();
}

Status VIcebergDeleteSink::_build_position_delete_block(const std::string& file_path,
                                                        const std::vector<int64_t>& positions,
                                                        Block& output_block) {
    // Create file_path column (repeated for each position)
    auto file_path_col = ColumnString::create();
    for (size_t i = 0; i < positions.size(); ++i) {
        file_path_col->insert_data(file_path.data(), file_path.size());
    }

    // Create pos column
    auto pos_col = ColumnVector<TYPE_BIGINT>::create();
    pos_col->get_data().assign(positions.begin(), positions.end());

    // Build block
    output_block.insert(ColumnWithTypeAndName(std::move(file_path_col),
                                              std::make_shared<DataTypeString>(), "file_path"));
    output_block.insert(
            ColumnWithTypeAndName(std::move(pos_col), std::make_shared<DataTypeInt64>(), "pos"));

    return Status::OK();
}

std::string VIcebergDeleteSink::_get_file_extension() const {
    std::string compress_name;
    switch (_compress_type) {
    case TFileCompressType::SNAPPYBLOCK: {
        compress_name = ".snappy";
        break;
    }
    case TFileCompressType::ZLIB: {
        compress_name = ".zlib";
        break;
    }
    case TFileCompressType::ZSTD: {
        compress_name = ".zstd";
        break;
    }
    default: {
        compress_name = "";
        break;
    }
    }

    std::string file_format_name;
    switch (_file_format_type) {
    case TFileFormatType::FORMAT_PARQUET: {
        file_format_name = ".parquet";
        break;
    }
    case TFileFormatType::FORMAT_ORC: {
        file_format_name = ".orc";
        break;
    }
    default: {
        file_format_name = "";
        break;
    }
    }
    return fmt::format("{}{}", compress_name, file_format_name);
}

Status VIcebergDeleteSink::_write_deletion_vector_files(
        const std::map<std::string, IcebergFileDeletion>& file_deletions) {
    std::vector<DeletionVectorBlob> blobs;
    for (const auto& [data_file_path, deletion] : file_deletions) {
        if (deletion.rows_to_delete.isEmpty()) {
            continue;
        }
        roaring::Roaring64Map merged_rows = deletion.rows_to_delete;
        DeletionVectorBlob blob;
        blob.delete_count = static_cast<int64_t>(merged_rows.cardinality());
        auto previous_delete_it = _rewritable_delete_files.find(data_file_path);
        if (previous_delete_it != _rewritable_delete_files.end()) {
            roaring::Roaring64Map previous_rows;
            RETURN_IF_ERROR(load_rewritable_delete_rows(
                    _state, _state->runtime_profile(), data_file_path, previous_delete_it->second,
                    _hadoop_conf, _file_type, _broker_addresses, &previous_rows));
            merged_rows |= previous_rows;
        }

        size_t bitmap_size = merged_rows.getSizeInBytes();
        blob.referenced_data_file = data_file_path;
        blob.partition_spec_id = deletion.partition_spec_id;
        blob.partition_data_json = deletion.partition_data_json;
        blob.merged_count = static_cast<int64_t>(merged_rows.cardinality());
        RETURN_IF_ERROR(calculate_iceberg_deletion_vector_content_size(
                bitmap_size, &blob.content_size_in_bytes));
        blob.blob_data.resize(static_cast<size_t>(blob.content_size_in_bytes));
        merged_rows.write(blob.blob_data.data() + 8);

        uint32_t total_length = static_cast<uint32_t>(4 + bitmap_size);
        BigEndian::Store32(blob.blob_data.data(), total_length);

        constexpr char DV_MAGIC[] = {'\xD1', '\xD3', '\x39', '\x64'};
        memcpy(blob.blob_data.data() + 4, DV_MAGIC, 4);

        uint32_t crc = static_cast<uint32_t>(
                ::crc32(0, reinterpret_cast<const Bytef*>(blob.blob_data.data() + 4),
                        static_cast<uInt>(4 + bitmap_size)));
        BigEndian::Store32(blob.blob_data.data() + 8 + bitmap_size, crc);
        blobs.emplace_back(std::move(blob));
    }

    if (blobs.empty()) {
        return Status::OK();
    }

    std::string puffin_path = _generate_puffin_file_path();
    int64_t puffin_file_size = 0;
    RETURN_IF_ERROR(_write_puffin_file(puffin_path, &blobs, &puffin_file_size));

    for (const auto& blob : blobs) {
        TIcebergCommitData commit_data;
        commit_data.__set_file_path(puffin_path);
        commit_data.__set_row_count(blob.merged_count);
        commit_data.__set_affected_rows(blob.delete_count);
        commit_data.__set_file_size(puffin_file_size);
        commit_data.__set_file_content(TFileContent::DELETION_VECTOR);
        commit_data.__set_content_offset(blob.content_offset);
        commit_data.__set_content_size_in_bytes(blob.content_size_in_bytes);
        commit_data.__set_referenced_data_file_path(blob.referenced_data_file);
        if (blob.partition_spec_id != 0 || !blob.partition_data_json.empty()) {
            commit_data.__set_partition_spec_id(blob.partition_spec_id);
            commit_data.__set_partition_data_json(blob.partition_data_json);
        }

        _commit_data_list.push_back(commit_data);
        _delete_file_count++;
    }
    return Status::OK();
}

Status VIcebergDeleteSink::_write_puffin_file(const std::string& puffin_path,
                                              std::vector<DeletionVectorBlob>* blobs,
                                              int64_t* out_file_size) {
    DCHECK(blobs != nullptr);
    DCHECK(!blobs->empty());

    io::FSPropertiesRef fs_properties(_file_type);
    fs_properties.properties = &_hadoop_conf;
    if (!_broker_addresses.empty()) {
        fs_properties.broker_addresses = &_broker_addresses;
    }
    io::FileDescription file_description = {.path = puffin_path, .fs_name {}};
    auto fs = DORIS_TRY(FileFactory::create_fs(fs_properties, file_description));
    io::FileWriterOptions file_writer_options = {.used_by_s3_committer = false};
    io::FileWriterPtr file_writer;
    _created_files.emplace_back(fs, puffin_path);
    RETURN_IF_ERROR(fs->create_file(file_description.path, &file_writer, &file_writer_options));

    constexpr char PUFFIN_MAGIC[] = {'\x50', '\x46', '\x41', '\x31'};
    RETURN_IF_ERROR(file_writer->append(Slice(reinterpret_cast<const uint8_t*>(PUFFIN_MAGIC), 4)));
    int64_t current_offset = 4;
    for (auto& blob : *blobs) {
        blob.content_offset = current_offset;
        RETURN_IF_ERROR(file_writer->append(Slice(
                reinterpret_cast<const uint8_t*>(blob.blob_data.data()), blob.blob_data.size())));
        current_offset += static_cast<int64_t>(blob.blob_data.size());
    }
    RETURN_IF_ERROR(file_writer->append(Slice(reinterpret_cast<const uint8_t*>(PUFFIN_MAGIC), 4)));

    std::string footer_json = _build_puffin_footer_json(*blobs);
    RETURN_IF_ERROR(file_writer->append(
            Slice(reinterpret_cast<const uint8_t*>(footer_json.data()), footer_json.size())));

    char footer_size_buf[4];
    LittleEndian::Store32(footer_size_buf, static_cast<uint32_t>(footer_json.size()));
    RETURN_IF_ERROR(file_writer->append(
            Slice(reinterpret_cast<const uint8_t*>(footer_size_buf), sizeof(footer_size_buf))));

    char flags[4] = {0, 0, 0, 0};
    RETURN_IF_ERROR(
            file_writer->append(Slice(reinterpret_cast<const uint8_t*>(flags), sizeof(flags))));
    RETURN_IF_ERROR(file_writer->append(Slice(reinterpret_cast<const uint8_t*>(PUFFIN_MAGIC), 4)));
    RETURN_IF_ERROR(file_writer->close());

    *out_file_size = current_offset + 4 + static_cast<int64_t>(footer_json.size()) + 4 + 4 + 4;
    return Status::OK();
}

std::string VIcebergDeleteSink::_build_puffin_footer_json(
        const std::vector<DeletionVectorBlob>& blobs) {
    rapidjson::StringBuffer buffer;
    rapidjson::Writer<rapidjson::StringBuffer> writer(buffer);
    writer.StartObject();
    writer.Key("blobs");
    writer.StartArray();
    for (const auto& blob : blobs) {
        writer.StartObject();
        writer.Key("type");
        writer.String("deletion-vector-v1");
        writer.Key("fields");
        writer.StartArray();
        writer.EndArray();
        writer.Key("snapshot-id");
        writer.Int64(-1);
        writer.Key("sequence-number");
        writer.Int64(-1);
        writer.Key("offset");
        writer.Int64(blob.content_offset);
        writer.Key("length");
        writer.Int64(blob.content_size_in_bytes);
        writer.Key("properties");
        writer.StartObject();
        writer.Key("referenced-data-file");
        writer.String(blob.referenced_data_file.c_str(),
                      static_cast<rapidjson::SizeType>(blob.referenced_data_file.size()));
        std::string cardinality = std::to_string(blob.merged_count);
        writer.Key("cardinality");
        writer.String(cardinality.c_str(), static_cast<rapidjson::SizeType>(cardinality.size()));
        writer.EndObject();
        writer.EndObject();
    }
    writer.EndArray();
    writer.Key("properties");
    writer.StartObject();
    writer.Key("created-by");
    writer.String("doris-puffin-v1");
    writer.EndObject();
    writer.EndObject();
    return {buffer.GetString(), buffer.GetSize()};
}

std::string VIcebergDeleteSink::_generate_delete_file_path(
        const std::string& referenced_data_file) {
    // Generate unique delete file name using UUID
    std::string uuid = generate_uuid_string();
    std::string file_name;

    std::string file_extension = _get_file_extension();
    file_name =
            fmt::format("delete_pos_{}_{}{}", uuid,
                        std::hash<std::string> {}(referenced_data_file) % 10000000, file_extension);

    // Combine with output path or table location
    std::string base_path = _output_path.empty() ? _table_location : _output_path;

    // Ensure base path ends with /
    if (!base_path.empty() && base_path.back() != '/') {
        base_path += '/';
    }

    // Delete files are data files in Iceberg, write under data location
    return fmt::format("{}{}", base_path, file_name);
}

std::string VIcebergDeleteSink::_generate_puffin_file_path() {
    std::string uuid = generate_uuid_string();
    std::string file_name = fmt::format("delete_dv_{}.puffin", uuid);
    std::string base_path = _output_path.empty() ? _table_location : _output_path;
    if (!base_path.empty() && base_path.back() != '/') {
        base_path += '/';
    }
    return fmt::format("{}{}", base_path, file_name);
}

} // namespace doris