diff --git a/be/src/exec/scan/file_scanner.cpp b/be/src/exec/scan/file_scanner.cpp index 78d6421adb5054..ee14b4210f8158 100644 --- a/be/src/exec/scan/file_scanner.cpp +++ b/be/src/exec/scan/file_scanner.cpp @@ -69,6 +69,7 @@ #include "format/table/hive_reader.h" #include "format/table/hudi_jni_reader.h" #include "format/table/hudi_reader.h" +#include "format/table/iceberg_position_delete_sys_table_reader.h" #include "format/table/iceberg_reader.h" #include "format/table/iceberg_sys_table_jni_reader.h" #include "format/table/jdbc_jni_reader.h" @@ -98,6 +99,20 @@ class ShardedKVCache; namespace doris { using namespace ErrorCode; +namespace { +constexpr int kIcebergPositionDeleteContent = 1; +constexpr int kIcebergDeletionVectorContent = 3; + +bool is_iceberg_position_deletes_sys_table(const TFileRangeDesc& range) { + return range.__isset.table_format_params && + range.table_format_params.table_format_type == "iceberg" && + range.table_format_params.__isset.iceberg_params && + range.table_format_params.iceberg_params.__isset.content && + (range.table_format_params.iceberg_params.content == kIcebergPositionDeleteContent || + range.table_format_params.iceberg_params.content == kIcebergDeletionVectorContent); +} +} // namespace + const std::string FileScanner::FileReadBytesProfile = "FileReadBytes"; const std::string FileScanner::FileReadTimeProfile = "FileReadTime"; @@ -1038,6 +1053,7 @@ Status FileScanner::_get_next_reader() { // create reader for specific format Status init_status = Status::OK(); TFileFormatType::type format_type = _get_current_format_type(); + const bool is_position_deletes_sys_table = is_iceberg_position_deletes_sys_table(range); // for compatibility, this logic is deprecated in 3.1 if (format_type == TFileFormatType::FORMAT_JNI && range.__isset.table_format_params) { if (range.table_format_params.table_format_type == "paimon" && @@ -1163,6 +1179,17 @@ Status FileScanner::_get_next_reader() { auto file_meta_cache_ptr = _should_enable_file_meta_cache() ? ExecEnv::GetInstance()->file_meta_cache() : nullptr; + if (is_position_deletes_sys_table) { + ReaderInitContext ctx; + _fill_base_init_context(&ctx); + auto reader = IcebergPositionDeleteSysTableReader::create_unique( + _file_slot_descs, _state, _profile, range, _params, _io_ctx, + file_meta_cache_ptr); + init_status = static_cast(reader.get())->init_reader(&ctx); + _cur_reader = std::move(reader); + need_to_get_parsed_schema = false; + break; + } if (push_down_predicates) { RETURN_IF_ERROR(_process_late_arrival_conjuncts()); } @@ -1175,6 +1202,17 @@ Status FileScanner::_get_next_reader() { auto file_meta_cache_ptr = _should_enable_file_meta_cache() ? ExecEnv::GetInstance()->file_meta_cache() : nullptr; + if (is_position_deletes_sys_table) { + ReaderInitContext ctx; + _fill_base_init_context(&ctx); + auto reader = IcebergPositionDeleteSysTableReader::create_unique( + _file_slot_descs, _state, _profile, range, _params, _io_ctx, + file_meta_cache_ptr); + init_status = static_cast(reader.get())->init_reader(&ctx); + _cur_reader = std::move(reader); + need_to_get_parsed_schema = false; + break; + } if (push_down_predicates) { RETURN_IF_ERROR(_process_late_arrival_conjuncts()); } diff --git a/be/src/exec/scan/file_scanner_v2.cpp b/be/src/exec/scan/file_scanner_v2.cpp index 10b5f850ea36f7..fc60aaa2205188 100644 --- a/be/src/exec/scan/file_scanner_v2.cpp +++ b/be/src/exec/scan/file_scanner_v2.cpp @@ -54,6 +54,7 @@ #include "format_v2/jni/trino_connector_jni_reader.h" #include "format_v2/table/hive_reader.h" #include "format_v2/table/hudi_reader.h" +#include "format_v2/table/iceberg_position_delete_sys_table_reader.h" #include "format_v2/table/iceberg_reader.h" #include "format_v2/table/paimon_reader.h" #include "format_v2/table/remote_doris_reader.h" @@ -70,6 +71,9 @@ namespace doris { namespace { +constexpr int kIcebergPositionDeleteContent = 1; +constexpr int kIcebergDeletionVectorContent = 3; + std::string table_format_name(const TFileRangeDesc& range) { return range.__isset.table_format_params ? range.table_format_params.table_format_type : "NotSet"; @@ -110,6 +114,15 @@ bool is_supported_jni_table_format(const TFileRangeDesc& range) { table_format == "max_compute" || table_format == "trino_connector"; } +bool is_iceberg_position_deletes_sys_table(const TFileRangeDesc& range) { + return range.__isset.table_format_params && + range.table_format_params.table_format_type == "iceberg" && + range.table_format_params.__isset.iceberg_params && + range.table_format_params.iceberg_params.__isset.content && + (range.table_format_params.iceberg_params.content == kIcebergPositionDeleteContent || + range.table_format_params.iceberg_params.content == kIcebergDeletionVectorContent); +} + bool is_csv_format(TFileFormatType::type format_type) { switch (format_type) { case TFileFormatType::FORMAT_CSV_PLAIN: @@ -462,7 +475,9 @@ Status FileScannerV2::_create_table_reader_for_format( } else if (table_format == "hive") { *reader = format::hive::HiveReader::create_unique(); } else if (table_format == "iceberg") { - if (get_range_format_type(*_params, range) == TFileFormatType::FORMAT_JNI) { + if (is_iceberg_position_deletes_sys_table(range)) { + *reader = std::make_unique(); + } else if (get_range_format_type(*_params, range) == TFileFormatType::FORMAT_JNI) { *reader = std::make_unique(); } else { *reader = std::make_unique(); diff --git a/be/src/format/table/iceberg_position_delete_sys_table_reader.cpp b/be/src/format/table/iceberg_position_delete_sys_table_reader.cpp new file mode 100644 index 00000000000000..c6e97465e2dd44 --- /dev/null +++ b/be/src/format/table/iceberg_position_delete_sys_table_reader.cpp @@ -0,0 +1,608 @@ +// 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/iceberg_position_delete_sys_table_reader.h" + +#include +#include + +#include +#include +#include + +#include "common/cast_set.h" +#include "core/block/block.h" +#include "core/column/column_nullable.h" +#include "core/column/column_string.h" +#include "core/column/column_vector.h" +#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_serde/data_type_serde.h" +#include "core/types.h" +#include "format/orc/vorc_reader.h" +#include "format/parquet/schema_desc.h" +#include "format/parquet/vparquet_reader.h" +#include "format/table/parquet_utils.h" +#include "format/table/table_schema_change_helper.h" +#include "runtime/runtime_state.h" + +namespace doris { + +namespace { + +constexpr const char* kFilePathColumn = "file_path"; +constexpr const char* kPosColumn = "pos"; +constexpr const char* kRowColumn = "row"; +constexpr const char* kPartitionColumn = "partition"; +constexpr const char* kSpecIdColumn = "spec_id"; +constexpr const char* kDeleteFilePathColumn = "delete_file_path"; +constexpr const char* kContentOffsetColumn = "content_offset"; +constexpr const char* kContentSizeInBytesColumn = "content_size_in_bytes"; +constexpr const char* kIcebergOrcAttribute = "iceberg.id"; +constexpr int kPositionDeleteContent = 1; + +bool block_has_row(const Block& block, size_t row) { + return block.columns() > 0 && row < block.rows(); +} + +void insert_int64_nullable(MutableColumnPtr& column, const int64_t* value) { + if (value == nullptr) { + parquet_utils::insert_null(column); + } else { + parquet_utils::insert_int64(column, *value); + } +} + +// Fail loudly if the filled output block is malformed. Each output column must be produced by a +// file slot; a projected system-table column that is not backed by a file slot would be skipped +// during fill and left shorter than the rest, producing a block with inconsistent column lengths. +void check_output_columns_aligned(const MutableColumns& columns) { + if (columns.empty()) { + return; + } + const size_t expected_rows = columns.front()->size(); + for (const auto& column : columns) { + DORIS_CHECK(column->size() == expected_rows) + << "Iceberg position delete system table output block has inconsistent column " + "sizes; a projected column is not backed by a file slot"; + } +} + +const ColumnString* get_string_column(const Block& block, const std::string& name) { + auto pos = block.get_position_by_name(name); + if (pos < 0) { + return nullptr; + } + return check_and_get_column(block.get_by_position(pos).column.get()); +} + +const ColumnInt64* get_int64_column(const Block& block, const std::string& name) { + auto pos = block.get_position_by_name(name); + if (pos < 0) { + return nullptr; + } + return check_and_get_column(block.get_by_position(pos).column.get()); +} + +const schema::external::TField* get_field_ptr(const schema::external::TFieldPtr& field_ptr) { + if (!field_ptr.__isset.field_ptr || field_ptr.field_ptr == nullptr) { + return nullptr; + } + return field_ptr.field_ptr.get(); +} + +const schema::external::TField* find_current_schema_field(const TFileScanRangeParams* params, + const std::string& name) { + if (params == nullptr || !params->__isset.history_schema_info || + params->history_schema_info.empty()) { + return nullptr; + } + const schema::external::TSchema* schema = ¶ms->history_schema_info.front(); + if (params->__isset.current_schema_id) { + for (const auto& candidate : params->history_schema_info) { + if (candidate.__isset.schema_id && candidate.schema_id == params->current_schema_id) { + schema = &candidate; + break; + } + } + } + if (!schema->__isset.root_field || !schema->root_field.__isset.fields) { + return nullptr; + } + for (const auto& field_ptr : schema->root_field.fields) { + const auto* field = get_field_ptr(field_ptr); + if (field != nullptr && field->__isset.name && field->name == name) { + return field; + } + } + return nullptr; +} + +template +std::shared_ptr create_position_delete_root_node( + const ReadColumns& read_columns) { + auto root_node = std::make_shared(); + for (const auto& column : read_columns) { + if (column.name == kRowColumn) { + continue; + } + root_node->add_children(column.name, column.name, + TableSchemaChangeHelper::ConstNode::get_instance()); + } + return root_node; +} + +} // namespace + +IcebergPositionDeleteSysTableReader::IcebergPositionDeleteSysTableReader( + const std::vector& file_slot_descs, RuntimeState* state, + RuntimeProfile* profile, const TFileRangeDesc& range, + const TFileScanRangeParams* range_params, std::shared_ptr io_ctx, + FileMetaCache* meta_cache) + : _file_slot_descs(file_slot_descs), + _state(state), + _profile(profile), + _range(range), + _range_params(range_params), + _io_ctx(std::move(io_ctx)) { + _meta_cache = meta_cache; +} + +IcebergPositionDeleteSysTableReader::~IcebergPositionDeleteSysTableReader() = default; + +Status IcebergPositionDeleteSysTableReader::_do_init_reader(ReaderInitContext* /*ctx*/) { + if (_state == nullptr || _profile == nullptr || _range_params == nullptr || + _io_ctx == nullptr) { + return Status::InvalidArgument( + "invalid Iceberg position delete system table reader context"); + } + if (!_range.__isset.table_format_params || !_range.table_format_params.__isset.iceberg_params) { + return Status::InternalError("Iceberg position delete system table range misses params"); + } + + _iceberg_file_desc = &_range.table_format_params.iceberg_params; + if (!_iceberg_file_desc->__isset.delete_files || _iceberg_file_desc->delete_files.size() != 1) { + return Status::InternalError( + "Iceberg position delete system table range should contain exactly one delete " + "file"); + } + _delete_file_desc = _iceberg_file_desc->delete_files.data(); + if (is_iceberg_deletion_vector(*_delete_file_desc)) { + _delete_file_kind = DeleteFileKind::DELETION_VECTOR; + } else if (_delete_file_desc->__isset.content && + _delete_file_desc->content == kPositionDeleteContent) { + _delete_file_kind = DeleteFileKind::POSITION_DELETE; + } else if (!_delete_file_desc->__isset.content) { + return Status::InternalError( + "Iceberg position delete system table delete file misses content"); + } else { + return Status::InternalError( + "Iceberg position delete system table does not support delete file content {}", + _delete_file_desc->content); + } + _batch_size = _state->batch_size(); + + if (_delete_file_kind == DeleteFileKind::DELETION_VECTOR) { + return _init_deletion_vector_reader(); + } + return _init_position_delete_reader(); +} + +Status IcebergPositionDeleteSysTableReader::_get_columns_impl( + std::unordered_map* name_to_type) { + for (const auto* slot : _file_slot_descs) { + name_to_type->emplace(slot->col_name(), slot->get_data_type_ptr()); + } + return Status::OK(); +} + +bool IcebergPositionDeleteSysTableReader::count_read_rows() { + return _delete_file_kind == DeleteFileKind::POSITION_DELETE; +} + +void IcebergPositionDeleteSysTableReader::_collect_profile_before_close() { + if (_position_reader != nullptr) { + _position_reader->collect_profile_before_close(); + } +} + +Status IcebergPositionDeleteSysTableReader::close() { + if (_position_reader != nullptr) { + RETURN_IF_ERROR(_position_reader->close()); + } + _partition_value.reset(); + _next_dv_position.reset(); + _dv_positions = roaring::Roaring64Map(); + return Status::OK(); +} + +Status IcebergPositionDeleteSysTableReader::_init_position_delete_reader() { + if (!_delete_file_desc->__isset.file_format) { + return Status::InternalError("Iceberg position delete file misses file format"); + } + + // `row` is optional in position delete files and expensive to read, so only read it when the + // query actually projects it. Whether the delete file physically stores `row` is decided from + // the reader's own footer/type below, reusing the same reader instance that is initialized + // afterwards so the delete file is opened and its footer parsed only once. + const bool row_requested = _output_column_requested(kRowColumn); + + if (_delete_file_desc->file_format == TFileFormatType::FORMAT_PARQUET) { + auto parquet_reader = + ParquetReader::create_unique(_profile, *_range_params, _range, _batch_size, + &_state->timezone_obj(), _io_ctx, _state, _meta_cache); + + const FieldDescriptor* schema = nullptr; + int row_index = -1; + if (row_requested) { + RETURN_IF_ERROR(parquet_reader->get_file_metadata_schema(&schema)); + DORIS_CHECK(schema != nullptr); + row_index = schema->get_column_index(kRowColumn); + } + const bool read_row = row_requested && row_index >= 0; + _init_read_columns(read_row); + std::vector read_column_names; + read_column_names.reserve(_read_columns.size()); + for (const auto& column : _read_columns) { + read_column_names.push_back(column.name); + } + + ParquetInitContext pctx; + pctx.column_names = read_column_names; + pctx.col_name_to_block_idx = &_read_col_name_to_block_idx; + pctx.state = _state; + pctx.params = _range_params; + pctx.range = &_range; + pctx.filter_groups = false; + if (read_row) { + const auto* table_row_field = find_current_schema_field(_range_params, kRowColumn); + if (table_row_field == nullptr) { + return Status::InternalError( + "Iceberg position delete system table row schema is missing"); + } + const auto* file_row_field = schema->get_column(static_cast(row_index)); + std::shared_ptr row_node; + RETURN_IF_ERROR(TableSchemaChangeHelper::BuildTableInfoUtil:: + by_parquet_field_id_with_name_mapping( + *table_row_field, *file_row_field, row_node)); + auto root_node = create_position_delete_root_node(_read_columns); + root_node->add_children(kRowColumn, file_row_field->name, row_node); + pctx.table_info_node = std::move(root_node); + } + RETURN_IF_ERROR(static_cast(parquet_reader.get())->init_reader(&pctx)); + _position_reader = std::move(parquet_reader); + return Status::OK(); + } + + if (_delete_file_desc->file_format == TFileFormatType::FORMAT_ORC) { + auto orc_reader = + OrcReader::create_unique(_profile, _state, *_range_params, _range, _batch_size, + _state->timezone(), _io_ctx, _meta_cache); + + const orc::Type* row_type = nullptr; + if (row_requested) { + const orc::Type* root_type = nullptr; + RETURN_IF_ERROR(orc_reader->get_file_type(&root_type)); + DORIS_CHECK(root_type != nullptr); + for (uint64_t i = 0; i < root_type->getSubtypeCount(); ++i) { + if (root_type->getFieldName(i) == kRowColumn) { + row_type = root_type->getSubtype(i); + break; + } + } + } + const bool read_row = row_requested && row_type != nullptr; + _init_read_columns(read_row); + std::vector read_column_names; + read_column_names.reserve(_read_columns.size()); + for (const auto& column : _read_columns) { + read_column_names.push_back(column.name); + } + + OrcInitContext octx; + octx.column_names = read_column_names; + octx.col_name_to_block_idx = &_read_col_name_to_block_idx; + octx.state = _state; + octx.params = _range_params; + octx.range = &_range; + if (read_row) { + const auto* table_row_field = find_current_schema_field(_range_params, kRowColumn); + if (table_row_field == nullptr) { + return Status::InternalError( + "Iceberg position delete system table row schema is missing"); + } + std::shared_ptr row_node; + RETURN_IF_ERROR( + TableSchemaChangeHelper::BuildTableInfoUtil::by_orc_field_id_with_name_mapping( + *table_row_field, row_type, kIcebergOrcAttribute, row_node)); + auto root_node = create_position_delete_root_node(_read_columns); + root_node->add_children(kRowColumn, kRowColumn, row_node); + octx.table_info_node = std::move(root_node); + } + RETURN_IF_ERROR(static_cast(orc_reader.get())->init_reader(&octx)); + _position_reader = std::move(orc_reader); + return Status::OK(); + } + + return Status::NotSupported("Unsupported Iceberg position delete file format {}", + _delete_file_desc->file_format); +} + +Status IcebergPositionDeleteSysTableReader::_init_deletion_vector_reader() { + if (!_delete_file_desc->__isset.referenced_data_file_path || + _delete_file_desc->referenced_data_file_path.empty()) { + return Status::InternalError("Iceberg deletion vector misses referenced data file path"); + } + + IcebergDeleteFileReaderOptions options; + options.state = _state; + options.profile = _profile; + options.scan_params = _range_params; + options.io_ctx = _io_ctx.get(); + options.meta_cache = _meta_cache; + if (_range.__isset.fs_name) { + options.fs_name = &_range.fs_name; + } + + _dv_positions = roaring::Roaring64Map(); + RETURN_IF_ERROR(read_iceberg_deletion_vector(*_delete_file_desc, options, &_dv_positions)); + _next_dv_position.emplace(_dv_positions.begin()); + return Status::OK(); +} + +Status IcebergPositionDeleteSysTableReader::_do_get_next_block(Block* block, size_t* read_rows, + bool* eof) { + DORIS_CHECK(_io_ctx != nullptr); + if (_io_ctx->should_stop) { + *read_rows = 0; + *eof = true; + return Status::OK(); + } + + if (_delete_file_kind == DeleteFileKind::DELETION_VECTOR) { + return _append_deletion_vector_block(block, read_rows, eof); + } + + if (_position_reader == nullptr) { + return Status::InternalError("Iceberg position delete reader is not initialized"); + } + + while (true) { + Block delete_block = _create_delete_block(); + size_t delete_rows = 0; + bool position_reader_eof = false; + RETURN_IF_ERROR(_position_reader->get_next_block(&delete_block, &delete_rows, + &position_reader_eof)); + if (delete_rows > 0) { + RETURN_IF_ERROR( + _append_position_delete_block(block, delete_block, delete_rows, read_rows)); + *eof = position_reader_eof; + return Status::OK(); + } + if (position_reader_eof) { + *read_rows = 0; + *eof = true; + return Status::OK(); + } + } +} + +Status IcebergPositionDeleteSysTableReader::_append_position_delete_block(Block* output_block, + const Block& delete_block, + size_t delete_rows, + size_t* appended_rows) { + auto columns_guard = output_block->mutate_columns_scoped(); + auto& columns = columns_guard.mutable_columns(); + auto name_to_pos = output_block->get_name_to_pos_map(); + + for (size_t row = 0; row < delete_rows; ++row) { + for (const auto* slot : _file_slot_descs) { + auto it = name_to_pos.find(slot->col_name()); + if (it == name_to_pos.end()) { + continue; + } + RETURN_IF_ERROR(_append_sys_column(columns[it->second], *slot, &delete_block, row, 0)); + } + } + // Every output column must be produced by a file slot above; a projected system-table column + // that is not backed by a file slot would be left short and yield a malformed block with + // inconsistent column lengths. + check_output_columns_aligned(columns); + *appended_rows = delete_rows; + return Status::OK(); +} + +Status IcebergPositionDeleteSysTableReader::_append_deletion_vector_block(Block* block, + size_t* read_rows, + bool* eof) { + if (!_next_dv_position.has_value() || *_next_dv_position == _dv_positions.end()) { + *read_rows = 0; + *eof = true; + return Status::OK(); + } + const size_t rows_limit = std::max(_batch_size, 1); + + auto columns_guard = block->mutate_columns_scoped(); + auto& columns = columns_guard.mutable_columns(); + auto name_to_pos = block->get_name_to_pos_map(); + + size_t rows = 0; + while (rows < rows_limit && *_next_dv_position != _dv_positions.end()) { + const uint64_t dv_pos = **_next_dv_position; + for (const auto* slot : _file_slot_descs) { + auto it = name_to_pos.find(slot->col_name()); + if (it == name_to_pos.end()) { + continue; + } + RETURN_IF_ERROR(_append_sys_column(columns[it->second], *slot, nullptr, 0, dv_pos)); + } + ++(*_next_dv_position); + ++rows; + } + check_output_columns_aligned(columns); + *read_rows = rows; + *eof = *_next_dv_position == _dv_positions.end(); + return Status::OK(); +} + +Status IcebergPositionDeleteSysTableReader::_append_sys_column(MutableColumnPtr& column, + const SlotDescriptor& slot, + const Block* delete_block, + size_t source_row, uint64_t dv_pos) { + const std::string& name = slot.col_name(); + if (name == kFilePathColumn) { + if (_delete_file_kind == DeleteFileKind::DELETION_VECTOR) { + parquet_utils::insert_string(column, _delete_file_desc->referenced_data_file_path); + return Status::OK(); + } + const auto* path_column = get_string_column(*delete_block, kFilePathColumn); + if (path_column == nullptr || !block_has_row(*delete_block, source_row)) { + return Status::InternalError("Iceberg position delete file_path column is missing"); + } + parquet_utils::insert_string(column, path_column->get_data_at(source_row).to_string()); + return Status::OK(); + } + + if (name == kPosColumn) { + if (_delete_file_kind == DeleteFileKind::DELETION_VECTOR) { + parquet_utils::insert_int64(column, cast_set(dv_pos)); + return Status::OK(); + } + const auto* pos_column = get_int64_column(*delete_block, kPosColumn); + if (pos_column == nullptr || !block_has_row(*delete_block, source_row)) { + return Status::InternalError("Iceberg position delete pos column is missing"); + } + parquet_utils::insert_int64(column, pos_column->get_element(source_row)); + return Status::OK(); + } + + if (name == kRowColumn) { + if (delete_block != nullptr) { + auto row_pos = delete_block->get_position_by_name(kRowColumn); + if (row_pos >= 0) { + const auto& row_column = *delete_block->get_by_position(row_pos).column; + if (source_row < row_column.size()) { + column->insert_from(row_column, source_row); + return Status::OK(); + } + } + } + parquet_utils::insert_null(column); + return Status::OK(); + } + + if (name == kPartitionColumn) { + return _append_partition_column(column, slot); + } + + if (name == kSpecIdColumn) { + if (_iceberg_file_desc->__isset.partition_spec_id) { + parquet_utils::insert_int32(column, + cast_set(_iceberg_file_desc->partition_spec_id)); + } else { + parquet_utils::insert_null(column); + } + return Status::OK(); + } + + if (name == kDeleteFilePathColumn) { + parquet_utils::insert_string(column, _delete_file_output_path()); + return Status::OK(); + } + + if (name == kContentOffsetColumn) { + const int64_t* value = _delete_file_desc->__isset.content_offset + ? &_delete_file_desc->content_offset + : nullptr; + insert_int64_nullable(column, value); + return Status::OK(); + } + + if (name == kContentSizeInBytesColumn) { + const int64_t* value = _delete_file_desc->__isset.content_size_in_bytes + ? &_delete_file_desc->content_size_in_bytes + : nullptr; + insert_int64_nullable(column, value); + return Status::OK(); + } + + return Status::InternalError("Unknown Iceberg position delete system table column: {}", name); +} + +Status IcebergPositionDeleteSysTableReader::_append_partition_column(MutableColumnPtr& column, + const SlotDescriptor& slot) { + if (!_iceberg_file_desc->__isset.partition_data_json || + _iceberg_file_desc->partition_data_json.empty()) { + parquet_utils::insert_null(column); + return Status::OK(); + } + + if (!_partition_value) { + auto partition_value = slot.get_data_type_ptr()->create_column(); + auto serde = slot.get_data_type_ptr()->get_serde(); + StringRef partition_data(_iceberg_file_desc->partition_data_json.data(), + _iceberg_file_desc->partition_data_json.size()); + DataTypeSerDe::FormatOptions options = DataTypeSerDe::get_default_format_options(); + RETURN_IF_ERROR(serde->from_string(partition_data, *partition_value, options)); + DORIS_CHECK(partition_value->size() == 1); + _partition_value = std::move(partition_value); + } + column->insert_from(*_partition_value, 0); + return Status::OK(); +} + +Block IcebergPositionDeleteSysTableReader::_create_delete_block() const { + Block block; + for (const auto& column : _read_columns) { + block.insert(ColumnWithTypeAndName(column.type->create_column(), column.type, column.name)); + } + return block; +} + +bool IcebergPositionDeleteSysTableReader::_output_column_requested(const std::string& name) const { + return std::any_of(_file_slot_descs.begin(), _file_slot_descs.end(), + [&name](const SlotDescriptor* slot) { return slot->col_name() == name; }); +} + +void IcebergPositionDeleteSysTableReader::_init_read_columns(bool read_row) { + _read_columns.clear(); + _read_columns.push_back({kFilePathColumn, std::make_shared()}); + _read_columns.push_back({kPosColumn, std::make_shared()}); + if (read_row) { + for (const auto* slot : _file_slot_descs) { + if (slot->col_name() == kRowColumn) { + _read_columns.push_back({kRowColumn, slot->get_data_type_ptr()}); + break; + } + } + } + + _read_col_name_to_block_idx.clear(); + for (size_t i = 0; i < _read_columns.size(); ++i) { + _read_col_name_to_block_idx.emplace(_read_columns[i].name, cast_set(i)); + } +} + +const std::string& IcebergPositionDeleteSysTableReader::_delete_file_output_path() const { + if (_delete_file_desc->__isset.original_path && !_delete_file_desc->original_path.empty()) { + return _delete_file_desc->original_path; + } + return _delete_file_desc->path; +} + +} // namespace doris diff --git a/be/src/format/table/iceberg_position_delete_sys_table_reader.h b/be/src/format/table/iceberg_position_delete_sys_table_reader.h new file mode 100644 index 00000000000000..ce82b3d40cd847 --- /dev/null +++ b/be/src/format/table/iceberg_position_delete_sys_table_reader.h @@ -0,0 +1,110 @@ +// 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. + +#pragma once + +#include + +#include +#include +#include +#include +#include +#include +#include + +#include "common/factory_creator.h" +#include "common/status.h" +#include "core/block/column_with_type_and_name.h" +#include "format/generic_reader.h" +#include "format/table/iceberg_delete_file_reader_helper.h" +#include "runtime/descriptors.h" + +namespace doris { + +class FileMetaCache; +class RuntimeProfile; +class RuntimeState; + +class IcebergPositionDeleteSysTableReader : public GenericReader { + ENABLE_FACTORY_CREATOR(IcebergPositionDeleteSysTableReader); + +public: + IcebergPositionDeleteSysTableReader(const std::vector& file_slot_descs, + RuntimeState* state, RuntimeProfile* profile, + const TFileRangeDesc& range, + const TFileScanRangeParams* range_params, + std::shared_ptr io_ctx, + FileMetaCache* meta_cache); + ~IcebergPositionDeleteSysTableReader() override; + + Status _get_columns_impl(std::unordered_map* name_to_type) override; + void set_batch_size(size_t batch_size) override { _batch_size = batch_size; } + size_t get_batch_size() const override { return _batch_size; } + bool count_read_rows() override; + Status close() override; + +protected: + Status _do_init_reader(ReaderInitContext* ctx) override; + Status _do_get_next_block(Block* block, size_t* read_rows, bool* eof) override; + void _collect_profile_before_close() override; + +private: + enum class DeleteFileKind { + POSITION_DELETE, + DELETION_VECTOR, + }; + + struct ReadColumn { + std::string name; + DataTypePtr type; + }; + + Status _init_position_delete_reader(); + Status _init_deletion_vector_reader(); + Status _append_position_delete_block(Block* output_block, const Block& delete_block, + size_t delete_rows, size_t* appended_rows); + Status _append_deletion_vector_block(Block* block, size_t* read_rows, bool* eof); + Status _append_sys_column(MutableColumnPtr& column, const SlotDescriptor& slot, + const Block* delete_block, size_t source_row, uint64_t dv_pos); + Status _append_partition_column(MutableColumnPtr& column, const SlotDescriptor& slot); + Block _create_delete_block() const; + bool _output_column_requested(const std::string& name) const; + void _init_read_columns(bool read_row); + const std::string& _delete_file_output_path() const; + + const std::vector& _file_slot_descs; + RuntimeState* _state = nullptr; + RuntimeProfile* _profile = nullptr; + const TFileRangeDesc& _range; + const TFileScanRangeParams* _range_params = nullptr; + + std::shared_ptr _io_ctx; + const TIcebergFileDesc* _iceberg_file_desc = nullptr; + const TIcebergDeleteFileDesc* _delete_file_desc = nullptr; + DeleteFileKind _delete_file_kind = DeleteFileKind::POSITION_DELETE; + + size_t _batch_size = 102400; + std::unique_ptr _position_reader; + std::vector _read_columns; + std::unordered_map _read_col_name_to_block_idx; + ColumnPtr _partition_value; + roaring::Roaring64Map _dv_positions; + std::optional _next_dv_position; +}; + +} // namespace doris diff --git a/be/src/format_v2/table/iceberg_position_delete_sys_table_reader.cpp b/be/src/format_v2/table/iceberg_position_delete_sys_table_reader.cpp new file mode 100644 index 00000000000000..ef13f585fb6a6d --- /dev/null +++ b/be/src/format_v2/table/iceberg_position_delete_sys_table_reader.cpp @@ -0,0 +1,596 @@ +// 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_v2/table/iceberg_position_delete_sys_table_reader.h" + +#include +#include +#include +#include + +#include "common/cast_set.h" +#include "core/block/block.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_vector.h" +#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_serde/data_type_serde.h" +#include "core/types.h" +#include "format/table/iceberg_delete_file_reader_helper.h" +#include "format/table/parquet_utils.h" +#include "runtime/descriptors.h" +#include "runtime/runtime_state.h" + +namespace doris::format::iceberg { + +namespace { + +constexpr const char* kFilePathColumn = "file_path"; +constexpr const char* kPosColumn = "pos"; +constexpr const char* kRowColumn = "row"; +constexpr const char* kPartitionColumn = "partition"; +constexpr const char* kSpecIdColumn = "spec_id"; +constexpr const char* kDeleteFilePathColumn = "delete_file_path"; +constexpr const char* kContentOffsetColumn = "content_offset"; +constexpr const char* kContentSizeInBytesColumn = "content_size_in_bytes"; +constexpr int kPositionDeleteContent = 1; +constexpr int32_t kDeleteFilePathFieldId = 2147483546; +constexpr int32_t kDeleteFilePosFieldId = 2147483545; +constexpr int32_t kDeleteFileRowFieldId = 2147483544; + +bool block_has_row(const Block& block, size_t row) { + return block.columns() > 0 && row < block.rows(); +} + +void insert_int64_nullable(MutableColumnPtr& column, const int64_t* value) { + if (value == nullptr) { + parquet_utils::insert_null(column); + } else { + parquet_utils::insert_int64(column, *value); + } +} + +// Fail loudly if the filled output block is malformed. Each output column must be produced by a +// file slot; a projected system-table column that is not backed by a file slot would be skipped +// during fill and left shorter than the rest, producing a block with inconsistent column lengths. +void check_output_columns_aligned(const MutableColumns& columns) { + if (columns.empty()) { + return; + } + const size_t expected_rows = columns.front()->size(); + for (const auto& column : columns) { + DORIS_CHECK(column->size() == expected_rows) + << "Iceberg position delete system table output block has inconsistent column " + "sizes; a projected column is not backed by a file slot"; + } +} + +const IColumn* get_column(const Block& block, const std::string& name) { + auto pos = block.get_position_by_name(name); + if (pos < 0) { + return nullptr; + } + return block.get_by_position(pos).column.get(); +} + +const IColumn* nested_column(const IColumn* column) { + if (column == nullptr) { + return nullptr; + } + if (const auto* nullable = check_and_get_column(column)) { + return nullable->get_nested_column_ptr().get(); + } + return column; +} + +bool column_is_null_at(const IColumn* column, size_t row) { + const auto* nullable = check_and_get_column(column); + return nullable != nullptr && nullable->is_null_at(row); +} + +const ColumnString* get_string_column(const IColumn* column) { + return check_and_get_column(nested_column(column)); +} + +const ColumnInt64* get_int64_column(const IColumn* column) { + return check_and_get_column(nested_column(column)); +} + +ColumnDefinition build_delete_file_column(const std::string& name, DataTypePtr type) { + ColumnDefinition column; + column.identifier = Field::create_field(name); + column.name = name; + column.type = std::move(type); + return column; +} + +void set_iceberg_delete_field_id(ColumnDefinition* column) { + DORIS_CHECK(column != nullptr); + if (column->name == kFilePathColumn) { + column->identifier = Field::create_field(kDeleteFilePathFieldId); + } else if (column->name == kPosColumn) { + column->identifier = Field::create_field(kDeleteFilePosFieldId); + } else if (column->name == kRowColumn && !column->has_identifier_field_id()) { + column->identifier = Field::create_field(kDeleteFileRowFieldId); + } +} + +bool has_field_id(const std::vector& schema) { + for (const auto& field : schema) { + if (!field.has_identifier_field_id()) { + return false; + } + if (!has_field_id(field.children)) { + return false; + } + } + return true; +} + +class PositionDeleteFileTableReader final : public format::TableReader { +protected: + format::TableColumnMappingMode mapping_mode() const override { + return !_data_reader.file_schema.empty() && has_field_id(_data_reader.file_schema) + ? format::TableColumnMappingMode::BY_FIELD_ID + : format::TableColumnMappingMode::BY_NAME; + } +}; + +} // namespace + +IcebergPositionDeleteSysTableV2Reader::~IcebergPositionDeleteSysTableV2Reader() = default; + +Status IcebergPositionDeleteSysTableV2Reader::prepare_split( + const format::SplitReadOptions& options) { + RETURN_IF_ERROR(close()); + RETURN_IF_ERROR(format::TableReader::prepare_split(options)); + _current_range = options.current_range; + _has_split = true; + return _init_split(); +} + +Status IcebergPositionDeleteSysTableV2Reader::get_block(Block* block, bool* eos) { + SCOPED_TIMER(_profile.exec_timer); + DORIS_CHECK(block != nullptr); + DORIS_CHECK(eos != nullptr); + DORIS_CHECK(block->columns() == _projected_columns.size()); + block->clear_column_data(_projected_columns.size()); + + if (*eos) { + return Status::OK(); + } + if (_io_ctx != nullptr && _io_ctx->should_stop) { + *eos = true; + return Status::OK(); + } + if (!_has_split) { + *eos = true; + return Status::OK(); + } + + size_t read_rows = 0; + if (_delete_file_kind == DeleteFileKind::DELETION_VECTOR) { + return _append_deletion_vector_block(block, &read_rows, eos); + } + + DORIS_CHECK(_position_reader != nullptr); + if (_batch_size > 0) { + _position_reader->set_batch_size(_batch_size); + } + + while (true) { + Block delete_block = _create_delete_block(); + bool position_reader_eof = false; + RETURN_IF_ERROR(_position_reader->get_block(&delete_block, &position_reader_eof)); + const size_t delete_rows = delete_block.rows(); + if (delete_rows > 0) { + RETURN_IF_ERROR( + _append_position_delete_block(block, delete_block, delete_rows, &read_rows)); + *eos = false; + return Status::OK(); + } + if (position_reader_eof) { + RETURN_IF_ERROR(close()); + *eos = true; + return Status::OK(); + } + } +} + +Status IcebergPositionDeleteSysTableV2Reader::close() { + Status close_status = Status::OK(); + if (_position_reader != nullptr) { + close_status = _position_reader->close(); + _position_reader.reset(); + } + auto base_status = format::TableReader::close(); + if (!base_status.ok() && close_status.ok()) { + close_status = std::move(base_status); + } + _iceberg_file_desc = nullptr; + _delete_file_desc = nullptr; + _read_columns.clear(); + _partition_value.reset(); + _next_dv_position.reset(); + _dv_positions = roaring::Roaring64Map(); + _has_split = false; + return close_status; +} + +std::string IcebergPositionDeleteSysTableV2Reader::debug_string() const { + std::ostringstream out; + out << "IcebergPositionDeleteSysTableV2Reader{base=" << format::TableReader::debug_string() + << ", has_split=" << _has_split << ", delete_file_kind=" + << (_delete_file_kind == DeleteFileKind::DELETION_VECTOR ? "DELETION_VECTOR" + : "POSITION_DELETE") + << ", read_column_count=" << _read_columns.size() + << ", dv_position_count=" << _dv_positions.cardinality() << "}"; + return out.str(); +} + +Status IcebergPositionDeleteSysTableV2Reader::_init_split() { + if (_runtime_state == nullptr || _scanner_profile == nullptr || _scan_params == nullptr) { + return Status::InvalidArgument( + "invalid Iceberg position delete system table v2 reader context"); + } + if (_file_slot_descs == nullptr) { + return Status::InvalidArgument( + "Iceberg position delete system table v2 reader requires file slot descriptors"); + } + if (!_current_range.__isset.table_format_params || + !_current_range.table_format_params.__isset.iceberg_params) { + return Status::InternalError("Iceberg position delete system table range misses params"); + } + + _iceberg_file_desc = &_current_range.table_format_params.iceberg_params; + if (!_iceberg_file_desc->__isset.delete_files || _iceberg_file_desc->delete_files.size() != 1) { + return Status::InternalError( + "Iceberg position delete system table range should contain exactly one delete " + "file"); + } + _delete_file_desc = _iceberg_file_desc->delete_files.data(); + if (is_iceberg_deletion_vector(*_delete_file_desc)) { + _delete_file_kind = DeleteFileKind::DELETION_VECTOR; + } else if (_delete_file_desc->__isset.content && + _delete_file_desc->content == kPositionDeleteContent) { + _delete_file_kind = DeleteFileKind::POSITION_DELETE; + } else if (!_delete_file_desc->__isset.content) { + return Status::InternalError( + "Iceberg position delete system table delete file misses content"); + } else { + return Status::InternalError( + "Iceberg position delete system table does not support delete file content {}", + _delete_file_desc->content); + } + + if (_delete_file_kind == DeleteFileKind::DELETION_VECTOR) { + return _init_deletion_vector_reader(); + } + return _init_position_delete_reader(); +} + +Status IcebergPositionDeleteSysTableV2Reader::_init_position_delete_reader() { + if (!_delete_file_desc->__isset.file_format) { + return Status::InternalError("Iceberg position delete file misses file format"); + } + format::FileFormat file_format; + if (_delete_file_desc->file_format == TFileFormatType::FORMAT_PARQUET) { + file_format = format::FileFormat::PARQUET; + } else if (_delete_file_desc->file_format == TFileFormatType::FORMAT_ORC) { + file_format = format::FileFormat::ORC; + } else { + return Status::NotSupported( + "Iceberg position delete system table v2 reader only supports Parquet and ORC " + "delete files, file_format={}", + _delete_file_desc->file_format); + } + + const bool read_row = _output_column_requested(kRowColumn); + _init_read_columns(read_row); + std::vector projected_columns; + RETURN_IF_ERROR(_build_delete_file_projected_columns(&projected_columns)); + + _position_reader = std::make_unique(); + RETURN_IF_ERROR(_position_reader->init({ + .projected_columns = std::move(projected_columns), + .conjuncts = {}, + .format = file_format, + .scan_params = _scan_params, + .io_ctx = _io_ctx, + .runtime_state = _runtime_state, + .scanner_profile = _scanner_profile, + .file_slot_descs = nullptr, + .push_down_agg_type = TPushAggOp::type::NONE, + .condition_cache_digest = 0, + })); + // Keep standalone-reader defaults for scanner-only fields that may be added to + // SplitReadOptions. + auto split_options = format::SplitReadOptions {}; + split_options.partition_values = {}; + split_options.partition_prune_conjuncts = {}; + split_options.cache = nullptr; + split_options.current_range = _current_range; + split_options.current_split_format = file_format; + split_options.global_rowid_context = std::nullopt; + RETURN_IF_ERROR(_position_reader->prepare_split(split_options)); + return Status::OK(); +} + +Status IcebergPositionDeleteSysTableV2Reader::_init_deletion_vector_reader() { + if (!_delete_file_desc->__isset.referenced_data_file_path || + _delete_file_desc->referenced_data_file_path.empty()) { + return Status::InternalError("Iceberg deletion vector misses referenced data file path"); + } + if (_io_ctx == nullptr) { + return Status::InvalidArgument( + "Iceberg position delete system table v2 reader requires IO context"); + } + + IcebergDeleteFileReaderOptions options; + options.state = _runtime_state; + options.profile = _scanner_profile; + options.scan_params = _scan_params; + options.io_ctx = _io_ctx.get(); + if (_current_range.__isset.fs_name) { + options.fs_name = &_current_range.fs_name; + } + + _dv_positions = roaring::Roaring64Map(); + RETURN_IF_ERROR(read_iceberg_deletion_vector(*_delete_file_desc, options, &_dv_positions)); + _next_dv_position.emplace(_dv_positions.begin()); + return Status::OK(); +} + +Status IcebergPositionDeleteSysTableV2Reader::_append_position_delete_block( + Block* output_block, const Block& delete_block, size_t delete_rows, size_t* appended_rows) { + auto columns_guard = output_block->mutate_columns_scoped(); + auto& columns = columns_guard.mutable_columns(); + auto name_to_pos = output_block->get_name_to_pos_map(); + + for (size_t row = 0; row < delete_rows; ++row) { + for (const auto* slot : *_file_slot_descs) { + auto it = name_to_pos.find(slot->col_name()); + if (it == name_to_pos.end()) { + continue; + } + RETURN_IF_ERROR(_append_sys_column(columns[it->second], *slot, &delete_block, row, 0)); + } + } + check_output_columns_aligned(columns); + *appended_rows = delete_rows; + return Status::OK(); +} + +Status IcebergPositionDeleteSysTableV2Reader::_append_deletion_vector_block(Block* block, + size_t* read_rows, + bool* eof) { + const size_t batch_size = std::max( + _batch_size > 0 ? _batch_size + : (_runtime_state == nullptr + ? static_cast(102400) + : static_cast(_runtime_state->batch_size())), + 1); + if (!_next_dv_position.has_value() || *_next_dv_position == _dv_positions.end()) { + *read_rows = 0; + *eof = true; + RETURN_IF_ERROR(close()); + return Status::OK(); + } + + auto columns_guard = block->mutate_columns_scoped(); + auto& columns = columns_guard.mutable_columns(); + auto name_to_pos = block->get_name_to_pos_map(); + + size_t rows = 0; + while (rows < batch_size && *_next_dv_position != _dv_positions.end()) { + const uint64_t dv_pos = **_next_dv_position; + for (const auto* slot : *_file_slot_descs) { + auto it = name_to_pos.find(slot->col_name()); + if (it == name_to_pos.end()) { + continue; + } + RETURN_IF_ERROR(_append_sys_column(columns[it->second], *slot, nullptr, 0, dv_pos)); + } + ++(*_next_dv_position); + ++rows; + } + check_output_columns_aligned(columns); + *read_rows = rows; + _record_scan_rows(rows); + // FileScannerV2 treats eof=true as "advance to the next split" without returning the + // current block. Keep eof false after appending rows and report EOF on the next empty call. + *eof = false; + return Status::OK(); +} + +Status IcebergPositionDeleteSysTableV2Reader::_append_sys_column(MutableColumnPtr& column, + const SlotDescriptor& slot, + const Block* delete_block, + size_t source_row, + uint64_t dv_pos) { + const std::string& name = slot.col_name(); + if (name == kFilePathColumn) { + if (_delete_file_kind == DeleteFileKind::DELETION_VECTOR) { + parquet_utils::insert_string(column, _delete_file_desc->referenced_data_file_path); + return Status::OK(); + } + const auto* source_column = get_column(*delete_block, kFilePathColumn); + const auto* path_column = get_string_column(source_column); + if (path_column == nullptr || !block_has_row(*delete_block, source_row) || + column_is_null_at(source_column, source_row)) { + return Status::InternalError("Iceberg position delete file_path column is missing"); + } + parquet_utils::insert_string(column, path_column->get_data_at(source_row).to_string()); + return Status::OK(); + } + + if (name == kPosColumn) { + if (_delete_file_kind == DeleteFileKind::DELETION_VECTOR) { + parquet_utils::insert_int64(column, cast_set(dv_pos)); + return Status::OK(); + } + const auto* source_column = get_column(*delete_block, kPosColumn); + const auto* pos_column = get_int64_column(source_column); + if (pos_column == nullptr || !block_has_row(*delete_block, source_row) || + column_is_null_at(source_column, source_row)) { + return Status::InternalError("Iceberg position delete pos column is missing"); + } + parquet_utils::insert_int64(column, pos_column->get_element(source_row)); + return Status::OK(); + } + + if (name == kRowColumn) { + if (delete_block != nullptr) { + auto row_pos = delete_block->get_position_by_name(kRowColumn); + if (row_pos >= 0) { + auto row_column = delete_block->get_by_position(row_pos) + .column->convert_to_full_column_if_const(); + if (source_row < row_column->size()) { + column->insert_from(*row_column, source_row); + return Status::OK(); + } + } + } + parquet_utils::insert_null(column); + return Status::OK(); + } + + if (name == kPartitionColumn) { + return _append_partition_column(column, slot); + } + + if (name == kSpecIdColumn) { + if (_iceberg_file_desc->__isset.partition_spec_id) { + parquet_utils::insert_int32(column, + cast_set(_iceberg_file_desc->partition_spec_id)); + } else { + parquet_utils::insert_null(column); + } + return Status::OK(); + } + + if (name == kDeleteFilePathColumn) { + parquet_utils::insert_string(column, _delete_file_output_path()); + return Status::OK(); + } + + if (name == kContentOffsetColumn) { + const int64_t* value = _delete_file_desc->__isset.content_offset + ? &_delete_file_desc->content_offset + : nullptr; + insert_int64_nullable(column, value); + return Status::OK(); + } + + if (name == kContentSizeInBytesColumn) { + const int64_t* value = _delete_file_desc->__isset.content_size_in_bytes + ? &_delete_file_desc->content_size_in_bytes + : nullptr; + insert_int64_nullable(column, value); + return Status::OK(); + } + + return Status::InternalError("Unknown Iceberg position delete system table column: {}", name); +} + +Status IcebergPositionDeleteSysTableV2Reader::_append_partition_column(MutableColumnPtr& column, + const SlotDescriptor& slot) { + if (!_iceberg_file_desc->__isset.partition_data_json || + _iceberg_file_desc->partition_data_json.empty()) { + parquet_utils::insert_null(column); + return Status::OK(); + } + + if (!_partition_value) { + auto partition_value = slot.get_data_type_ptr()->create_column(); + auto serde = slot.get_data_type_ptr()->get_serde(); + StringRef partition_data(_iceberg_file_desc->partition_data_json.data(), + _iceberg_file_desc->partition_data_json.size()); + DataTypeSerDe::FormatOptions options = DataTypeSerDe::get_default_format_options(); + RETURN_IF_ERROR(serde->from_string(partition_data, *partition_value, options)); + DORIS_CHECK(partition_value->size() == 1); + _partition_value = std::move(partition_value); + } + column->insert_from(*_partition_value, 0); + return Status::OK(); +} + +Block IcebergPositionDeleteSysTableV2Reader::_create_delete_block() const { + Block block; + for (const auto& column : _read_columns) { + block.insert(ColumnWithTypeAndName(column.type->create_column(), column.type, column.name)); + } + return block; +} + +bool IcebergPositionDeleteSysTableV2Reader::_output_column_requested( + const std::string& name) const { + return std::any_of(_file_slot_descs->begin(), _file_slot_descs->end(), + [&name](const SlotDescriptor* slot) { return slot->col_name() == name; }); +} + +void IcebergPositionDeleteSysTableV2Reader::_init_read_columns(bool read_row) { + _read_columns.clear(); + _read_columns.push_back({kFilePathColumn, make_nullable(std::make_shared())}); + _read_columns.push_back({kPosColumn, make_nullable(std::make_shared())}); + if (read_row) { + for (const auto* slot : *_file_slot_descs) { + if (slot->col_name() == kRowColumn) { + _read_columns.push_back({kRowColumn, slot->get_data_type_ptr()}); + break; + } + } + } +} + +Status IcebergPositionDeleteSysTableV2Reader::_build_delete_file_projected_columns( + std::vector* columns) const { + DORIS_CHECK(columns != nullptr); + columns->clear(); + columns->reserve(_read_columns.size()); + for (const auto& column : _read_columns) { + if (column.name == kRowColumn) { + auto it = std::ranges::find_if(_projected_columns, [](const ColumnDefinition& field) { + return field.name == kRowColumn; + }); + if (it == _projected_columns.end()) { + return Status::InternalError( + "Iceberg position delete system table row schema is missing"); + } + columns->push_back(*it); + columns->back().type = column.type; + set_iceberg_delete_field_id(&columns->back()); + continue; + } + auto field = build_delete_file_column(column.name, column.type); + set_iceberg_delete_field_id(&field); + columns->push_back(std::move(field)); + } + return Status::OK(); +} + +const std::string& IcebergPositionDeleteSysTableV2Reader::_delete_file_output_path() const { + if (_delete_file_desc->__isset.original_path && !_delete_file_desc->original_path.empty()) { + return _delete_file_desc->original_path; + } + return _delete_file_desc->path; +} + +} // namespace doris::format::iceberg diff --git a/be/src/format_v2/table/iceberg_position_delete_sys_table_reader.h b/be/src/format_v2/table/iceberg_position_delete_sys_table_reader.h new file mode 100644 index 00000000000000..9c802e726053cb --- /dev/null +++ b/be/src/format_v2/table/iceberg_position_delete_sys_table_reader.h @@ -0,0 +1,86 @@ +// 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. + +#pragma once + +#include +#include +#include +#include +#include +#include + +#include "common/status.h" +#include "core/data_type/data_type.h" +#include "format_v2/table_reader.h" +#include "gen_cpp/PlanNodes_types.h" +#include "roaring/roaring64map.hh" + +namespace doris { +class SlotDescriptor; +} // namespace doris + +namespace doris::format::iceberg { + +class IcebergPositionDeleteSysTableV2Reader final : public format::TableReader { +public: + ~IcebergPositionDeleteSysTableV2Reader() override; + + Status prepare_split(const format::SplitReadOptions& options) override; + Status get_block(Block* block, bool* eos) override; + Status close() override; + std::string debug_string() const override; + +private: + enum class DeleteFileKind { + POSITION_DELETE, + DELETION_VECTOR, + }; + + struct ReadColumn { + std::string name; + DataTypePtr type; + }; + + Status _init_split(); + Status _init_position_delete_reader(); + Status _init_deletion_vector_reader(); + Status _append_position_delete_block(Block* output_block, const Block& delete_block, + size_t delete_rows, size_t* appended_rows); + Status _append_deletion_vector_block(Block* block, size_t* read_rows, bool* eof); + Status _append_sys_column(MutableColumnPtr& column, const SlotDescriptor& slot, + const Block* delete_block, size_t source_row, uint64_t dv_pos); + Status _append_partition_column(MutableColumnPtr& column, const SlotDescriptor& slot); + Block _create_delete_block() const; + bool _output_column_requested(const std::string& name) const; + void _init_read_columns(bool read_row); + Status _build_delete_file_projected_columns(std::vector* columns) const; + const std::string& _delete_file_output_path() const; + + TFileRangeDesc _current_range; + const TIcebergFileDesc* _iceberg_file_desc = nullptr; + const TIcebergDeleteFileDesc* _delete_file_desc = nullptr; + DeleteFileKind _delete_file_kind = DeleteFileKind::POSITION_DELETE; + std::unique_ptr _position_reader; + std::vector _read_columns; + ColumnPtr _partition_value; + roaring::Roaring64Map _dv_positions; + std::optional _next_dv_position; + bool _has_split = false; +}; + +} // namespace doris::format::iceberg diff --git a/be/test/exec/scan/file_scanner_v2_test.cpp b/be/test/exec/scan/file_scanner_v2_test.cpp index 9df8dcb54f4934..dbc101150fde64 100644 --- a/be/test/exec/scan/file_scanner_v2_test.cpp +++ b/be/test/exec/scan/file_scanner_v2_test.cpp @@ -40,6 +40,9 @@ namespace doris { namespace { +constexpr int kIcebergPositionDeleteContent = 1; +constexpr int kIcebergDeletionVectorContent = 3; + TFileRangeDesc range_with_format(std::string table_format, TFileFormatType::type format_type) { TFileRangeDesc range; range.__set_format_type(format_type); @@ -51,6 +54,14 @@ TFileRangeDesc range_with_format(std::string table_format, TFileFormatType::type return range; } +TFileRangeDesc iceberg_position_deletes_range(TFileFormatType::type format_type, int content) { + auto range = range_with_format("iceberg", format_type); + TIcebergFileDesc iceberg_params; + iceberg_params.__set_content(content); + range.table_format_params.__set_iceberg_params(std::move(iceberg_params)); + return range; +} + TFileRangeDesc hudi_range_with_delta_logs() { auto range = range_with_format("hudi", TFileFormatType::FORMAT_PARQUET); THudiFileDesc hudi_params; @@ -150,6 +161,27 @@ TEST(FileScannerV2Test, SupportedFormatMatrix) { EXPECT_FALSE(FileScannerV2::is_supported(params, hudi_range_with_delta_logs())); } +// Scenario: Iceberg position-delete system table splits use FileScannerV2 for both native delete +// formats and V3 deletion vectors. Avro remains unsupported and is rejected by FE before routing. +TEST(FileScannerV2Test, IcebergPositionDeletesSupportNativeFormats) { + TFileScanRangeParams params; + params.__set_format_type(TFileFormatType::FORMAT_PARQUET); + + const auto parquet_position_delete = iceberg_position_deletes_range( + TFileFormatType::FORMAT_PARQUET, kIcebergPositionDeleteContent); + const auto parquet_deletion_vector = iceberg_position_deletes_range( + TFileFormatType::FORMAT_PARQUET, kIcebergDeletionVectorContent); + const auto orc_position_delete = iceberg_position_deletes_range(TFileFormatType::FORMAT_ORC, + kIcebergPositionDeleteContent); + const auto avro_position_delete = iceberg_position_deletes_range(TFileFormatType::FORMAT_AVRO, + kIcebergPositionDeleteContent); + + EXPECT_TRUE(FileScannerV2::is_supported(params, parquet_position_delete)); + EXPECT_TRUE(FileScannerV2::is_supported(params, parquet_deletion_vector)); + EXPECT_TRUE(FileScannerV2::is_supported(params, orc_position_delete)); + EXPECT_FALSE(FileScannerV2::is_supported(params, avro_position_delete)); +} + TEST(FileScannerV2Test, FileScanLocalStateSelectsV2ForSupportedQueriesOnly) { TQueryOptions query_options; TFileScanRangeParams params; diff --git a/be/test/format/table/iceberg/iceberg_position_delete_sys_table_reader_test.cpp b/be/test/format/table/iceberg/iceberg_position_delete_sys_table_reader_test.cpp new file mode 100644 index 00000000000000..bc3f916281de7d --- /dev/null +++ b/be/test/format/table/iceberg/iceberg_position_delete_sys_table_reader_test.cpp @@ -0,0 +1,606 @@ +// 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/iceberg_position_delete_sys_table_reader.h" + +#include + +#include +#include +#include +#include +#include + +#include "common/object_pool.h" +#include "core/column/column_nullable.h" +#include "core/column/column_string.h" +#include "core/column/column_struct.h" +#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 "format/table/parquet_utils.h" +#include "format_v2/table/iceberg_position_delete_sys_table_reader.h" +#include "io/io_common.h" +#include "runtime/runtime_profile.h" +#include "runtime/runtime_state.h" + +namespace doris { + +namespace { + +class ProfileTrackingReader final : public GenericReader { +public: + int collect_calls = 0; + +protected: + Status _do_get_next_block(Block* /*block*/, size_t* /*read_rows*/, bool* /*eof*/) override { + return Status::OK(); + } + + void _collect_profile_before_close() override { ++collect_calls; } +}; + +SlotDescriptor* make_slot(ObjectPool* pool, int id, std::string name, DataTypePtr type) { + TSlotDescriptor slot_desc; + slot_desc.__set_id(id); + slot_desc.__set_parent(0); + slot_desc.__set_slotType(type->to_thrift()); + slot_desc.__set_columnPos(id); + slot_desc.__set_byteOffset(0); + slot_desc.__set_nullIndicatorByte(id / 8); + slot_desc.__set_nullIndicatorBit(id % 8); + slot_desc.__set_slotIdx(id); + slot_desc.__set_isMaterialized(true); + slot_desc.__set_colName(std::move(name)); + return pool->add(new SlotDescriptor(slot_desc)); +} + +Block make_output_block(const std::vector& slots) { + Block block; + for (const auto* slot : slots) { + auto type = slot->get_data_type_ptr(); + block.insert(ColumnWithTypeAndName(type->create_column(), type, slot->col_name())); + } + return block; +} + +const IColumn& nested_column(const Block& block, const std::string& name) { + const auto position = block.get_position_by_name(name); + DORIS_CHECK(position >= 0); + const auto& column = *block.get_by_position(position).column; + if (const auto* nullable = check_and_get_column(&column)) { + return nullable->get_nested_column(); + } + return column; +} + +bool is_null_at(const Block& block, const std::string& name, size_t row) { + const auto position = block.get_position_by_name(name); + DORIS_CHECK(position >= 0); + const auto* nullable = + check_and_get_column(block.get_by_position(position).column.get()); + DORIS_CHECK(nullable != nullptr); + return nullable->is_null_at(row); +} + +std::string string_at(const Block& block, const std::string& name, size_t row) { + const auto* column = check_and_get_column(&nested_column(block, name)); + DORIS_CHECK(column != nullptr); + return column->get_data_at(row).to_string(); +} + +Int64 int_at(const Block& block, const std::string& name, size_t row) { + return nested_column(block, name).get_int(row); +} + +Int64 struct_int_at(const Block& block, const std::string& name, size_t child_index, size_t row) { + const auto& struct_column = assert_cast(nested_column(block, name)); + const auto& child = assert_cast(struct_column.get_column(child_index)); + DORIS_CHECK(!child.is_null_at(row)); + return child.get_nested_column().get_int(row); +} + +TFileRangeDesc range_with_delete_file(const TIcebergDeleteFileDesc& delete_file) { + TIcebergFileDesc iceberg_desc; + iceberg_desc.__set_delete_files({delete_file}); + TTableFormatFileDesc table_format_desc; + table_format_desc.__set_iceberg_params(std::move(iceberg_desc)); + TFileRangeDesc range; + range.__set_table_format_params(std::move(table_format_desc)); + return range; +} + +} // namespace + +TEST(IcebergPositionDeleteSysTableReaderTest, UsesScannerIOContext) { + RuntimeState state {TQueryOptions(), TQueryGlobals()}; + RuntimeProfile profile("test_profile"); + TFileRangeDesc range; + TFileScanRangeParams params; + std::vector file_slot_descs; + auto scanner_io_ctx = std::make_shared(); + io::FileReaderStats file_reader_stats; + scanner_io_ctx->file_reader_stats = &file_reader_stats; + + IcebergPositionDeleteSysTableReader reader(file_slot_descs, &state, &profile, range, ¶ms, + scanner_io_ctx, nullptr); + + EXPECT_EQ(scanner_io_ctx.get(), reader._io_ctx.get()); + EXPECT_EQ(&file_reader_stats, reader._io_ctx->file_reader_stats); + scanner_io_ctx->should_stop = true; + EXPECT_TRUE(reader._io_ctx->should_stop); + + EXPECT_TRUE(reader.count_read_rows()); + reader._delete_file_kind = IcebergPositionDeleteSysTableReader::DeleteFileKind::DELETION_VECTOR; + EXPECT_FALSE(reader.count_read_rows()); +} + +TEST(IcebergPositionDeleteSysTableReaderTest, ForwardsProfileCollectionToNestedReader) { + RuntimeState state {TQueryOptions(), TQueryGlobals()}; + RuntimeProfile profile("test_profile"); + TFileRangeDesc range; + TFileScanRangeParams params; + std::vector file_slot_descs; + auto scanner_io_ctx = std::make_shared(); + + IcebergPositionDeleteSysTableReader reader(file_slot_descs, &state, &profile, range, ¶ms, + scanner_io_ctx, nullptr); + auto nested_reader = std::make_unique(); + auto* nested_reader_ptr = nested_reader.get(); + reader._position_reader = std::move(nested_reader); + + reader.collect_profile_before_close(); + reader.collect_profile_before_close(); + + EXPECT_EQ(1, nested_reader_ptr->collect_calls); + reader._dv_positions.add(uint64_t {1}); + reader._next_dv_position.emplace(reader._dv_positions.begin()); + reader._partition_value = std::make_shared()->create_column(); + ASSERT_TRUE(reader.close().ok()); + EXPECT_TRUE(reader._dv_positions.isEmpty()); + EXPECT_FALSE(reader._next_dv_position.has_value()); + EXPECT_EQ(nullptr, reader._partition_value.get()); +} + +TEST(IcebergPositionDeleteSysTableReaderTest, StopsBeforeExpandingDeletionVector) { + RuntimeState state {TQueryOptions(), TQueryGlobals()}; + RuntimeProfile profile("test_profile"); + TFileRangeDesc range; + TFileScanRangeParams params; + std::vector file_slot_descs; + auto scanner_io_ctx = std::make_shared(); + scanner_io_ctx->should_stop = true; + + IcebergPositionDeleteSysTableReader reader(file_slot_descs, &state, &profile, range, ¶ms, + scanner_io_ctx, nullptr); + Block block; + size_t read_rows = 1; + bool eof = false; + + ASSERT_TRUE(reader._do_get_next_block(&block, &read_rows, &eof).ok()); + EXPECT_EQ(0, read_rows); + EXPECT_TRUE(eof); +} + +TEST(IcebergPositionDeleteSysTableReaderTest, ValidatesRangeAndDeleteFileMetadata) { + RuntimeState state {TQueryOptions(), TQueryGlobals()}; + RuntimeProfile profile("test_profile"); + TFileScanRangeParams params; + std::vector file_slot_descs; + auto scanner_io_ctx = std::make_shared(); + + TFileRangeDesc empty_range; + IcebergPositionDeleteSysTableReader invalid_context_reader( + file_slot_descs, &state, &profile, empty_range, ¶ms, nullptr, nullptr); + auto status = invalid_context_reader._do_init_reader(nullptr); + EXPECT_TRUE(status.is()) << status; + + IcebergPositionDeleteSysTableReader missing_params_reader( + file_slot_descs, &state, &profile, empty_range, ¶ms, scanner_io_ctx, nullptr); + status = missing_params_reader._do_init_reader(nullptr); + EXPECT_NE(std::string::npos, status.to_string().find("range misses params")); + + TTableFormatFileDesc table_format_desc; + table_format_desc.__set_iceberg_params(TIcebergFileDesc()); + TFileRangeDesc no_delete_file_range; + no_delete_file_range.__set_table_format_params(std::move(table_format_desc)); + IcebergPositionDeleteSysTableReader no_delete_file_reader(file_slot_descs, &state, &profile, + no_delete_file_range, ¶ms, + scanner_io_ctx, nullptr); + status = no_delete_file_reader._do_init_reader(nullptr); + EXPECT_NE(std::string::npos, status.to_string().find("exactly one delete file")); + + TIcebergDeleteFileDesc missing_content; + auto missing_content_range = range_with_delete_file(missing_content); + IcebergPositionDeleteSysTableReader missing_content_reader(file_slot_descs, &state, &profile, + missing_content_range, ¶ms, + scanner_io_ctx, nullptr); + status = missing_content_reader._do_init_reader(nullptr); + EXPECT_NE(std::string::npos, status.to_string().find("misses content")); + + TIcebergDeleteFileDesc equality_delete; + equality_delete.__set_content(2); + auto equality_delete_range = range_with_delete_file(equality_delete); + IcebergPositionDeleteSysTableReader equality_delete_reader(file_slot_descs, &state, &profile, + equality_delete_range, ¶ms, + scanner_io_ctx, nullptr); + status = equality_delete_reader._do_init_reader(nullptr); + EXPECT_NE(std::string::npos, status.to_string().find("does not support delete file content 2")); + + TIcebergDeleteFileDesc missing_format; + missing_format.__set_content(1); + auto missing_format_range = range_with_delete_file(missing_format); + IcebergPositionDeleteSysTableReader missing_format_reader(file_slot_descs, &state, &profile, + missing_format_range, ¶ms, + scanner_io_ctx, nullptr); + status = missing_format_reader._do_init_reader(nullptr); + EXPECT_NE(std::string::npos, status.to_string().find("misses file format")); + + TIcebergDeleteFileDesc unsupported_format; + unsupported_format.__set_content(1); + unsupported_format.__set_file_format(TFileFormatType::FORMAT_CSV_PLAIN); + auto unsupported_format_range = range_with_delete_file(unsupported_format); + IcebergPositionDeleteSysTableReader unsupported_format_reader(file_slot_descs, &state, &profile, + unsupported_format_range, ¶ms, + scanner_io_ctx, nullptr); + status = unsupported_format_reader._do_init_reader(nullptr); + EXPECT_TRUE(status.is()) << status; + + TIcebergDeleteFileDesc invalid_dv; + invalid_dv.__set_content(3); + auto invalid_dv_range = range_with_delete_file(invalid_dv); + IcebergPositionDeleteSysTableReader invalid_dv_reader( + file_slot_descs, &state, &profile, invalid_dv_range, ¶ms, scanner_io_ctx, nullptr); + status = invalid_dv_reader._do_init_reader(nullptr); + EXPECT_NE(std::string::npos, status.to_string().find("misses referenced data file path")); +} + +TEST(IcebergPositionDeleteSysTableReaderTest, AppendsDeletionVectorMetadataAndCachesPartition) { + RuntimeState state {TQueryOptions(), TQueryGlobals()}; + RuntimeProfile profile("test_profile"); + TFileRangeDesc range; + TFileScanRangeParams params; + ObjectPool pool; + const auto nullable_string = make_nullable(std::make_shared()); + const auto nullable_int32 = make_nullable(std::make_shared()); + const auto nullable_int64 = make_nullable(std::make_shared()); + const auto partition_type = make_nullable( + std::make_shared(DataTypes {nullable_int32}, Strings {"p"})); + std::vector slots { + make_slot(&pool, 0, "file_path", nullable_string), + make_slot(&pool, 1, "pos", nullable_int64), + make_slot(&pool, 2, "row", nullable_int32), + make_slot(&pool, 3, "partition", partition_type), + make_slot(&pool, 4, "spec_id", nullable_int32), + make_slot(&pool, 5, "delete_file_path", nullable_string), + make_slot(&pool, 6, "content_offset", nullable_int64), + make_slot(&pool, 7, "content_size_in_bytes", nullable_int64), + }; + auto scanner_io_ctx = std::make_shared(); + IcebergPositionDeleteSysTableReader reader(slots, &state, &profile, range, ¶ms, + scanner_io_ctx, nullptr); + + TIcebergFileDesc iceberg_desc; + iceberg_desc.__set_partition_spec_id(7); + iceberg_desc.__set_partition_data_json(R"({"p":42})"); + TIcebergDeleteFileDesc delete_file; + delete_file.__set_path("/physical-delete.puffin"); + delete_file.__set_original_path("s3://bucket/delete.puffin"); + delete_file.__set_referenced_data_file_path("s3://bucket/data.parquet"); + delete_file.__set_content_offset(12); + delete_file.__set_content_size_in_bytes(34); + reader._iceberg_file_desc = &iceberg_desc; + reader._delete_file_desc = &delete_file; + reader._delete_file_kind = IcebergPositionDeleteSysTableReader::DeleteFileKind::DELETION_VECTOR; + reader._batch_size = 1; + reader._dv_positions.add(uint64_t {5}); + reader._dv_positions.add(uint64_t {9}); + reader._next_dv_position.emplace(reader._dv_positions.begin()); + + Block block = make_output_block(slots); + size_t read_rows = 0; + bool eof = false; + ASSERT_TRUE(reader._append_deletion_vector_block(&block, &read_rows, &eof).ok()); + EXPECT_EQ(1, read_rows); + EXPECT_FALSE(eof); + ASSERT_EQ(1, block.rows()); + ColumnPtr cached_partition = reader._partition_value; + ASSERT_NE(nullptr, cached_partition.get()); + EXPECT_EQ("s3://bucket/data.parquet", string_at(block, "file_path", 0)); + EXPECT_EQ(5, int_at(block, "pos", 0)); + EXPECT_TRUE(is_null_at(block, "row", 0)); + EXPECT_FALSE(is_null_at(block, "partition", 0)); + EXPECT_EQ(42, struct_int_at(block, "partition", 0, 0)); + EXPECT_EQ(7, int_at(block, "spec_id", 0)); + EXPECT_EQ("s3://bucket/delete.puffin", string_at(block, "delete_file_path", 0)); + EXPECT_EQ(12, int_at(block, "content_offset", 0)); + EXPECT_EQ(34, int_at(block, "content_size_in_bytes", 0)); + + ASSERT_TRUE(reader._append_deletion_vector_block(&block, &read_rows, &eof).ok()); + EXPECT_EQ(1, read_rows); + EXPECT_TRUE(eof); + ASSERT_EQ(2, block.rows()); + EXPECT_EQ(cached_partition.get(), reader._partition_value.get()); + EXPECT_EQ(9, int_at(block, "pos", 1)); + EXPECT_FALSE(is_null_at(block, "partition", 1)); + EXPECT_EQ(42, struct_int_at(block, "partition", 0, 1)); + + ASSERT_TRUE(reader._append_deletion_vector_block(&block, &read_rows, &eof).ok()); + EXPECT_EQ(0, read_rows); + EXPECT_TRUE(eof); + + reader._dv_positions = roaring::Roaring64Map(); + reader._dv_positions.add(uint64_t {13}); + reader._next_dv_position.emplace(reader._dv_positions.begin()); + std::vector partial_slots {slots[0], slots[1]}; + Block partial_block = make_output_block(partial_slots); + ASSERT_TRUE(reader._append_deletion_vector_block(&partial_block, &read_rows, &eof).ok()); + EXPECT_EQ(1, read_rows); + EXPECT_TRUE(eof); + EXPECT_EQ(13, int_at(partial_block, "pos", 0)); +} + +TEST(IcebergPositionDeleteSysTableReaderTest, AppendsPositionDeleteRowsAndValidatesColumns) { + RuntimeState state {TQueryOptions(), TQueryGlobals()}; + RuntimeProfile profile("test_profile"); + TFileRangeDesc range; + TFileScanRangeParams params; + ObjectPool pool; + const auto nullable_string = make_nullable(std::make_shared()); + const auto nullable_int32 = make_nullable(std::make_shared()); + const auto nullable_int64 = make_nullable(std::make_shared()); + std::vector slots { + make_slot(&pool, 0, "file_path", nullable_string), + make_slot(&pool, 1, "pos", nullable_int64), + make_slot(&pool, 2, "row", nullable_int32), + }; + auto scanner_io_ctx = std::make_shared(); + IcebergPositionDeleteSysTableReader reader(slots, &state, &profile, range, ¶ms, + scanner_io_ctx, nullptr); + TIcebergFileDesc iceberg_desc; + TIcebergDeleteFileDesc delete_file; + delete_file.__set_path("/delete.parquet"); + reader._iceberg_file_desc = &iceberg_desc; + reader._delete_file_desc = &delete_file; + reader._delete_file_kind = IcebergPositionDeleteSysTableReader::DeleteFileKind::POSITION_DELETE; + + reader.set_batch_size(17); + EXPECT_EQ(17, reader.get_batch_size()); + EXPECT_TRUE(reader._output_column_requested("row")); + EXPECT_FALSE(reader._output_column_requested("missing")); + reader._init_read_columns(true); + ASSERT_EQ(3, reader._read_columns.size()); + EXPECT_EQ("row", reader._read_columns.back().name); + + std::unordered_map name_to_type; + ASSERT_TRUE(reader._get_columns_impl(&name_to_type).ok()); + EXPECT_EQ(3, name_to_type.size()); + + Block delete_block = reader._create_delete_block(); + { + auto columns_guard = delete_block.mutate_columns_scoped(); + auto& columns = columns_guard.mutable_columns(); + parquet_utils::insert_string(columns[0], "s3://bucket/data.parquet"); + parquet_utils::insert_int64(columns[1], 19); + parquet_utils::insert_int32(columns[2], 23); + } + Block output_block = make_output_block(slots); + size_t appended_rows = 0; + ASSERT_TRUE(reader._append_position_delete_block(&output_block, delete_block, 1, &appended_rows) + .ok()); + EXPECT_EQ(1, appended_rows); + ASSERT_EQ(1, output_block.rows()); + EXPECT_EQ("s3://bucket/data.parquet", string_at(output_block, "file_path", 0)); + EXPECT_EQ(19, int_at(output_block, "pos", 0)); + EXPECT_EQ(23, int_at(output_block, "row", 0)); + + Block empty_delete_block = reader._create_delete_block(); + auto path_column = nullable_string->create_column(); + auto status = reader._append_sys_column(path_column, *slots[0], &empty_delete_block, 0, 0); + EXPECT_NE(std::string::npos, status.to_string().find("file_path column is missing")); + auto pos_column = nullable_int64->create_column(); + status = reader._append_sys_column(pos_column, *slots[1], &empty_delete_block, 0, 0); + EXPECT_NE(std::string::npos, status.to_string().find("pos column is missing")); + auto row_column = nullable_int32->create_column(); + ASSERT_TRUE(reader._append_sys_column(row_column, *slots[2], &empty_delete_block, 0, 0).ok()); + ASSERT_EQ(1, row_column->size()); + + Block missing_columns; + path_column = nullable_string->create_column(); + status = reader._append_sys_column(path_column, *slots[0], &missing_columns, 0, 0); + EXPECT_NE(std::string::npos, status.to_string().find("file_path column is missing")); + pos_column = nullable_int64->create_column(); + status = reader._append_sys_column(pos_column, *slots[1], &missing_columns, 0, 0); + EXPECT_NE(std::string::npos, status.to_string().find("pos column is missing")); + + std::vector partial_slots {slots[0], slots[1]}; + Block partial_output_block = make_output_block(partial_slots); + appended_rows = 0; + ASSERT_TRUE(reader._append_position_delete_block(&partial_output_block, delete_block, 1, + &appended_rows) + .ok()); + EXPECT_EQ(1, appended_rows); + EXPECT_EQ(1, partial_output_block.rows()); + + auto* unknown_slot = make_slot(&pool, 3, "unknown", nullable_string); + auto unknown_column = nullable_string->create_column(); + status = reader._append_sys_column(unknown_column, *unknown_slot, &delete_block, 0, 0); + EXPECT_NE(std::string::npos, status.to_string().find("Unknown Iceberg")); + + Block unused_block; + size_t read_rows = 0; + bool eof = false; + status = reader._do_get_next_block(&unused_block, &read_rows, &eof); + EXPECT_NE(std::string::npos, status.to_string().find("reader is not initialized")); +} + +TEST(IcebergPositionDeleteSysTableReaderTest, AppendsNullMetadataAndUsesDeletePathFallback) { + RuntimeState state {TQueryOptions(), TQueryGlobals()}; + RuntimeProfile profile("test_profile"); + TFileRangeDesc range; + TFileScanRangeParams params; + ObjectPool pool; + const auto nullable_string = make_nullable(std::make_shared()); + const auto nullable_int32 = make_nullable(std::make_shared()); + const auto nullable_int64 = make_nullable(std::make_shared()); + const auto partition_type = make_nullable( + std::make_shared(DataTypes {nullable_int32}, Strings {"p"})); + std::vector slots { + make_slot(&pool, 0, "partition", partition_type), + make_slot(&pool, 1, "spec_id", nullable_int32), + make_slot(&pool, 2, "delete_file_path", nullable_string), + make_slot(&pool, 3, "content_offset", nullable_int64), + make_slot(&pool, 4, "content_size_in_bytes", nullable_int64), + }; + auto scanner_io_ctx = std::make_shared(); + IcebergPositionDeleteSysTableReader reader(slots, &state, &profile, range, ¶ms, + scanner_io_ctx, nullptr); + TIcebergFileDesc iceberg_desc; + TIcebergDeleteFileDesc delete_file; + delete_file.__set_path("/fallback-delete-file"); + reader._iceberg_file_desc = &iceberg_desc; + reader._delete_file_desc = &delete_file; + reader._delete_file_kind = IcebergPositionDeleteSysTableReader::DeleteFileKind::DELETION_VECTOR; + reader._dv_positions.add(uint64_t {1}); + reader._next_dv_position.emplace(reader._dv_positions.begin()); + + Block block = make_output_block(slots); + size_t read_rows = 0; + bool eof = false; + ASSERT_TRUE(reader._append_deletion_vector_block(&block, &read_rows, &eof).ok()); + EXPECT_EQ(1, read_rows); + EXPECT_TRUE(eof); + EXPECT_TRUE(is_null_at(block, "partition", 0)); + EXPECT_TRUE(is_null_at(block, "spec_id", 0)); + EXPECT_EQ("/fallback-delete-file", string_at(block, "delete_file_path", 0)); + EXPECT_TRUE(is_null_at(block, "content_offset", 0)); + EXPECT_TRUE(is_null_at(block, "content_size_in_bytes", 0)); + + std::vector empty_slots; + IcebergPositionDeleteSysTableReader empty_reader(empty_slots, &state, &profile, range, ¶ms, + scanner_io_ctx, nullptr); + empty_reader._dv_positions.add(uint64_t {2}); + empty_reader._next_dv_position.emplace(empty_reader._dv_positions.begin()); + Block empty_block; + ASSERT_TRUE(empty_reader._append_deletion_vector_block(&empty_block, &read_rows, &eof).ok()); + EXPECT_EQ(1, read_rows); + EXPECT_TRUE(eof); +} + +TEST(IcebergPositionDeleteSysTableV2ReaderTest, RecordsDeletionVectorRows) { + io::FileReaderStats file_reader_stats; + auto scanner_io_ctx = std::make_shared(); + scanner_io_ctx->file_reader_stats = &file_reader_stats; + std::vector file_slot_descs; + + format::iceberg::IcebergPositionDeleteSysTableV2Reader reader; + reader._io_ctx = scanner_io_ctx; + reader._file_slot_descs = &file_slot_descs; + reader._batch_size = 2; + reader._dv_positions.add(uint64_t {7}); + reader._dv_positions.add(uint64_t {9}); + reader._dv_positions.add(uint64_t {11}); + reader._next_dv_position.emplace(reader._dv_positions.begin()); + + Block block; + size_t read_rows = 0; + bool eof = true; + ASSERT_TRUE(reader._append_deletion_vector_block(&block, &read_rows, &eof).ok()); + EXPECT_EQ(2, read_rows); + EXPECT_FALSE(eof); + EXPECT_EQ(2, file_reader_stats.read_rows); + + ASSERT_TRUE(reader._append_deletion_vector_block(&block, &read_rows, &eof).ok()); + EXPECT_EQ(1, read_rows); + EXPECT_FALSE(eof); + EXPECT_EQ(3, file_reader_stats.read_rows); +} + +TEST(IcebergPositionDeleteSysTableV2ReaderTest, CachesAndClearsPartitionValue) { + ObjectPool pool; + const auto nullable_int32 = make_nullable(std::make_shared()); + const auto partition_type = make_nullable( + std::make_shared(DataTypes {nullable_int32}, Strings {"p"})); + auto* partition_slot = make_slot(&pool, 0, "partition", partition_type); + + TIcebergFileDesc iceberg_desc; + iceberg_desc.__set_partition_data_json(R"({"p":42})"); + + format::iceberg::IcebergPositionDeleteSysTableV2Reader reader; + reader._iceberg_file_desc = &iceberg_desc; + auto partition_column = partition_type->create_column(); + ASSERT_TRUE(reader._append_partition_column(partition_column, *partition_slot).ok()); + ColumnPtr cached_partition = reader._partition_value; + ASSERT_NE(nullptr, cached_partition.get()); + + ASSERT_TRUE(reader._append_partition_column(partition_column, *partition_slot).ok()); + EXPECT_EQ(cached_partition.get(), reader._partition_value.get()); + + Block block; + block.insert(ColumnWithTypeAndName(std::move(partition_column), partition_type, "partition")); + ASSERT_EQ(2, block.rows()); + EXPECT_FALSE(is_null_at(block, "partition", 0)); + EXPECT_EQ(42, struct_int_at(block, "partition", 0, 0)); + EXPECT_FALSE(is_null_at(block, "partition", 1)); + EXPECT_EQ(42, struct_int_at(block, "partition", 0, 1)); + + reader._has_split = true; + reader._dv_positions.add(uint64_t {1}); + reader._next_dv_position.emplace(reader._dv_positions.begin()); + ASSERT_TRUE(reader.close().ok()); + EXPECT_EQ(nullptr, reader._iceberg_file_desc); + EXPECT_EQ(nullptr, reader._partition_value.get()); + EXPECT_TRUE(reader._dv_positions.isEmpty()); + EXPECT_FALSE(reader._next_dv_position.has_value()); + EXPECT_FALSE(reader._has_split); +} + +TEST(IcebergPositionDeleteSysTableV2ReaderTest, ValidatesDeleteFileContentAfterBindingDescriptor) { + RuntimeState state {TQueryOptions(), TQueryGlobals()}; + RuntimeProfile profile("test_profile"); + TFileScanRangeParams params; + std::vector file_slot_descs; + TIcebergDeleteFileDesc equality_delete; + equality_delete.__set_content(2); + + format::iceberg::IcebergPositionDeleteSysTableV2Reader reader; + reader._runtime_state = &state; + reader._scanner_profile = &profile; + reader._scan_params = ¶ms; + reader._file_slot_descs = &file_slot_descs; + reader._current_range = range_with_delete_file(equality_delete); + + auto status = reader._init_split(); + EXPECT_NE(std::string::npos, status.to_string().find("does not support delete file content 2")); + ASSERT_NE(nullptr, reader._iceberg_file_desc); + ASSERT_NE(nullptr, reader._delete_file_desc); + EXPECT_EQ(reader._iceberg_file_desc->delete_files.data(), reader._delete_file_desc); +} + +TEST(IcebergPositionDeleteSysTableV2ReaderTest, StopsBeforeExpandingDeletionVector) { + format::iceberg::IcebergPositionDeleteSysTableV2Reader reader; + reader._io_ctx = std::make_shared(); + reader._io_ctx->should_stop = true; + + Block block; + bool eof = false; + ASSERT_TRUE(reader.get_block(&block, &eof).ok()); + EXPECT_TRUE(eof); +} + +} // namespace doris diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergExternalTable.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergExternalTable.java index c6ee8e96707220..008c12582d78c5 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergExternalTable.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergExternalTable.java @@ -343,19 +343,6 @@ public Map getSupportedSysTables() { return IcebergSysTable.SUPPORTED_SYS_TABLES; } - @Override - public Optional findSysTable(String tableNameWithSysTableName) { - Optional sysTable = MTMVRelatedTableIf.super.findSysTable(tableNameWithSysTableName); - if (sysTable.isPresent()) { - return sysTable; - } - String sysTableName = SysTable.getTableNameWithSysTableName(tableNameWithSysTableName).second; - if (IcebergSysTable.POSITION_DELETES.equals(sysTableName)) { - return Optional.of(IcebergSysTable.UNSUPPORTED_POSITION_DELETES_TABLE); - } - return Optional.empty(); - } - @Override public boolean isView() { makeSureInitialized(); diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java index 2769a1ddb8e43e..ba232854670603 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java @@ -49,11 +49,13 @@ import org.apache.doris.datasource.iceberg.source.IcebergDeleteFileFilter.EqualityDelete; import org.apache.doris.datasource.property.storage.StorageProperties; import org.apache.doris.nereids.exceptions.NotSupportedException; +import org.apache.doris.persist.gson.GsonUtils; import org.apache.doris.planner.PlanNodeId; import org.apache.doris.planner.ScanContext; import org.apache.doris.qe.ConnectContext; import org.apache.doris.qe.SessionVariable; import org.apache.doris.spi.Split; +import org.apache.doris.system.Backend; import org.apache.doris.thrift.TColumnCategory; import org.apache.doris.thrift.TExplainLevel; import org.apache.doris.thrift.TFileFormatType; @@ -67,9 +69,12 @@ import com.google.common.annotations.VisibleForTesting; import com.google.common.base.Preconditions; import com.google.common.collect.Lists; +import com.google.gson.JsonObject; import org.apache.commons.lang3.exception.ExceptionUtils; import org.apache.iceberg.BaseFileScanTask; import org.apache.iceberg.BaseTable; +import org.apache.iceberg.BatchScan; +import org.apache.iceberg.ContentScanTask; import org.apache.iceberg.DataFile; import org.apache.iceberg.DeleteFile; import org.apache.iceberg.DeleteFileIndex; @@ -79,10 +84,14 @@ import org.apache.iceberg.ManifestContent; import org.apache.iceberg.ManifestFile; import org.apache.iceberg.PartitionData; +import org.apache.iceberg.PartitionField; import org.apache.iceberg.PartitionSpec; import org.apache.iceberg.PartitionSpecParser; +import org.apache.iceberg.PositionDeletesScanTask; +import org.apache.iceberg.ScanTask; import org.apache.iceberg.SchemaParser; import org.apache.iceberg.Snapshot; +import org.apache.iceberg.SplittableScanTask; import org.apache.iceberg.Table; import org.apache.iceberg.TableProperties; import org.apache.iceberg.TableScan; @@ -97,6 +106,8 @@ import org.apache.iceberg.mapping.MappedFields; import org.apache.iceberg.mapping.NameMapping; import org.apache.iceberg.mapping.NameMappingParser; +import org.apache.iceberg.types.Type; +import org.apache.iceberg.types.Types.NestedField; import org.apache.iceberg.util.ScanTaskUtil; import org.apache.iceberg.util.SerializationUtil; import org.apache.iceberg.util.TableScanUtil; @@ -104,6 +115,7 @@ import org.apache.logging.log4j.Logger; import java.io.IOException; +import java.math.BigDecimal; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; @@ -184,6 +196,17 @@ public IcebergScanNode(PlanNodeId id, TupleDescriptor desc, boolean needCheckCol super(id, desc, "ICEBERG_SCAN_NODE", scanContext, needCheckColumnPriv, sv); ExternalTable table = (ExternalTable) desc.getTable(); + initIcebergSource(table); + } + + public IcebergScanNode(PlanNodeId id, TupleDescriptor desc, IcebergSysExternalTable sysExternalTable, + SessionVariable sv, ScanContext scanContext) { + super(id, desc, "ICEBERG_SCAN_NODE", scanContext, false, sv); + isSystemTable = true; + initIcebergSource(sysExternalTable); + } + + private void initIcebergSource(ExternalTable table) { if (table instanceof HMSExternalTable) { source = new IcebergHMSSource((HMSExternalTable) table, desc); } else if (table instanceof IcebergExternalTable || table instanceof IcebergSysExternalTable) { @@ -286,12 +309,19 @@ private void setIcebergParams(TFileRangeDesc rangeDesc, IcebergSplit icebergSpli TTableFormatFileDesc tableFormatFileDesc = new TTableFormatFileDesc(); tableFormatFileDesc.setTableFormatType(icebergSplit.getTableFormatType().value()); TIcebergFileDesc fileDesc = new TIcebergFileDesc(); + if (isSystemTable && icebergSplit.isPositionDeleteSystemTableSplit()) { + setIcebergPositionDeleteSysTableParams(rangeDesc, icebergSplit, tableFormatFileDesc, fileDesc); + return; + } if (isSystemTable) { rangeDesc.setFormatType(TFileFormatType.FORMAT_JNI); tableFormatFileDesc.setTableLevelRowCount(-1); fileDesc.setSerializedSplit(icebergSplit.getSerializedSplit()); tableFormatFileDesc.setIcebergParams(fileDesc); rangeDesc.setTableFormatParams(tableFormatFileDesc); + rangeDesc.unsetColumnsFromPath(); + rangeDesc.unsetColumnsFromPathKeys(); + rangeDesc.unsetColumnsFromPathIsNull(); return; } if (tableLevelPushDownCount) { @@ -394,6 +424,41 @@ private void setIcebergParams(TFileRangeDesc rangeDesc, IcebergSplit icebergSpli rangeDesc.setTableFormatParams(tableFormatFileDesc); } + private void setIcebergPositionDeleteSysTableParams(TFileRangeDesc rangeDesc, IcebergSplit icebergSplit, + TTableFormatFileDesc tableFormatFileDesc, TIcebergFileDesc fileDesc) { + rangeDesc.setFormatType(icebergSplit.getPositionDeleteFileFormat()); + tableFormatFileDesc.setTableLevelRowCount(-1); + fileDesc.setContent(icebergSplit.getPositionDeleteContent()); + + if (icebergSplit.getPartitionSpecId() != null) { + fileDesc.setPartitionSpecId(icebergSplit.getPartitionSpecId()); + } + if (icebergSplit.getPartitionDataJson() != null) { + fileDesc.setPartitionDataJson(icebergSplit.getPartitionDataJson()); + } + + TIcebergDeleteFileDesc deleteFileDesc = new TIcebergDeleteFileDesc(); + deleteFileDesc.setPath(rangeDesc.getPath()); + deleteFileDesc.setOriginalPath(icebergSplit.getPositionDeleteOriginalPath()); + deleteFileDesc.setFileFormat(icebergSplit.getPositionDeleteFileFormat()); + deleteFileDesc.setContent(icebergSplit.getPositionDeleteContent()); + if (icebergSplit.getPositionDeleteContentOffset() != null) { + deleteFileDesc.setContentOffset(icebergSplit.getPositionDeleteContentOffset()); + } + if (icebergSplit.getPositionDeleteContentSizeInBytes() != null) { + deleteFileDesc.setContentSizeInBytes(icebergSplit.getPositionDeleteContentSizeInBytes()); + } + if (icebergSplit.getPositionDeleteReferencedDataFilePath() != null) { + deleteFileDesc.setReferencedDataFilePath(icebergSplit.getPositionDeleteReferencedDataFilePath()); + } + fileDesc.setDeleteFiles(Lists.newArrayList(deleteFileDesc)); + tableFormatFileDesc.setIcebergParams(fileDesc); + rangeDesc.setTableFormatParams(tableFormatFileDesc); + rangeDesc.unsetColumnsFromPath(); + rangeDesc.unsetColumnsFromPathKeys(); + rangeDesc.unsetColumnsFromPathIsNull(); + } + @Override protected List getDeleteFiles(TFileRangeDesc rangeDesc) { List deleteFiles = new ArrayList<>(); @@ -641,11 +706,11 @@ private CloseableIterable splitFiles(TableScan scan) { return TableScanUtil.splitFiles(CloseableIterable.withNoopClose(fileScanTaskList), targetSplitSize); } - private long determineTargetFileSplitSize(Iterable tasks) { + private long determineTargetFileSplitSize(Iterable> tasks) { long result = sessionVariable.getMaxInitialSplitSize(); long accumulatedTotalFileSize = 0; boolean exceedInitialThreshold = false; - for (FileScanTask task : tasks) { + for (ContentScanTask task : tasks) { accumulatedTotalFileSize += ScanTaskUtil.contentSizeInBytes(task.file()); if (!exceedInitialThreshold && accumulatedTotalFileSize >= sessionVariable.getMaxSplitSize() * sessionVariable.getMaxInitialSplitNum()) { @@ -659,6 +724,13 @@ private long determineTargetFileSplitSize(Iterable tasks) { return result; } + private long determinePositionDeleteTargetSplitSize(Iterable tasks) { + if (sessionVariable.getFileSplitSize() > 0) { + return sessionVariable.getFileSplitSize(); + } + return determineTargetFileSplitSize(tasks); + } + private CloseableIterable planFileScanTaskWithManifestCache(TableScan scan) throws IOException { // Get the snapshot from the scan; return empty if no snapshot exists Snapshot snapshot = scan.snapshot(); @@ -896,14 +968,134 @@ private Split createIcebergSplit(FileScanTask fileScanTask) { return split; } - private Split createIcebergSysSplit(FileScanTask fileScanTask) { - long rowCount = fileScanTask.file() == null ? 1 : fileScanTask.file().recordCount(); + private Split createIcebergSysSplit(ScanTask scanTask) { + long rowCount = Math.max(scanTask.estimatedRowsCount(), 1L); + if (scanTask.isFileScanTask() && scanTask.asFileScanTask().file() != null) { + rowCount = Math.max(scanTask.asFileScanTask().file().recordCount(), 1L); + } IcebergSplit split = IcebergSplit.newSysTableSplit( - SerializationUtil.serializeToBase64(fileScanTask), rowCount); + SerializationUtil.serializeToBase64(scanTask), rowCount); split.setTableFormatType(TableFormatType.ICEBERG); return split; } + private Split createIcebergPositionDeleteSysSplit(PositionDeletesScanTask task) throws UserException { + DeleteFile deleteFile = task.file(); + String originalPath = deleteFile.path().toString(); + LocationPath locationPath = createLocationPathWithCache(originalPath); + IcebergSplit split = IcebergSplit.newPositionDeleteSysTableSplit( + locationPath, task.start(), task.length(), deleteFile.fileSizeInBytes(), + storagePropertiesMap, originalPath); + split.setTableFormatType(TableFormatType.ICEBERG); + split.setPositionDeleteFileFormat(getNativePositionDeleteFileFormat(deleteFile.format())); + split.setPositionDeleteOriginalPath(originalPath); + if (deleteFile.format() == FileFormat.PUFFIN) { + split.setPositionDeleteContent(IcebergDeleteFileFilter.DeletionVector.type()); + split.setPositionDeleteReferencedDataFilePath(deleteFile.referencedDataFile()); + split.setPositionDeleteContentOffset(deleteFile.contentOffset()); + split.setPositionDeleteContentSizeInBytes(deleteFile.contentSizeInBytes()); + } else { + split.setPositionDeleteContent(IcebergDeleteFileFilter.PositionDelete.type()); + } + + split.setPartitionSpecId(deleteFile.specId()); + PartitionSpec partitionSpec = icebergTable.specs().get(deleteFile.specId()); + Preconditions.checkNotNull(partitionSpec, "Partition spec with specId %s not found for table %s", + deleteFile.specId(), icebergTable.name()); + if (partitionSpec.isPartitioned() && deleteFile.partition() != null + && isPositionDeletesPartitionColumnRequested()) { + split.setPartitionDataJson(getPartitionDataObjectJson( + (PartitionData) deleteFile.partition(), partitionSpec, + getPositionDeletesOutputPartitionFields())); + } + return split; + } + + @SuppressWarnings("unchecked") + private Iterable splitPositionDeleteScanTask(PositionDeletesScanTask task) { + return ((SplittableScanTask) task).split(targetSplitSize); + } + + private TFileFormatType getNativePositionDeleteFileFormat(FileFormat fileFormat) { + if (fileFormat == FileFormat.PARQUET || fileFormat == FileFormat.PUFFIN) { + return TFileFormatType.FORMAT_PARQUET; + } else if (fileFormat == FileFormat.ORC) { + return TFileFormatType.FORMAT_ORC; + } + throw new UnsupportedOperationException( + "Unsupported Iceberg position delete file format: " + fileFormat); + } + + private List getPositionDeletesOutputPartitionFields() { + NestedField partitionField = icebergTable.schema().findField("partition"); + Preconditions.checkNotNull(partitionField, + "Partition field not found in Iceberg position_deletes metadata table schema"); + return partitionField.type().asNestedType().fields(); + } + + private boolean isPositionDeletesPartitionColumnRequested() { + return desc.getSlots().stream() + .anyMatch(slot -> "partition".equalsIgnoreCase(slot.getColumn().getName())); + } + + private String getPartitionDataObjectJson(PartitionData partitionData, PartitionSpec partitionSpec, + List outputPartitionFields) throws UserException { + List partitionTypes = partitionData.getPartitionType().asNestedType().fields(); + boolean enableMappingVarbinary = getEnableMappingVarbinary(); + for (int i = 0; i < partitionTypes.size(); i++) { + Type type = partitionTypes.get(i).type(); + if (partitionData.get(i) != null && (type.typeId() == Type.TypeID.BINARY + || type.typeId() == Type.TypeID.FIXED + || (type.typeId() == Type.TypeID.UUID && enableMappingVarbinary))) { + throw new UserException("Iceberg position_deletes cannot materialize non-null partition field '" + + partitionTypes.get(i).name() + "' of type " + type + + " without a binary-safe partition transport"); + } + } + List partitionValues = IcebergUtils.getPartitionValues( + partitionData, partitionSpec, sessionVariable.getTimeZone()); + Map partitionValueByFieldId = new HashMap<>(); + List fields = partitionSpec.fields(); + for (int i = 0; i < fields.size(); i++) { + partitionValueByFieldId.put(fields.get(i).fieldId(), + getPartitionJsonValue(partitionTypes.get(i).type(), partitionValues.get(i))); + } + JsonObject partitionJson = new JsonObject(); + for (NestedField outputPartitionField : outputPartitionFields) { + partitionJson.add(outputPartitionField.name(), + GsonUtils.GSON.toJsonTree(partitionValueByFieldId.get(outputPartitionField.fieldId()))); + } + return GsonUtils.GSON.toJson(partitionJson); + } + + private static Object getPartitionJsonValue(Type type, String partitionValue) { + if (partitionValue == null) { + return null; + } + switch (type.typeId()) { + case BOOLEAN: + return Boolean.parseBoolean(partitionValue); + case INTEGER: + return Integer.parseInt(partitionValue); + case LONG: + return Long.parseLong(partitionValue); + case FLOAT: + return Float.parseFloat(partitionValue); + case DOUBLE: + return Double.parseDouble(partitionValue); + case DECIMAL: + return new BigDecimal(partitionValue); + case STRING: + case UUID: + case DATE: + case TIME: + case TIMESTAMP: + return partitionValue; + default: + return partitionValue; + } + } + @Override protected TColumnCategory classifyColumn(SlotDescriptor slot, List partitionKeys) { if (Column.ICEBERG_ROWID_COL.equalsIgnoreCase(slot.getColumn().getName())) { @@ -972,6 +1164,9 @@ private List doGetSplits(int numBackends) throws UserException { } private List doGetSystemTableSplits() throws UserException { + if (isPositionDeletesSystemTable()) { + return doGetPositionDeletesSystemTableSplits(); + } List splits = new ArrayList<>(); TableScan scan = createTableScan(); long startTime = System.currentTimeMillis(); @@ -988,6 +1183,75 @@ private List doGetSystemTableSplits() throws UserException { return splits; } + private boolean isPositionDeletesSystemTable() { + TableIf targetTable = source.getTargetTable(); + return targetTable instanceof IcebergSysExternalTable + && "position_deletes".equalsIgnoreCase(((IcebergSysExternalTable) targetTable).getSysTableType()); + } + + private List doGetPositionDeletesSystemTableSplits() throws UserException { + checkPositionDeletesBackendCompatibility(backendPolicy.getBackends()); + List splits = new ArrayList<>(); + List positionDeleteTasks = new ArrayList<>(); + BatchScan scan = icebergTable.newBatchScan().metricsReporter(new IcebergMetricsReporter()); + + IcebergTableQueryInfo info = getSpecifiedSnapshot(); + if (info != null) { + if (info.getRef() != null) { + scan = scan.useRef(info.getRef()); + } else { + scan = scan.useSnapshot(info.getSnapshotId()); + } + } + + List expressions = new ArrayList<>(); + for (Expr conjunct : conjuncts) { + Expression expression = IcebergUtils.convertToIcebergExpr(conjunct, icebergTable.schema()); + if (expression != null) { + expressions.add(expression); + } + } + for (Expression predicate : expressions) { + scan = scan.filter(predicate); + this.pushdownIcebergPredicates.add(predicate.toString()); + } + + long startTime = System.currentTimeMillis(); + scan = scan.planWith(source.getCatalog().getThreadPoolWithPreAuth()); + try (CloseableIterable scanTasks = scan.planFiles()) { + for (ScanTask task : scanTasks) { + if (!(task instanceof PositionDeletesScanTask)) { + throw new UserException("Unexpected Iceberg position_deletes scan task: " + task); + } + positionDeleteTasks.add((PositionDeletesScanTask) task); + } + } catch (IOException e) { + throw new UserException(e.getMessage(), e); + } finally { + if (getSummaryProfile() != null) { + getSummaryProfile().addExternalTableGetFileScanTasksTime(System.currentTimeMillis() - startTime); + } + } + targetSplitSize = determinePositionDeleteTargetSplitSize(positionDeleteTasks); + for (PositionDeletesScanTask task : positionDeleteTasks) { + for (PositionDeletesScanTask splitTask : splitPositionDeleteScanTask(task)) { + splits.add(createIcebergPositionDeleteSysSplit(splitTask)); + } + } + selectedPartitionNum = 0; + return splits; + } + + @VisibleForTesting + static void checkPositionDeletesBackendCompatibility(Iterable backends) throws UserException { + for (Backend backend : backends) { + if (backend.isSmoothUpgradeSrc()) { + throw new UserException("Iceberg position_deletes system table is unavailable while backend " + + backend.getId() + " is a smooth upgrade source"); + } + } + } + @Override public boolean isBatchMode() { if (isSystemTable) { diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergSplit.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergSplit.java index 3af484abd6d282..59b4f483f019e3 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergSplit.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergSplit.java @@ -20,6 +20,7 @@ import org.apache.doris.common.util.LocationPath; import org.apache.doris.datasource.FileSplit; import org.apache.doris.datasource.property.storage.StorageProperties; +import org.apache.doris.thrift.TFileFormatType; import lombok.Data; import org.apache.iceberg.DeleteFile; @@ -52,6 +53,13 @@ public class IcebergSplit extends FileSplit { private Long firstRowId = null; private Long lastUpdatedSequenceNumber = null; private String serializedSplit; + private boolean positionDeleteSystemTableSplit = false; + private TFileFormatType positionDeleteFileFormat; + private int positionDeleteContent; + private String positionDeleteOriginalPath; + private String positionDeleteReferencedDataFilePath; + private Long positionDeleteContentOffset; + private Long positionDeleteContentSizeInBytes; // File path will be changed if the file is modified, so there's no need to get modification time. public IcebergSplit(LocationPath file, long start, long length, long fileLength, String[] hosts, @@ -78,4 +86,13 @@ public static IcebergSplit newSysTableSplit(String serializedSplit, long rowCoun split.setSelfSplitWeight(Math.max(rowCount, 1L)); return split; } + + public static IcebergSplit newPositionDeleteSysTableSplit(LocationPath file, long start, long length, + long fileLength, Map config, String originalPath) { + IcebergSplit split = new IcebergSplit(file, start, length, fileLength, null, null, config, + Collections.emptyList(), originalPath); + split.setPositionDeleteSystemTableSplit(true); + split.setSelfSplitWeight(Math.max(length, 1L)); + return split; + } } diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/systable/IcebergSysTable.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/systable/IcebergSysTable.java index 72a739e7b9a64b..5115c2f9ec6f95 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/systable/IcebergSysTable.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/systable/IcebergSysTable.java @@ -20,7 +20,6 @@ import org.apache.doris.datasource.ExternalTable; import org.apache.doris.datasource.iceberg.IcebergExternalTable; import org.apache.doris.datasource.iceberg.IcebergSysExternalTable; -import org.apache.doris.nereids.exceptions.AnalysisException; import org.apache.iceberg.MetadataTableType; @@ -40,27 +39,20 @@ * @see org.apache.iceberg.MetadataTableType for all supported system table types */ public class IcebergSysTable extends NativeSysTable { - public static final String POSITION_DELETES = MetadataTableType.POSITION_DELETES.name().toLowerCase(Locale.ROOT); - /** * All supported Iceberg system tables. * Key is the system table name (e.g., "snapshots", "history"). */ public static final Map SUPPORTED_SYS_TABLES = Collections.unmodifiableMap( Arrays.stream(MetadataTableType.values()) - .filter(type -> type != MetadataTableType.POSITION_DELETES) - .map(type -> new IcebergSysTable(type.name().toLowerCase(Locale.ROOT), true)) + .map(type -> new IcebergSysTable(type.name().toLowerCase(Locale.ROOT))) .collect(Collectors.toMap(SysTable::getSysTableName, Function.identity()))); - public static final SysTable UNSUPPORTED_POSITION_DELETES_TABLE = - new IcebergSysTable(POSITION_DELETES, false); private final String tableName; - private final boolean supported; - private IcebergSysTable(String tableName, boolean supported) { + private IcebergSysTable(String tableName) { super(tableName); this.tableName = tableName; - this.supported = supported; } @Override @@ -70,9 +62,6 @@ public String getSysTableName() { @Override public ExternalTable createSysExternalTable(ExternalTable sourceTable) { - if (!supported) { - throw new AnalysisException("SysTable " + tableName + " is not supported yet"); - } if (!(sourceTable instanceof IcebergExternalTable)) { throw new IllegalArgumentException( "Expected IcebergExternalTable but got " + sourceTable.getClass().getSimpleName()); diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/source/IcebergScanNodeTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/source/IcebergScanNodeTest.java index f55b4f6f8e8027..c154d43d53763b 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/source/IcebergScanNodeTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/source/IcebergScanNodeTest.java @@ -19,38 +19,61 @@ import org.apache.doris.analysis.TupleDescriptor; import org.apache.doris.analysis.TupleId; +import org.apache.doris.common.UserException; import org.apache.doris.common.util.LocationPath; import org.apache.doris.datasource.TableFormatType; import org.apache.doris.planner.PlanNodeId; import org.apache.doris.planner.ScanContext; import org.apache.doris.qe.SessionVariable; +import org.apache.doris.system.Backend; import org.apache.doris.thrift.TFileRangeDesc; import org.apache.doris.thrift.TIcebergDeleteFileDesc; import org.apache.iceberg.DataFile; +import org.apache.iceberg.FileFormat; import org.apache.iceberg.FileScanTask; +import org.apache.iceberg.PartitionData; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.Schema; +import org.apache.iceberg.types.Types; import org.apache.iceberg.util.ScanTaskUtil; import org.junit.Assert; import org.junit.Test; import org.mockito.Mockito; import java.lang.reflect.Field; +import java.lang.reflect.InvocationTargetException; import java.lang.reflect.Method; +import java.nio.ByteBuffer; import java.util.ArrayList; import java.util.Collections; +import java.util.List; +import java.util.UUID; public class IcebergScanNodeTest { private static final long MB = 1024L * 1024L; private static class TestIcebergScanNode extends IcebergScanNode { + private final boolean enableMappingVarbinary; + TestIcebergScanNode(SessionVariable sv) { + this(sv, false); + } + + TestIcebergScanNode(SessionVariable sv, boolean enableMappingVarbinary) { super(new PlanNodeId(0), new TupleDescriptor(new TupleId(0)), sv, ScanContext.EMPTY); + this.enableMappingVarbinary = enableMappingVarbinary; } @Override public boolean isBatchMode() { return false; } + + @Override + protected boolean getEnableMappingVarbinary() { + return enableMappingVarbinary; + } } @Test @@ -143,4 +166,98 @@ public void testSetIcebergParamsPropagatesPositionDeleteFileFormat() throws Exce .get(0); Assert.assertEquals(org.apache.doris.thrift.TFileFormatType.FORMAT_ORC, deleteFileDesc.getFileFormat()); } + + @Test + public void testPartitionDataJsonMatchesRenamedFieldById() throws Exception { + SessionVariable sv = new SessionVariable(); + TestIcebergScanNode node = new TestIcebergScanNode(sv); + + Schema schema = new Schema(Types.NestedField.required(1, "p", Types.IntegerType.get())); + PartitionSpec oldSpec = PartitionSpec.builderFor(schema).identity("p").build(); + PartitionData partitionData = new PartitionData(oldSpec.partitionType()); + partitionData.set(0, 10); + int partitionFieldId = oldSpec.fields().get(0).fieldId(); + List outputPartitionFields = Collections.singletonList( + Types.NestedField.optional(partitionFieldId, "p2", Types.IntegerType.get())); + + Method method = IcebergScanNode.class.getDeclaredMethod("getPartitionDataObjectJson", + PartitionData.class, PartitionSpec.class, List.class); + method.setAccessible(true); + + Assert.assertEquals("{\"p2\":10}", method.invoke(node, partitionData, oldSpec, outputPartitionFields)); + } + + @Test + public void testRejectBinaryPartitionValueWithoutBinarySafeTransport() throws Exception { + assertUnsupportedPositionDeletesPartitionValue( + Types.BinaryType.get(), ByteBuffer.wrap(new byte[] {0, (byte) 0xff}), false, "binary"); + assertUnsupportedPositionDeletesPartitionValue( + Types.FixedType.ofLength(2), ByteBuffer.wrap(new byte[] {0, (byte) 0xff}), false, "fixed[2]"); + } + + @Test + public void testRejectUuidPartitionValueWhenMappedToVarbinary() throws Exception { + assertUnsupportedPositionDeletesPartitionValue( + Types.UUIDType.get(), UUID.fromString("123e4567-e89b-12d3-a456-426614174000"), true, "uuid"); + } + + private void assertUnsupportedPositionDeletesPartitionValue( + org.apache.iceberg.types.Type type, Object value, boolean enableMappingVarbinary, + String expectedType) throws Exception { + TestIcebergScanNode node = new TestIcebergScanNode(new SessionVariable(), enableMappingVarbinary); + Schema schema = new Schema(Types.NestedField.required(1, "p", type)); + PartitionSpec spec = PartitionSpec.builderFor(schema).identity("p").build(); + PartitionData partitionData = new PartitionData(spec.partitionType()); + partitionData.set(0, value); + + Method method = IcebergScanNode.class.getDeclaredMethod("getPartitionDataObjectJson", + PartitionData.class, PartitionSpec.class, List.class); + method.setAccessible(true); + try { + method.invoke(node, partitionData, spec, spec.partitionType().fields()); + Assert.fail("Binary partition values must not be silently materialized as NULL"); + } catch (InvocationTargetException e) { + Assert.assertTrue(e.getCause() instanceof UserException); + Assert.assertTrue(e.getCause().getMessage().contains("partition field 'p'")); + Assert.assertTrue(e.getCause().getMessage().contains(expectedType)); + } + } + + @Test + public void testRejectUnsupportedPositionDeleteFileFormat() throws Exception { + TestIcebergScanNode node = new TestIcebergScanNode(new SessionVariable()); + Method method = IcebergScanNode.class.getDeclaredMethod( + "getNativePositionDeleteFileFormat", FileFormat.class); + method.setAccessible(true); + + try { + method.invoke(node, FileFormat.AVRO); + Assert.fail("AVRO position delete files should be rejected explicitly"); + } catch (InvocationTargetException e) { + Assert.assertTrue(e.getCause() instanceof UnsupportedOperationException); + Assert.assertEquals("Unsupported Iceberg position delete file format: AVRO", + e.getCause().getMessage()); + } + } + + @Test + public void testRejectSmoothUpgradeSourceBackendForPositionDeletes() throws Exception { + Backend currentBackend = Mockito.mock(Backend.class); + Mockito.when(currentBackend.isSmoothUpgradeSrc()).thenReturn(false); + IcebergScanNode.checkPositionDeletesBackendCompatibility(Collections.singletonList(currentBackend)); + + Backend smoothUpgradeSource = Mockito.mock(Backend.class); + Mockito.when(smoothUpgradeSource.isSmoothUpgradeSrc()).thenReturn(true); + Mockito.when(smoothUpgradeSource.getId()).thenReturn(10001L); + List backends = new ArrayList<>(); + backends.add(currentBackend); + backends.add(smoothUpgradeSource); + + try { + IcebergScanNode.checkPositionDeletesBackendCompatibility(backends); + Assert.fail("smooth upgrade source backend should reject native position_deletes planning"); + } catch (UserException e) { + Assert.assertTrue(e.getMessage().contains("backend 10001 is a smooth upgrade source")); + } + } } diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/systable/IcebergSysTableResolverTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/systable/IcebergSysTableResolverTest.java index 097d8111fe4c90..ef7380a6a7f708 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/datasource/systable/IcebergSysTableResolverTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/systable/IcebergSysTableResolverTest.java @@ -21,7 +21,6 @@ import org.apache.doris.datasource.iceberg.IcebergExternalDatabase; import org.apache.doris.datasource.iceberg.IcebergExternalTable; import org.apache.doris.datasource.iceberg.IcebergSysExternalTable; -import org.apache.doris.nereids.exceptions.AnalysisException; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; @@ -34,8 +33,8 @@ public class IcebergSysTableResolverTest { private IcebergExternalDatabase db = Mockito.mock(IcebergExternalDatabase.class); @Test - public void testSupportedSysTablesExcludePositionDeletes() { - Assertions.assertFalse(IcebergSysTable.SUPPORTED_SYS_TABLES.containsKey(IcebergSysTable.POSITION_DELETES)); + public void testSupportedSysTablesIncludePositionDeletes() { + Assertions.assertTrue(IcebergSysTable.SUPPORTED_SYS_TABLES.containsKey("position_deletes")); } @Test @@ -58,10 +57,14 @@ public void testResolveForPlanAndDescribeUseNativePath() throws Exception { } @Test - public void testPositionDeletesKeepsUnsupportedError() throws Exception { + public void testPositionDeletesUseNativePath() throws Exception { IcebergExternalTable sourceTable = newIcebergTable(); - Assertions.assertThrows(AnalysisException.class, () -> - SysTableResolver.resolveForPlan(sourceTable, "test_ctl", "test_db", "tbl$position_deletes")); + Optional plan = SysTableResolver.resolveForPlan( + sourceTable, "test_ctl", "test_db", "tbl$position_deletes"); + Assertions.assertTrue(plan.isPresent()); + Assertions.assertTrue(plan.get().isNative()); + Assertions.assertTrue(plan.get().getSysExternalTable() instanceof IcebergSysExternalTable); + Assertions.assertEquals("tbl$position_deletes", plan.get().getSysExternalTable().getName()); } private IcebergExternalTable newIcebergTable() throws Exception { diff --git a/gensrc/thrift/PlanNodes.thrift b/gensrc/thrift/PlanNodes.thrift index c2dd42bd522a16..b546649933c239 100644 --- a/gensrc/thrift/PlanNodes.thrift +++ b/gensrc/thrift/PlanNodes.thrift @@ -324,6 +324,10 @@ struct TIcebergDeleteFileDesc { 6: optional i64 content_offset; 7: optional i64 content_size_in_bytes; 8: optional TFileFormatType file_format; + // Original Iceberg delete file path before Doris storage path normalization. + 9: optional string original_path; + // Referenced data file path. Required to materialize rows from deletion vectors. + 10: optional string referenced_data_file_path; } struct TIcebergFileDesc { diff --git a/regression-test/data/external_table_p0/iceberg/test_iceberg_position_deletes_sys_table.out b/regression-test/data/external_table_p0/iceberg/test_iceberg_position_deletes_sys_table.out new file mode 100644 index 00000000000000..d7f3395f293c2a --- /dev/null +++ b/regression-test/data/external_table_p0/iceberg/test_iceberg_position_deletes_sys_table.out @@ -0,0 +1,4 @@ +-- This file is automatically generated. You should know what you did if you want to edit this +-- !partition_rename_field -- +0 10 +1 10 diff --git a/regression-test/suites/external_table_p0/iceberg/test_iceberg_position_deletes_sys_table.groovy b/regression-test/suites/external_table_p0/iceberg/test_iceberg_position_deletes_sys_table.groovy new file mode 100644 index 00000000000000..dae16a9f3c7b66 --- /dev/null +++ b/regression-test/suites/external_table_p0/iceberg/test_iceberg_position_deletes_sys_table.groovy @@ -0,0 +1,659 @@ +// 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. + +suite("test_iceberg_position_deletes_sys_table", "p0,external") { + String enabled = context.config.otherConfigs.get("enableIcebergTest") + if (enabled == null || !enabled.equalsIgnoreCase("true")) { + logger.info("disable iceberg test.") + return + } + + String restPort = context.config.otherConfigs.get("iceberg_rest_uri_port") + String minioPort = context.config.otherConfigs.get("iceberg_minio_port") + String externalEnvIp = context.config.otherConfigs.get("externalEnvIp") + String catalogName = "test_iceberg_position_deletes_sys_table" + String dbName = "position_deletes_sys_table_db" + + sql """drop catalog if exists ${catalogName}""" + sql """ + CREATE CATALOG ${catalogName} PROPERTIES ( + 'type'='iceberg', + 'iceberg.catalog.type'='rest', + 'uri' = 'http://${externalEnvIp}:${restPort}', + "s3.access_key" = "admin", + "s3.secret_key" = "password", + "s3.endpoint" = "http://${externalEnvIp}:${minioPort}", + "s3.region" = "us-east-1", + "meta.cache.iceberg.table.ttl-second" = "0", + "meta.cache.iceberg.schema.ttl-second" = "0" + );""" + + spark_iceberg_multi """ + SET spark.sql.shuffle.partitions=1; + CREATE DATABASE IF NOT EXISTS demo.${dbName}; + DROP TABLE IF EXISTS demo.${dbName}.pd_unpartitioned; + CREATE TABLE demo.${dbName}.pd_unpartitioned ( + id INT, + name STRING + ) USING iceberg + TBLPROPERTIES ( + 'format-version'='2', + 'write.delete.mode'='merge-on-read', + 'write.update.mode'='merge-on-read', + 'write.merge.mode'='merge-on-read', + 'write.distribution-mode'='none', + 'write.target-file-size-bytes'='134217728' + ); + INSERT INTO demo.${dbName}.pd_unpartitioned + SELECT /*+ COALESCE(1) */ id, name FROM VALUES + (1, 'a'), (2, 'b'), (3, 'c'), (4, 'd') AS t(id, name); + DELETE FROM demo.${dbName}.pd_unpartitioned WHERE id = 2; + DROP TABLE IF EXISTS demo.${dbName}.pd_orc_unpartitioned; + CREATE TABLE demo.${dbName}.pd_orc_unpartitioned ( + id INT, + name STRING + ) USING iceberg + TBLPROPERTIES ( + 'format-version'='2', + 'write.format.default'='orc', + 'write.delete.format.default'='orc', + 'write.delete.mode'='merge-on-read', + 'write.update.mode'='merge-on-read', + 'write.merge.mode'='merge-on-read', + 'write.distribution-mode'='none', + 'write.target-file-size-bytes'='134217728' + ); + INSERT INTO demo.${dbName}.pd_orc_unpartitioned + SELECT /*+ COALESCE(1) */ id, name FROM VALUES + (1, 'a'), (2, 'b'), (3, 'c') AS t(id, name); + DELETE FROM demo.${dbName}.pd_orc_unpartitioned WHERE id = 2; + DROP TABLE IF EXISTS demo.${dbName}.pd_partitioned; + CREATE TABLE demo.${dbName}.pd_partitioned ( + id INT, + name STRING, + dt STRING + ) USING iceberg + PARTITIONED BY (dt) + TBLPROPERTIES ( + 'format-version'='2', + 'write.delete.mode'='merge-on-read', + 'write.update.mode'='merge-on-read', + 'write.merge.mode'='merge-on-read', + 'write.distribution-mode'='none', + 'write.target-file-size-bytes'='134217728' + ); + INSERT INTO demo.${dbName}.pd_partitioned + SELECT /*+ COALESCE(1) */ id, name, dt FROM VALUES + (1, 'a', '2026-06-26'), + (2, 'b', '2026-06-26'), + (3, 'c', '2026-06-27'), + (4, 'd', '2026-06-27') AS t(id, name, dt); + DELETE FROM demo.${dbName}.pd_partitioned WHERE id = 2; + DROP TABLE IF EXISTS demo.${dbName}.pd_int_partitioned; + CREATE TABLE demo.${dbName}.pd_int_partitioned ( + id INT, + name STRING, + p INT + ) USING iceberg + PARTITIONED BY (p) + TBLPROPERTIES ( + 'format-version'='2', + 'write.delete.mode'='merge-on-read', + 'write.update.mode'='merge-on-read', + 'write.merge.mode'='merge-on-read', + 'write.distribution-mode'='none', + 'write.target-file-size-bytes'='134217728' + ); + INSERT INTO demo.${dbName}.pd_int_partitioned + SELECT /*+ COALESCE(1) */ id, name, p FROM VALUES + (1, 'a', 10), + (2, 'b', 10), + (3, 'c', 20) AS t(id, name, p); + DELETE FROM demo.${dbName}.pd_int_partitioned WHERE id = 2; + DROP TABLE IF EXISTS demo.${dbName}.pd_date_partitioned; + CREATE TABLE demo.${dbName}.pd_date_partitioned ( + id INT, + name STRING, + dt DATE + ) USING iceberg + PARTITIONED BY (dt) + TBLPROPERTIES ( + 'format-version'='2', + 'write.delete.mode'='merge-on-read', + 'write.update.mode'='merge-on-read', + 'write.merge.mode'='merge-on-read', + 'write.distribution-mode'='none', + 'write.target-file-size-bytes'='134217728' + ); + INSERT INTO demo.${dbName}.pd_date_partitioned + SELECT /*+ COALESCE(1) */ id, name, CAST(dt AS DATE) FROM VALUES + (1, 'a', '2026-01-02'), + (2, 'b', '2026-01-02'), + (3, 'c', '2026-03-04') AS t(id, name, dt); + DELETE FROM demo.${dbName}.pd_date_partitioned WHERE id = 2; + DROP TABLE IF EXISTS demo.${dbName}.pd_partition_evolution; + CREATE TABLE demo.${dbName}.pd_partition_evolution ( + id INT, + name STRING, + p INT + ) USING iceberg + PARTITIONED BY (p) + TBLPROPERTIES ( + 'format-version'='2', + 'write.delete.mode'='merge-on-read', + 'write.update.mode'='merge-on-read', + 'write.merge.mode'='merge-on-read', + 'write.distribution-mode'='none', + 'write.target-file-size-bytes'='134217728' + ); + INSERT INTO demo.${dbName}.pd_partition_evolution + SELECT /*+ COALESCE(1) */ id, name, p FROM VALUES + (1, 'a', 10), + (2, 'b', 10), + (3, 'c', 20), + (4, 'd', 20) AS t(id, name, p); + DELETE FROM demo.${dbName}.pd_partition_evolution WHERE id = 2; + ALTER TABLE demo.${dbName}.pd_partition_evolution ADD PARTITION FIELD bucket(1, id); + INSERT INTO demo.${dbName}.pd_partition_evolution + SELECT /*+ COALESCE(1) */ id, name, p FROM VALUES + (5, 'e', 10), + (6, 'f', 10) AS t(id, name, p); + DELETE FROM demo.${dbName}.pd_partition_evolution WHERE id = 6; + DROP TABLE IF EXISTS demo.${dbName}.pd_partition_rename; + CREATE TABLE demo.${dbName}.pd_partition_rename ( + id INT, + name STRING, + p INT + ) USING iceberg + PARTITIONED BY (p) + TBLPROPERTIES ( + 'format-version'='2', + 'write.delete.mode'='merge-on-read', + 'write.update.mode'='merge-on-read', + 'write.merge.mode'='merge-on-read', + 'write.distribution-mode'='none', + 'write.target-file-size-bytes'='134217728' + ); + INSERT INTO demo.${dbName}.pd_partition_rename + SELECT /*+ COALESCE(1) */ id, name, p FROM VALUES + (1, 'a', 10), + (2, 'b', 10), + (3, 'c', 20), + (4, 'd', 20) AS t(id, name, p); + DELETE FROM demo.${dbName}.pd_partition_rename WHERE id = 2; + -- Replacing an identity transform with the same transform and an alias renames the + -- partition field while preserving its Iceberg field ID. + ALTER TABLE demo.${dbName}.pd_partition_rename REPLACE PARTITION FIELD p WITH p AS p2; + INSERT INTO demo.${dbName}.pd_partition_rename + SELECT /*+ COALESCE(1) */ id, name, p FROM VALUES + (5, 'e', 10), + (6, 'f', 10) AS t(id, name, p); + DELETE FROM demo.${dbName}.pd_partition_rename WHERE id = 6; + DROP TABLE IF EXISTS demo.${dbName}.pd_v3_unpartitioned; + CREATE TABLE demo.${dbName}.pd_v3_unpartitioned ( + id INT, + name STRING + ) USING iceberg + TBLPROPERTIES ( + 'format-version'='3', + 'write.delete.mode'='merge-on-read', + 'write.update.mode'='merge-on-read', + 'write.merge.mode'='merge-on-read', + 'write.distribution-mode'='none', + 'write.target-file-size-bytes'='134217728' + ); + INSERT INTO demo.${dbName}.pd_v3_unpartitioned + SELECT /*+ COALESCE(1) */ id, name FROM VALUES + (1, 'a'), (2, 'b'), (3, 'c') AS t(id, name); + DELETE FROM demo.${dbName}.pd_v3_unpartitioned WHERE id = 2; + DROP TABLE IF EXISTS demo.${dbName}.pd_v3_partitioned; + CREATE TABLE demo.${dbName}.pd_v3_partitioned ( + id INT, + name STRING, + dt STRING + ) USING iceberg + PARTITIONED BY (dt) + TBLPROPERTIES ( + 'format-version'='3', + 'write.delete.mode'='merge-on-read', + 'write.update.mode'='merge-on-read', + 'write.merge.mode'='merge-on-read', + 'write.distribution-mode'='none', + 'write.target-file-size-bytes'='134217728' + ); + INSERT INTO demo.${dbName}.pd_v3_partitioned + SELECT /*+ COALESCE(1) */ id, name, dt FROM VALUES + (1, 'a', '2026-07-01'), + (2, 'b', '2026-07-01'), + (3, 'c', '2026-07-02') AS t(id, name, dt); + DELETE FROM demo.${dbName}.pd_v3_partitioned WHERE id = 2; + DROP TABLE IF EXISTS demo.${dbName}.pd_schema_time_travel; + CREATE TABLE demo.${dbName}.pd_schema_time_travel ( + id INT, + name STRING + ) USING iceberg + TBLPROPERTIES ( + 'format-version'='2', + 'write.delete.mode'='merge-on-read', + 'write.update.mode'='merge-on-read', + 'write.merge.mode'='merge-on-read', + 'write.distribution-mode'='none', + 'write.target-file-size-bytes'='134217728' + ); + INSERT INTO demo.${dbName}.pd_schema_time_travel + SELECT /*+ COALESCE(1) */ id, name FROM VALUES + (1, 'a'), (2, 'b'), (3, 'c') AS t(id, name); + DELETE FROM demo.${dbName}.pd_schema_time_travel WHERE id = 2; + ALTER TABLE demo.${dbName}.pd_schema_time_travel ADD COLUMN note STRING; + INSERT INTO demo.${dbName}.pd_schema_time_travel + SELECT /*+ COALESCE(1) */ id, name, note FROM VALUES + (4, 'd', 'new'), (5, 'e', 'new') AS t(id, name, note); + DELETE FROM demo.${dbName}.pd_schema_time_travel WHERE id = 4; + DROP TABLE IF EXISTS demo.${dbName}.pd_no_deletes; + CREATE TABLE demo.${dbName}.pd_no_deletes ( + id INT, + name STRING + ) USING iceberg + TBLPROPERTIES ('format-version'='2'); + INSERT INTO demo.${dbName}.pd_no_deletes VALUES (1, 'a'), (2, 'b'); + """ + + sql """switch ${catalogName}""" + sql """use ${dbName}""" + + def assertPositionDeletesSchema = { String tableName, boolean partitioned, List extraColumns = [] -> + List> descRows = sql """desc ${tableName}\$position_deletes""" + List columns = descRows.collect { it[0].toString() } + List expectedColumns = ["file_path", "pos", "row"] + if (partitioned) { + expectedColumns.add("partition") + } + expectedColumns.addAll(["spec_id", "delete_file_path"]) + expectedColumns.addAll(extraColumns) + assertEquals(expectedColumns, columns) + } + + def countRows = { String query -> + List> rows = sql query + assertEquals(1, rows.size()) + return ((Number) rows[0][0]).longValue() + } + + def assertSparkDorisPositionDeletes = { String tableName, List columns -> + String columnList = columns.join(", ") + String orderBy = "file_path, pos, delete_file_path" + List> sparkRows = spark_iceberg """ + select ${columnList} + from demo.${dbName}.${tableName}.position_deletes + order by ${orderBy} + """ + List> dorisRows = sql """ + select ${columnList} + from ${tableName}\$position_deletes + order by ${orderBy} + """ + assertSparkDorisResultEquals(sparkRows, dorisRows) + } + + def assertSparkDorisTableRows = { String tableName, List> expectedRows -> + spark_iceberg """REFRESH TABLE demo.${dbName}.${tableName}""" + List> sparkRows = spark_iceberg """ + select id, name, dt + from demo.${dbName}.${tableName} + order by id + """ + List> dorisRows = sql """ + select id, name, dt + from ${tableName} + order by id + """ + assertSparkDorisResultEquals(sparkRows, dorisRows) + assertEquals(expectedRows, dorisRows) + } + + List v3ExtraColumns = ["content_offset", "content_size_in_bytes"] + List commonCompareColumns = ["file_path", "pos", "spec_id", "delete_file_path"] + List v3CompareColumns = commonCompareColumns + v3ExtraColumns + + assertPositionDeletesSchema("pd_unpartitioned", false) + long unpartitionedCount = countRows("""select count(*) from pd_unpartitioned\$position_deletes""") + assertTrue(unpartitionedCount > 0) + assertSparkDorisPositionDeletes("pd_unpartitioned", commonCompareColumns) + assertEquals(unpartitionedCount, countRows("""select count(pos) from pd_unpartitioned\$position_deletes""")) + assertEquals(unpartitionedCount, (long) sql( + """select * from pd_unpartitioned\$position_deletes""").size()) + assertEquals(unpartitionedCount, (long) sql( + """select pos from pd_unpartitioned\$position_deletes""").size()) + assertEquals(unpartitionedCount, (long) sql( + """select delete_file_path, pos from pd_unpartitioned\$position_deletes""").size()) + assertEquals(unpartitionedCount, (long) sql( + """select file_path, pos, delete_file_path from pd_unpartitioned\$position_deletes where pos >= 0""").size()) + assertEquals(unpartitionedCount, countRows( + """select count(*) from pd_unpartitioned\$position_deletes where spec_id = 0""")) + assertEquals(unpartitionedCount, countRows( + """select count(*) from pd_unpartitioned\$position_deletes where delete_file_path is not null""")) + assertEquals(0L, countRows("""select count(*) from pd_unpartitioned\$position_deletes where pos < 0""")) + List> unpartitionedRows = sql """select `row` from pd_unpartitioned\$position_deletes""" + assertEquals(unpartitionedCount, (long) unpartitionedRows.size()) + + assertPositionDeletesSchema("pd_orc_unpartitioned", false) + long orcUnpartitionedCount = countRows( + """select count(*) from pd_orc_unpartitioned\$position_deletes""") + assertTrue(orcUnpartitionedCount > 0) + assertSparkDorisPositionDeletes("pd_orc_unpartitioned", commonCompareColumns) + List> orcUnpartitionedRows = sql """select `row` from pd_orc_unpartitioned\$position_deletes""" + assertEquals(orcUnpartitionedCount, (long) orcUnpartitionedRows.size()) + assertTrue(orcUnpartitionedRows.every { it[0] == null }) + try { + sql """set file_split_size=1""" + assertSparkDorisPositionDeletes("pd_unpartitioned", commonCompareColumns) + assertEquals(unpartitionedCount, countRows( + """select count(*) from pd_unpartitioned\$position_deletes where pos >= 0""")) + assertEquals(orcUnpartitionedCount, countRows( + """select count(*) from pd_orc_unpartitioned\$position_deletes where pos >= 0""")) + } finally { + sql """unset variable file_split_size""" + } + assertEquals([[1, "a"], [3, "c"], [4, "d"]], sql("""select * from pd_unpartitioned order by id""")) + + assertPositionDeletesSchema("pd_partitioned", true) + long partitionedCount = countRows("""select count(*) from pd_partitioned\$position_deletes""") + assertTrue(partitionedCount > 0) + assertSparkDorisPositionDeletes("pd_partitioned", commonCompareColumns) + assertEquals(partitionedCount, countRows("""select count(pos) from pd_partitioned\$position_deletes""")) + assertEquals(partitionedCount, (long) sql( + """select * from pd_partitioned\$position_deletes""").size()) + assertEquals(partitionedCount, (long) sql( + """select file_path, pos, delete_file_path from pd_partitioned\$position_deletes where pos >= 0""").size()) + assertEquals(partitionedCount, countRows( + """select count(*) from pd_partitioned\$position_deletes where `partition` is not null""")) + assertEquals(0L, countRows("""select count(*) from pd_partitioned\$position_deletes where pos < 0""")) + List> partitionedRows = sql """select `row`, `partition` from pd_partitioned\$position_deletes""" + assertEquals(partitionedCount, (long) partitionedRows.size()) + assertTrue(partitionedRows.every { it[0] == null && it[1] != null }) + assertEquals([[1, "a", "2026-06-26"], [3, "c", "2026-06-27"], [4, "d", "2026-06-27"]], + sql("""select * from pd_partitioned order by id""")) + + assertPositionDeletesSchema("pd_int_partitioned", true) + long intPartitionedCount = countRows("""select count(*) from pd_int_partitioned\$position_deletes""") + assertTrue(intPartitionedCount > 0) + assertEquals(intPartitionedCount, countRows("""select count(pos) from pd_int_partitioned\$position_deletes""")) + assertEquals(intPartitionedCount, countRows( + """select count(*) from pd_int_partitioned\$position_deletes where `partition` is not null""")) + List> intPartitionedRows = sql """select `row`, `partition` from pd_int_partitioned\$position_deletes""" + assertEquals(intPartitionedCount, (long) intPartitionedRows.size()) + assertTrue(intPartitionedRows.every { it[0] == null && it[1] != null }) + assertTrue(intPartitionedRows.every { + String partitionValue = it[1].toString() + partitionValue.contains("\"p\":10") && !partitionValue.contains("\"p\":\"10\"") + }) + assertEquals([[1, "a", 10], [3, "c", 20]], sql("""select * from pd_int_partitioned order by id""")) + + // DATE-typed partition column. Iceberg serializes the partition value to an ISO date string + // (`2026-01-02`) which the BE parses back into the metadata table `partition` struct DATE field. + // Verifies the typed partition round-trip is not limited to STRING/INT columns. + assertPositionDeletesSchema("pd_date_partitioned", true) + long datePartitionedCount = countRows("""select count(*) from pd_date_partitioned\$position_deletes""") + assertEquals(1L, datePartitionedCount) + assertSparkDorisPositionDeletes("pd_date_partitioned", commonCompareColumns) + assertEquals(datePartitionedCount, countRows( + """select count(*) from pd_date_partitioned\$position_deletes where `partition` is not null""")) + List> datePartitionedRows = sql """ + select `row`, cast(`partition` as string) from pd_date_partitioned\$position_deletes + """ + assertEquals(datePartitionedCount, (long) datePartitionedRows.size()) + assertTrue(datePartitionedRows.every { it[0] == null && it[1] != null }) + // The deleted row (id=2) lives in partition dt=2026-01-02, so the partition value must render + // that exact date rather than a null / days-since-epoch integer / mangled string. + assertTrue(datePartitionedRows.every { it[1].toString().contains("2026-01-02") }) + assertEquals([[1, "a", "2026-01-02"], [3, "c", "2026-03-04"]], + sql("""select id, name, cast(dt as string) from pd_date_partitioned order by id""")) + + assertPositionDeletesSchema("pd_partition_evolution", true) + long partitionEvolutionCount = countRows("""select count(*) from pd_partition_evolution\$position_deletes""") + assertEquals(2L, partitionEvolutionCount) + assertSparkDorisPositionDeletes("pd_partition_evolution", commonCompareColumns) + assertEquals(partitionEvolutionCount, countRows( + """select count(pos) from pd_partition_evolution\$position_deletes""")) + assertEquals(partitionEvolutionCount, countRows( + """select count(*) from pd_partition_evolution\$position_deletes where `partition` is not null""")) + List> partitionEvolutionRows = sql """ + select spec_id, cast(`partition` as string) + from pd_partition_evolution\$position_deletes order by spec_id + """ + assertEquals(2, partitionEvolutionRows.size()) + assertEquals(0, ((Number) partitionEvolutionRows[0][0]).intValue()) + assertEquals(1, ((Number) partitionEvolutionRows[1][0]).intValue()) + String oldSpecPartition = partitionEvolutionRows[0][1].toString() + String newSpecPartition = partitionEvolutionRows[1][1].toString() + assertTrue(oldSpecPartition.contains("\"p\":10")) + assertTrue(oldSpecPartition.contains(":null")) + assertTrue(newSpecPartition.contains("\"p\":10")) + assertFalse(newSpecPartition.contains(":null")) + assertEquals([[1, "a", 10], [3, "c", 20], [4, "d", 20], [5, "e", 10]], + sql("""select * from pd_partition_evolution order by id""")) + + assertPositionDeletesSchema("pd_partition_rename", true) + long partitionRenameCount = countRows("""select count(*) from pd_partition_rename\$position_deletes""") + order_qt_partition_rename_field """ + select spec_id, struct_element(`partition`, 'p2') + from pd_partition_rename\$position_deletes + order by spec_id + """ + List> sparkPartitionRenameRows = spark_iceberg """ + select spec_id, partition.p2 + from demo.${dbName}.pd_partition_rename.position_deletes + order by spec_id + """ + List> dorisPartitionRenameRows = sql """ + select spec_id, struct_element(`partition`, 'p2') + from pd_partition_rename\$position_deletes + order by spec_id + """ + assertSparkDorisResultEquals(sparkPartitionRenameRows, dorisPartitionRenameRows) + assertEquals(partitionRenameCount, countRows(""" + select count(*) from pd_partition_rename\$position_deletes + where struct_element(`partition`, 'p2') = 10 + """)) + + assertPositionDeletesSchema("pd_v3_unpartitioned", false, v3ExtraColumns) + long v3UnpartitionedCount = countRows("""select count(*) from pd_v3_unpartitioned\$position_deletes""") + assertEquals(1L, v3UnpartitionedCount) + assertSparkDorisPositionDeletes("pd_v3_unpartitioned", v3CompareColumns) + assertEquals(v3UnpartitionedCount, countRows( + """select count(pos) from pd_v3_unpartitioned\$position_deletes""")) + assertEquals(v3UnpartitionedCount, countRows( + """select count(content_offset) from pd_v3_unpartitioned\$position_deletes""")) + assertEquals(v3UnpartitionedCount, countRows( + """select count(content_size_in_bytes) from pd_v3_unpartitioned\$position_deletes""")) + assertEquals(v3UnpartitionedCount, (long) sql( + """select file_path, pos, delete_file_path, content_offset, content_size_in_bytes + from pd_v3_unpartitioned\$position_deletes""").size()) + assertEquals(v3UnpartitionedCount, countRows( + """select count(*) from pd_v3_unpartitioned\$position_deletes + where content_offset >= 0 and content_size_in_bytes > 0""")) + assertEquals(v3UnpartitionedCount, countRows( + """select count(*) from pd_v3_unpartitioned\$position_deletes where spec_id = 0""")) + assertEquals(v3UnpartitionedCount, countRows( + """select count(*) from pd_v3_unpartitioned\$position_deletes where delete_file_path is not null""")) + List> v3UnpartitionedRows = sql """select `row` from pd_v3_unpartitioned\$position_deletes""" + assertEquals(v3UnpartitionedCount, (long) v3UnpartitionedRows.size()) + assertTrue(v3UnpartitionedRows.every { it[0] == null }) + assertEquals([[1, "a"], [3, "c"]], sql("""select * from pd_v3_unpartitioned order by id""")) + + assertPositionDeletesSchema("pd_v3_partitioned", true, v3ExtraColumns) + long v3PartitionedCount = countRows("""select count(*) from pd_v3_partitioned\$position_deletes""") + assertEquals(1L, v3PartitionedCount) + assertSparkDorisPositionDeletes("pd_v3_partitioned", v3CompareColumns) + assertEquals(v3PartitionedCount, countRows( + """select count(pos) from pd_v3_partitioned\$position_deletes""")) + assertEquals(v3PartitionedCount, countRows( + """select count(content_offset) from pd_v3_partitioned\$position_deletes""")) + assertEquals(v3PartitionedCount, countRows( + """select count(content_size_in_bytes) from pd_v3_partitioned\$position_deletes""")) + assertEquals(v3PartitionedCount, (long) sql( + """select file_path, pos, delete_file_path, content_offset, content_size_in_bytes + from pd_v3_partitioned\$position_deletes""").size()) + assertEquals(v3PartitionedCount, countRows( + """select count(*) from pd_v3_partitioned\$position_deletes + where content_offset >= 0 and content_size_in_bytes > 0""")) + assertEquals(v3PartitionedCount, countRows( + """select count(*) from pd_v3_partitioned\$position_deletes where `partition` is not null""")) + assertEquals(v3PartitionedCount, countRows( + """select count(*) from pd_v3_partitioned\$position_deletes where spec_id = 0""")) + assertEquals(v3PartitionedCount, countRows( + """select count(*) from pd_v3_partitioned\$position_deletes where delete_file_path is not null""")) + List> v3PartitionedRows = sql """select `row`, `partition` from pd_v3_partitioned\$position_deletes""" + assertEquals(v3PartitionedCount, (long) v3PartitionedRows.size()) + assertTrue(v3PartitionedRows.every { it[0] == null && it[1] != null }) + assertEquals([[1, "a", "2026-07-01"], [3, "c", "2026-07-02"]], + sql("""select * from pd_v3_partitioned order by id""")) + + assertPositionDeletesSchema("pd_schema_time_travel", false) + long schemaChangeCount = countRows("""select count(*) from pd_schema_time_travel\$position_deletes""") + assertEquals(2L, schemaChangeCount) + assertEquals(schemaChangeCount, countRows( + """select count(*) from pd_schema_time_travel\$position_deletes where pos >= 0""")) + assertEquals(schemaChangeCount, countRows( + """select count(*) from pd_schema_time_travel\$position_deletes where spec_id = 0""")) + assertEquals(schemaChangeCount, countRows( + """select count(*) from pd_schema_time_travel\$position_deletes where delete_file_path is not null""")) + assertEquals(0L, countRows("""select count(*) from pd_schema_time_travel\$position_deletes where pos < 0""")) + List> schemaChangeRows = sql """select `row` from pd_schema_time_travel\$position_deletes""" + assertEquals(schemaChangeCount, (long) schemaChangeRows.size()) + assertTrue(schemaChangeRows.every { it[0] == null }) + assertEquals([[1, "a", null], [3, "c", null], [5, "e", "new"]], + sql("""select * from pd_schema_time_travel order by id""")) + + List> schemaChangeSnapshots = sql """ + select snapshot_id from pd_schema_time_travel\$snapshots order by committed_at + """ + assertEquals(4, schemaChangeSnapshots.size()) + Object afterInsertSnapshot = schemaChangeSnapshots.get(0)[0] + Object afterFirstDeleteSnapshot = schemaChangeSnapshots.get(1)[0] + Object afterSecondDeleteSnapshot = schemaChangeSnapshots.get(3)[0] + assertEquals(0L, countRows(""" + select count(*) from pd_schema_time_travel\$position_deletes + for version as of ${afterInsertSnapshot} + """)) + assertEquals(1L, countRows(""" + select count(*) from pd_schema_time_travel\$position_deletes + for version as of ${afterFirstDeleteSnapshot} + """)) + assertEquals(2L, countRows(""" + select count(*) from pd_schema_time_travel\$position_deletes + for version as of ${afterSecondDeleteSnapshot} + """)) + assertEquals([[1, "a"], [2, "b"], [3, "c"]], + sql("""select * from pd_schema_time_travel for version as of ${afterInsertSnapshot} order by id""")) + assertEquals([[1, "a"], [3, "c"]], + sql("""select * from pd_schema_time_travel for version as of ${afterFirstDeleteSnapshot} order by id""")) + assertEquals([[1, "a", null], [3, "c", null], [5, "e", "new"]], + sql("""select * from pd_schema_time_travel for version as of ${afterSecondDeleteSnapshot} order by id""")) + + assertPositionDeletesSchema("pd_no_deletes", false) + assertSparkDorisPositionDeletes("pd_no_deletes", commonCompareColumns) + assertEquals(0L, countRows("""select count(*) from pd_no_deletes\$position_deletes""")) + + def assertPositionDeletesScannerPath = { boolean enableFileScannerV2 -> + sql """set enable_file_scanner_v2 = ${enableFileScannerV2}""" + assertEquals(unpartitionedCount, countRows( + """select count(*) from pd_unpartitioned\$position_deletes where pos >= 0""")) + assertEquals(partitionedCount, countRows( + """select count(*) from pd_partitioned\$position_deletes where `partition` is not null""")) + assertEquals(partitionRenameCount, countRows( + """select count(*) from pd_partition_rename\$position_deletes + where struct_element(`partition`, 'p2') is not null""")) + assertEquals(v3UnpartitionedCount, countRows( + """select count(*) from pd_v3_unpartitioned\$position_deletes + where content_offset >= 0 and content_size_in_bytes > 0""")) + assertSparkDorisPositionDeletes("pd_unpartitioned", commonCompareColumns) + assertSparkDorisPositionDeletes("pd_orc_unpartitioned", commonCompareColumns) + assertSparkDorisPositionDeletes("pd_v3_unpartitioned", v3CompareColumns) + } + assertPositionDeletesScannerPath(false) + assertPositionDeletesScannerPath(true) + sql """set enable_file_scanner_v2 = true""" + + spark_iceberg_multi """ + DROP TABLE IF EXISTS demo.${dbName}.pd_doris_spark_interop; + CREATE TABLE demo.${dbName}.pd_doris_spark_interop ( + id INT, + name STRING, + dt STRING + ) USING iceberg + PARTITIONED BY (dt) + TBLPROPERTIES ( + 'format-version'='2', + 'write.delete.mode'='merge-on-read', + 'write.update.mode'='merge-on-read', + 'write.merge.mode'='merge-on-read', + 'write.distribution-mode'='none', + 'write.target-file-size-bytes'='134217728' + ); + INSERT INTO demo.${dbName}.pd_doris_spark_interop + SELECT /*+ COALESCE(1) */ id, name, dt FROM VALUES + (1, 'spark_a', '2026-07-03'), + (2, 'spark_b', '2026-07-03') AS t(id, name, dt); + """ + + assertSparkDorisTableRows("pd_doris_spark_interop", [ + [1, "spark_a", "2026-07-03"], + [2, "spark_b", "2026-07-03"] + ]) + + sql """insert into pd_doris_spark_interop values (3, 'doris_c', '2026-07-04')""" + assertSparkDorisTableRows("pd_doris_spark_interop", [ + [1, "spark_a", "2026-07-03"], + [2, "spark_b", "2026-07-03"], + [3, "doris_c", "2026-07-04"] + ]) + + spark_iceberg """ + INSERT INTO demo.${dbName}.pd_doris_spark_interop + SELECT /*+ COALESCE(1) */ id, name, dt FROM VALUES + (4, 'spark_d', '2026-07-04') AS t(id, name, dt) + """ + assertSparkDorisTableRows("pd_doris_spark_interop", [ + [1, "spark_a", "2026-07-03"], + [2, "spark_b", "2026-07-03"], + [3, "doris_c", "2026-07-04"], + [4, "spark_d", "2026-07-04"] + ]) + + sql """delete from pd_doris_spark_interop where id = 2""" + assertSparkDorisTableRows("pd_doris_spark_interop", [ + [1, "spark_a", "2026-07-03"], + [3, "doris_c", "2026-07-04"], + [4, "spark_d", "2026-07-04"] + ]) + + spark_iceberg """DELETE FROM demo.${dbName}.pd_doris_spark_interop WHERE id = 1""" + assertSparkDorisTableRows("pd_doris_spark_interop", [ + [3, "doris_c", "2026-07-04"], + [4, "spark_d", "2026-07-04"] + ]) + + assertPositionDeletesSchema("pd_doris_spark_interop", true) + long interopPositionDeleteCount = countRows( + """select count(*) from pd_doris_spark_interop\$position_deletes""") + assertEquals(2L, interopPositionDeleteCount) + assertEquals(interopPositionDeleteCount, countRows( + """select count(*) from pd_doris_spark_interop\$position_deletes + where `partition` is not null and delete_file_path is not null""")) + assertSparkDorisPositionDeletes("pd_doris_spark_interop", commonCompareColumns) +} diff --git a/regression-test/suites/external_table_p0/iceberg/test_iceberg_sys_table.groovy b/regression-test/suites/external_table_p0/iceberg/test_iceberg_sys_table.groovy index de4d582a83abf2..719d894253b448 100644 --- a/regression-test/suites/external_table_p0/iceberg/test_iceberg_sys_table.groovy +++ b/regression-test/suites/external_table_p0/iceberg/test_iceberg_sys_table.groovy @@ -262,7 +262,7 @@ suite("test_iceberg_sys_table", "p0,external") { systableName) } - def test_table_systables = { table -> + def test_table_systables = { table, boolean partitioned -> test_systable_entries(table, "entries") test_systable_entries(table, "all_entries") test_systable_files(table, "files") @@ -278,17 +278,21 @@ suite("test_iceberg_sys_table", "p0,external") { test_systable_manifests(table, "manifests") test_systable_manifests(table, "all_manifests") test_systable_partitions(table) - // TODO: these table will be supportted in future - // test_systable_position_deletes(table) - test { - sql """select * from ${table}\$position_deletes""" - exception "SysTable position_deletes is not supported yet" + List> positionDeleteDesc = sql """desc ${table}\$position_deletes""" + List positionDeleteColumns = positionDeleteDesc.collect { it[0].toString() } + List expectedPositionDeleteColumns = ["file_path", "pos", "row"] + if (partitioned) { + expectedPositionDeleteColumns.add("partition") } + expectedPositionDeleteColumns.addAll(["spec_id", "delete_file_path"]) + assertEquals(expectedPositionDeleteColumns, positionDeleteColumns) + List> positionDeleteCount = sql """select count(*) from ${table}\$position_deletes""" + assertEquals(1, positionDeleteCount.size()) } - test_table_systables("test_iceberg_systable_unpartitioned") - test_table_systables("test_iceberg_systable_partitioned") + test_table_systables("test_iceberg_systable_unpartitioned", false) + test_table_systables("test_iceberg_systable_partitioned", true) sql """drop table if exists test_iceberg_systable_tbl1;""" sql """create table test_iceberg_systable_tbl1 (id int);""" @@ -319,10 +323,17 @@ suite("test_iceberg_sys_table", "p0,external") { """ exception "denied" } + test { + sql """ + select file_path, pos from ${catalog_name}.${db_name}.test_iceberg_systable_tbl1\$position_deletes + """ + exception "denied" + } } sql """grant select_priv on ${catalog_name}.${db_name}.test_iceberg_systable_tbl1 to ${user}""" connect(user, "${pwd}", context.config.jdbcUrl) { sql """select committed_at, snapshot_id, parent_id, operation from ${catalog_name}.${db_name}.test_iceberg_systable_tbl1\$snapshots""" + sql """select file_path, pos from ${catalog_name}.${db_name}.test_iceberg_systable_tbl1\$position_deletes""" } try_sql("DROP USER ${user}")