| // 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 "exprs/vexpr.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/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 = ¶ms; |
| 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 |
| |
| 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; |
| |
| // 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()); |
| return close_status; |
| } |
| |
| if (_delete_type == TFileContent::POSITION_DELETES && !_file_deletions.empty()) { |
| SCOPED_TIMER(_write_delete_files_timer); |
| if (_format_version >= 3) { |
| RETURN_IF_ERROR(_write_deletion_vector_files(_file_deletions)); |
| } else { |
| RETURN_IF_ERROR(_write_position_delete_files(_file_deletions)); |
| } |
| } |
| |
| // 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 (const auto& commit_data : _commit_data_list) { |
| _state->add_iceberg_commit_datas(commit_data); |
| } |
| } |
| |
| return Status::OK(); |
| } |
| |
| 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 |
| RETURN_IF_ERROR(writer->open(_state, _state->runtime_profile(), |
| _position_delete_output_expr_ctxs, column_names, _hadoop_conf, |
| _file_type, _broker_addresses)); |
| |
| // 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()); |
| blob.content_size_in_bytes = static_cast<int64_t>(4 + 4 + bitmap_size + 4); |
| 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), |
| 4 + (uInt)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; |
| 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 |
