github-actions[bot] commented on code in PR #65135: URL: https://github.com/apache/doris/pull/65135#discussion_r3568047943
########## be/src/format/table/iceberg_position_delete_sys_table_reader.cpp: ########## @@ -0,0 +1,582 @@ +// 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 <gen_cpp/ExternalTableSchema_types.h> +#include <gen_cpp/PlanNodes_types.h> + +#include <algorithm> +#include <memory> +#include <utility> + +#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<ColumnString>(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<ColumnInt64>(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 <typename ReadColumns> +std::shared_ptr<TableSchemaChangeHelper::StructNode> create_position_delete_root_node( + const ReadColumns& read_columns) { + auto root_node = std::make_shared<TableSchemaChangeHelper::StructNode>(); + 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<SlotDescriptor*>& file_slot_descs, RuntimeState* state, + RuntimeProfile* profile, const TFileRangeDesc& range, + const TFileScanRangeParams* range_params, FileMetaCache* meta_cache) + : _file_slot_descs(file_slot_descs), + _state(state), + _profile(profile), + _range(range), + _range_params(range_params), + _io_context(state) { + _meta_cache = meta_cache; +} + +IcebergPositionDeleteSysTableReader::~IcebergPositionDeleteSysTableReader() = default; + +Status IcebergPositionDeleteSysTableReader::_do_init_reader(ReaderInitContext* /*ctx*/) { + if (_state == nullptr || _profile == nullptr || _range_params == 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[0]; + 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<std::string, DataTypePtr>* 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(); +} + +Status IcebergPositionDeleteSysTableReader::close() { + if (_position_reader != nullptr) { + RETURN_IF_ERROR(_position_reader->close()); + } + _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_context.io_ctx, _state, _meta_cache); Review Comment: This V1 system-table reader is passing the nested Parquet reader a private `_io_context` instead of the scanner IO context. The same member is used for the ORC and Puffin DV paths, but `FileScanner::try_stop()` only flips the scanner `_io_ctx->should_stop`, and the scanner counters/profile are populated from the scanner-owned reader/cache stats. Existing V1 Iceberg delete-vector reads pass `this->get_io_ctx()` into the helper, so this new path can keep reading large delete files after a LIMIT/cancellation stop and underreport the actual delete-file IO. Please thread the scanner `IOContext` into this reader, or mirror `should_stop` and merge the private stats back before counter collection. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
