[branch-4.1](fix) Backport external write correctness and Paimon struct reads

be/src/agent/be_exec_version_manager.cpp

@@ -124,8 +124,10 @@ void BeExecVersionManager::check_function_compatibility(int current_be_exec_vers
 
 // 10: start from doris 4.0.3
 //   a. use new fixed object serialization way.
+// 11: start from master
+//   a. enforce Iceberg SQL MERGE cardinality only when every executing BE supports it.
 
-const int BeExecVersionManager::max_be_exec_version = 10;
+const int BeExecVersionManager::max_be_exec_version = 11;
 const int BeExecVersionManager::min_be_exec_version = 0;
 std::map<std::string, std::set<int>> BeExecVersionManager::_function_change_map {};
 std::set<std::string> BeExecVersionManager::_function_restrict_map;

be/src/agent/be_exec_version_manager.h

@@ -33,6 +33,7 @@ constexpr inline int AGGREGATION_2_1_VERSION =
 constexpr inline int USE_CONST_SERDE =
         8; // support const column in serialize/deserialize function: PR #41175
 constexpr inline int USE_NEW_FIXED_OBJECT_SERIALIZATION_VERSION = 10;
+constexpr inline int SUPPORT_ICEBERG_MERGE_CARDINALITY_VERSION = 11;
 
 class BeExecVersionManager {
 public:

be/src/exec/sink/viceberg_delete_sink.cpp

@@ -40,6 +40,7 @@
 #include "format/transformer/vfile_format_transformer.h"
 #include "io/file_factory.h"
 #include "runtime/runtime_state.h"
+#include "util/debug_points.h"
 #include "util/slice.h"
 #include "util/string_util.h"
 #include "util/uid_util.h"
@@ -252,17 +253,26 @@ Status VIcebergDeleteSink::close(Status close_status) {
     if (!close_status.ok()) {
         LOG(WARNING) << fmt::format("VIcebergDeleteSink close with error: {}",
                                     close_status.to_string());
+        _cleanup_created_files();
         return close_status;
     }
 
+    Status write_status = Status::OK();
+    DBUG_EXECUTE_IF("VIcebergDeleteSink.close.inject_failure", {
+        write_status = Status::InternalError("injected Iceberg delete close failure");
+    });
+
     if (_delete_type == TFileContent::POSITION_DELETES && !_file_deletions.empty()) {
         SCOPED_TIMER(_write_delete_files_timer);
-        if (_format_version >= 3) {
-            RETURN_IF_ERROR(_write_deletion_vector_files(_file_deletions));
-        } else {
-            RETURN_IF_ERROR(_write_position_delete_files(_file_deletions));
+        if (write_status.ok()) {
+            write_status = _format_version >= 3 ? _write_deletion_vector_files(_file_deletions)
+                                                : _write_position_delete_files(_file_deletions);
         }
     }
+    if (!write_status.ok()) {
+        _cleanup_created_files();
+        return write_status;
+    }
 
     // Update counters
     if (_delete_file_count_counter) {
@@ -278,9 +288,33 @@ Status VIcebergDeleteSink::close(Status close_status) {
         }
     }
 
+    if (!_defer_file_cleanup_until_outer_close) {
+        _created_files.clear();
+    }
+
     return Status::OK();
 }
 
+void VIcebergDeleteSink::finish_deferred_file_cleanup(Status outer_status) {
+    if (!outer_status.ok()) {
+        _cleanup_created_files();
+    } else {
+        _created_files.clear();
+    }
+    _defer_file_cleanup_until_outer_close = false;
+}
+
+void VIcebergDeleteSink::_cleanup_created_files() {
+    for (const auto& [fs, path] : _created_files) {
+        Status delete_status = fs->delete_file(path);
+        if (!delete_status.ok() && !delete_status.is<ErrorCode::NOT_FOUND>()) {
+            LOG(WARNING) << fmt::format("Failed to delete Iceberg delete file {}: {}", path,
+                                        delete_status.to_string());
+        }
+    }
+    _created_files.clear();
+}
+
 int VIcebergDeleteSink::_get_row_id_column_index(const Block& block) {
     // Find __DORIS_ICEBERG_ROWID_COL__ column in block
     for (size_t i = 0; i < block.columns(); ++i) {
@@ -448,9 +482,13 @@ Status VIcebergDeleteSink::_write_position_delete_files(
         }
 
         // Open writer
-        RETURN_IF_ERROR(writer->open(_state, _state->runtime_profile(),
-                                     _position_delete_output_expr_ctxs, column_names, _hadoop_conf,
-                                     _file_type, _broker_addresses));
+        Status open_status =
+                writer->open(_state, _state->runtime_profile(), _position_delete_output_expr_ctxs,
+                             column_names, _hadoop_conf, _file_type, _broker_addresses);
+        if (writer->file_system() != nullptr) {
+            _created_files.emplace_back(writer->file_system(), delete_file_path);
+        }
+        RETURN_IF_ERROR(open_status);
 
         // Build block with (file_path, pos) columns
         std::vector<int64_t> positions;
@@ -669,6 +707,7 @@ Status VIcebergDeleteSink::_write_puffin_file(const std::string& puffin_path,
     auto fs = DORIS_TRY(FileFactory::create_fs(fs_properties, file_description));
     io::FileWriterOptions file_writer_options = {.used_by_s3_committer = false};
     io::FileWriterPtr file_writer;
+    _created_files.emplace_back(fs, puffin_path);
     RETURN_IF_ERROR(fs->create_file(file_description.path, &file_writer, &file_writer_options));
 
     constexpr char PUFFIN_MAGIC[] = {'\x50', '\x46', '\x41', '\x31'};

be/src/exec/sink/viceberg_delete_sink.h

@@ -77,6 +77,10 @@ class VIcebergDeleteSink final : public AsyncResultWriter {
 
     Status close(Status) override;
 
+    void defer_file_cleanup_until_outer_close() { _defer_file_cleanup_until_outer_close = true; }
+
+    void finish_deferred_file_cleanup(Status outer_status);
+
 private:
     /**
      * Extract $row_id column from block and group by file_path.
@@ -129,6 +133,7 @@ class VIcebergDeleteSink final : public AsyncResultWriter {
                                         const std::vector<int64_t>& positions, Block& output_block);
     Status _init_position_delete_output_exprs();
     std::string _get_file_extension() const;
+    void _cleanup_created_files();
 
     TDataSink _t_sink;
     RuntimeState* _state = nullptr;
@@ -141,6 +146,9 @@ class VIcebergDeleteSink final : public AsyncResultWriter {
 
     // Collected commit data from all writers
     std::vector<TIcebergCommitData> _commit_data_list;
+    // MERGE owns both data and delete objects until both inner sinks close successfully.
+    std::vector<std::pair<std::shared_ptr<io::FileSystem>, std::string>> _created_files;
+    bool _defer_file_cleanup_until_outer_close = false;
     // TODO: All deletions are held in memory until close(). Consider flushing
     //  per-file when the upstream guarantees file_path ordering, or flushing
     //  when estimated memory exceeds a threshold, to reduce peak memory usage.

be/src/exec/sink/viceberg_merge_sink.cpp

@@ -19,12 +19,17 @@
 
 #include <fmt/format.h>
 
+#include "agent/be_exec_version_manager.h"
 #include "common/consts.h"
 #include "common/exception.h"
 #include "common/logging.h"
 #include "core/block/block.h"
 #include "core/column/column_nullable.h"
+#include "core/column/column_string.h"
+#include "core/column/column_struct.h"
 #include "core/column/column_vector.h"
+#include "core/data_type/data_type_nullable.h"
+#include "core/data_type/data_type_struct.h"
 #include "exec/sink/sink_common.h"
 #include "exec/sink/viceberg_delete_sink.h"
 #include "exec/sink/writer/iceberg/viceberg_table_writer.h"
@@ -52,8 +57,10 @@ Status VIcebergMergeSink::init_properties(ObjectPool* pool, const RowDescriptor&
 
     _table_writer = std::make_unique<VIcebergTableWriter>(_table_sink, _table_output_expr_ctxs,
                                                           nullptr, nullptr);
+    _table_writer->defer_file_cleanup_until_outer_close();
     _delete_writer = std::make_unique<VIcebergDeleteSink>(_delete_sink, _delete_output_expr_ctxs,
                                                           nullptr, nullptr);
+    _delete_writer->defer_file_cleanup_until_outer_close();
     RETURN_IF_ERROR(_table_writer->init_properties(pool, row_desc));
     RETURN_IF_ERROR(_delete_writer->init_properties(pool));
     return Status::OK();
@@ -65,6 +72,14 @@ Status VIcebergMergeSink::open(RuntimeState* state, RuntimeProfile* profile) {
     _written_rows_counter = ADD_COUNTER(profile, "RowsWritten", TUnit::UNIT);
     _insert_rows_counter = ADD_COUNTER(profile, "InsertRows", TUnit::UNIT);
     _delete_rows_counter = ADD_COUNTER(profile, "DeleteRows", TUnit::UNIT);
+    // The query-wide version keeps validation all-or-nothing during a rolling BE upgrade.
+    _require_merge_cardinality_check =
+            _require_merge_cardinality_check &&
+            state->be_exec_version() >= SUPPORT_ICEBERG_MERGE_CARDINALITY_VERSION;
+    if (_require_merge_cardinality_check) {
+        _matched_row_id_state_bytes_counter =
+                ADD_COUNTER(profile, "MatchedRowIdStateBytes", TUnit::BYTES);
+    }
     _send_data_timer = ADD_TIMER(profile, "SendDataTime");
     _open_timer = ADD_TIMER(profile, "OpenTime");
     _close_timer = ADD_TIMER(profile, "CloseTime");
@@ -94,8 +109,6 @@ Status VIcebergMergeSink::write(RuntimeState* state, Block& block) {
         return Status::OK();
     }
 
-    _row_count += output_block.rows();
-
     if (_operation_idx < 0 || _row_id_idx < 0) {
         return Status::InternalError("Iceberg merge sink missing operation/row_id columns");
     }
@@ -120,17 +133,26 @@ Status VIcebergMergeSink::write(RuntimeState* state, Block& block) {
         if (delete_op) {
             delete_filter[i] = 1;
             has_delete = true;
-            ++_delete_row_count;
             ++delete_rows;
         }
         if (insert_op) {
             insert_filter[i] = 1;
             has_insert = true;
-            ++_insert_row_count;
             ++insert_rows;
         }
     }
 
+    if (_require_merge_cardinality_check) {
+        // The physical sink hashes matched rows by row_id, so exact state retained across blocks
+        // enforces SQL MERGE cardinality for the whole query without changing UPDATE semantics.
+        RETURN_IF_ERROR(_validate_matched_row_ids(output_block, delete_filter.data()));
+        COUNTER_SET(_matched_row_id_state_bytes_counter,
+                    static_cast<int64_t>(_matched_row_id_state_size));
+    }
+    _row_count += output_block.rows();
+    _delete_row_count += delete_rows;
+    _insert_row_count += insert_rows;
+
     bool skip_io = false;
 #ifdef BE_TEST
     skip_io = _skip_io;
@@ -167,6 +189,105 @@ Status VIcebergMergeSink::write(RuntimeState* state, Block& block) {
     return Status::OK();
 }
 
+Status VIcebergMergeSink::_validate_matched_row_ids(const Block& block,
+                                                    const uint8_t* delete_filter) {
+    const auto& row_id = block.get_by_position(_row_id_idx);
+    const IColumn* row_id_data = row_id.column.get();
+    const IDataType* row_id_type = row_id.type.get();
+    const auto* nullable_row_id = check_and_get_column<ColumnNullable>(row_id_data);
+    if (nullable_row_id != nullptr) {
+        row_id_data = nullable_row_id->get_nested_column_ptr().get();
+    }
+    if (const auto* nullable_type = check_and_get_data_type<DataTypeNullable>(row_id_type)) {
+        row_id_type = nullable_type->get_nested_type().get();
+    }
+
+    const auto* struct_column = check_and_get_column<ColumnStruct>(row_id_data);
+    const auto* struct_type = check_and_get_data_type<DataTypeStruct>(row_id_type);
+    if (struct_column == nullptr || struct_type == nullptr) {
+        return Status::InternalError("Iceberg merge row_id column is not a struct");
+    }
+
+    int file_path_idx = -1;
+    int row_position_idx = -1;
+    const auto& field_names = struct_type->get_element_names();
+    for (size_t i = 0; i < field_names.size(); ++i) {
+        std::string field_name = doris::to_lower(field_names[i]);
+        if (field_name == "file_path") {
+            file_path_idx = static_cast<int>(i);
+        } else if (field_name == "row_position") {
+            row_position_idx = static_cast<int>(i);
+        }
+    }
+    if (file_path_idx < 0 || row_position_idx < 0) {
+        return Status::InternalError(
+                "Iceberg merge row_id must contain file_path and row_position fields");
+    }
+
+    const auto& file_path_column = struct_column->get_column_ptr(file_path_idx);
+    const auto& row_position_column = struct_column->get_column_ptr(row_position_idx);
+    const auto* nullable_file_path = check_and_get_column<ColumnNullable>(file_path_column.get());
+    const auto* nullable_row_position =
+            check_and_get_column<ColumnNullable>(row_position_column.get());
+    const auto* file_paths =
+            check_and_get_column<ColumnString>(remove_nullable(file_path_column).get());
+    const auto* row_positions = check_and_get_column<ColumnVector<TYPE_BIGINT>>(
+            remove_nullable(row_position_column).get());
+    if (file_paths == nullptr || row_positions == nullptr) {
+        return Status::InternalError("Iceberg merge row_id fields have incorrect types");
+    }
+
+    std::map<roaring::Roaring64Map*, size_t> touched_bitmap_sizes;
+    for (size_t i = 0; i < block.rows(); ++i) {
+        if (delete_filter[i] == 0) {
+            continue;
+        }
+        if ((nullable_row_id != nullptr && nullable_row_id->is_null_at(i)) ||
+            (nullable_file_path != nullptr && nullable_file_path->is_null_at(i)) ||
+            (nullable_row_position != nullptr && nullable_row_position->is_null_at(i))) {
+            return Status::InternalError("Iceberg merge matched row_id cannot be null");
+        }
+
+        int64_t row_position = row_positions->get_element(i);
+        if (row_position < 0) {
+            return Status::InternalError("Invalid row_position {} in Iceberg merge row_id",
+                                         row_position);
+        }
+        // Intern each file path once and keep exact positions in a compressed bitmap; retaining a
+        // full path string per matched row makes MERGE memory grow with path_length * row_count.
+        auto [file_it, inserted] =
+                _matched_row_positions.try_emplace(file_paths->get_data_at(i).to_string());
+        auto* positions = &file_it->second;
+        auto touched_it = touched_bitmap_sizes.find(positions);
+        if (touched_it == touched_bitmap_sizes.end()) {
+            touched_it = touched_bitmap_sizes.emplace(positions, positions->getSizeInBytes()).first;
+        }
+        if (inserted) {
+            _matched_row_id_state_size +=
+                    sizeof(std::pair<const std::string, roaring::Roaring64Map>);
+            _matched_row_id_state_size += file_it->first.capacity();
+            _matched_row_id_state_size += touched_it->second;
+        }
+        if (!positions->addChecked(static_cast<uint64_t>(row_position))) {
+            return Status::InvalidArgument(
+                    "Iceberg MERGE failed because multiple source rows matched the same target "
+                    "row");
+        }
+    }
+
+    // Measure only bitmaps touched by this block; rescanning all retained files on every write
+    // makes a many-file MERGE quadratic in the number of input blocks.
+    for (const auto& [positions, previous_size] : touched_bitmap_sizes) {
+        size_t current_size = positions->getSizeInBytes();
+        if (current_size >= previous_size) {
+            _matched_row_id_state_size += current_size - previous_size;
+        } else {
+            _matched_row_id_state_size -= previous_size - current_size;
+        }
+    }
+    return Status::OK();
+}
+
 Status VIcebergMergeSink::close(Status close_status) {
     SCOPED_TIMER(_close_timer);
 
@@ -201,10 +322,14 @@ Status VIcebergMergeSink::close(Status close_status) {
         COUNTER_SET(_delete_rows_counter, static_cast<int64_t>(_delete_row_count));
     }
 
-    if (!table_status.ok()) {
-        return table_status;
+    Status result_status = table_status.ok() ? delete_status : table_status;
+    if (_table_writer) {
+        _table_writer->finish_deferred_file_cleanup(result_status);
+    }
+    if (_delete_writer) {
+        _delete_writer->finish_deferred_file_cleanup(result_status);
     }
-    return delete_status;
+    return result_status;
 }
 
 Status VIcebergMergeSink::_build_inner_sinks() {
@@ -213,6 +338,9 @@ Status VIcebergMergeSink::_build_inner_sinks() {
     }
 
     const auto& merge_sink = _t_sink.iceberg_merge_sink;
+    // Missing means an old FE plan, which predates SQL MERGE cardinality validation.
+    _require_merge_cardinality_check = merge_sink.__isset.require_merge_cardinality_check &&
+                                       merge_sink.require_merge_cardinality_check;
 
     TIcebergTableSink table_sink;
     if (merge_sink.__isset.db_name) {