diff --git a/be/src/exec/sink/viceberg_delete_sink.cpp b/be/src/exec/sink/viceberg_delete_sink.cpp index 55b732d6527..3cb11884898 100644 --- a/be/src/exec/sink/viceberg_delete_sink.cpp +++ b/be/src/exec/sink/viceberg_delete_sink.cpp @@ -148,14 +148,16 @@ Status VIcebergDeleteSink::init_properties(ObjectPool* pool) { _compress_type = delete_sink.compress_type; } - // Get output path and table location - if (delete_sink.__isset.output_path) { - _output_path = delete_sink.output_path; - } - - if (delete_sink.__isset.table_location) { - _table_location = delete_sink.table_location; + // Every FE version sends both forms of the data location (see IcebergOutputLocation). Without + // the table's own form the commit could only record the address BE wrote the file at, e.g. + // s3://b/db/t/data/delete_pos_x.parquet for an oss:// table, which other engines resolve with + // the wrong file system. + if (delete_sink.output_path.empty() || delete_sink.table_location.empty()) { + return Status::InternalError( + "Iceberg delete sink must carry output_path and table_location, got '{}' and '{}'", + delete_sink.output_path, delete_sink.table_location); } + _output_location.emplace(delete_sink.output_path, delete_sink.table_location); // Get Hadoop configuration if (delete_sink.__isset.hadoop_config) { @@ -222,7 +224,7 @@ Status VIcebergDeleteSink::open(RuntimeState* state, RuntimeProfile* profile) { LOG(INFO) << fmt::format( "VIcebergDeleteSink opened: delete_type={}, output_path={}, format_version={}", - to_string(_delete_type), _output_path, _format_version); + to_string(_delete_type), _t_sink.iceberg_delete_sink.output_path, _format_version); return Status::OK(); } @@ -481,18 +483,16 @@ Status VIcebergDeleteSink::_collect_position_deletes( Status VIcebergDeleteSink::_write_position_delete_files( const std::map& file_deletions) { + DORIS_CHECK(_output_location.has_value()); constexpr size_t kBatchSize = 4096; for (const auto& [data_file_path, deletion] : file_deletions) { if (deletion.rows_to_delete.isEmpty()) { continue; } - // Generate unique delete file path - std::string delete_file_path = _generate_delete_file_path(data_file_path); - // Create delete file writer auto writer = VIcebergDeleteFileWriterFactory::create_writer( - TFileContent::POSITION_DELETES, delete_file_path, _file_format_type, - _compress_type); + TFileContent::POSITION_DELETES, *_output_location, + _delete_file_name(data_file_path), _file_format_type, _compress_type); // Build column names for position delete std::vector column_names = {"file_path", "pos"}; @@ -506,7 +506,7 @@ Status VIcebergDeleteSink::_write_position_delete_files( writer->open(_state, _state->runtime_profile(), _position_delete_output_expr_ctxs, column_names, _hadoop_conf, _file_type, _broker_addresses); if (writer->file_system() != nullptr) { - _created_files.emplace_back(writer->file_system(), delete_file_path); + _created_files.emplace_back(writer->file_system(), writer->write_path()); } RETURN_IF_ERROR(open_status); @@ -543,7 +543,7 @@ Status VIcebergDeleteSink::_write_position_delete_files( _delete_file_count++; VLOG(1) << fmt::format("Written position delete file: path={}, rows={}, referenced_file={}", - delete_file_path, commit_data.row_count, data_file_path); + writer->write_path(), commit_data.row_count, data_file_path); } return Status::OK(); @@ -643,6 +643,7 @@ std::string VIcebergDeleteSink::_get_file_extension() const { Status VIcebergDeleteSink::_write_deletion_vector_files( const std::map& file_deletions) { + DORIS_CHECK(_output_location.has_value()); std::vector blobs; for (const auto& [data_file_path, deletion] : file_deletions) { if (deletion.rows_to_delete.isEmpty()) { @@ -687,13 +688,15 @@ Status VIcebergDeleteSink::_write_deletion_vector_files( return Status::OK(); } - std::string puffin_path = _generate_puffin_file_path(); + const std::string puffin_file_name = _puffin_file_name(); int64_t puffin_file_size = 0; - RETURN_IF_ERROR(_write_puffin_file(puffin_path, &blobs, &puffin_file_size)); + RETURN_IF_ERROR(_write_puffin_file(_output_location->write_path(puffin_file_name), &blobs, + &puffin_file_size)); + const std::string commit_path = _output_location->commit_path(puffin_file_name); for (const auto& blob : blobs) { TIcebergCommitData commit_data; - commit_data.__set_file_path(puffin_path); + commit_data.__set_file_path(commit_path); commit_data.__set_row_count(blob.merged_count); commit_data.__set_affected_rows(blob.delete_count); commit_data.__set_file_size(puffin_file_size); @@ -803,37 +806,15 @@ std::string VIcebergDeleteSink::_build_puffin_footer_json( return {buffer.GetString(), buffer.GetSize()}; } -std::string VIcebergDeleteSink::_generate_delete_file_path( - const std::string& referenced_data_file) { - // Generate unique delete file name using UUID - std::string uuid = generate_uuid_string(); - std::string file_name; - - std::string file_extension = _get_file_extension(); - file_name = - fmt::format("delete_pos_{}_{}{}", uuid, - std::hash {}(referenced_data_file) % 10000000, file_extension); - - // Combine with output path or table location - std::string base_path = _output_path.empty() ? _table_location : _output_path; - - // Ensure base path ends with / - if (!base_path.empty() && base_path.back() != '/') { - base_path += '/'; - } - - // Delete files are data files in Iceberg, write under data location - return fmt::format("{}{}", base_path, file_name); +std::string VIcebergDeleteSink::_delete_file_name(const std::string& referenced_data_file) const { + // Delete files live in the data location itself, next to the partition directories. + return fmt::format("delete_pos_{}_{}{}", generate_uuid_string(), + std::hash {}(referenced_data_file) % 10000000, + _get_file_extension()); } -std::string VIcebergDeleteSink::_generate_puffin_file_path() { - std::string uuid = generate_uuid_string(); - std::string file_name = fmt::format("delete_dv_{}.puffin", uuid); - std::string base_path = _output_path.empty() ? _table_location : _output_path; - if (!base_path.empty() && base_path.back() != '/') { - base_path += '/'; - } - return fmt::format("{}{}", base_path, file_name); +std::string VIcebergDeleteSink::_puffin_file_name() const { + return fmt::format("delete_dv_{}.puffin", generate_uuid_string()); } } // namespace doris diff --git a/be/src/exec/sink/viceberg_delete_sink.h b/be/src/exec/sink/viceberg_delete_sink.h index 625134ba3a3..a4bb50d20f7 100644 --- a/be/src/exec/sink/viceberg_delete_sink.h +++ b/be/src/exec/sink/viceberg_delete_sink.h @@ -24,12 +24,14 @@ #include #include #include +#include #include #include #include "common/status.h" #include "core/block/block.h" #include "exec/sink/writer/async_result_writer.h" +#include "exec/sink/writer/iceberg/iceberg_output_location.h" #include "exec/sink/writer/iceberg/viceberg_delete_file_writer.h" #include "exprs/vexpr_fwd.h" #include "roaring/roaring64map.hh" @@ -114,12 +116,9 @@ private: std::string _build_puffin_footer_json(const std::vector& blobs); - std::string _generate_puffin_file_path(); - - /** - * Generate unique delete file path - */ - std::string _generate_delete_file_path(const std::string& referenced_data_file = ""); + // Unique file names, relative to the data location (_output_location). + std::string _puffin_file_name() const; + std::string _delete_file_name(const std::string& referenced_data_file) const; /** * Get $row_id column index from block @@ -163,9 +162,8 @@ private: TFileFormatType::type _file_format_type = TFileFormatType::FORMAT_PARQUET; TFileCompressType::type _compress_type = TFileCompressType::SNAPPYBLOCK; - // Output directory for delete files - std::string _output_path; - std::string _table_location; + // The data location delete files are written to and committed under, set by init_properties(). + std::optional _output_location; TFileType::type _file_type = TFileType::FILE_HDFS; std::vector _broker_addresses; diff --git a/be/src/exec/sink/writer/iceberg/iceberg_output_location.h b/be/src/exec/sink/writer/iceberg/iceberg_output_location.h new file mode 100644 index 00000000000..58982f73156 --- /dev/null +++ b/be/src/exec/sink/writer/iceberg/iceberg_output_location.h @@ -0,0 +1,81 @@ +// 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 "common/check.h" + +namespace doris { + +// The directory an Iceberg writer puts its files in, in both forms of the same location. BE +// creates, writes and deletes a file under write_dir, the form BE addresses the storage with. +// Iceberg metadata records the file under commit_dir, the table's own form, which every engine +// reading the table resolves. FE sends both: TIcebergTableSink output_path / original_output_path +// and TIcebergDeleteSink output_path / table_location. Example: an OSS table with data location +// oss://b/db/t/data reaches BE as write_dir s3://b/db/t/data and commit_dir oss://b/db/t/data, so +// delete_pos_1.parquet is written to s3://b/db/t/data/delete_pos_1.parquet and committed as +// oss://b/db/t/data/delete_pos_1.parquet. +// +// Every file a writer reports takes both of its paths from one IcebergOutputLocation, so it +// cannot commit the form it wrote: delete files used to be committed as s3://..., and other +// engines resolved them with a file system the table is not on. +class IcebergOutputLocation { +public: + IcebergOutputLocation(std::string write_dir, std::string commit_dir) + : _write_dir(std::move(write_dir)), _commit_dir(std::move(commit_dir)) { + DORIS_CHECK(!_write_dir.empty()); + DORIS_CHECK(!_commit_dir.empty()); + } + + // The same subdirectory in both forms, e.g. the partition directory "p=1/q=2". + IcebergOutputLocation subdirectory(std::string_view relative_dir) const { + DORIS_CHECK(!relative_dir.empty()); + return {_join(_write_dir, relative_dir), _join(_commit_dir, relative_dir)}; + } + + // The path BE writes file_name at. + std::string write_path(std::string_view file_name) const { + return _join(_write_dir, file_name); + } + + // The path Iceberg metadata records file_name at. + std::string commit_path(std::string_view file_name) const { + return _join(_commit_dir, file_name); + } + +private: + // FE takes write.data.path verbatim, so a directory may end with '/'. Both forms join with + // exactly one separator, as Iceberg's own location provider does: oss://b/t/data and + // oss://b/t/data/ both join "p=1" into oss://b/t/data/p=1. + static std::string _join(const std::string& dir, std::string_view name) { + std::string path = dir; + if (path.back() != '/') { + path.push_back('/'); + } + path.append(name); + return path; + } + + std::string _write_dir; + std::string _commit_dir; +}; + +} // namespace doris diff --git a/be/src/exec/sink/writer/iceberg/viceberg_delete_file_writer.cpp b/be/src/exec/sink/writer/iceberg/viceberg_delete_file_writer.cpp index 7f7f2c6f2f4..1233d09dae5 100644 --- a/be/src/exec/sink/writer/iceberg/viceberg_delete_file_writer.cpp +++ b/be/src/exec/sink/writer/iceberg/viceberg_delete_file_writer.cpp @@ -44,11 +44,13 @@ std::unique_ptr build_position_delete_schema() { } VIcebergDeleteFileWriter::VIcebergDeleteFileWriter(TFileContent::type delete_type, - const std::string& output_path, + const IcebergOutputLocation& location, + const std::string& file_name, TFileFormatType::type file_format, TFileCompressType::type compress_type) : _delete_type(delete_type), - _output_path(output_path), + _write_path(location.write_path(file_name)), + _commit_path(location.commit_path(file_name)), _file_format(file_format), _compress_type(compress_type) {} @@ -74,7 +76,7 @@ Status VIcebergDeleteFileWriter::open(RuntimeState* state, RuntimeProfile* profi fs_properties.broker_addresses = &broker_addresses; } - io::FileDescription file_description = {.path = _output_path, .fs_name {}}; + io::FileDescription file_description = {.path = _write_path, .fs_name {}}; _fs = DORIS_TRY(FileFactory::create_fs(fs_properties, file_description)); io::FileWriterOptions file_writer_options = {.used_by_s3_committer = false}; @@ -131,7 +133,7 @@ Status VIcebergDeleteFileWriter::close(TIcebergCommitData& commit_data) { } // Fill commit data (use __set_ to mark optional fields as present) - commit_data.__set_file_path(_output_path); + commit_data.__set_file_path(_commit_path); commit_data.__set_row_count(_written_rows); commit_data.__set_affected_rows(_written_rows); commit_data.__set_file_size(_file_size); @@ -152,9 +154,10 @@ Status VIcebergDeleteFileWriter::close(TIcebergCommitData& commit_data) { // Factory method std::unique_ptr VIcebergDeleteFileWriterFactory::create_writer( - TFileContent::type delete_type, const std::string& output_path, - TFileFormatType::type file_format, TFileCompressType::type compress_type) { - return std::make_unique(delete_type, output_path, file_format, + TFileContent::type delete_type, const IcebergOutputLocation& location, + const std::string& file_name, TFileFormatType::type file_format, + TFileCompressType::type compress_type) { + return std::make_unique(delete_type, location, file_name, file_format, compress_type); } diff --git a/be/src/exec/sink/writer/iceberg/viceberg_delete_file_writer.h b/be/src/exec/sink/writer/iceberg/viceberg_delete_file_writer.h index 4ceb7080a3c..b216b3ba16f 100644 --- a/be/src/exec/sink/writer/iceberg/viceberg_delete_file_writer.h +++ b/be/src/exec/sink/writer/iceberg/viceberg_delete_file_writer.h @@ -23,6 +23,7 @@ #include "common/status.h" #include "core/block/block.h" +#include "exec/sink/writer/iceberg/iceberg_output_location.h" #include "exprs/vexpr_fwd.h" #include "gen_cpp/DataSinks_types.h" #include "io/fs/file_writer.h" @@ -50,8 +51,10 @@ class VFileFormatTransformer; */ class VIcebergDeleteFileWriter { public: - VIcebergDeleteFileWriter(TFileContent::type delete_type, const std::string& output_path, - TFileFormatType::type file_format, + // Writes file_name under location; the commit data reports the file under the table's own form + // of that directory, not the form it was written at. + VIcebergDeleteFileWriter(TFileContent::type delete_type, const IcebergOutputLocation& location, + const std::string& file_name, TFileFormatType::type file_format, TFileCompressType::type compress_type); ~VIcebergDeleteFileWriter(); @@ -96,9 +99,13 @@ public: std::shared_ptr file_system() const { return _fs; } + // The path the file is written at, which is what cleanup must delete. + const std::string& write_path() const { return _write_path; } + private: TFileContent::type _delete_type; - std::string _output_path; + std::string _write_path; + std::string _commit_path; TFileFormatType::type _file_format; TFileCompressType::type _compress_type = TFileCompressType::SNAPPYBLOCK; @@ -122,8 +129,9 @@ private: class VIcebergDeleteFileWriterFactory { public: static std::unique_ptr create_writer( - TFileContent::type delete_type, const std::string& output_path, - TFileFormatType::type file_format, TFileCompressType::type compress_type); + TFileContent::type delete_type, const IcebergOutputLocation& location, + const std::string& file_name, TFileFormatType::type file_format, + TFileCompressType::type compress_type); }; } // namespace doris diff --git a/be/src/exec/sink/writer/iceberg/viceberg_partition_writer.cpp b/be/src/exec/sink/writer/iceberg/viceberg_partition_writer.cpp index 363776dcf17..249cd15dcf1 100644 --- a/be/src/exec/sink/writer/iceberg/viceberg_partition_writer.cpp +++ b/be/src/exec/sink/writer/iceberg/viceberg_partition_writer.cpp @@ -93,7 +93,7 @@ Status VIcebergPartitionWriter::open(RuntimeState* state, RuntimeProfile* profil if (!_write_info.broker_addresses.empty()) { fs_properties.broker_addresses = &(_write_info.broker_addresses); } - _path = fmt::format("{}/{}", _write_info.write_path, _get_target_file_name()); + _path = _write_info.location.write_path(_get_target_file_name()); io::FileDescription file_description = {.path = _path, .fs_name {}}; _fs = DORIS_TRY(FileFactory::create_fs(fs_properties, file_description)); io::FileWriterOptions file_writer_options = {.used_by_s3_committer = false}; @@ -210,8 +210,7 @@ Status VIcebergPartitionWriter::_build_iceberg_commit_data(TIcebergCommitData* c DCHECK(commit_data != nullptr); DCHECK(_file_format_transformer != nullptr); - commit_data->__set_file_path( - fmt::format("{}/{}", _write_info.original_write_path, _get_target_file_name())); + commit_data->__set_file_path(_write_info.location.commit_path(_get_target_file_name())); commit_data->__set_row_count(_row_count); commit_data->__set_affected_rows(_row_count); commit_data->__set_file_size(_file_format_transformer->written_len()); diff --git a/be/src/exec/sink/writer/iceberg/viceberg_table_writer.cpp b/be/src/exec/sink/writer/iceberg/viceberg_table_writer.cpp index 61cb105460d..d20eb8a005f 100644 --- a/be/src/exec/sink/writer/iceberg/viceberg_table_writer.cpp +++ b/be/src/exec/sink/writer/iceberg/viceberg_table_writer.cpp @@ -61,6 +61,18 @@ Status VIcebergTableWriter::open(RuntimeState* state, RuntimeProfile* profile) { RETURN_IF_ERROR(validate_external_file_report_ack(state->query_options())); + // Every FE version sends both forms of the data location (see IcebergOutputLocation). A plan + // without the table's own form leaves no path to commit the files under: the commit used to + // record a relative path such as /p=1/xxx.parquet that no engine can open. + const auto& iceberg_sink = _t_sink.iceberg_table_sink; + if (iceberg_sink.output_path.empty() || iceberg_sink.original_output_path.empty()) { + return Status::InternalError( + "Iceberg table sink must carry output_path and original_output_path, got '{}' " + "and '{}'", + iceberg_sink.output_path, iceberg_sink.original_output_path); + } + _output_location.emplace(iceberg_sink.output_path, iceberg_sink.original_output_path); + // Get target file size from query options // If value is 0 or not set, use config::iceberg_sink_max_file_size _target_file_size_bytes = config::iceberg_sink_max_file_size; @@ -822,47 +834,29 @@ std::vector VIcebergTableWriter::_partition_values( std::shared_ptr VIcebergTableWriter::_create_partition_writer( Block* transformed_block, int position, const std::string* file_name, int file_name_index) { + DORIS_CHECK(_output_location.has_value()); auto& iceberg_table_sink = _t_sink.iceberg_table_sink; std::vector partition_values; - const std::string& output_path = iceberg_table_sink.output_path; - std::string write_path; - std::string original_write_path; - std::string target_path; + // Relative to the data location, e.g. "year=2023/month=12"; empty when the file goes to the + // data location itself. + std::string partition_path; // Case 1: Full static partition - use pre-computed static partition path and values if (_is_full_static_partition) { partition_values = _static_partition_value_list; - if (!_static_partition_path.empty()) { - original_write_path = fmt::format("{}/{}", iceberg_table_sink.original_output_path, - _static_partition_path); - target_path = fmt::format("{}/{}", output_path, _static_partition_path); - write_path = fmt::format("{}/{}", output_path, _static_partition_path); - } else { - original_write_path = iceberg_table_sink.original_output_path; - target_path = output_path; - write_path = output_path; - } + partition_path = _static_partition_path; } else if (transformed_block != nullptr) { // Case 2: Dynamic partition or Hybrid mode (partial static + partial dynamic) // _partition_to_path and _partition_values already handle hybrid mode internally IcebergPartitionData partition_data = _get_partition_data(transformed_block, position); - std::string partition_path = _partition_to_path(partition_data); + partition_path = _partition_to_path(partition_data); partition_values = _partition_values(partition_data); - original_write_path = - fmt::format("{}/{}", iceberg_table_sink.original_output_path, partition_path); - target_path = fmt::format("{}/{}", output_path, partition_path); - write_path = fmt::format("{}/{}", output_path, partition_path); - } else { - // Case 3: Non-partitioned table - original_write_path = iceberg_table_sink.original_output_path; - target_path = output_path; - write_path = output_path; } + // Case 3: Non-partitioned table, written to the data location itself IPartitionWriterBase::WriteInfo write_info = { - .write_path = std::move(write_path), - .original_write_path = std::move(original_write_path), - .target_path = std::move(target_path), + .location = partition_path.empty() ? *_output_location + : _output_location->subdirectory(partition_path), .file_type = iceberg_table_sink.file_type, .broker_addresses = {}}; if (iceberg_table_sink.__isset.broker_addresses) { diff --git a/be/src/exec/sink/writer/iceberg/viceberg_table_writer.h b/be/src/exec/sink/writer/iceberg/viceberg_table_writer.h index 36eb5c08aa0..5c9a8dddb9f 100644 --- a/be/src/exec/sink/writer/iceberg/viceberg_table_writer.h +++ b/be/src/exec/sink/writer/iceberg/viceberg_table_writer.h @@ -20,10 +20,13 @@ #include #include +#include + #include "common/atomic_shared_ptr.h" #include "common/status.h" #include "core/block/block.h" #include "core/column/column.h" +#include "exec/sink/writer/iceberg/iceberg_output_location.h" #include "exec/sink/writer/iceberg/partition_data.h" #include "exec/sink/writer/iceberg/partition_transformers.h" #include "exprs/vexpr_fwd.h" @@ -164,6 +167,8 @@ private: TDataSink _t_sink; RuntimeState* _state = nullptr; RuntimeProfile* _operator_profile = nullptr; + // The table's data location, set by open(); partition writers write under its subdirectories. + std::optional _output_location; // Target file size in bytes for controlling when to split files int64_t _target_file_size_bytes = 0; diff --git a/be/src/exec/sink/writer/iceberg/vpartition_writer_base.h b/be/src/exec/sink/writer/iceberg/vpartition_writer_base.h index dc70f646795..0c1980ca252 100644 --- a/be/src/exec/sink/writer/iceberg/vpartition_writer_base.h +++ b/be/src/exec/sink/writer/iceberg/vpartition_writer_base.h @@ -20,6 +20,7 @@ #include #include "common/status.h" +#include "exec/sink/writer/iceberg/iceberg_output_location.h" #include "runtime/descriptors.h" namespace doris { @@ -34,9 +35,9 @@ class RuntimeProfile; class IPartitionWriterBase { public: struct WriteInfo { - std::string write_path; - std::string original_write_path; - std::string target_path; + // The directory of the partition this writer writes, in the form BE writes the file in + // and the form the commit records it in. + IcebergOutputLocation location; TFileType::type file_type; std::vector broker_addresses; }; diff --git a/be/test/exec/sink/viceberg_delete_sink_test.cpp b/be/test/exec/sink/viceberg_delete_sink_test.cpp index e2897125e49..2eaa762350b 100644 --- a/be/test/exec/sink/viceberg_delete_sink_test.cpp +++ b/be/test/exec/sink/viceberg_delete_sink_test.cpp @@ -17,7 +17,10 @@ #include "exec/sink/viceberg_delete_sink.h" +#include +#include #include +#include #include #include @@ -36,7 +39,9 @@ #include "format/table/deletion_vector.h" #include "gen_cpp/DataSinks_types.h" #include "gen_cpp/Types_types.h" +#include "runtime/exec_env.h" #include "runtime/runtime_state.h" +#include "util/defer_op.h" #include "util/uid_util.h" namespace doris { @@ -471,26 +476,57 @@ TEST_F(VIcebergDeleteSinkTest, TestBuildPositionDeleteBlock) { ASSERT_EQ(40, pos_column->get_element(3)); } -TEST_F(VIcebergDeleteSinkTest, TestGenerateDeleteFilePath) { +TEST_F(VIcebergDeleteSinkTest, TestDeleteFileNames) { VExprContextSPtrs output_exprs; auto sink = std::make_shared(_t_data_sink, output_exprs, nullptr, nullptr); ObjectPool pool; - Status status = sink->init_properties(&pool); - ASSERT_TRUE(status.ok()); + ASSERT_TRUE(sink->init_properties(&pool).ok()); - std::string data_file_path = "data/file1.parquet"; - std::string delete_file_path = sink->_generate_delete_file_path(data_file_path); + // Names are relative to the data location; the location supplies both forms of the directory. + std::string delete_file_name = sink->_delete_file_name("data/file1.parquet"); + EXPECT_TRUE(delete_file_name.starts_with("delete_pos_")) << delete_file_name; + EXPECT_TRUE(delete_file_name.ends_with(".snappy.parquet")) << delete_file_name; + EXPECT_EQ(std::string::npos, delete_file_name.find('/')); + EXPECT_NE(delete_file_name, sink->_delete_file_name("data/file1.parquet")); - // Verify the path format - ASSERT_FALSE(delete_file_path.empty()); - const auto& delete_sink = _t_data_sink.iceberg_delete_sink; - std::string expected_base = - delete_sink.__isset.output_path ? delete_sink.output_path : delete_sink.table_location; - if (!expected_base.empty() && expected_base.back() != '/') { - expected_base += '/'; - } - ASSERT_TRUE(delete_file_path.rfind(expected_base, 0) == 0); - ASSERT_NE(std::string::npos, delete_file_path.find("delete_pos_")); + std::string puffin_file_name = sink->_puffin_file_name(); + EXPECT_TRUE(puffin_file_name.starts_with("delete_dv_")) << puffin_file_name; + EXPECT_TRUE(puffin_file_name.ends_with(".puffin")) << puffin_file_name; + EXPECT_EQ(std::string::npos, puffin_file_name.find('/')); +} + +TEST_F(VIcebergDeleteSinkTest, TestInitPropertiesTakesBothFormsOfTheDataLocation) { + // FE writes through the normalized output_path but the table records its own location. + TDataSink t_data_sink = _t_data_sink; + t_data_sink.iceberg_delete_sink.__set_output_path("s3://b/db/t/data"); + t_data_sink.iceberg_delete_sink.__set_table_location("oss://b/db/t/data/"); + VExprContextSPtrs output_exprs; + auto sink = std::make_shared(t_data_sink, output_exprs, nullptr, nullptr); + ObjectPool pool; + ASSERT_TRUE(sink->init_properties(&pool).ok()); + ASSERT_TRUE(sink->_output_location.has_value()); + EXPECT_EQ("s3://b/db/t/data/delete_dv_1.puffin", + sink->_output_location->write_path("delete_dv_1.puffin")); + EXPECT_EQ("oss://b/db/t/data/delete_dv_1.puffin", + sink->_output_location->commit_path("delete_dv_1.puffin")); + + // Without the table's own form the commit could only record the s3:// address BE wrote. + t_data_sink.iceberg_delete_sink.__isset.table_location = false; + t_data_sink.iceberg_delete_sink.table_location.clear(); + auto no_table_location = + std::make_shared(t_data_sink, output_exprs, nullptr, nullptr); + Status status = no_table_location->init_properties(&pool); + EXPECT_TRUE(status.is()) << status; + EXPECT_NE(std::string::npos, status.to_string().find("table_location")); + + t_data_sink.iceberg_delete_sink.__set_table_location("oss://b/db/t/data/"); + t_data_sink.iceberg_delete_sink.__isset.output_path = false; + t_data_sink.iceberg_delete_sink.output_path.clear(); + auto no_output_path = + std::make_shared(t_data_sink, output_exprs, nullptr, nullptr); + status = no_output_path->init_properties(&pool); + EXPECT_TRUE(status.is()) << status; + EXPECT_NE(std::string::npos, status.to_string().find("output_path")); } TEST_F(VIcebergDeleteSinkTest, TestUnsupportedDeleteType) { @@ -533,12 +569,88 @@ TEST_F(VIcebergDeleteSinkTest, TestValidateDeletionVectorContentSize) { ASSERT_TRUE(unsupported_status.is()) << unsupported_status; } +TEST_F(VIcebergDeleteSinkTest, TestWritePositionDeletesCommitsTableLocation) { + // The Parquet writer's dictionary encoder allocates from ExecEnv's Arrow pool, which this + // fixture does not otherwise initialize. + auto* env = ExecEnv::GetInstance(); + auto* previous_pool = env->_arrow_memory_pool; + env->_arrow_memory_pool = arrow::default_memory_pool(); + Defer restore_pool([&] { env->_arrow_memory_pool = previous_pool; }); + + std::filesystem::path temp_dir = std::filesystem::temp_directory_path() / + ("iceberg_delete_sink_test_" + generate_uuid_string()); + ASSERT_TRUE(std::filesystem::create_directories(temp_dir)); + + // Write locally, but record the table's original URI in the commit metadata. + const std::string table_location = "oss://bucket/warehouse/table/data/"; + const std::string data_file_path = table_location + "file1.parquet"; + TDataSink t_data_sink = build_local_delete_sink(temp_dir.string(), 2); + t_data_sink.iceberg_delete_sink.__set_table_location(table_location); + RuntimeState state; + VExprContextSPtrs output_exprs; + auto sink = std::make_shared(t_data_sink, output_exprs, nullptr, nullptr); + ObjectPool pool; + ASSERT_TRUE(sink->init_properties(&pool).ok()); + sink->_state = &state; + + std::map file_deletions; + auto [it, inserted] = + file_deletions.emplace(data_file_path, IcebergFileDeletion(1, "[\"p=1\"]")); + ASSERT_TRUE(inserted); + it->second.rows_to_delete.add(uint32_t {10}); + it->second.rows_to_delete.add(uint32_t {20}); + + const auto status = sink->_write_position_delete_files(file_deletions); + ASSERT_TRUE(status.ok()) << status; + ASSERT_EQ(1, sink->_commit_data_list.size()); + ASSERT_EQ(1, sink->_created_files.size()); + ASSERT_EQ(1, sink->_delete_file_count); + + const std::filesystem::path write_path = sink->_created_files[0].second; + ASSERT_EQ(temp_dir, write_path.parent_path()); + ASSERT_TRUE(std::filesystem::exists(write_path)); + const auto& commit_data = sink->_commit_data_list[0]; + EXPECT_EQ(table_location + write_path.filename().string(), commit_data.file_path); + EXPECT_EQ(data_file_path, commit_data.referenced_data_file_path); + EXPECT_EQ(TFileContent::POSITION_DELETES, commit_data.file_content); + EXPECT_EQ(2, commit_data.row_count); + EXPECT_EQ(2, commit_data.affected_rows); + EXPECT_EQ(1, commit_data.partition_spec_id); + EXPECT_EQ("[\"p=1\"]", commit_data.partition_data_json); + EXPECT_EQ(std::filesystem::file_size(write_path), commit_data.file_size); + + // The referenced data-file URI inside the position-delete file must stay unchanged too. + auto input = arrow::io::ReadableFile::Open(write_path.string()); + ASSERT_TRUE(input.ok()) << input.status(); + auto reader = ::parquet::arrow::OpenFile(*input, arrow::default_memory_pool()); + ASSERT_TRUE(reader.ok()) << reader.status(); + auto table = (*reader)->ReadTable(); + ASSERT_TRUE(table.ok()) << table.status(); + ASSERT_EQ(2, (*table)->num_rows()); + ASSERT_EQ(2, (*table)->num_columns()); + for (int64_t row = 0; row < 2; ++row) { + auto file_path = (*table)->column(0)->GetScalar(row); + ASSERT_TRUE(file_path.ok()) << file_path.status(); + EXPECT_EQ(data_file_path, (*file_path)->ToString()); + auto position = (*table)->column(1)->GetScalar(row); + ASSERT_TRUE(position.ok()) << position.status(); + EXPECT_EQ(std::to_string((row + 1) * 10), (*position)->ToString()); + } + + std::filesystem::remove_all(temp_dir); +} + TEST_F(VIcebergDeleteSinkTest, TestWriteDeletionVectorsToSingleSharedPuffin) { std::filesystem::path temp_dir = std::filesystem::temp_directory_path() / ("iceberg_delete_sink_test_" + generate_uuid_string()); ASSERT_TRUE(std::filesystem::create_directories(temp_dir)); - TDataSink t_data_sink = build_local_delete_sink(temp_dir.string(), 3); + const std::string table_location = + "abfss://container@account.dfs.core.windows.net/warehouse/table/data"; + const std::string first_data_file = table_location + "/file1.parquet"; + const std::string second_data_file = table_location + "/file2.parquet"; + TDataSink t_data_sink = build_local_delete_sink(temp_dir.string() + "/", 3); + t_data_sink.iceberg_delete_sink.__set_table_location(table_location); VExprContextSPtrs output_exprs; auto sink = std::make_shared(t_data_sink, output_exprs, nullptr, nullptr); ObjectPool pool; @@ -546,23 +658,38 @@ TEST_F(VIcebergDeleteSinkTest, TestWriteDeletionVectorsToSingleSharedPuffin) { std::map file_deletions; auto [file1_it, file1_inserted] = - file_deletions.emplace("file1.parquet", IcebergFileDeletion(1, "[\"p=1\"]")); + file_deletions.emplace(first_data_file, IcebergFileDeletion(1, "[\"p=1\"]")); ASSERT_TRUE(file1_inserted); file1_it->second.rows_to_delete.add((uint32_t)10); file1_it->second.rows_to_delete.add((uint32_t)20); auto [file2_it, file2_inserted] = - file_deletions.emplace("file2.parquet", IcebergFileDeletion(2, "[\"p=2\"]")); + file_deletions.emplace(second_data_file, IcebergFileDeletion(2, "[\"p=2\"]")); ASSERT_TRUE(file2_inserted); file2_it->second.rows_to_delete.add((uint32_t)30); - ASSERT_TRUE(sink->_write_deletion_vector_files(file_deletions).ok()); + const auto status = sink->_write_deletion_vector_files(file_deletions); + ASSERT_TRUE(status.ok()) << status; ASSERT_EQ(2, sink->_commit_data_list.size()); ASSERT_EQ(2, sink->_delete_file_count); + ASSERT_EQ(1, sink->_created_files.size()); + + const std::filesystem::path write_path = sink->_created_files[0].second; + ASSERT_EQ(temp_dir, write_path.parent_path()); + ASSERT_TRUE(std::filesystem::exists(write_path)); const auto& first_commit = sink->_commit_data_list[0]; const auto& second_commit = sink->_commit_data_list[1]; + EXPECT_EQ(table_location + "/" + write_path.filename().string(), first_commit.file_path); ASSERT_EQ(first_commit.file_path, second_commit.file_path); + EXPECT_EQ(first_data_file, first_commit.referenced_data_file_path); + EXPECT_EQ(second_data_file, second_commit.referenced_data_file_path); + EXPECT_EQ(TFileContent::DELETION_VECTOR, first_commit.file_content); + EXPECT_EQ(TFileContent::DELETION_VECTOR, second_commit.file_content); + EXPECT_EQ(1, first_commit.partition_spec_id); + EXPECT_EQ(2, second_commit.partition_spec_id); + EXPECT_EQ("[\"p=1\"]", first_commit.partition_data_json); + EXPECT_EQ("[\"p=2\"]", second_commit.partition_data_json); ASSERT_EQ(first_commit.file_size, second_commit.file_size); ASSERT_LT(first_commit.content_offset, second_commit.content_offset); @@ -574,7 +701,7 @@ TEST_F(VIcebergDeleteSinkTest, TestWriteDeletionVectorsToSingleSharedPuffin) { } ASSERT_EQ(1, puffin_file_count); - std::ifstream input(first_commit.file_path, std::ios::binary); + std::ifstream input(write_path, std::ios::binary); ASSERT_TRUE(input.good()); std::string file_bytes((std::istreambuf_iterator(input)), std::istreambuf_iterator()); diff --git a/be/test/exec/sink/viceberg_merge_sink_test.cpp b/be/test/exec/sink/viceberg_merge_sink_test.cpp index 991f7bc2650..a262e06d8f2 100644 --- a/be/test/exec/sink/viceberg_merge_sink_test.cpp +++ b/be/test/exec/sink/viceberg_merge_sink_test.cpp @@ -211,6 +211,44 @@ TEST_F(VIcebergMergeSinkTest, TestUpdateProducesDeleteAndInsert) { ASSERT_TRUE(sink->close(Status::OK()).ok()); } +TEST_F(VIcebergMergeSinkTest, TestInnerSinksWriteAndCommitUnderBothFormsOfTheDataLocation) { + ObjectPool pool; + IcebergWriteMockRuntimeState state; + + DataTypes types {std::make_shared(), + std::make_shared(DataTypes {std::make_shared(), + std::make_shared()}, + Strings {"file_path", "row_position"}), + std::make_shared(), std::make_shared()}; + MockRowDescriptor row_desc(types, &pool); + auto output_exprs = build_output_exprs(&pool, &state, row_desc); + // UPDATE / MERGE write data files and delete files through one normalized location, and both + // kinds must be committed under the table's own form of it. + TDataSink t_sink = build_sink(); + t_sink.iceberg_merge_sink.__set_output_path("s3://b/db/t/data"); + t_sink.iceberg_merge_sink.__set_original_output_path("oss://b/db/t/data"); + t_sink.iceberg_merge_sink.__set_table_location("oss://b/db/t/data"); + + auto sink = std::make_shared(t_sink, output_exprs, nullptr, nullptr); + sink->set_skip_io(true); + ASSERT_TRUE(sink->init_properties(&pool, row_desc).ok()); + RuntimeProfile profile("iceberg_merge_sink_locations"); + ASSERT_TRUE(sink->open(&state, &profile).ok()); + + ASSERT_TRUE(sink->_table_writer->_output_location.has_value()); + EXPECT_EQ("s3://b/db/t/data/f.parquet", + sink->_table_writer->_output_location->write_path("f.parquet")); + EXPECT_EQ("oss://b/db/t/data/f.parquet", + sink->_table_writer->_output_location->commit_path("f.parquet")); + ASSERT_TRUE(sink->_delete_writer->_output_location.has_value()); + EXPECT_EQ("s3://b/db/t/data/f.parquet", + sink->_delete_writer->_output_location->write_path("f.parquet")); + EXPECT_EQ("oss://b/db/t/data/f.parquet", + sink->_delete_writer->_output_location->commit_path("f.parquet")); + + ASSERT_TRUE(sink->close(Status::OK()).ok()); +} + TEST_F(VIcebergMergeSinkTest, TestDeleteOnlySkipsVariantDataWriter) { ObjectPool pool; // Opening the retained delete writer still requires the coordinator ownership handshake. diff --git a/be/test/exec/sink/writer/iceberg/iceberg_output_location_test.cpp b/be/test/exec/sink/writer/iceberg/iceberg_output_location_test.cpp new file mode 100644 index 00000000000..1abd5d71481 --- /dev/null +++ b/be/test/exec/sink/writer/iceberg/iceberg_output_location_test.cpp @@ -0,0 +1,56 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +#include "exec/sink/writer/iceberg/iceberg_output_location.h" + +#include + +namespace doris { + +TEST(IcebergOutputLocationTest, WritesUnderTheNormalizedFormAndCommitsUnderTheTableForm) { + IcebergOutputLocation location("s3://b/db/t/data", "oss://b/db/t/data"); + + EXPECT_EQ("s3://b/db/t/data/delete_pos_1.parquet", location.write_path("delete_pos_1.parquet")); + EXPECT_EQ("oss://b/db/t/data/delete_pos_1.parquet", + location.commit_path("delete_pos_1.parquet")); +} + +TEST(IcebergOutputLocationTest, JoinsDirectoriesWithAndWithoutTrailingSlashOnce) { + // write.data.path is taken verbatim, so either form may end with '/'. + IcebergOutputLocation location("s3://b/db/t/data/", "oss://b/db/t/data"); + + EXPECT_EQ("s3://b/db/t/data/f.parquet", location.write_path("f.parquet")); + EXPECT_EQ("oss://b/db/t/data/f.parquet", location.commit_path("f.parquet")); + + IcebergOutputLocation partition = location.subdirectory("p=1/q=2"); + EXPECT_EQ("s3://b/db/t/data/p=1/q=2/f.parquet", partition.write_path("f.parquet")); + EXPECT_EQ("oss://b/db/t/data/p=1/q=2/f.parquet", partition.commit_path("f.parquet")); +} + +TEST(IcebergOutputLocationTest, KeepsTheTableFormWhenNormalizationRewritesTheAuthority) { + // FE rewrites more than the scheme for Azure: the account moves out of the authority. + IcebergOutputLocation location( + "s3://container/warehouse/t/data", + "abfss://container@account.dfs.core.windows.net/warehouse/t/data"); + + IcebergOutputLocation partition = location.subdirectory("dt=2024-01-01"); + EXPECT_EQ("s3://container/warehouse/t/data/dt=2024-01-01/f.orc", partition.write_path("f.orc")); + EXPECT_EQ("abfss://container@account.dfs.core.windows.net/warehouse/t/data/dt=2024-01-01/f.orc", + partition.commit_path("f.orc")); +} + +} // namespace doris diff --git a/be/test/exec/sink/writer/iceberg/iceberg_partition_writer_test.cpp b/be/test/exec/sink/writer/iceberg/iceberg_partition_writer_test.cpp index 590e1953c67..ba3f13b6086 100644 --- a/be/test/exec/sink/writer/iceberg/iceberg_partition_writer_test.cpp +++ b/be/test/exec/sink/writer/iceberg/iceberg_partition_writer_test.cpp @@ -82,9 +82,12 @@ protected: const TDataSink& sink, const VExprContextSPtrs& output_exprs, const iceberg::Schema& schema, const std::string* schema_json, const std::map& hadoop_conf, - TFileFormatType::type format = TFileFormatType::FORMAT_ORC) { - IPartitionWriterBase::WriteInfo write_info; - write_info.file_type = TFileType::FILE_LOCAL; + TFileFormatType::type format = TFileFormatType::FORMAT_ORC, + IcebergOutputLocation location = IcebergOutputLocation( + "/tmp/iceberg_partition_writer_test", "/tmp/iceberg_partition_writer_test")) { + IPartitionWriterBase::WriteInfo write_info {.location = std::move(location), + .file_type = TFileType::FILE_LOCAL, + .broker_addresses = {}}; return std::make_unique( sink, std::vector {}, output_exprs, schema, schema_json, std::vector {}, std::move(write_info), "data", 0, format, @@ -128,6 +131,24 @@ TEST_F(VIcebergPartitionWriterTest, OrcSkipsFooterCollectionWhenMetricsAreDisabl EXPECT_FALSE(commit_data.__isset.column_stats); } +TEST_F(VIcebergPartitionWriterTest, CommitDataRecordsTheTableFormOfTheLocation) { + // The file is written through FE's normalized form; the commit must keep the table's own. + VExprContextSPtrs output_exprs; + iceberg::Schema schema(std::vector {}); + std::string schema_json; + std::map hadoop_conf; + auto writer = + make_writer(make_table_sink(false), output_exprs, schema, &schema_json, hadoop_conf, + TFileFormatType::FORMAT_ORC, + IcebergOutputLocation("s3://b/db/t/data/p=1", "oss://b/db/t/data/p=1")); + install_fake_transformer(writer.get(), output_exprs); + + TIcebergCommitData commit_data; + ASSERT_TRUE(build_commit_data(writer.get(), &commit_data).ok()); + EXPECT_EQ("oss://b/db/t/data/p=1/data-0.zlib.orc", commit_data.file_path); + EXPECT_EQ(TFileContent::DATA, commit_data.file_content); +} + TEST_F(VIcebergPartitionWriterTest, ParquetKeepsSplitOffsetsWhenColumnMetricsAreDisabled) { TimezoneUtils::load_timezones_to_cache(); auto* env = ExecEnv::GetInstance(); diff --git a/be/test/exec/sink/writer/iceberg/viceberg_table_writer_test.cpp b/be/test/exec/sink/writer/iceberg/viceberg_table_writer_test.cpp index 2f136adcdae..41a126a9cbf 100644 --- a/be/test/exec/sink/writer/iceberg/viceberg_table_writer_test.cpp +++ b/be/test/exec/sink/writer/iceberg/viceberg_table_writer_test.cpp @@ -32,6 +32,7 @@ #include "core/data_type/data_type_timestamptz.h" #include "core/data_type/data_type_varbinary.h" #include "exec/sink/writer/iceberg/partition_transformers.h" +#include "exec/sink/writer/iceberg/viceberg_partition_writer.h" #include "exec/sink/writer/iceberg/vpartition_writer_base.h" #include "exprs/vexpr_context.h" #include "exprs/vslot_ref.h" @@ -327,6 +328,65 @@ TEST(VIcebergTableWriterTest, BinaryIdentityStaticAndDynamicRouting) { EXPECT_EQ(std::vector({"0xdead", "7"}), writer._partition_values(hybrid)); } +TEST(VIcebergTableWriterTest, OpenRejectsASinkWithoutTheTableFormOfTheDataLocation) { + // Without original_output_path the commit used to record relative paths such as + // /p=1/xxx.parquet; now the write fails before any file is created. + TIcebergTableSink sink; + sink.__set_output_path("s3://b/db/t/data"); + TDataSink data_sink; + data_sink.__set_iceberg_table_sink(sink); + VIcebergTableWriter writer(data_sink, {}); + RuntimeState state; + TQueryOptions query_options; + query_options.__set_supports_external_file_report_ack(true); + state.set_query_options(query_options); + RuntimeProfile profile("test"); + + Status status = writer.open(&state, &profile); + + EXPECT_TRUE(status.is()) << status; + EXPECT_NE(std::string::npos, status.to_string().find("original_output_path")); + EXPECT_FALSE(writer._output_location.has_value()); +} + +TEST(VIcebergTableWriterTest, PartitionWritersWriteAndCommitUnderBothFormsOfTheDataLocation) { + // FE normalizes the location BE writes to (oss:// becomes s3://); the commit must keep the + // table's own form in the data location and in every partition directory under it. + std::vector columns; + columns.emplace_back(false, 1, "p", std::make_unique(), std::nullopt); + TIcebergTableSink sink; + sink.__set_output_path("s3://b/db/t/data"); + sink.__set_original_output_path("oss://b/db/t/data/"); + sink.__set_file_format(TFileFormatType::FORMAT_PARQUET); + sink.__set_compression_type(TFileCompressType::SNAPPYBLOCK); + TDataSink data_sink; + data_sink.__set_iceberg_table_sink(sink); + VIcebergTableWriter writer(data_sink, {}); + writer._schema = std::make_shared(std::move(columns)); + // What open() derives from the two location fields. + writer._output_location.emplace(sink.output_path, sink.original_output_path); + + const std::string file_name = "f"; + auto expect_paths = [&](const std::string& partition_dir) { + auto writer_base = writer._create_partition_writer(nullptr, -1, &file_name, 0); + auto* partition_writer = dynamic_cast(writer_base.get()); + ASSERT_NE(nullptr, partition_writer); + const auto& location = partition_writer->_write_info.location; + const std::string target_file = partition_writer->_get_target_file_name(); + EXPECT_EQ("s3://b/db/t/data/" + partition_dir + target_file, + location.write_path(target_file)); + EXPECT_EQ("oss://b/db/t/data/" + partition_dir + target_file, + location.commit_path(target_file)); + }; + // A non-partitioned table writes into the data location itself. + expect_paths(""); + // A partition writes into its directory under the data location. + writer._is_full_static_partition = true; + writer._static_partition_path = "p=1"; + writer._static_partition_value_list = {"1"}; + expect_paths("p=1/"); +} + TEST(VIcebergTableWriterTest, RejectMissingPartitionSource) { std::vector columns; columns.emplace_back(false, 3, "id", std::make_unique(), std::nullopt); diff --git a/be/test/format/transformer/vparquet_writer_test.cpp b/be/test/format/transformer/vparquet_writer_test.cpp index 6591ca24f57..1c53c4b6916 100644 --- a/be/test/format/transformer/vparquet_writer_test.cpp +++ b/be/test/format/transformer/vparquet_writer_test.cpp @@ -594,9 +594,7 @@ TEST_F(VParquetWriterTest, IcebergNanosecondsRoundTripWithoutCoercion) { const auto name = "iceberg_ns_" + UniqueId::gen_uid().to_string(); _file_path = "./" + name + "-0.parquet"; VIcebergPartitionWriter writer(sink, {}, exprs, *schema, &json, {"ts"}, - {.write_path = ".", - .original_write_path = ".", - .target_path = ".", + {.location = IcebergOutputLocation(".", "."), .file_type = TFileType::FILE_LOCAL, .broker_addresses = {}}, name, 0, TFileFormatType::FORMAT_PARQUET, diff --git a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorTransaction.java b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorTransaction.java index ad17c6d353e..b6ca61f312f 100644 --- a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorTransaction.java +++ b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorTransaction.java @@ -144,6 +144,10 @@ public class IcebergConnectorTransaction implements ConnectorTransaction, Rewrit private final IcebergCatalogResourceTracker resourceTracker; private final String catalogFlavor; private final List commitDataList = new ArrayList<>(); + // The raw data locations the sinks of this write commit their files under, registered by planWrite for + // every sink it builds (several for a rewrite_data_files whose groups each plan one). Guarded by this, + // like commitDataList. + private final Set commitLocations = new HashSet<>(); // The single SDK transaction / table, opened lazily by beginWrite (the write plan binds the target // table only when the sink is planned). volatile: addCommitData / commit may run on different threads. @@ -626,10 +630,46 @@ public class IcebergConnectorTransaction implements ConnectorTransaction, Rewrit throw new DorisConnectorException("failed to deserialize Iceberg commit data", e); } synchronized (this) { + checkCommitLocation(data); commitDataList.add(data); } } + /** + * Registers a raw data location a sink of this write carries ({@code TIcebergTableSink.original_output_path} + * or {@code TIcebergDeleteSink.table_location}): BE writes the sink's files under the normalized form of it + * and must report them under it. Called by planWrite for every sink it builds. + */ + void addCommitLocation(String location) { + if (location == null || location.isEmpty()) { + throw new DorisConnectorException("An iceberg sink must carry the raw location it commits files under"); + } + synchronized (this) { + commitLocations.add(location); + } + } + + /** + * Rejects a fragment whose file is not recorded under a location a sink of this write carries. BE writes + * through the normalized location (s3://b/db/t/data for an oss:// table) and must report the table's own + * form (oss://b/db/t/data/x.parquet): Iceberg metadata, other engines and {@link #rollback} resolve the + * reported path with the table's file system, so the s3:// form names a file no other engine can open, and + * nothing failed when delete files were committed that way. Throwing here makes FE reject the report: the + * statement fails and BE deletes the files under the paths it wrote them at. + */ + private void checkCommitLocation(TIcebergCommitData data) { + String path = data.getFilePath(); + if (path != null) { + for (String location : commitLocations) { + if (path.startsWith(location.endsWith("/") ? location : location + "/")) { + return; + } + } + } + throw new DorisConnectorException("Iceberg commit data reports file " + path + + " outside the locations its sinks commit files under: " + commitLocations); + } + /** * O5-2: stashes the engine-extracted target-only write constraint (neutral form). It is converted to an * iceberg {@link Expression} lazily at commit time ({@link #buildWriteConstraintExpression}) — the table diff --git a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergWritePlanProvider.java b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergWritePlanProvider.java index 80a5e3e429c..53794216318 100644 --- a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergWritePlanProvider.java +++ b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergWritePlanProvider.java @@ -236,21 +236,32 @@ public class IcebergWritePlanProvider implements ConnectorWritePlanProvider { Map> rewritableDeletes = IcebergStatementScope.rewritableDeleteSupply(session); + TDataSink dataSink = buildDataSink(writeContext.getWriteOperation(), table, tableHandle, handle, + rewritableDeletes, schemaContext); + // BE reports every file of the sink under the raw location the sink carries, whichever writer + // produced it; the transaction rejects a reported file anywhere else. + commitLocationsOf(dataSink).forEach(transaction::addCommitLocation); + return new ConnectorSinkPlan(dataSink); + } + + private TDataSink buildDataSink(WriteOperation writeOperation, Table table, IcebergTableHandle tableHandle, + ConnectorWriteHandle handle, Map> rewritableDeletes, + IcebergWriteSchemaContext schemaContext) { // Dispatch on the write operation to the matching BE sink dialect (each is a distinct TDataSinkType, // byte-identical to the legacy fe-core planner sink). OVERWRITE shares TIcebergTableSink with INSERT // (the overwrite flag is read from the handle); UPDATE shares TIcebergMergeSink with MERGE. - switch (writeContext.getWriteOperation()) { + switch (writeOperation) { case INSERT: case OVERWRITE: { TDataSink dataSink = new TDataSink(TDataSinkType.ICEBERG_TABLE_SINK); dataSink.setIcebergTableSink(buildSink(table, tableHandle, handle, schemaContext)); - return new ConnectorSinkPlan(dataSink); + return dataSink; } case DELETE: { TDataSink dataSink = new TDataSink(TDataSinkType.ICEBERG_DELETE_SINK); dataSink.setIcebergDeleteSink( buildDeleteSink(table, tableHandle, rewritableDeletes, schemaContext)); - return new ConnectorSinkPlan(dataSink); + return dataSink; } case UPDATE: case MERGE: { @@ -260,7 +271,7 @@ public class IcebergWritePlanProvider implements ConnectorWritePlanProvider { handle.getBoundTargetColumns().stream() .anyMatch(column -> containsVariant(column.getType())), schemaContext)); - return new ConnectorSinkPlan(dataSink); + return dataSink; } case REWRITE: { // Compaction rewrite (ALTER TABLE ... EXECUTE rewrite_data_files): same TIcebergTableSink @@ -269,11 +280,29 @@ public class IcebergWritePlanProvider implements ConnectorWritePlanProvider { // is already evicted above. TDataSink dataSink = new TDataSink(TDataSinkType.ICEBERG_TABLE_SINK); dataSink.setIcebergTableSink(buildRewriteSink(table, tableHandle, handle)); - return new ConnectorSinkPlan(dataSink); + return dataSink; } default: - throw new DorisConnectorException( - "Unsupported iceberg write operation: " + writeContext.getWriteOperation()); + throw new DorisConnectorException("Unsupported iceberg write operation: " + writeOperation); + } + } + + /** + * The raw data locations BE commits the files of {@code sink} under, read from the field each dialect + * carries them in: data files under {@code original_output_path}, delete files and deletion vectors under + * {@code table_location}. The merge sink writes both kinds and carries both fields. + */ + private static List commitLocationsOf(TDataSink sink) { + switch (sink.getType()) { + case ICEBERG_TABLE_SINK: + return Collections.singletonList(sink.getIcebergTableSink().getOriginalOutputPath()); + case ICEBERG_DELETE_SINK: + return Collections.singletonList(sink.getIcebergDeleteSink().getTableLocation()); + case ICEBERG_MERGE_SINK: + return Arrays.asList(sink.getIcebergMergeSink().getOriginalOutputPath(), + sink.getIcebergMergeSink().getTableLocation()); + default: + throw new DorisConnectorException("Unsupported iceberg sink type: " + sink.getType()); } } diff --git a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorTransactionTest.java b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorTransactionTest.java index ba5ce1cc586..771588ecca5 100644 --- a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorTransactionTest.java +++ b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorTransactionTest.java @@ -131,8 +131,14 @@ public class IcebergConnectorTransactionTest { return ops; } + // Stands in for the raw location planWrite registers for the sink it builds; every fragment fed to a + // transaction from txnFor lives under it. + private static final String COMMIT_LOCATION = "s3://b"; + private static IcebergConnectorTransaction txnFor(RecordingIcebergCatalogOps ops, RecordingConnectorContext ctx) { - return new IcebergConnectorTransaction(42L, ops, ctx); + IcebergConnectorTransaction txn = new IcebergConnectorTransaction(42L, ops, ctx); + txn.addCommitLocation(COMMIT_LOCATION); + return txn; } // The overwrite tests below predate the static default (iceberg_partition_overwrite_mode defaults to @@ -249,6 +255,7 @@ public class IcebergConnectorTransactionTest { private static TIcebergCommitData dataItem(long affectedRows, long rowCount, TFileContent content) { TIcebergCommitData d = new TIcebergCommitData(); + d.setFilePath(COMMIT_LOCATION + "/db/t/f.parquet"); d.setRowCount(rowCount); if (affectedRows >= 0) { d.setAffectedRows(affectedRows); @@ -396,6 +403,63 @@ public class IcebergConnectorTransactionTest { () -> txn.addCommitData(new byte[] {(byte) 0xDE, (byte) 0xAD, (byte) 0xBE, (byte) 0xEF})); } + // ─────────────────── addCommitData: files must be reported under a location a sink carries ─────────────────── + + private static IcebergConnectorTransaction txnCommittingUnder(String... locations) { + IcebergConnectorTransaction txn = + new IcebergConnectorTransaction(42L, opsReturning(null), new RecordingConnectorContext()); + for (String location : locations) { + txn.addCommitLocation(location); + } + return txn; + } + + @Test + public void addCommitDataAcceptsFilesUnderEveryRegisteredLocation() { + // One location without and one with a trailing '/': write.data.path is taken verbatim. + IcebergConnectorTransaction txn = txnCommittingUnder("oss://b/db/t/data", "oss://b/db/t2/data/"); + txn.addCommitData(commitBytes(dataFileItem("oss://b/db/t/data/id=1/f.parquet", 1L, 64L))); + txn.addCommitData(commitBytes( + positionDeleteItem("oss://b/db/t2/data/delete_pos_1.parquet", 1L, "oss://b/db/t2/data/f.parquet"))); + Assertions.assertEquals(2, txn.getCommitDataList().size()); + } + + @Test + public void addCommitDataRejectsTheNormalizedAddressBeWroteTheFileAt() { + // BE writes an oss:// table through s3://; reporting that form records a file no other engine opens. + IcebergConnectorTransaction txn = txnCommittingUnder("oss://b/db/t/data"); + DorisConnectorException ex = Assertions.assertThrows(DorisConnectorException.class, + () -> txn.addCommitData(commitBytes(positionDeleteItem( + "s3://b/db/t/data/delete_pos_1.parquet", 1L, "oss://b/db/t/data/f.parquet")))); + Assertions.assertTrue(ex.getMessage().contains("s3://b/db/t/data/delete_pos_1.parquet"), ex.getMessage()); + Assertions.assertTrue(ex.getMessage().contains("oss://b/db/t/data"), ex.getMessage()); + Assertions.assertTrue(txn.getCommitDataList().isEmpty(), "a rejected fragment must not be accumulated"); + } + + @Test + public void addCommitDataRejectsASiblingDirectorySharingTheLocationPrefix() { + IcebergConnectorTransaction txn = txnCommittingUnder("oss://b/db/t/data"); + Assertions.assertThrows(DorisConnectorException.class, + () -> txn.addCommitData(commitBytes(dataFileItem("oss://b/db/t/data_old/f.parquet", 1L, 64L)))); + } + + @Test + public void addCommitDataRejectsAFragmentWithoutAPathOrBeforeAnySinkWasPlanned() { + IcebergConnectorTransaction planned = txnCommittingUnder("oss://b/db/t/data"); + Assertions.assertThrows(DorisConnectorException.class, + () -> planned.addCommitData(commitBytes(dataItem(1L, 1L, TFileContent.DATA).setFilePath(null)))); + IcebergConnectorTransaction unplanned = txnCommittingUnder(); + Assertions.assertThrows(DorisConnectorException.class, + () -> unplanned.addCommitData(commitBytes(dataFileItem("oss://b/db/t/data/f.parquet", 1L, 64L)))); + } + + @Test + public void addCommitLocationRejectsAnEmptyLocation() { + IcebergConnectorTransaction txn = txnCommittingUnder(); + Assertions.assertThrows(DorisConnectorException.class, () -> txn.addCommitLocation("")); + Assertions.assertThrows(DorisConnectorException.class, () -> txn.addCommitLocation(null)); + } + // ─────────────────── getUpdateCnt: data/delete split, affectedRows priority (T03, regression) ─────────────────── @Test @@ -1180,7 +1244,7 @@ public class IcebergConnectorTransactionTest { long snapshotId = table.currentSnapshot().snapshotId(); IcebergConnectorTransaction txn = txnFor(opsReturning(table), new RecordingConnectorContext()); txn.beginWrite(SESSION, "db1", "t1", overwriteStaticCtx(table, Collections.singletonMap("ts", "2024-01-02"))); - txn.addCommitData(commitBytes(dataFileItem("s3://bucket/replacement.parquet", 1L, 100L, + txn.addCommitData(commitBytes(dataFileItem("s3://b/replacement.parquet", 1L, 100L, Collections.singletonList("2024-01-02 00:00:00.000000000")))); // DATE metrics are days, so silently committing a nanos equality would retain the old file. DorisConnectorException ex = Assertions.assertThrows(DorisConnectorException.class, txn::commit); @@ -1206,7 +1270,7 @@ public class IcebergConnectorTransactionTest { } IcebergConnectorTransaction txn = txnFor(opsReturning(table), new RecordingConnectorContext()); txn.beginWrite(SESSION, "db1", "t1", overwriteStaticCtx(table, Collections.singletonMap("ts", "2024-01-01"))); - txn.addCommitData(commitBytes(dataFileItem("s3://bucket/replacement.parquet", 1L, 1024L, + txn.addCommitData(commitBytes(dataFileItem("s3://b/replacement.parquet", 1L, 1024L, Collections.singletonList("2024-01-01 00:00:00.000000000")))); txn.commit(); List remaining = currentDataFiles(catalog.loadTable(id)); @@ -1233,7 +1297,7 @@ public class IcebergConnectorTransactionTest { IcebergConnectorTransaction txn = txnFor(opsReturning(table), new RecordingConnectorContext()); txn.beginWrite(SESSION, "db1", "t1", overwriteStaticCtx(table, Collections.singletonMap("ts", "2024-01-01 00:00:00+08:00"))); - txn.addCommitData(commitBytes(dataFileItem("s3://bucket/replacement.parquet", 1L, 1024L, + txn.addCommitData(commitBytes(dataFileItem("s3://b/replacement.parquet", 1L, 1024L, Collections.singletonList("2023-12-31 16:00:00.000000000")))); txn.commit(); List remaining = currentDataFiles(catalog.loadTable(id)); diff --git a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWriteKnownBugsTest.java b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWriteKnownBugsTest.java index f18a79a4e95..a1ae89bc1ef 100644 --- a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWriteKnownBugsTest.java +++ b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWriteKnownBugsTest.java @@ -45,6 +45,9 @@ public class IcebergWriteKnownBugsTest { private static final String SEED = "s3://bucket/events/seed.parquet"; private static final String ADDED = "s3://bucket/events/added.parquet"; private static final String DELETED = "s3://bucket/events/deleted.parquet"; + // Stands in for the raw location planWrite registers for the sink it builds; every reported file + // above lives under it. + private static final String COMMIT_LOCATION = "s3://bucket/events"; @Test @Disabled("Known issue: DELETE ignores the standard snapshot isolation property.") @@ -113,7 +116,10 @@ public class IcebergWriteKnownBugsTest { private static IcebergConnectorTransaction transaction(Table table) { RecordingIcebergCatalogOps operations = new RecordingIcebergCatalogOps(); operations.table = table; - return new IcebergConnectorTransaction(9, operations, new RecordingConnectorContext()); + IcebergConnectorTransaction transaction = + new IcebergConnectorTransaction(9, operations, new RecordingConnectorContext()); + transaction.addCommitLocation(COMMIT_LOCATION); + return transaction; } private static TIcebergCommitData fragment(String path, long count, TFileContent content) { diff --git a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWritePlanProviderTest.java b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWritePlanProviderTest.java index b0eb9bee169..26ae3e6fd06 100644 --- a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWritePlanProviderTest.java +++ b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWritePlanProviderTest.java @@ -44,6 +44,7 @@ import org.apache.doris.thrift.TFileCompressType; import org.apache.doris.thrift.TFileContent; import org.apache.doris.thrift.TFileFormatType; import org.apache.doris.thrift.TFileType; +import org.apache.doris.thrift.TIcebergCommitData; import org.apache.doris.thrift.TIcebergDeleteFileDesc; import org.apache.doris.thrift.TIcebergDeleteSink; import org.apache.doris.thrift.TIcebergMergeSink; @@ -83,6 +84,9 @@ import org.apache.iceberg.io.OutputFile; import org.apache.iceberg.types.Type; import org.apache.iceberg.types.Types; import org.apache.iceberg.util.DateTimeUtil; +import org.apache.thrift.TException; +import org.apache.thrift.TSerializer; +import org.apache.thrift.protocol.TBinaryProtocol; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; @@ -2491,6 +2495,68 @@ public class IcebergWritePlanProviderTest { Assertions.assertTrue(update.isWritesDataFiles()); } + // BE writes position-delete and deletion-vector files under output_path and commits them under + // table_location (VIcebergDeleteSink::_commit_path_of), so both must name the same directory, the + // data location, differing only in its normalized vs raw form. If table_location became the table + // root, BE would record delete files without the data directory, i.e. files that do not exist, + // and nothing would fail. See TIcebergDeleteSink.table_location in DataSinks.thrift. + @Test + public void planWriteDeleteAndMergeSinksKeepTableLocationInTheOutputDirectory() { + Table table = partitionedSortedTable(freshCatalog()); + TIcebergDeleteSink deleteSink = planDeleteSink(table, contextWithStorage(), + new WriteHandle(new IcebergTableHandle("db1", "t1")).writeOperation(WriteOperation.DELETE)); + TIcebergMergeSink mergeSink = planMergeSink(table, contextWithStorage(), + new WriteHandle(new IcebergTableHandle("db1", "t1")).writeOperation(WriteOperation.UPDATE)); + + Assertions.assertEquals(pathAfterScheme(deleteSink.getOutputPath()), + pathAfterScheme(deleteSink.getTableLocation()), + "delete table_location must be the raw form of output_path's directory"); + Assertions.assertEquals(pathAfterScheme(mergeSink.getOutputPath()), + pathAfterScheme(mergeSink.getTableLocation()), + "merge table_location must be the raw form of output_path's directory"); + } + + // "s3://bucket/wh/db1/t1/data" and "oss://bucket/wh/db1/t1/data" -> "bucket/wh/db1/t1/data". + private static String pathAfterScheme(String location) { + return location.substring(location.indexOf("://") + 3); + } + + // planWrite registers the raw location each sink dialect carries with the transaction, so a file BE + // reports under the normalized address it wrote (the s3:// form of this oss:// table, as delete files once + // were) fails the statement instead of being recorded where other engines cannot resolve it. + @Test + public void planWriteRegistersTheCommitLocationOfEverySinkDialect() { + for (WriteOperation operation : Arrays.asList(WriteOperation.INSERT, WriteOperation.REWRITE, + WriteOperation.DELETE, WriteOperation.UPDATE)) { + Table table = partitionedSortedTable(freshCatalog()); + RecordingConnectorContext ctx = contextWithStorage(); + RecordingIcebergCatalogOps ops = new RecordingIcebergCatalogOps(); + ops.table = table; + IcebergConnectorTransaction txn = new IcebergConnectorTransaction(42L, ops, ctx); + providerFor(table, ctx).planWrite(new WriteSession(txn), + new WriteHandle(new IcebergTableHandle("db1", "t1")).writeOperation(operation)); + + txn.addCommitData(commitFragment("oss://bucket/wh/db1/t1/data/id=1/f.parquet")); + DorisConnectorException rejected = Assertions.assertThrows(DorisConnectorException.class, + () -> txn.addCommitData(commitFragment("s3://bucket/wh/db1/t1/data/id=1/f.parquet")), + operation + " must reject a file reported under the normalized location"); + Assertions.assertTrue(rejected.getMessage().contains("s3://bucket/wh/db1/t1/data/id=1/f.parquet"), + rejected.getMessage()); + Assertions.assertEquals(1, txn.getCommitDataList().size(), operation.toString()); + } + } + + private static byte[] commitFragment(String filePath) { + TIcebergCommitData data = new TIcebergCommitData(); + data.setFilePath(filePath); + data.setFileContent(TFileContent.DATA); + try { + return new TSerializer(new TBinaryProtocol.Factory()).serialize(data); + } catch (TException e) { + throw new AssertionError(e); + } + } + @Test public void planWriteBuildsMergeSinkWithTableDerivedFields() { Table table = partitionedSortedTable(freshCatalog()); diff --git a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWriteTransactionCoverageTest.java b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWriteTransactionCoverageTest.java index 83412d30c0c..b29e8e3848d 100644 --- a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWriteTransactionCoverageTest.java +++ b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWriteTransactionCoverageTest.java @@ -58,6 +58,9 @@ public class IcebergWriteTransactionCoverageTest { private static final String SEED = "s3://bucket/events/seed.parquet"; private static final String ADDED = "s3://bucket/events/added.parquet"; private static final String DELETED = "s3://bucket/events/deleted.parquet"; + // Stands in for the raw location planWrite registers for the sink it builds; every reported file + // above lives under it. + private static final String COMMIT_LOCATION = "s3://bucket/events"; @Test public void appendV1PreservesHistoricalSnapshot() throws IOException { @@ -252,6 +255,7 @@ public class IcebergWriteTransactionCoverageTest { operations.throwOnLoadTable = true; IcebergConnectorTransaction transaction = new IcebergConnectorTransaction( 7, operations, new RecordingConnectorContext()); + transaction.addCommitLocation(COMMIT_LOCATION); Assertions.assertThrows(DorisConnectorException.class, () -> transaction.beginWrite(session(null), "db", "events", context(WriteOperation.INSERT))); Assertions.assertNull(transaction.getTransaction()); @@ -321,7 +325,10 @@ public class IcebergWriteTransactionCoverageTest { private static IcebergConnectorTransaction transaction(Table table) { RecordingIcebergCatalogOps operations = new RecordingIcebergCatalogOps(); operations.table = table; - return new IcebergConnectorTransaction(7, operations, new RecordingConnectorContext()); + IcebergConnectorTransaction transaction = + new IcebergConnectorTransaction(7, operations, new RecordingConnectorContext()); + transaction.addCommitLocation(COMMIT_LOCATION); + return transaction; } private static IcebergWriteContext context(WriteOperation operation) { diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/QeProcessorImpl.java b/fe/fe-core/src/main/java/org/apache/doris/qe/QeProcessorImpl.java index babfb4b2fcb..06f13007bcb 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/qe/QeProcessorImpl.java +++ b/fe/fe-core/src/main/java/org/apache/doris/qe/QeProcessorImpl.java @@ -324,13 +324,15 @@ public final class QeProcessorImpl implements QeProcessor { try { boolean accepted = info.getCoord().updateFragmentExecStatus(params); if (hasExternalCommitData && !accepted) { - return rejectedExternalFileReport(result, "FE has not accepted the external-file report"); + return rejectExternalFileReport(info.getCoord(), result, + "FE has not accepted the external-file report"); } } catch (Exception e) { LOG.warn("Exception during handle report, response: {}, query: {}, instance: {}", result.toString(), DebugUtil.printId(params.query_id), DebugUtil.printId(params.fragment_instance_id), e); return hasExternalCommitData - ? rejectedExternalFileReport(result, "FE did not accept the external-file report") + ? rejectExternalFileReport(info.getCoord(), result, + "FE did not accept the external-file report: " + e.getMessage()) : result; } result.setStatus(new TStatus(TStatusCode.OK)); @@ -354,6 +356,14 @@ public final class QeProcessorImpl implements QeProcessor { + params.getFragmentId() + ":" + params.getBackendId(); } + // A rejected final report is terminal: BE drops its files and never resends it, so the statement + // must fail now instead of waiting for a fragment that can no longer finish. + private static TReportExecStatusResult rejectExternalFileReport( + CoordInterface coord, TReportExecStatusResult result, String message) { + coord.cancel(new Status(TStatusCode.INTERNAL_ERROR, message)); + return rejectedExternalFileReport(result, message); + } + private static TReportExecStatusResult rejectedExternalFileReport( TReportExecStatusResult result, String message) { TStatus status = new TStatus(TStatusCode.INTERNAL_ERROR); diff --git a/fe/fe-core/src/test/java/org/apache/doris/qe/QeProcessorImplReportAckTest.java b/fe/fe-core/src/test/java/org/apache/doris/qe/QeProcessorImplReportAckTest.java index c4abc116445..103fffa297f 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/qe/QeProcessorImplReportAckTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/qe/QeProcessorImplReportAckTest.java @@ -17,6 +17,7 @@ package org.apache.doris.qe; +import org.apache.doris.common.Status; import org.apache.doris.common.profile.ExecutionProfile; import org.apache.doris.planner.PlanFragmentId; import org.apache.doris.system.Backend; @@ -34,6 +35,7 @@ import com.google.common.cache.Cache; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; import org.mockito.Mockito; import java.lang.reflect.Field; @@ -82,17 +84,34 @@ class QeProcessorImplReportAckTest { Assertions.assertEquals(TStatusCode.INTERNAL_ERROR, result.getStatus().getStatusCode()); Assertions.assertFalse(result.isExternalFileCommitDataAccepted()); + // BE drops the files of a rejected final report, so the statement must fail instead of waiting. + Status cancelReason = verifyCancelled(coordinator); + Assertions.assertEquals(TStatusCode.INTERNAL_ERROR, cancelReason.getErrorCode()); + Assertions.assertTrue(cancelReason.getErrorMsg().contains("injected acceptance failure")); } @Test void rejectsExternalReportWhenHandlerDoesNotAcceptIt() throws Exception { TUniqueId queryId = new TUniqueId(12345, 4); - register(queryId); + Coordinator coordinator = register(queryId); TReportExecStatusResult result = report(params(queryId)); Assertions.assertEquals(TStatusCode.INTERNAL_ERROR, result.getStatus().getStatusCode()); Assertions.assertFalse(result.isExternalFileCommitDataAccepted()); + Assertions.assertEquals(TStatusCode.INTERNAL_ERROR, verifyCancelled(coordinator).getErrorCode()); + } + + @Test + void acceptedExternalReportDoesNotCancelCoordinator() throws Exception { + TUniqueId queryId = new TUniqueId(12345, 8); + Coordinator coordinator = register(queryId); + Mockito.when(coordinator.updateFragmentExecStatus(Mockito.any())).thenReturn(true); + + TReportExecStatusResult result = report(params(queryId)); + + Assertions.assertTrue(result.isExternalFileCommitDataAccepted()); + Mockito.verify(coordinator, Mockito.never()).cancel(Mockito.any()); } @Test @@ -186,6 +205,12 @@ class QeProcessorImplReportAckTest { return coordinator; } + private static Status verifyCancelled(Coordinator coordinator) { + ArgumentCaptor reason = ArgumentCaptor.forClass(Status.class); + Mockito.verify(coordinator, Mockito.times(1)).cancel(reason.capture()); + return reason.getValue(); + } + private static TReportExecStatusParams params(TUniqueId queryId) { return new TReportExecStatusParams() .setQueryId(queryId) diff --git a/gensrc/thrift/DataSinks.thrift b/gensrc/thrift/DataSinks.thrift index dd0dbabc51a..c4f303c118b 100644 --- a/gensrc/thrift/DataSinks.thrift +++ b/gensrc/thrift/DataSinks.thrift @@ -439,6 +439,9 @@ struct TIcebergColumnStats { } struct TIcebergCommitData { + // The path Iceberg metadata records for the file: under the table's own form of the location + // the sink carries (TIcebergTableSink.original_output_path, TIcebergDeleteSink.table_location), + // never under the normalized output_path BE wrote it at. FE rejects a report outside it. 1: optional string file_path // File-level record count used to construct Iceberg data/delete file metadata. 2: optional i64 row_count @@ -493,6 +496,12 @@ struct TIcebergTableSink { 9: optional map hadoop_config 10: optional bool overwrite 11: optional Types.TFileType file_type + // The raw (un-normalized) data location: the same directory as output_path, in the table's + // own form. BE writes data files under output_path and commits them under + // original_output_path, e.g. written to s3://b/db/t/data/p=1/x.parquet, committed as + // oss://b/db/t/data/p=1/x.parquet. "Same directory" is not "same text after the scheme": + // normalization can also rewrite the authority, e.g. abfss://c@acct.dfs.core.windows.net/t/data + // becomes s3://c/t/data. 12: optional string original_output_path 13: optional PlanNodes.TFileCompressType compression_type 14: optional list broker_addresses; @@ -530,6 +539,11 @@ struct TIcebergDeleteSink { 5: optional PlanNodes.TFileFormatType file_format 6: optional PlanNodes.TFileCompressType compress_type 7: optional string output_path + // The raw (un-normalized) data location: the same directory as output_path, in the table's + // own form. BE writes delete files under output_path and commits them under table_location, + // e.g. written to s3://b/db/t/data/delete_x.parquet, committed as oss://b/db/t/data/delete_x.parquet. + // It must not be the table root, or committed paths would miss the data directory. See + // TIcebergTableSink.original_output_path for what "same directory" means. 8: optional string table_location 9: optional map hadoop_config 10: optional Types.TFileType file_type @@ -554,6 +568,8 @@ struct TIcebergMergeSink { 7: optional PlanNodes.TFileFormatType file_format 8: optional PlanNodes.TFileCompressType compression_type 9: optional string output_path + // Same contract as TIcebergTableSink.original_output_path, for the data files; FE sets it to + // the same value as table_location, which the delete files are committed under. 10: optional string original_output_path 11: optional map hadoop_config 12: optional Types.TFileType file_type @@ -573,6 +589,7 @@ struct TIcebergMergeSink { // delete side (position delete only) 20: optional TFileContent delete_type + // Same contract as TIcebergDeleteSink.table_location: the raw form of output_path's directory. 21: optional string table_location 22: optional i32 partition_spec_id_for_delete 23: optional string partition_data_json_for_delete diff --git a/regression-test/data/external_table_p0/iceberg/dml/test_iceberg_update_delete_basic.out.org b/regression-test/data/external_table_p0/iceberg/dml/test_iceberg_update_delete_basic.out.org deleted file mode 100644 index c42c6138a81..00000000000 --- a/regression-test/data/external_table_p0/iceberg/dml/test_iceberg_update_delete_basic.out.org +++ /dev/null @@ -1,77 +0,0 @@ --- This file is automatically generated. You should know what you did if you want to edit this --- !parquet_q01 -- -1 - --- !order_parquet_q02 -- -1 Updated 25 -2 Bob 30 -3 Charlie 35 - --- !parquet_q03 -- -1 - --- !order_parquet_q04 -- -3 Charlie 35 -1 Updated 25 - --- !order_parquet_q05 -- -1 s3://warehouse/wh/test_update_delete_basic_db/test_update_delete_basic_tbl_parquet/data/delete_pos_5066988_0553b2d5.zstd.parquet PARQUET 0 1 2078 \N \N \N \N \N \N \N \N \N \N \N \N \N \N {"age":{"column_size":null, "value_count":null, "null_value_count":null, "nan_value_count":null, "lower_bound":null, "upper_bound":null}, "id":{"column_size":null, "value_count":null, "null_value_count":null, "nan_value_count":null, "lower_bound":null, "upper_bound":null}, "name":{"column_size":null, "value_count":null, "null_value_count":null, "nan_value_count":null, "lower_bound":null, "upper_bound":null}} -1 s3://warehouse/wh/test_update_delete_basic_db/test_update_delete_basic_tbl_parquet/data/delete_pos_5066988_3638addc.zstd.parquet PARQUET 0 1 2078 \N \N \N \N \N \N \N \N \N \N \N \N \N \N {"age":{"column_size":null, "value_count":null, "null_value_count":null, "nan_value_count":null, "lower_bound":null, "upper_bound":null}, "id":{"column_size":null, "value_count":null, "null_value_count":null, "nan_value_count":null, "lower_bound":null, "upper_bound":null}, "name":{"column_size":null, "value_count":null, "null_value_count":null, "nan_value_count":null, "lower_bound":null, "upper_bound":null}} - --- !parquet_q06 -- -1 - --- !order_parquet_q07 -- -12 Cat 22 2024-01-03 -11 Ben 21 2024-01-02 -10 UpdatedP 20 2024-01-01 - --- !parquet_q08 -- -1 - --- !order_parquet_q09 -- -12 Cat 22 2024-01-03 -10 UpdatedP 20 2024-01-01 - --- !order_parquet_q10 -- -1 s3://warehouse/wh/test_update_delete_basic_db/test_update_delete_basic_tbl_par_parquet/data/delete_pos_5092880_cd626510.zstd.parquet PARQUET 0 {"dt_day":"2024-01-01"} 1 2226 \N \N \N \N \N \N \N \N \N \N \N \N \N \N {"age":{"column_size":null, "value_count":null, "null_value_count":null, "nan_value_count":null, "lower_bound":null, "upper_bound":null}, "dt":{"column_size":null, "value_count":null, "null_value_count":null, "nan_value_count":null, "lower_bound":null, "upper_bound":null}, "id":{"column_size":null, "value_count":null, "null_value_count":null, "nan_value_count":null, "lower_bound":null, "upper_bound":null}, "name":{"column_size":null, "value_count":null, "null_value_count":null, "nan_value_count":null, "lower_bound":null, "upper_bound":null}} -1 s3://warehouse/wh/test_update_delete_basic_db/test_update_delete_basic_tbl_par_parquet/data/delete_pos_7425526_6dd1173c.zstd.parquet PARQUET 0 {"dt_day":"2024-01-02"} 1 2225 \N \N \N \N \N \N \N \N \N \N \N \N \N \N {"age":{"column_size":null, "value_count":null, "null_value_count":null, "nan_value_count":null, "lower_bound":null, "upper_bound":null}, "dt":{"column_size":null, "value_count":null, "null_value_count":null, "nan_value_count":null, "lower_bound":null, "upper_bound":null}, "id":{"column_size":null, "value_count":null, "null_value_count":null, "nan_value_count":null, "lower_bound":null, "upper_bound":null}, "name":{"column_size":null, "value_count":null, "null_value_count":null, "nan_value_count":null, "lower_bound":null, "upper_bound":null}} - --- !orc_q01 -- -1 - --- !order_orc_q02 -- -1 Updated 25 -2 Bob 30 -3 Charlie 35 - --- !orc_q03 -- -1 - --- !order_orc_q04 -- -3 Charlie 35 -1 Updated 25 - --- !order_orc_q05 -- -1 s3://warehouse/wh/test_update_delete_basic_db/test_update_delete_basic_tbl_orc/data/delete_pos_4451312_c1ffb646.zlib.orc ORC 0 1 926 \N \N \N \N \N \N \N \N \N \N \N \N \N \N {"age":{"column_size":null, "value_count":null, "null_value_count":null, "nan_value_count":null, "lower_bound":null, "upper_bound":null}, "id":{"column_size":null, "value_count":null, "null_value_count":null, "nan_value_count":null, "lower_bound":null, "upper_bound":null}, "name":{"column_size":null, "value_count":null, "null_value_count":null, "nan_value_count":null, "lower_bound":null, "upper_bound":null}} -1 s3://warehouse/wh/test_update_delete_basic_db/test_update_delete_basic_tbl_orc/data/delete_pos_4451312_c5f91315.zlib.orc ORC 0 1 924 \N \N \N \N \N \N \N \N \N \N \N \N \N \N {"age":{"column_size":null, "value_count":null, "null_value_count":null, "nan_value_count":null, "lower_bound":null, "upper_bound":null}, "id":{"column_size":null, "value_count":null, "null_value_count":null, "nan_value_count":null, "lower_bound":null, "upper_bound":null}, "name":{"column_size":null, "value_count":null, "null_value_count":null, "nan_value_count":null, "lower_bound":null, "upper_bound":null}} - --- !orc_q06 -- -1 - --- !order_orc_q07 -- -11 Ben 21 2024-01-02 -12 Cat 22 2024-01-03 -10 UpdatedP 20 2024-01-01 - --- !orc_q08 -- -1 - --- !order_orc_q09 -- -12 Cat 22 2024-01-03 -10 UpdatedP 20 2024-01-01 - --- !order_orc_q10 -- -1 s3://warehouse/wh/test_update_delete_basic_db/test_update_delete_basic_tbl_par_orc/data/delete_pos_7724320_d7587f50.zlib.orc ORC 0 {"dt_day":"2024-01-01"} 1 986 \N \N \N \N \N \N \N \N \N \N \N \N \N \N {"age":{"column_size":null, "value_count":null, "null_value_count":null, "nan_value_count":null, "lower_bound":null, "upper_bound":null}, "dt":{"column_size":null, "value_count":null, "null_value_count":null, "nan_value_count":null, "lower_bound":null, "upper_bound":null}, "id":{"column_size":null, "value_count":null, "null_value_count":null, "nan_value_count":null, "lower_bound":null, "upper_bound":null}, "name":{"column_size":null, "value_count":null, "null_value_count":null, "nan_value_count":null, "lower_bound":null, "upper_bound":null}} -1 s3://warehouse/wh/test_update_delete_basic_db/test_update_delete_basic_tbl_par_orc/data/delete_pos_817404_2d2aa46c.zlib.orc ORC 0 {"dt_day":"2024-01-02"} 1 990 \N \N \N \N \N \N \N \N \N \N \N \N \N \N {"age":{"column_size":null, "value_count":null, "null_value_count":null, "nan_value_count":null, "lower_bound":null, "upper_bound":null}, "dt":{"column_size":null, "value_count":null, "null_value_count":null, "nan_value_count":null, "lower_bound":null, "upper_bound":null}, "id":{"column_size":null, "value_count":null, "null_value_count":null, "nan_value_count":null, "lower_bound":null, "upper_bound":null}, "name":{"column_size":null, "value_count":null, "null_value_count":null, "nan_value_count":null, "lower_bound":null, "upper_bound":null}} - diff --git a/regression-test/suites/external_table_p0/iceberg/dml/test_iceberg_update_delete_basic.groovy b/regression-test/suites/external_table_p0/iceberg/dml/test_iceberg_update_delete_basic.groovy index 21384862153..9ef8c30b1c8 100644 --- a/regression-test/suites/external_table_p0/iceberg/dml/test_iceberg_update_delete_basic.groovy +++ b/regression-test/suites/external_table_p0/iceberg/dml/test_iceberg_update_delete_basic.groovy @@ -47,6 +47,15 @@ suite("test_iceberg_update_delete_basic", "p0,external,iceberg,external_docker,e sql """create database if not exists ${dbName}""" sql """use ${dbName}""" + // Delete files must be recorded under the table's own data location (s3a:// in this fixture), not + // under the s3:// address BE writes them at. + def dataLocationOf = { String table -> + String ddl = sql("""show create table ${table}""")[0][1] + def matcher = ddl =~ /LOCATION '([^']+)'/ + assert matcher.find() : "no LOCATION in ${ddl}" + return matcher.group(1) + "/data/" + } + def formats = ["parquet", "orc"] for (String format : formats) { logger.info("Run update/delete test with format ${format}") @@ -94,10 +103,11 @@ suite("test_iceberg_update_delete_basic", "p0,external,iceberg,external_docker,e FROM ${catalogName}.${dbName}.${formatTableName}\$delete_files """ assert deleteFiles.size() > 0 : "Delete files should be created for ${formatTableName}" + String dataLocation = dataLocationOf(formatTableName) for (def row : deleteFiles) { String filePath = row[0].toString() String fileFormat = row[1].toString() - assert filePath.contains("/data/delete_pos_") + assert filePath.startsWith(dataLocation + "delete_pos_") : "${filePath} is not under ${dataLocation}" assert filePath.endsWith(format == "parquet" ? ".parquet" : ".orc") assert fileFormat.equalsIgnoreCase(format) } @@ -142,10 +152,12 @@ suite("test_iceberg_update_delete_basic", "p0,external,iceberg,external_docker,e FROM ${catalogName}.${dbName}.${formatTableNamePartition}\$delete_files """ assert partitionDeleteFiles.size() > 0 : "Delete files should be created for ${formatTableNamePartition}" + String partitionDataLocation = dataLocationOf(formatTableNamePartition) for (def row : partitionDeleteFiles) { String filePath = row[0].toString() String fileFormat = row[1].toString() - assert filePath.contains("/data/delete_pos_") + assert filePath.startsWith(partitionDataLocation + "delete_pos_") : + "${filePath} is not under ${partitionDataLocation}" assert filePath.endsWith(format == "parquet" ? ".parquet" : ".orc") assert fileFormat.equalsIgnoreCase(format) } diff --git a/regression-test/suites/external_table_p0/iceberg/write/ICEBERG_WRITE_P0_COVERAGE.md b/regression-test/suites/external_table_p0/iceberg/write/ICEBERG_WRITE_P0_COVERAGE.md index 06a7281bfb9..f740652d960 100644 --- a/regression-test/suites/external_table_p0/iceberg/write/ICEBERG_WRITE_P0_COVERAGE.md +++ b/regression-test/suites/external_table_p0/iceberg/write/ICEBERG_WRITE_P0_COVERAGE.md @@ -53,6 +53,7 @@ Coverage status: | R16 | CTAS behaves inconsistently for complex types, NULL, partition transforms, file formats and failure cleanup | Black box: DDL and writer committed together | Schema, file format, atomicity | P0 | | R17 | Combining sort order, distribution mode, multiple file flushes and concurrent commits causes misordering, lost rows or duplicate commits | White box: exchange, sort writer, optimistic commit | Distributed correctness, stability | P0 | | R18 | STRING identity/bucket/truncate computes wrong physical partition values for empty strings, Chinese text, emoji, combining characters and NULL | Black box: UTF-8 transform metadata | Partition routing, pruning | P0 | +| R19 | Written data files, position deletes or deletion vectors are recorded in metadata under the normalized address BE actually wrote them at (such as s3://) instead of the table's own location (such as s3a:// or oss://) | White box: sink write path versus commit path | Whether other engines and maintenance operations can open the files | P0 | ## Combined coverage @@ -87,6 +88,7 @@ Coverage status: | Sort and distribution properties | Multi-column sort order, NULL ordering, none/hash/range distribution, forced multi-file flush | Covered (verified) | `test_iceberg_write_order_distribution_properties` | | Concurrent writes | Serializability invariant of conflicting MERGEs on the same row, non-conflicting distributed appends | Covered (verified) | `test_iceberg_write_concurrent_merge_invariants` | | Distributed execution | Multi-bucket source tables, multi-partition Iceberg sinks, multiple BE writers, no catalog/database shared between suites | Covered (verified) | All suites added by this work | +| File path recording | Every file written by INSERT, partition OVERWRITE, branch INSERT, DELETE, UPDATE, MERGE and rewrite_data_files (v2 position deletes, v3 deletion vectors) is recorded under the table's data location | Covered (regular regression) | `test_iceberg_write_file_locations` | | Spark cross-check | Fifteen positive suites compare row by row after the last write; the STRING transform suite also compares physical partition metadata before and after evolution | Covered (verified) | The fifteen positive suites; physical metadata in `test_iceberg_write_string_transform_metadata` | ## Design of the added cases @@ -112,9 +114,10 @@ Coverage status: | W17 | Verify sort order, distribution mode and multi-file flush | R03, R11, R17 | Correctness, stability | Multi-BE Doris | NULL sort keys, multi-column ascending/descending order, none/hash/range, low target file size | The plan contains the declared sort; the total row count across files is correct; the three distribution modes agree | | W18 | Verify commit invariants of concurrent MERGE and append | R11, R13, R17 | Concurrency, atomicity | Multi-BE Doris | A readiness barrier makes two independent sessions ready to dispatch at the same time; updates of the same row and non-conflicting appends | Commits on the same row are serializable with cardinality one; only Iceberg validation/commit conflicts are accepted; non-conflicting writes are neither lost nor duplicated | | W19 | Verify type safety of the MERGE source projection entering the truncate transform | R12, R18 | Regular regression, stability | Iceberg v2 MOR | Matched updates and unmatched inserts on a required STRING truncate column | MERGE accepts non-NULL values in the nullable physical projection and produces correct physical partitions | +| W20 | Verify that every write operation records its file paths under the table's own location | R19 | Correctness, compatibility | The REST fixture's warehouse is s3a://, which BE normalizes to s3:// | INSERT, partition OVERWRITE, branch INSERT, DELETE, UPDATE, MERGE and rewrite_data_files in turn on v2 and v3 tables | Every file in `$all_files` starts with the LOCATION from `SHOW CREATE TABLE` plus `/data/`; new write operations are added to this case too | ## P0 coverage check -Each of R01-R18 maps to at least one P0 regression. After the last successful write, all fifteen positive suites cross-check the logical result of the same table between Spark and Doris; transforms, rollover, CTAS properties and failure atomicity additionally have physical metadata or state oracles. W08, W10 and W19 run as regular regressions; W07, which may still write unreadable required-null files, keeps its isolation switch so that it does not damage a shared cluster. +Each of R01-R19 maps to at least one P0 regression. After the last successful write, all fifteen positive suites cross-check the logical result of the same table between Spark and Doris; transforms, rollover, CTAS properties and failure atomicity additionally have physical metadata or state oracles. W08, W10 and W19 run as regular regressions; W07, which may still write unreadable required-null files, keeps its isolation switch so that it does not damage a shared cluster. This matrix has no uncovered items. COW row-level DML, branch row-level DML, tag writes and Avro writes are explicit current capability boundaries, each pinned by an expected-rejection case that fixes the error semantics and failure atomicity; only the known required-null defect keeps using an isolated negative regression. diff --git a/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_file_locations.groovy b/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_file_locations.groovy new file mode 100644 index 00000000000..b8cc5203d6c --- /dev/null +++ b/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_file_locations.groovy @@ -0,0 +1,161 @@ +// 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. + +// Every file Doris writes into an Iceberg table must be recorded under the table's own location, not +// under the address BE wrote it at. The REST fixture's warehouse is s3a://warehouse/wh/, which FE +// normalizes to s3:// for BE, so a writer that records the s3:// address is caught here: delete files +// and deletion vectors were once committed that way. Each write operation Doris supports runs below; +// a new one belongs here too. +suite("test_iceberg_write_file_locations", "p0,external,iceberg,external_docker,external_docker_iceberg") { + 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_write_file_locations" + String dbName = "iceberg_write_file_locations_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" + ) + """ + sql """switch ${catalogName}""" + sql """drop database if exists ${dbName} force""" + sql """create database ${dbName}""" + sql """use ${dbName}""" + + // The table's data location, taken from its metadata rather than from any file Doris wrote. + def dataLocationOf = { String table -> + String ddl = sql("""show create table ${table}""")[0][1] + def matcher = ddl =~ /LOCATION '([^']+)'/ + assertTrue(matcher.find(), "no LOCATION in ${ddl}") + return matcher.group(1) + "/data/" + } + + // Every file of every snapshot, branches included: data files, position deletes and deletion vectors. + def assertFilesUnderDataLocation = { String table, String operation -> + String dataLocation = dataLocationOf(table) + def files = sql """select content, file_format, file_path from ${table}\$all_files""" + assertTrue(files.size() > 0, "no files in ${table} after ${operation}") + for (def file : files) { + String path = file[2].toString() + assertTrue(path.startsWith(dataLocation), + "${operation} recorded ${file[1]} file ${path} (content ${file[0]}) outside ${dataLocation}") + } + } + + def countFiles = { String table, String predicate -> + return (sql """select count(*) from ${table}\$all_files where ${predicate}""")[0][0] as long + } + + // Format v2: data files and position delete files. + sql """drop table if exists file_locations_v2""" + sql """ + create table file_locations_v2 ( + id int, + region string, + payload string + ) + partition by list (region) () + properties ( + "format-version" = "2", + "write.delete.mode" = "merge-on-read", + "write.update.mode" = "merge-on-read", + "write.merge.mode" = "merge-on-read" + ) + """ + sql """insert into file_locations_v2 values (1, 'A', 'a1'), (2, 'B', 'b1'), (3, 'C', 'c1')""" + assertFilesUnderDataLocation("file_locations_v2", "INSERT") + sql """insert into file_locations_v2 values (4, 'A', 'a2'), (5, 'B', 'b2')""" + assertFilesUnderDataLocation("file_locations_v2", "second INSERT") + sql """insert overwrite table file_locations_v2 partition (region = 'C') values (6, 'c2')""" + assertFilesUnderDataLocation("file_locations_v2", "INSERT OVERWRITE PARTITION") + sql """alter table file_locations_v2 create branch location_branch""" + sql """insert into file_locations_v2@branch(location_branch) values (7, 'A', 'branch')""" + assertFilesUnderDataLocation("file_locations_v2", "branch INSERT") + sql """delete from file_locations_v2 where id = 1""" + assertFilesUnderDataLocation("file_locations_v2", "DELETE") + sql """update file_locations_v2 set payload = 'b1-updated' where id = 2""" + assertFilesUnderDataLocation("file_locations_v2", "UPDATE") + sql """ + merge into file_locations_v2 t + using ( + select 4 as id, 'A' as region, 'a2-merged' as payload + union all + select 8, 'D', 'd1' + ) s + on t.id = s.id + when matched then update set payload = s.payload + when not matched then insert (id, region, payload) values (s.id, s.region, s.payload) + """ + assertFilesUnderDataLocation("file_locations_v2", "MERGE") + sql """ + alter table ${catalogName}.${dbName}.file_locations_v2 + execute rewrite_data_files("target-file-size-bytes" = "10485760", "min-input-files" = "2") + """ + assertFilesUnderDataLocation("file_locations_v2", "rewrite_data_files") + assertTrue(countFiles("file_locations_v2", "content = 1") > 0, "DELETE / UPDATE / MERGE wrote no delete files") + + // Format v3: row-level operations write deletion vectors into Puffin files. + sql """drop table if exists file_locations_v3""" + sql """ + create table file_locations_v3 ( + id int, + region string, + payload string + ) + partition by list (region) () + properties ( + "format-version" = "3", + "write.delete.mode" = "merge-on-read", + "write.update.mode" = "merge-on-read", + "write.merge.mode" = "merge-on-read" + ) + """ + sql """insert into file_locations_v3 values (1, 'A', 'a1'), (2, 'B', 'b1'), (3, 'C', 'c1')""" + assertFilesUnderDataLocation("file_locations_v3", "INSERT") + sql """delete from file_locations_v3 where id = 1""" + assertFilesUnderDataLocation("file_locations_v3", "DELETE") + sql """update file_locations_v3 set payload = 'b1-updated' where id = 2""" + assertFilesUnderDataLocation("file_locations_v3", "UPDATE") + sql """ + merge into file_locations_v3 t + using ( + select 3 as id, 'C' as region, 'c1-merged' as payload + union all + select 4, 'D', 'd1' + ) s + on t.id = s.id + when matched then update set payload = s.payload + when not matched then insert (id, region, payload) values (s.id, s.region, s.payload) + """ + assertFilesUnderDataLocation("file_locations_v3", "MERGE") + assertTrue(countFiles("file_locations_v3", "upper(file_format) = 'PUFFIN'") > 0, + "DELETE / UPDATE / MERGE wrote no deletion vectors") +}