This is an automated email from the ASF dual-hosted git repository.
Gabriel39 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 8305fe71867 [fix](iceberg) Enforce external write correctness (#66112)
8305fe71867 is described below
commit 8305fe71867fb2acab5a769fe9a098ada59aff50
Author: Gabriel <[email protected]>
AuthorDate: Wed Jul 29 15:49:02 2026 +0800
[fix](iceberg) Enforce external write correctness (#66112)
## Proposed changes
- Validate unsupported CTAS sinks before publishing table metadata, so
setup failures cannot leave an unusable table or remove a concurrent
replacement.
- Preserve nullable string wrappers in truncate partition transforms.
- Enforce Iceberg MERGE cardinality routing independently of
`enable_strict_consistency_dml`.
- Detect duplicate target matches with file-path-interned Roaring
bitmaps and expose retained validation-state bytes in the sink profile.
- Activate the negative regressions for the three retained Jira fixes
and align the duplicate-match oracle with the emitted diagnostic.
## Test
- Full build: `./build.sh --fe --be -j 48`
- FE: `CreateTableCommandTest`, `IcebergDDLAndDMLPlanTest` (19 tests)
- BE: `VIcebergMergeSinkTest` (8 tests, including 100,000 matched rows)
- Regression: `test_paimon_ctas_atomicity_negative` (1 suite, passed)
- Iceberg end-to-end SQL: duplicate-source MERGE rejected atomically;
nullable/merge truncate writes and logical rows verified; physical
partition values verified from the same Iceberg tables through Spark
metadata.
The Doris-side `$partitions`/`$snapshots` checks in the Iceberg suites
currently hit an unrelated JNI scanner initialization abort on the
current master test binary; the equivalent table state and metadata
assertions above passed.
---
be/src/agent/be_exec_version_manager.cpp | 4 +-
be/src/agent/be_exec_version_manager.h | 1 +
be/src/exec/sink/viceberg_delete_sink.cpp | 53 +++-
be/src/exec/sink/viceberg_delete_sink.h | 8 +
be/src/exec/sink/viceberg_merge_sink.cpp | 142 ++++++++++-
be/src/exec/sink/viceberg_merge_sink.h | 7 +
.../sink/writer/iceberg/partition_transformers.h | 5 +-
.../writer/iceberg/viceberg_delete_file_writer.h | 2 +
.../writer/iceberg/viceberg_partition_writer.cpp | 30 ++-
.../writer/iceberg/viceberg_partition_writer.h | 9 +-
.../sink/writer/iceberg/viceberg_table_writer.cpp | 33 ++-
.../sink/writer/iceberg/viceberg_table_writer.h | 12 +
be/test/exec/sink/viceberg_merge_sink_test.cpp | 275 ++++++++++++++++++++-
.../writer/iceberg/partition_transformers_test.cpp | 26 ++
.../main/java/org/apache/doris/common/Config.java | 2 +-
.../doris/datasource/paimon/PaimonMetadataOps.java | 27 +-
.../glue/translator/PhysicalPlanTranslator.java | 3 +-
.../nereids/properties/RequestPropertyDeriver.java | 5 +-
...IcebergMergeSinkToPhysicalIcebergMergeSink.java | 1 +
.../trees/plans/commands/CreateTableCommand.java | 35 ++-
.../trees/plans/commands/IcebergMergeCommand.java | 1 +
.../trees/plans/commands/IcebergUpdateCommand.java | 1 +
.../plans/logical/LogicalIcebergMergeSink.java | 26 +-
.../plans/physical/PhysicalIcebergMergeSink.java | 40 ++-
.../org/apache/doris/planner/IcebergMergeSink.java | 7 +-
.../iceberg/IcebergDDLAndDMLPlanTest.java | 149 ++++++++++-
.../datasource/paimon/PaimonMetadataOpsTest.java | 64 ++++-
.../plans/commands/CreateTableCommandTest.java | 133 ++++++++++
.../commands/insert/IcebergMergeExecutorTest.java | 3 +-
.../apache/doris/planner/IcebergMergeSinkTest.java | 18 +-
gensrc/thrift/DataSinks.thrift | 2 +
...eberg_write_merge_duplicate_source_negative.out | 2 +
.../PAIMON_ICEBERG_READ_WRITE_P0_COVERAGE.md | 9 +-
.../iceberg/write/ICEBERG_WRITE_P0_COVERAGE.md | 16 +-
...rg_write_merge_duplicate_source_negative.groovy | 93 ++++++-
...st_iceberg_write_merge_truncate_negative.groovy | 15 +-
...iceberg_write_nullable_truncate_negative.groovy | 8 -
.../test_paimon_ctas_atomicity_negative.groovy | 30 ++-
38 files changed, 1175 insertions(+), 122 deletions(-)
diff --git a/be/src/agent/be_exec_version_manager.cpp
b/be/src/agent/be_exec_version_manager.cpp
index 9ad9d9e9a59..4c3bdbd9f31 100644
--- a/be/src/agent/be_exec_version_manager.cpp
+++ b/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;
diff --git a/be/src/agent/be_exec_version_manager.h
b/be/src/agent/be_exec_version_manager.h
index 92c2c945f25..c1a40e35a07 100644
--- a/be/src/agent/be_exec_version_manager.h
+++ b/be/src/agent/be_exec_version_manager.h
@@ -26,6 +26,7 @@
namespace doris {
constexpr inline int USE_NEW_FIXED_OBJECT_SERIALIZATION_VERSION = 10;
+constexpr inline int SUPPORT_ICEBERG_MERGE_CARDINALITY_VERSION = 11;
class BeExecVersionManager {
public:
diff --git a/be/src/exec/sink/viceberg_delete_sink.cpp
b/be/src/exec/sink/viceberg_delete_sink.cpp
index ef167ecdbca..172fdd28177 100644
--- a/be/src/exec/sink/viceberg_delete_sink.cpp
+++ b/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'};
diff --git a/be/src/exec/sink/viceberg_delete_sink.h
b/be/src/exec/sink/viceberg_delete_sink.h
index 647ee92f525..55698ae0404 100644
--- a/be/src/exec/sink/viceberg_delete_sink.h
+++ b/be/src/exec/sink/viceberg_delete_sink.h
@@ -77,6 +77,10 @@ public:
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 @@ private:
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 @@ private:
// 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.
diff --git a/be/src/exec/sink/viceberg_merge_sink.cpp
b/be/src/exec/sink/viceberg_merge_sink.cpp
index e9f4d6bf5c6..5ff5a0a1f28 100644
--- a/be/src/exec/sink/viceberg_merge_sink.cpp
+++ b/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) {
diff --git a/be/src/exec/sink/viceberg_merge_sink.h
b/be/src/exec/sink/viceberg_merge_sink.h
index d7f87b87012..f3733a33182 100644
--- a/be/src/exec/sink/viceberg_merge_sink.h
+++ b/be/src/exec/sink/viceberg_merge_sink.h
@@ -19,6 +19,7 @@
#include <gen_cpp/DataSinks_types.h>
+#include <map>
#include <memory>
#include <string>
#include <vector>
@@ -26,6 +27,7 @@
#include "common/status.h"
#include "exec/sink/writer/async_result_writer.h"
#include "exprs/vexpr_fwd.h"
+#include "roaring/roaring64map.hh"
#include "runtime/descriptors.h"
#include "runtime/runtime_profile.h"
@@ -60,6 +62,7 @@ public:
private:
Status _build_inner_sinks();
Status _prepare_output_layout();
+ Status _validate_matched_row_ids(const Block& block, const uint8_t*
delete_filter);
TDataSink _t_sink;
TDataSink _table_sink;
@@ -73,6 +76,9 @@ private:
int _operation_idx = -1;
int _row_id_idx = -1;
std::vector<int> _data_column_indices;
+ std::map<std::string, roaring::Roaring64Map> _matched_row_positions;
+ size_t _matched_row_id_state_size = sizeof(std::map<std::string,
roaring::Roaring64Map>);
+ bool _require_merge_cardinality_check = false;
VExprContextSPtrs _table_output_expr_ctxs;
VExprContextSPtrs _delete_output_expr_ctxs;
@@ -84,6 +90,7 @@ private:
RuntimeProfile::Counter* _written_rows_counter = nullptr;
RuntimeProfile::Counter* _insert_rows_counter = nullptr;
RuntimeProfile::Counter* _delete_rows_counter = nullptr;
+ RuntimeProfile::Counter* _matched_row_id_state_bytes_counter = nullptr;
RuntimeProfile::Counter* _send_data_timer = nullptr;
RuntimeProfile::Counter* _open_timer = nullptr;
RuntimeProfile::Counter* _close_timer = nullptr;
diff --git a/be/src/exec/sink/writer/iceberg/partition_transformers.h
b/be/src/exec/sink/writer/iceberg/partition_transformers.h
index b9dcceb3e80..51bca7000ad 100644
--- a/be/src/exec/sink/writer/iceberg/partition_transformers.h
+++ b/be/src/exec/sink/writer/iceberg/partition_transformers.h
@@ -167,7 +167,10 @@ public:
// Create a temp_block to execute substring function.
Block temp_block;
- temp_block.insert(column_with_type_and_name);
+ // Substring requires the physical ColumnString; preserve nullability
separately and
+ // restore the original null map after transforming the nested values.
+ temp_block.insert({string_column_ptr,
remove_nullable(column_with_type_and_name.type),
+ column_with_type_and_name.name});
temp_block.insert({int_type->create_column_const(temp_block.rows(),
to_field<TYPE_INT>(1)),
int_type, "const 1"});
temp_block.insert(
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 c242731dc15..4ceb7080a3c 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
@@ -94,6 +94,8 @@ public:
*/
int64_t get_file_size() const { return _file_size; }
+ std::shared_ptr<io::FileSystem> file_system() const { return _fs; }
+
private:
TFileContent::type _delete_type;
std::string _output_path;
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 5958165ad03..ba7644daec7 100644
--- a/be/src/exec/sink/writer/iceberg/viceberg_partition_writer.cpp
+++ b/be/src/exec/sink/writer/iceberg/viceberg_partition_writer.cpp
@@ -36,7 +36,8 @@ VIcebergPartitionWriter::VIcebergPartitionWriter(
const std::string* iceberg_schema_json, std::vector<std::string>
write_column_names,
WriteInfo write_info, std::string file_name, int file_name_index,
TFileFormatType::type file_format_type, TFileCompressType::type
compress_type,
- const std::map<std::string, std::string>& hadoop_conf)
+ const std::map<std::string, std::string>& hadoop_conf,
+ ClosedFileCallback closed_file_callback)
: _partition_values(std::move(partition_values)),
_write_output_expr_ctxs(write_output_expr_ctxs),
_schema(schema),
@@ -47,7 +48,8 @@ VIcebergPartitionWriter::VIcebergPartitionWriter(
_file_name_index(file_name_index),
_file_format_type(file_format_type),
_compress_type(compress_type),
- _hadoop_conf(hadoop_conf) {
+ _hadoop_conf(hadoop_conf),
+ _closed_file_callback(std::move(closed_file_callback)) {
if (t_sink.iceberg_table_sink.__isset.collect_column_stats) {
_collect_column_stats = t_sink.iceberg_table_sink.collect_column_stats;
}
@@ -61,9 +63,8 @@ Status VIcebergPartitionWriter::open(RuntimeState* state,
RuntimeProfile* profil
if (!_write_info.broker_addresses.empty()) {
fs_properties.broker_addresses = &(_write_info.broker_addresses);
}
- io::FileDescription file_description = {
- .path = fmt::format("{}/{}", _write_info.write_path,
_get_target_file_name()),
- .fs_name {}};
+ _path = fmt::format("{}/{}", _write_info.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};
RETURN_IF_ERROR(_fs->create_file(file_description.path, &_file_writer,
&file_writer_options));
@@ -129,16 +130,27 @@ Status VIcebergPartitionWriter::close(const Status&
status) {
bool status_ok = result_status.ok() && status.ok();
if (!status_ok && _fs != nullptr) {
// delete the actual created file, otherwise an orphan file is left
behind
- auto path = fmt::format("{}/{}", _write_info.write_path,
_get_target_file_name());
- Status st = _fs->delete_file(path);
+ Status st = _fs->delete_file(_path);
if (!st.ok()) {
- LOG(WARNING) << fmt::format("Delete file {} failed, reason: {}",
path, st.to_string());
+ LOG(WARNING) << fmt::format("Delete file {} failed, reason: {}",
_path, st.to_string());
}
}
if (status_ok) {
TIcebergCommitData commit_data;
- RETURN_IF_ERROR(_build_iceberg_commit_data(&commit_data));
+ Status commit_status = _build_iceberg_commit_data(&commit_data);
+ if (!commit_status.ok()) {
+ // A closed object without commit data can never be published, so
remove it immediately.
+ Status delete_status = _fs->delete_file(_path);
+ if (!delete_status.ok()) {
+ LOG(WARNING) << fmt::format("Delete file {} failed, reason:
{}", _path,
+ delete_status.to_string());
+ }
+ return commit_status;
+ }
_state->add_iceberg_commit_datas(commit_data);
+ if (_closed_file_callback) {
+ _closed_file_callback(_fs, _path);
+ }
}
return result_status;
}
diff --git a/be/src/exec/sink/writer/iceberg/viceberg_partition_writer.h
b/be/src/exec/sink/writer/iceberg/viceberg_partition_writer.h
index b0839a82ed3..f3a7a6002f8 100644
--- a/be/src/exec/sink/writer/iceberg/viceberg_partition_writer.h
+++ b/be/src/exec/sink/writer/iceberg/viceberg_partition_writer.h
@@ -19,6 +19,8 @@
#include <gen_cpp/DataSinks_types.h>
+#include <functional>
+
#include "exec/sink/writer/iceberg/vpartition_writer_base.h"
#include "exprs/vexpr_fwd.h"
#include "format/table/iceberg/schema.h"
@@ -43,6 +45,9 @@ class VFileFormatTransformer;
class VIcebergPartitionWriter : public IPartitionWriterBase {
public:
+ using ClosedFileCallback =
+ std::function<void(std::shared_ptr<io::FileSystem>, const
std::string&)>;
+
VIcebergPartitionWriter(const TDataSink& t_sink, std::vector<std::string>
partition_values,
const VExprContextSPtrs& write_output_expr_ctxs,
const doris::iceberg::Schema& schema,
@@ -51,7 +56,8 @@ public:
std::string file_name, int file_name_index,
TFileFormatType::type file_format_type,
TFileCompressType::type compress_type,
- const std::map<std::string, std::string>&
hadoop_conf);
+ const std::map<std::string, std::string>&
hadoop_conf,
+ ClosedFileCallback closed_file_callback = nullptr);
Status open(RuntimeState* state, RuntimeProfile* profile,
const RowDescriptor* row_desc) override;
@@ -93,6 +99,7 @@ private:
TFileFormatType::type _file_format_type;
TFileCompressType::type _compress_type;
const std::map<std::string, std::string>& _hadoop_conf;
+ ClosedFileCallback _closed_file_callback;
bool _collect_column_stats = true;
std::shared_ptr<io::FileSystem> _fs = nullptr;
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 0a3b0843115..80db0876a9b 100644
--- a/be/src/exec/sink/writer/iceberg/viceberg_table_writer.cpp
+++ b/be/src/exec/sink/writer/iceberg/viceberg_table_writer.cpp
@@ -33,6 +33,7 @@
#include "exprs/vexpr_context.h"
#include "format/table/iceberg/partition_spec_parser.h"
#include "format/table/iceberg/schema_parser.h"
+#include "io/fs/file_system.h"
#include "runtime/runtime_state.h"
namespace doris {
@@ -485,9 +486,36 @@ Status VIcebergTableWriter::close(Status status) {
COUNTER_SET(_close_timer, _close_ns);
COUNTER_SET(_write_file_counter, _write_file_count);
}
+ if (!status.ok() || !result_status.ok()) {
+ _cleanup_closed_files();
+ } else if (!_defer_file_cleanup_until_outer_close) {
+ _closed_files.clear();
+ }
return result_status;
}
+void VIcebergTableWriter::finish_deferred_file_cleanup(Status outer_status) {
+ // A successful inner close does not publish MERGE files; the sibling
delete close still
+ // decides whether these data objects must be removed or released to the
transaction.
+ if (!outer_status.ok()) {
+ _cleanup_closed_files();
+ } else {
+ _closed_files.clear();
+ }
+ _defer_file_cleanup_until_outer_close = false;
+}
+
+void VIcebergTableWriter::_cleanup_closed_files() {
+ for (const auto& [fs, path] : _closed_files) {
+ Status delete_status = fs->delete_file(path);
+ if (!delete_status.ok()) {
+ LOG(WARNING) << fmt::format("Delete rolled Iceberg file {} failed,
reason: {}", path,
+ delete_status.to_string());
+ }
+ }
+ _closed_files.clear();
+}
+
std::string VIcebergTableWriter::_partition_to_path(const
doris::iceberg::StructLike& data) {
std::stringstream ss;
for (size_t i = 0; i < _iceberg_partition_columns.size(); i++) {
@@ -611,7 +639,10 @@ std::shared_ptr<IPartitionWriterBase>
VIcebergTableWriter::_create_partition_wri
&_t_sink.iceberg_table_sink.schema_json, column_names,
write_info,
(file_name == nullptr) ? _compute_file_name() : *file_name,
file_name_index,
iceberg_table_sink.file_format,
iceberg_table_sink.compression_type,
- iceberg_table_sink.hadoop_config);
+ iceberg_table_sink.hadoop_config,
+ [this](std::shared_ptr<io::FileSystem> fs, const std::string&
path) {
+ _closed_files.emplace_back(std::move(fs), path);
+ });
};
auto partition_write = create_writer_lambda(file_name, file_name_index);
if (iceberg_table_sink.__isset.sort_info) {
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 cc7cec1fdad..070019d85db 100644
--- a/be/src/exec/sink/writer/iceberg/viceberg_table_writer.h
+++ b/be/src/exec/sink/writer/iceberg/viceberg_table_writer.h
@@ -32,6 +32,10 @@
namespace doris {
+namespace io {
+class FileSystem;
+}
+
class ObjectPool;
class RuntimeState;
@@ -60,6 +64,10 @@ public:
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);
+
bool is_rewrite_compaction() const { return _write_type ==
TIcebergWriteType::REWRITE; }
TIcebergWriteType::type write_type() const { return _write_type; }
@@ -137,6 +145,7 @@ private:
Status _write_prepared_block(Block& output_block);
Status _process_row_lineage_columns(Block& block);
+ void _cleanup_closed_files();
// Currently it is a copy, maybe it is better to use move semantics to
eliminate it.
TDataSink _t_sink;
@@ -171,6 +180,9 @@ private:
std::vector<std::string> _static_partition_value_list;
std::unordered_map<std::string, std::shared_ptr<IPartitionWriterBase>>
_partitions_to_writers;
+ // Rolled writers are no longer active, so retain their physical paths
until statement outcome is known.
+ std::vector<std::pair<std::shared_ptr<io::FileSystem>, std::string>>
_closed_files;
+ bool _defer_file_cleanup_until_outer_close = false;
VExprContextSPtrs _write_output_vexpr_ctxs;
size_t _row_count = 0;
const RowDescriptor* _row_desc = nullptr;
diff --git a/be/test/exec/sink/viceberg_merge_sink_test.cpp
b/be/test/exec/sink/viceberg_merge_sink_test.cpp
index d5c9ac1b891..eb7c0159d5f 100644
--- a/be/test/exec/sink/viceberg_merge_sink_test.cpp
+++ b/be/test/exec/sink/viceberg_merge_sink_test.cpp
@@ -19,6 +19,11 @@
#include <gtest/gtest.h>
+#include <filesystem>
+#include <fstream>
+
+#include "agent/be_exec_version_manager.h"
+#include "common/config.h"
#include "common/consts.h"
#include "common/object_pool.h"
#include "core/block/block.h"
@@ -28,13 +33,17 @@
#include "core/data_type/data_type_number.h"
#include "core/data_type/data_type_string.h"
#include "core/data_type/data_type_struct.h"
+#include "exec/sink/viceberg_delete_sink.h"
+#include "exec/sink/writer/iceberg/viceberg_table_writer.h"
#include "exprs/vexpr_context.h"
#include "gen_cpp/DataSinks_types.h"
#include "gen_cpp/Types_types.h"
+#include "io/fs/local_file_system.h"
#include "runtime/runtime_profile.h"
#include "testutil/mock/mock_descriptors.h"
#include "testutil/mock/mock_runtime_state.h"
#include "testutil/mock/mock_slot_ref.h"
+#include "util/debug_points.h"
namespace doris {
@@ -47,7 +56,7 @@ protected:
"]}";
}
- TDataSink build_sink() {
+ TDataSink build_sink(bool require_cardinality_check = true, bool
set_cardinality_check = true) {
TDataSink t_sink;
t_sink.__set_type(TDataSinkType::ICEBERG_MERGE_SINK);
@@ -64,6 +73,9 @@ protected:
merge_sink.__set_file_type(TFileType::FILE_LOCAL);
merge_sink.__set_delete_type(TFileContent::POSITION_DELETES);
merge_sink.__set_partition_spec_id_for_delete(0);
+ if (set_cardinality_check) {
+
merge_sink.__set_require_merge_cardinality_check(require_cardinality_check);
+ }
t_sink.__set_iceberg_merge_sink(merge_sink);
return t_sink;
@@ -106,7 +118,8 @@ protected:
return output_exprs;
}
- Block build_block_with_ops(const std::vector<int8_t>& ops) {
+ Block build_block_with_ops(const std::vector<int8_t>& ops, bool
distinct_files = true,
+ size_t file_index_offset = 0) {
Block block;
auto op_col = ColumnInt8::create();
@@ -121,9 +134,11 @@ protected:
auto id_col = ColumnInt32::create();
auto name_col = ColumnString::create();
for (size_t i = 0; i < ops.size(); ++i) {
- std::string file_path = "file" + std::to_string(i + 1) +
".parquet";
+ std::string file_path =
+ distinct_files ? "file" + std::to_string(file_index_offset
+ i + 1) + ".parquet"
+ : "shared-file.parquet";
file_path_col->insert_data(file_path.data(), file_path.size());
- row_pos_col->insert_value(static_cast<int64_t>((i + 1) * 10));
+ row_pos_col->insert_value(static_cast<int64_t>((file_index_offset
+ i + 1) * 10));
id_col->insert_value(static_cast<int32_t>(i + 1));
char name_value = static_cast<char>('a' + i);
name_col->insert_data(&name_value, 1);
@@ -315,4 +330,256 @@ TEST_F(VIcebergMergeSinkTest, TestSchemaMismatch) {
ASSERT_NE(std::string::npos, status.to_string().find("do not match schema
columns"));
}
+TEST_F(VIcebergMergeSinkTest, TestRejectsDuplicateMatchedTargetAcrossBlocks) {
+ ObjectPool pool;
+ MockRuntimeState state;
+
+ DataTypes types {std::make_shared<DataTypeInt8>(),
+ std::make_shared<DataTypeStruct>(DataTypes
{std::make_shared<DataTypeString>(),
+
std::make_shared<DataTypeInt64>()},
+ Strings {"file_path",
"row_position"}),
+ std::make_shared<DataTypeInt32>(),
std::make_shared<DataTypeString>()};
+ MockRowDescriptor row_desc(types, &pool);
+
+ auto output_exprs = build_output_exprs(&pool, &state, row_desc);
+ auto sink = std::make_shared<VIcebergMergeSink>(build_sink(),
output_exprs, nullptr, nullptr);
+ sink->set_skip_io(true);
+
+ ASSERT_TRUE(sink->init_properties(&pool, row_desc).ok());
+ RuntimeProfile profile("iceberg_merge_sink");
+ ASSERT_TRUE(sink->open(&state, &profile).ok());
+
+ Block first = build_block_with_ops({3});
+ ASSERT_TRUE(sink->write(&state, first).ok());
+ Block duplicate = build_block_with_ops({3});
+ Status status = sink->write(&state, duplicate);
+
+ ASSERT_FALSE(status.ok());
+ ASSERT_NE(std::string::npos,
+ status.to_string().find("multiple source rows matched the same
target row"));
+}
+
+TEST_F(VIcebergMergeSinkTest, TestUpdateSkipsCardinalityState) {
+ ObjectPool pool;
+ MockRuntimeState state;
+
+ DataTypes types {std::make_shared<DataTypeInt8>(),
+ std::make_shared<DataTypeStruct>(DataTypes
{std::make_shared<DataTypeString>(),
+
std::make_shared<DataTypeInt64>()},
+ Strings {"file_path",
"row_position"}),
+ std::make_shared<DataTypeInt32>(),
std::make_shared<DataTypeString>()};
+ MockRowDescriptor row_desc(types, &pool);
+
+ auto output_exprs = build_output_exprs(&pool, &state, row_desc);
+ auto sink =
+ std::make_shared<VIcebergMergeSink>(build_sink(false),
output_exprs, nullptr, nullptr);
+ sink->set_skip_io(true);
+
+ ASSERT_TRUE(sink->init_properties(&pool, row_desc).ok());
+ RuntimeProfile profile("iceberg_update_sink");
+ ASSERT_TRUE(sink->open(&state, &profile).ok());
+
+ Block first = build_block_with_ops({3});
+ ASSERT_TRUE(sink->write(&state, first).ok());
+ Block duplicate = build_block_with_ops({3});
+ ASSERT_TRUE(sink->write(&state, duplicate).ok());
+ EXPECT_TRUE(sink->_matched_row_positions.empty());
+ EXPECT_EQ(nullptr, profile.get_counter("MatchedRowIdStateBytes"));
+}
+
+TEST_F(VIcebergMergeSinkTest, TestOldFePlanSkipsCardinalityState) {
+ ObjectPool pool;
+ MockRuntimeState state;
+
+ DataTypes types {std::make_shared<DataTypeInt8>(),
+ std::make_shared<DataTypeStruct>(DataTypes
{std::make_shared<DataTypeString>(),
+
std::make_shared<DataTypeInt64>()},
+ Strings {"file_path",
"row_position"}),
+ std::make_shared<DataTypeInt32>(),
std::make_shared<DataTypeString>()};
+ MockRowDescriptor row_desc(types, &pool);
+
+ auto output_exprs = build_output_exprs(&pool, &state, row_desc);
+ auto sink = std::make_shared<VIcebergMergeSink>(build_sink(false, false),
output_exprs, nullptr,
+ nullptr);
+ sink->set_skip_io(true);
+
+ ASSERT_TRUE(sink->init_properties(&pool, row_desc).ok());
+ RuntimeProfile profile("old_fe_iceberg_update_sink");
+ ASSERT_TRUE(sink->open(&state, &profile).ok());
+ Block duplicate = build_block_with_ops({3, 3}, false);
+ ASSERT_TRUE(sink->write(&state, duplicate).ok());
+ EXPECT_TRUE(sink->_matched_row_positions.empty());
+}
+
+TEST_F(VIcebergMergeSinkTest, TestRollingUpgradeSkipsCardinalityState) {
+ ObjectPool pool;
+ MockRuntimeState state;
+ state.set_be_exec_version(SUPPORT_ICEBERG_MERGE_CARDINALITY_VERSION - 1);
+
+ DataTypes types {std::make_shared<DataTypeInt8>(),
+ std::make_shared<DataTypeStruct>(DataTypes
{std::make_shared<DataTypeString>(),
+
std::make_shared<DataTypeInt64>()},
+ Strings {"file_path",
"row_position"}),
+ std::make_shared<DataTypeInt32>(),
std::make_shared<DataTypeString>()};
+ MockRowDescriptor row_desc(types, &pool);
+
+ auto output_exprs = build_output_exprs(&pool, &state, row_desc);
+ auto sink = std::make_shared<VIcebergMergeSink>(build_sink(),
output_exprs, nullptr, nullptr);
+ sink->set_skip_io(true);
+
+ ASSERT_TRUE(sink->init_properties(&pool, row_desc).ok());
+ RuntimeProfile profile("rolling_upgrade_iceberg_merge_sink");
+ ASSERT_TRUE(sink->open(&state, &profile).ok());
+ Block duplicate = build_block_with_ops({3, 3}, false);
+ ASSERT_TRUE(sink->write(&state, duplicate).ok());
+ EXPECT_TRUE(sink->_matched_row_positions.empty());
+}
+
+TEST_F(VIcebergMergeSinkTest, TestErrorCloseRemovesRolledDataFiles) {
+ ObjectPool pool;
+ MockRuntimeState state;
+
+ DataTypes types {std::make_shared<DataTypeInt8>(),
+ std::make_shared<DataTypeStruct>(DataTypes
{std::make_shared<DataTypeString>(),
+
std::make_shared<DataTypeInt64>()},
+ Strings {"file_path",
"row_position"}),
+ std::make_shared<DataTypeInt32>(),
std::make_shared<DataTypeString>()};
+ MockRowDescriptor row_desc(types, &pool);
+ auto output_exprs = build_output_exprs(&pool, &state, row_desc);
+ auto sink = std::make_shared<VIcebergMergeSink>(build_sink(),
output_exprs, nullptr, nullptr);
+ sink->set_skip_io(true);
+
+ ASSERT_TRUE(sink->init_properties(&pool, row_desc).ok());
+ RuntimeProfile profile("iceberg_merge_cleanup_sink");
+ ASSERT_TRUE(sink->open(&state, &profile).ok());
+
+ std::filesystem::path path =
+ std::filesystem::temp_directory_path() /
"doris_iceberg_merge_rolled_file.parquet";
+ {
+ std::ofstream output(path);
+ output << "rolled-data";
+ }
+ ASSERT_TRUE(std::filesystem::exists(path));
+
sink->_table_writer->_closed_files.emplace_back(io::global_local_filesystem(),
path.string());
+
+ Status failure = Status::InvalidArgument("late duplicate");
+ EXPECT_FALSE(sink->close(failure).ok());
+ EXPECT_FALSE(std::filesystem::exists(path));
+}
+
+TEST_F(VIcebergMergeSinkTest, TestDeleteCloseFailureRemovesBothInnerSinkFiles)
{
+ ObjectPool pool;
+ MockRuntimeState state;
+
+ DataTypes types {std::make_shared<DataTypeInt8>(),
+ std::make_shared<DataTypeStruct>(DataTypes
{std::make_shared<DataTypeString>(),
+
std::make_shared<DataTypeInt64>()},
+ Strings {"file_path",
"row_position"}),
+ std::make_shared<DataTypeInt32>(),
std::make_shared<DataTypeString>()};
+ MockRowDescriptor row_desc(types, &pool);
+ auto output_exprs = build_output_exprs(&pool, &state, row_desc);
+ auto sink = std::make_shared<VIcebergMergeSink>(build_sink(),
output_exprs, nullptr, nullptr);
+ sink->set_skip_io(true);
+
+ ASSERT_TRUE(sink->init_properties(&pool, row_desc).ok());
+ RuntimeProfile profile("iceberg_merge_delete_close_cleanup_sink");
+ ASSERT_TRUE(sink->open(&state, &profile).ok());
+
+ std::filesystem::path data_path = std::filesystem::temp_directory_path() /
+
"doris_iceberg_merge_delete_close_data.parquet";
+ std::filesystem::path delete_path = std::filesystem::temp_directory_path()
/
+
"doris_iceberg_merge_delete_close_position.parquet";
+ {
+ std::ofstream output(data_path);
+ output << "closed-data";
+ }
+ {
+ std::ofstream output(delete_path);
+ output << "closed-delete";
+ }
+ ASSERT_TRUE(std::filesystem::exists(data_path));
+ ASSERT_TRUE(std::filesystem::exists(delete_path));
+
sink->_table_writer->_closed_files.emplace_back(io::global_local_filesystem(),
+ data_path.string());
+
sink->_delete_writer->_created_files.emplace_back(io::global_local_filesystem(),
+ delete_path.string());
+
+ bool previous_enable_debug_points = config::enable_debug_points;
+ config::enable_debug_points = true;
+ DebugPoints::instance()->add("VIcebergDeleteSink.close.inject_failure");
+ Status status = sink->close(Status::OK());
+ DebugPoints::instance()->clear();
+ config::enable_debug_points = previous_enable_debug_points;
+
+ EXPECT_FALSE(status.ok());
+ EXPECT_NE(std::string::npos, status.to_string().find("injected Iceberg
delete close failure"));
+ EXPECT_FALSE(std::filesystem::exists(data_path));
+ EXPECT_FALSE(std::filesystem::exists(delete_path));
+}
+
+TEST_F(VIcebergMergeSinkTest, TestMatchedRowIdsUseCompactRetainedState) {
+ ObjectPool pool;
+ MockRuntimeState state;
+
+ DataTypes types {std::make_shared<DataTypeInt8>(),
+ std::make_shared<DataTypeStruct>(DataTypes
{std::make_shared<DataTypeString>(),
+
std::make_shared<DataTypeInt64>()},
+ Strings {"file_path",
"row_position"}),
+ std::make_shared<DataTypeInt32>(),
std::make_shared<DataTypeString>()};
+ MockRowDescriptor row_desc(types, &pool);
+
+ auto output_exprs = build_output_exprs(&pool, &state, row_desc);
+ auto sink = std::make_shared<VIcebergMergeSink>(build_sink(),
output_exprs, nullptr, nullptr);
+ sink->set_skip_io(true);
+
+ ASSERT_TRUE(sink->init_properties(&pool, row_desc).ok());
+ RuntimeProfile profile("iceberg_merge_sink");
+ ASSERT_TRUE(sink->open(&state, &profile).ok());
+
+ constexpr size_t row_count = 100000;
+ std::vector<int8_t> operations(row_count, 3);
+ Block block = build_block_with_ops(operations, false);
+ ASSERT_TRUE(sink->write(&state, block).ok());
+
+ auto* retained_bytes = profile.get_counter("MatchedRowIdStateBytes");
+ ASSERT_NE(nullptr, retained_bytes);
+ EXPECT_LT(retained_bytes->value(), static_cast<int64_t>(row_count *
sizeof(int64_t)));
+}
+
+TEST_F(VIcebergMergeSinkTest, TestMatchedRowIdStateAcrossManyFilesAndWrites) {
+ ObjectPool pool;
+ MockRuntimeState state;
+
+ DataTypes types {std::make_shared<DataTypeInt8>(),
+ std::make_shared<DataTypeStruct>(DataTypes
{std::make_shared<DataTypeString>(),
+
std::make_shared<DataTypeInt64>()},
+ Strings {"file_path",
"row_position"}),
+ std::make_shared<DataTypeInt32>(),
std::make_shared<DataTypeString>()};
+ MockRowDescriptor row_desc(types, &pool);
+
+ auto output_exprs = build_output_exprs(&pool, &state, row_desc);
+ auto sink = std::make_shared<VIcebergMergeSink>(build_sink(),
output_exprs, nullptr, nullptr);
+ sink->set_skip_io(true);
+
+ ASSERT_TRUE(sink->init_properties(&pool, row_desc).ok());
+ RuntimeProfile profile("iceberg_merge_sink");
+ ASSERT_TRUE(sink->open(&state, &profile).ok());
+
+ auto* retained_bytes = profile.get_counter("MatchedRowIdStateBytes");
+ ASSERT_NE(nullptr, retained_bytes);
+ constexpr size_t files_per_write = 32;
+ constexpr size_t write_count = 64;
+ std::vector<int8_t> operations(files_per_write, 3);
+ int64_t previous_bytes = 0;
+ for (size_t write_index = 0; write_index < write_count; ++write_index) {
+ Block block = build_block_with_ops(operations, true, write_index *
files_per_write);
+ ASSERT_TRUE(sink->write(&state, block).ok());
+ EXPECT_GT(retained_bytes->value(), previous_bytes);
+ previous_bytes = retained_bytes->value();
+ }
+
+ EXPECT_EQ(files_per_write * write_count,
sink->_matched_row_positions.size());
+ EXPECT_EQ(static_cast<int64_t>(sink->_matched_row_id_state_size),
retained_bytes->value());
+}
+
} // namespace doris
diff --git a/be/test/exec/sink/writer/iceberg/partition_transformers_test.cpp
b/be/test/exec/sink/writer/iceberg/partition_transformers_test.cpp
index 94b8aec8c77..974eb817e88 100644
--- a/be/test/exec/sink/writer/iceberg/partition_transformers_test.cpp
+++ b/be/test/exec/sink/writer/iceberg/partition_transformers_test.cpp
@@ -520,4 +520,30 @@ TEST_F(PartitionTransformersTest,
test_nullable_column_integer_truncate_transfor
}
}
+TEST_F(PartitionTransformersTest,
test_nullable_column_string_truncate_transform) {
+ auto column = ColumnNullable::create(ColumnString::create(),
ColumnUInt8::create());
+ column->insert_data(nullptr, 0);
+ column->insert_data("iceberg", sizeof("iceberg") - 1);
+ column->insert_data("db", sizeof("db") - 1);
+ ColumnWithTypeAndName test_string(
+ column->get_ptr(),
+
std::make_shared<DataTypeNullable>(std::make_shared<DataTypeString>()),
"test_string");
+
+ Block block({test_string});
+ auto source_type =
DataTypeFactory::instance().create_data_type(TYPE_STRING, true);
+ StringTruncatePartitionColumnTransform transform(source_type, 3);
+
+ auto result = transform.apply(block, 0);
+
+ const auto* result_column = assert_cast<const
ColumnNullable*>(result.column.get());
+ const auto* result_strings =
+ assert_cast<const
ColumnString*>(result_column->get_nested_column_ptr().get());
+ EXPECT_EQ(3, result_column->size());
+ EXPECT_EQ(Field::create_field<TYPE_BOOLEAN>(1),
result_column->get_null_map_column()[0]);
+ EXPECT_EQ(Field::create_field<TYPE_BOOLEAN>(0),
result_column->get_null_map_column()[1]);
+ EXPECT_EQ(Field::create_field<TYPE_BOOLEAN>(0),
result_column->get_null_map_column()[2]);
+ EXPECT_EQ("ice", result_strings->get_data_at(1).to_string());
+ EXPECT_EQ("db", result_strings->get_data_at(2).to_string());
+}
+
} // namespace doris
diff --git a/fe/fe-common/src/main/java/org/apache/doris/common/Config.java
b/fe/fe-common/src/main/java/org/apache/doris/common/Config.java
index 66dc4cc881f..24a555fed33 100644
--- a/fe/fe-common/src/main/java/org/apache/doris/common/Config.java
+++ b/fe/fe-common/src/main/java/org/apache/doris/common/Config.java
@@ -2036,7 +2036,7 @@ public class Config extends ConfigBase {
* Max data version of backends serialize block.
*/
@ConfField(mutable = false)
- public static int max_be_exec_version = 10;
+ public static int max_be_exec_version = 11;
/**
* Min data version of backends serialize block.
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonMetadataOps.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonMetadataOps.java
index 2f0be3bc305..c1140f42cf3 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonMetadataOps.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonMetadataOps.java
@@ -191,6 +191,8 @@ public class PaimonMetadataOps implements
ExternalMetadataOps {
if (tableExist(db.getRemoteName(), tableName)) {
if (createTableInfo.isIfNotExists()) {
LOG.info("create table[{}] which already exists", tableName);
+ // Existing-table success skips the normal post-create hook,
so refresh names here.
+ resetTableNameCache(dbName);
return true;
} else {
ErrorReport.reportDdlException(ErrorCode.ERR_TABLE_EXISTS_ERROR, tableName);
@@ -208,6 +210,8 @@ public class PaimonMetadataOps implements
ExternalMetadataOps {
if (dorisTable != null) {
if (createTableInfo.isIfNotExists()) {
LOG.info("create table[{}] which already exists", tableName);
+ // Every successful no-op bypasses the normal post-create hook
and must refresh names.
+ resetTableNameCache(dbName);
return true;
} else {
ErrorReport.reportDdlException(ErrorCode.ERR_TABLE_EXISTS_ERROR, tableName);
@@ -223,9 +227,19 @@ public class PaimonMetadataOps implements
ExternalMetadataOps {
Schema schema = toPaimonSchema(structType, rootFieldNames,
createTableInfo.getPartitionDesc(),
createTableInfo.getProperties());
try {
+ // Let Paimon report a concurrent winner so callers can
distinguish an existing table
+ // from the table created by this statement before deciding
whether rollback is owned.
catalog.createTable(new Identifier(createTableInfo.getDbName(),
createTableInfo.getTableName()),
- schema, createTableInfo.isIfNotExists());
- } catch (TableAlreadyExistException | DatabaseNotExistException e) {
+ schema, false);
+ } catch (TableAlreadyExistException e) {
+ if (createTableInfo.isIfNotExists()) {
+ LOG.info("create table[{}] which already exists", tableName);
+ // A concurrent remote creator also bypasses the normal
post-create hook.
+ resetTableNameCache(dbName);
+ return true;
+ }
+ throw new RuntimeException(e);
+ } catch (DatabaseNotExistException e) {
throw new RuntimeException(e);
}
return false;
@@ -277,12 +291,17 @@ public class PaimonMetadataOps implements
ExternalMetadataOps {
@Override
public void afterCreateTable(String dbName, String tblName) {
+ Optional<ExternalDatabase<?>> db = resetTableNameCache(dbName);
+ LOG.info("after create table {}.{}.{}, is db exists: {}",
+ dorisCatalog.getName(), dbName, tblName, db.isPresent());
+ }
+
+ private Optional<ExternalDatabase<?>> resetTableNameCache(String dbName) {
Optional<ExternalDatabase<?>> db = dorisCatalog.getDbForReplay(dbName);
if (db.isPresent()) {
db.get().resetMetaCacheNames();
}
- LOG.info("after create table {}.{}.{}, is db exists: {}",
- dorisCatalog.getName(), dbName, tblName, db.isPresent());
+ return db;
}
@Override
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java
index f42112d1e11..4caeeb0883f 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java
@@ -638,7 +638,8 @@ public class PhysicalPlanTranslator extends
DefaultPlanVisitor<PlanFragment, Pla
rootFragment.setOutputExprs(outputExprs);
IcebergMergeSink sink = new IcebergMergeSink(
(IcebergExternalTable) icebergMergeSink.getTargetTable(),
- icebergMergeSink.getDeleteContext());
+ icebergMergeSink.getDeleteContext(),
+ icebergMergeSink.isRequireMergeCardinalityCheck());
rootFragment.setSink(sink);
return rootFragment;
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/RequestPropertyDeriver.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/RequestPropertyDeriver.java
index 6ea2601bbee..9c04507680f 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/RequestPropertyDeriver.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/RequestPropertyDeriver.java
@@ -202,9 +202,12 @@ public class RequestPropertyDeriver extends
PlanVisitor<Void, PlanContext> {
@Override
public Void visitPhysicalIcebergMergeSink(
PhysicalIcebergMergeSink<? extends Plan> icebergMergeSink,
PlanContext context) {
- if (connectContext != null &&
!connectContext.getSessionVariable().enableStrictConsistencyDml) {
+ if (!icebergMergeSink.isRequireMergeCardinalityCheck()
+ && connectContext != null
+ &&
!connectContext.getSessionVariable().enableStrictConsistencyDml) {
addRequestPropertyToChildren(PhysicalProperties.ANY);
} else {
+ // SQL MERGE cardinality is mandatory even when optional UPDATE
consistency is disabled.
addRequestPropertyToChildren(icebergMergeSink.getRequirePhysicalProperties());
}
return null;
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/implementation/LogicalIcebergMergeSinkToPhysicalIcebergMergeSink.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/implementation/LogicalIcebergMergeSinkToPhysicalIcebergMergeSink.java
index 9447aaf09d1..85b813b7fd6 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/implementation/LogicalIcebergMergeSinkToPhysicalIcebergMergeSink.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/implementation/LogicalIcebergMergeSinkToPhysicalIcebergMergeSink.java
@@ -39,6 +39,7 @@ public class
LogicalIcebergMergeSinkToPhysicalIcebergMergeSink extends OneImplem
sink.getCols(),
sink.getOutputExprs(),
sink.getDeleteContext(),
+ sink.isRequireMergeCardinalityCheck(),
Optional.empty(),
sink.getLogicalProperties(),
null,
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/CreateTableCommand.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/CreateTableCommand.java
index 7b9837183c6..3031e67be7d 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/CreateTableCommand.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/CreateTableCommand.java
@@ -18,10 +18,12 @@
package org.apache.doris.nereids.trees.plans.commands;
import org.apache.doris.analysis.StmtType;
+import org.apache.doris.catalog.DatabaseIf;
import org.apache.doris.catalog.Env;
import org.apache.doris.catalog.ScalarType;
import org.apache.doris.common.ErrorCode;
import org.apache.doris.common.FeConstants;
+import org.apache.doris.datasource.CatalogIf;
import org.apache.doris.nereids.NereidsPlanner;
import org.apache.doris.nereids.analyzer.UnboundResultSink;
import org.apache.doris.nereids.analyzer.UnboundTableSinkCreator;
@@ -99,6 +101,19 @@ public class CreateTableCommand extends Command implements
NeedAuditEncryption,
LOG.debug("Nereids start to execute the ctas command, query id:
{}, tableName: {}",
ctx.queryId(), createTableInfo.getTableName());
}
+ LogicalPlan sinkQuery = null;
+ if (!createTableInfo.isIfNotExists()) {
+ // An existence probe is used only to preserve the catalog
diagnostic; creation still
+ // goes through the atomic catalog API and is the sole proof of
ownership.
+ if (targetTableExists(ctx)) {
+ throw new
AnalysisException(ErrorCode.ERR_TABLE_EXISTS_ERROR.formatErrorMsg(
+ createTableInfo.getTableName()));
+ }
+ // Reject unsupported destinations before publishing metadata;
rollback by table name
+ // cannot distinguish this CTAS table from a concurrent
replacement with the same name.
+ sinkQuery =
UnboundTableSinkCreator.createUnboundTableSink(createTableInfo.getTableNameParts(),
+ ImmutableList.of(), ImmutableList.of(),
ImmutableList.of(), query);
+ }
try {
if (Env.getCurrentEnv().createTable(this.createTableInfo)) {
return;
@@ -107,12 +122,16 @@ public class CreateTableCommand extends Command
implements NeedAuditEncryption,
throw new AnalysisException(e.getMessage(), e.getCause());
}
- query =
UnboundTableSinkCreator.createUnboundTableSink(createTableInfo.getTableNameParts(),
- ImmutableList.of(), ImmutableList.of(), ImmutableList.of(),
query);
try {
+ if (sinkQuery == null) {
+ // IF NOT EXISTS must honor the catalog's atomic
existing-table result before sink
+ // validation, otherwise an unsupported connector turns the
required no-op into an error.
+ sinkQuery =
UnboundTableSinkCreator.createUnboundTableSink(createTableInfo.getTableNameParts(),
+ ImmutableList.of(), ImmutableList.of(),
ImmutableList.of(), query);
+ }
InsertIntoTableCommand insertCommand = null;
if (!FeConstants.runningUnitTest) {
- insertCommand = new InsertIntoTableCommand(query,
Optional.empty(),
+ insertCommand = new InsertIntoTableCommand(sinkQuery,
Optional.empty(),
Optional.empty(), Optional.empty(), true,
Optional.empty());
insertCommand.run(ctx, executor);
if (ctx.getState().getStateType() == MysqlStateType.OK) {
@@ -227,6 +246,16 @@ public class CreateTableCommand extends Command implements
NeedAuditEncryption,
}
}
+ boolean targetTableExists(ConnectContext ctx) {
+ List<String> qualifiedName = RelationUtil.getQualifierName(ctx,
createTableInfo.getTableNameParts());
+ CatalogIf<?> catalog =
Env.getCurrentEnv().getCatalogMgr().getCatalog(qualifiedName.get(0));
+ if (catalog == null) {
+ return false;
+ }
+ DatabaseIf<?> database = catalog.getDbNullable(qualifiedName.get(1));
+ return database != null && database.isTableExist(qualifiedName.get(2));
+ }
+
private String getAutoRangePartitionNameOrNull() {
try {
if (createTableInfo.getPartitionTableInfo().isAutoPartition()
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/IcebergMergeCommand.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/IcebergMergeCommand.java
index 59eb6a6d6ac..4d4b9b2014f 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/IcebergMergeCommand.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/IcebergMergeCommand.java
@@ -465,6 +465,7 @@ public class IcebergMergeCommand extends Command implements
ForwardWithSync, Exp
icebergTable.getBaseSchema(true),
outputExprs,
deleteCtx,
+ true,
Optional.empty(),
Optional.empty(),
projectPlan);
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/IcebergUpdateCommand.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/IcebergUpdateCommand.java
index 8759343e055..d9d0ab51b58 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/IcebergUpdateCommand.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/IcebergUpdateCommand.java
@@ -241,6 +241,7 @@ public class IcebergUpdateCommand extends Command
implements ForwardWithSync, Ex
icebergTable.getBaseSchema(true),
outputExprs,
deleteCtx,
+ false,
Optional.empty(),
Optional.empty(),
queryPlan);
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalIcebergMergeSink.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalIcebergMergeSink.java
index 7f528020890..c8198c27dcc 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalIcebergMergeSink.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalIcebergMergeSink.java
@@ -47,6 +47,7 @@ public class LogicalIcebergMergeSink<CHILD_TYPE extends Plan>
extends LogicalTab
private final IcebergExternalDatabase database;
private final IcebergExternalTable targetTable;
private final DeleteCommandContext deleteContext;
+ private final boolean requireMergeCardinalityCheck;
/**
* Constructor
@@ -56,6 +57,7 @@ public class LogicalIcebergMergeSink<CHILD_TYPE extends Plan>
extends LogicalTab
List<Column> cols,
List<NamedExpression> outputExprs,
DeleteCommandContext deleteContext,
+ boolean requireMergeCardinalityCheck,
Optional<GroupExpression> groupExpression,
Optional<LogicalProperties>
logicalProperties,
CHILD_TYPE child) {
@@ -63,6 +65,7 @@ public class LogicalIcebergMergeSink<CHILD_TYPE extends Plan>
extends LogicalTab
this.database = Objects.requireNonNull(database, "database != null in
LogicalIcebergMergeSink");
this.targetTable = Objects.requireNonNull(targetTable, "targetTable !=
null in LogicalIcebergMergeSink");
this.deleteContext = Objects.requireNonNull(deleteContext,
"deleteContext != null in LogicalIcebergMergeSink");
+ this.requireMergeCardinalityCheck = requireMergeCardinalityCheck;
}
public Plan withChildAndUpdateOutput(Plan child) {
@@ -70,19 +73,19 @@ public class LogicalIcebergMergeSink<CHILD_TYPE extends
Plan> extends LogicalTab
.map(NamedExpression.class::cast)
.collect(ImmutableList.toImmutableList());
return new LogicalIcebergMergeSink<>(database, targetTable, cols,
output,
- deleteContext, Optional.empty(), Optional.empty(), child);
+ deleteContext, requireMergeCardinalityCheck, Optional.empty(),
Optional.empty(), child);
}
@Override
public Plan withChildren(List<Plan> children) {
Preconditions.checkArgument(children.size() == 1,
"LogicalIcebergMergeSink only accepts one child");
return new LogicalIcebergMergeSink<>(database, targetTable, cols,
outputExprs,
- deleteContext, Optional.empty(), Optional.empty(),
children.get(0));
+ deleteContext, requireMergeCardinalityCheck, Optional.empty(),
Optional.empty(), children.get(0));
}
public LogicalIcebergMergeSink<CHILD_TYPE>
withOutputExprs(List<NamedExpression> outputExprs) {
return new LogicalIcebergMergeSink<>(database, targetTable, cols,
outputExprs,
- deleteContext, Optional.empty(), Optional.empty(), child());
+ deleteContext, requireMergeCardinalityCheck, Optional.empty(),
Optional.empty(), child());
}
public IcebergExternalDatabase getDatabase() {
@@ -97,6 +100,10 @@ public class LogicalIcebergMergeSink<CHILD_TYPE extends
Plan> extends LogicalTab
return deleteContext;
}
+ public boolean isRequireMergeCardinalityCheck() {
+ return requireMergeCardinalityCheck;
+ }
+
@Override
public boolean equals(Object o) {
if (this == o) {
@@ -112,12 +119,14 @@ public class LogicalIcebergMergeSink<CHILD_TYPE extends
Plan> extends LogicalTab
return Objects.equals(database, that.database)
&& Objects.equals(targetTable, that.targetTable)
&& Objects.equals(deleteContext, that.deleteContext)
+ && requireMergeCardinalityCheck ==
that.requireMergeCardinalityCheck
&& Objects.equals(cols, that.cols);
}
@Override
public int hashCode() {
- return Objects.hash(super.hashCode(), database, targetTable, cols,
deleteContext);
+ return Objects.hash(super.hashCode(), database, targetTable, cols,
deleteContext,
+ requireMergeCardinalityCheck);
}
@Override
@@ -127,7 +136,8 @@ public class LogicalIcebergMergeSink<CHILD_TYPE extends
Plan> extends LogicalTab
"database", database.getFullName(),
"targetTable", targetTable.getName(),
"cols", cols,
- "deleteFileType", deleteContext.getDeleteFileType());
+ "deleteFileType", deleteContext.getDeleteFileType(),
+ "requireMergeCardinalityCheck", requireMergeCardinalityCheck);
}
@Override
@@ -138,13 +148,15 @@ public class LogicalIcebergMergeSink<CHILD_TYPE extends
Plan> extends LogicalTab
@Override
public Plan withGroupExpression(Optional<GroupExpression> groupExpression)
{
return new LogicalIcebergMergeSink<>(database, targetTable, cols,
outputExprs,
- deleteContext, groupExpression,
Optional.of(getLogicalProperties()), child());
+ deleteContext, requireMergeCardinalityCheck,
+ groupExpression, Optional.of(getLogicalProperties()), child());
}
@Override
public Plan withGroupExprLogicalPropChildren(Optional<GroupExpression>
groupExpression,
Optional<LogicalProperties> logicalProperties, List<Plan>
children) {
return new LogicalIcebergMergeSink<>(database, targetTable, cols,
outputExprs,
- deleteContext, groupExpression, logicalProperties,
children.get(0));
+ deleteContext, requireMergeCardinalityCheck,
+ groupExpression, logicalProperties, children.get(0));
}
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalIcebergMergeSink.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalIcebergMergeSink.java
index 0281ad23243..e182c5adc06 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalIcebergMergeSink.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalIcebergMergeSink.java
@@ -22,7 +22,6 @@ import
org.apache.doris.datasource.iceberg.IcebergExternalDatabase;
import org.apache.doris.datasource.iceberg.IcebergExternalTable;
import org.apache.doris.datasource.iceberg.IcebergMergeOperation;
import org.apache.doris.nereids.memo.GroupExpression;
-import org.apache.doris.nereids.properties.DistributionSpecHash.ShuffleType;
import org.apache.doris.nereids.properties.DistributionSpecMerge;
import org.apache.doris.nereids.properties.LogicalProperties;
import org.apache.doris.nereids.properties.PhysicalProperties;
@@ -56,6 +55,7 @@ import java.util.TreeMap;
*/
public class PhysicalIcebergMergeSink<CHILD_TYPE extends Plan> extends
PhysicalBaseExternalTableSink<CHILD_TYPE> {
private final DeleteCommandContext deleteContext;
+ private final boolean requireMergeCardinalityCheck;
/**
* Constructor
@@ -65,10 +65,12 @@ public class PhysicalIcebergMergeSink<CHILD_TYPE extends
Plan> extends PhysicalB
List<Column> cols,
List<NamedExpression> outputExprs,
DeleteCommandContext deleteContext,
+ boolean requireMergeCardinalityCheck,
Optional<GroupExpression> groupExpression,
LogicalProperties logicalProperties,
CHILD_TYPE child) {
- this(database, targetTable, cols, outputExprs, deleteContext,
groupExpression, logicalProperties,
+ this(database, targetTable, cols, outputExprs, deleteContext,
requireMergeCardinalityCheck,
+ groupExpression, logicalProperties,
PhysicalProperties.GATHER, null, child);
}
@@ -80,6 +82,7 @@ public class PhysicalIcebergMergeSink<CHILD_TYPE extends
Plan> extends PhysicalB
List<Column> cols,
List<NamedExpression> outputExprs,
DeleteCommandContext deleteContext,
+ boolean requireMergeCardinalityCheck,
Optional<GroupExpression> groupExpression,
LogicalProperties logicalProperties,
PhysicalProperties physicalProperties,
@@ -89,17 +92,22 @@ public class PhysicalIcebergMergeSink<CHILD_TYPE extends
Plan> extends PhysicalB
logicalProperties, physicalProperties, statistics, child);
this.deleteContext = Objects.requireNonNull(
deleteContext, "deleteContext != null in
PhysicalIcebergMergeSink");
+ this.requireMergeCardinalityCheck = requireMergeCardinalityCheck;
}
public DeleteCommandContext getDeleteContext() {
return deleteContext;
}
+ public boolean isRequireMergeCardinalityCheck() {
+ return requireMergeCardinalityCheck;
+ }
+
@Override
public Plan withChildren(List<Plan> children) {
return new PhysicalIcebergMergeSink<>(
(IcebergExternalDatabase) database, (IcebergExternalTable)
targetTable,
- cols, outputExprs, deleteContext, groupExpression,
+ cols, outputExprs, deleteContext,
requireMergeCardinalityCheck, groupExpression,
getLogicalProperties(), physicalProperties, statistics,
children.get(0));
}
@@ -112,7 +120,8 @@ public class PhysicalIcebergMergeSink<CHILD_TYPE extends
Plan> extends PhysicalB
public Plan withGroupExpression(Optional<GroupExpression> groupExpression)
{
return new PhysicalIcebergMergeSink<>(
(IcebergExternalDatabase) database, (IcebergExternalTable)
targetTable, cols, outputExprs,
- deleteContext, groupExpression, getLogicalProperties(),
child());
+ deleteContext, requireMergeCardinalityCheck,
+ groupExpression, getLogicalProperties(), child());
}
@Override
@@ -120,14 +129,16 @@ public class PhysicalIcebergMergeSink<CHILD_TYPE extends
Plan> extends PhysicalB
Optional<LogicalProperties>
logicalProperties, List<Plan> children) {
return new PhysicalIcebergMergeSink<>(
(IcebergExternalDatabase) database, (IcebergExternalTable)
targetTable, cols, outputExprs,
- deleteContext, groupExpression, logicalProperties.get(),
children.get(0));
+ deleteContext, requireMergeCardinalityCheck,
+ groupExpression, logicalProperties.get(), children.get(0));
}
@Override
public PhysicalPlan withPhysicalPropertiesAndStats(PhysicalProperties
physicalProperties, Statistics statistics) {
return new PhysicalIcebergMergeSink<>(
(IcebergExternalDatabase) database, (IcebergExternalTable)
targetTable, cols, outputExprs,
- deleteContext, groupExpression, getLogicalProperties(),
physicalProperties, statistics, child());
+ deleteContext, requireMergeCardinalityCheck,
+ groupExpression, getLogicalProperties(), physicalProperties,
statistics, child());
}
@Override
@@ -142,12 +153,13 @@ public class PhysicalIcebergMergeSink<CHILD_TYPE extends
Plan> extends PhysicalB
return false;
}
PhysicalIcebergMergeSink<?> that = (PhysicalIcebergMergeSink<?>) o;
- return Objects.equals(deleteContext, that.deleteContext);
+ return Objects.equals(deleteContext, that.deleteContext)
+ && requireMergeCardinalityCheck ==
that.requireMergeCardinalityCheck;
}
@Override
public int hashCode() {
- return Objects.hash(super.hashCode(), deleteContext);
+ return Objects.hash(super.hashCode(), deleteContext,
requireMergeCardinalityCheck);
}
/**
@@ -172,8 +184,16 @@ public class PhysicalIcebergMergeSink<CHILD_TYPE extends
Plan> extends PhysicalB
ConnectContext ctx = ConnectContext.get();
if (ctx == null ||
!ctx.getSessionVariable().isEnableIcebergMergePartitioning()) {
- if (rowIdExprId != null) {
- return
PhysicalProperties.createHash(ImmutableList.of(rowIdExprId),
ShuffleType.REQUIRE);
+ if (rowIdExprId != null && operationExprId != null) {
+ // Route only delete images by row ID; unmatched inserts have
NULL row IDs and must
+ // remain distributable instead of collapsing onto one
exchange channel.
+ return new PhysicalProperties(new DistributionSpecMerge(
+ operationExprId,
+ ImmutableList.of(),
+ ImmutableList.of(rowIdExprId),
+ true,
+ ImmutableList.of(),
+ null));
}
return PhysicalProperties.GATHER;
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/planner/IcebergMergeSink.java
b/fe/fe-core/src/main/java/org/apache/doris/planner/IcebergMergeSink.java
index 9fd7dc046fa..bf52a777b04 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/IcebergMergeSink.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/planner/IcebergMergeSink.java
@@ -65,6 +65,7 @@ public class IcebergMergeSink extends
BaseExternalTableDataSink {
private final IcebergExternalTable targetTable;
private final DeleteCommandContext deleteContext;
+ private final boolean requireMergeCardinalityCheck;
private List<TIcebergRewritableDeleteFileSet> rewritableDeleteFileSets =
Collections.emptyList();
private static final HashSet<TFileFormatType> supportedTypes = new
HashSet<TFileFormatType>() {{
@@ -75,13 +76,15 @@ public class IcebergMergeSink extends
BaseExternalTableDataSink {
// Store PropertiesMap, including vended credentials or static credentials
private Map<StorageTypeId, StorageAdapter> storagePropertiesMap;
- public IcebergMergeSink(IcebergExternalTable targetTable,
DeleteCommandContext deleteContext) {
+ public IcebergMergeSink(IcebergExternalTable targetTable,
DeleteCommandContext deleteContext,
+ boolean requireMergeCardinalityCheck) {
super();
if (targetTable.isView()) {
throw new UnsupportedOperationException("UPDATE on iceberg view is
not supported");
}
this.targetTable = targetTable;
this.deleteContext = deleteContext;
+ this.requireMergeCardinalityCheck = requireMergeCardinalityCheck;
IcebergExternalCatalog catalog = (IcebergExternalCatalog)
targetTable.getCatalog();
storagePropertiesMap =
VendedCredentialsFactory.getStoragePropertiesMapWithVendedCredentials(
@@ -133,6 +136,8 @@ public class IcebergMergeSink extends
BaseExternalTableDataSink {
tSink.setFormatVersion(formatVersion);
tSink.setSchemaJson(SchemaParser.toJson(schema));
tSink.setCollectColumnStats(IcebergUtils.shouldCollectColumnStats(icebergTable,
schema));
+ // UPDATE and SQL MERGE share this sink, but only SQL MERGE has the
one-source-row invariant.
+ tSink.setRequireMergeCardinalityCheck(requireMergeCardinalityCheck);
// partition spec
if (icebergTable.spec().isPartitioned()) {
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/IcebergDDLAndDMLPlanTest.java
b/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/IcebergDDLAndDMLPlanTest.java
index b4a1c6a5d1b..556ff50225f 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/IcebergDDLAndDMLPlanTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/IcebergDDLAndDMLPlanTest.java
@@ -27,7 +27,6 @@ import org.apache.doris.nereids.NereidsPlanner;
import org.apache.doris.nereids.StatementContext;
import org.apache.doris.nereids.analyzer.UnboundAlias;
import org.apache.doris.nereids.glue.LogicalPlanAdapter;
-import org.apache.doris.nereids.properties.DistributionSpecHash;
import org.apache.doris.nereids.properties.DistributionSpecMerge;
import org.apache.doris.nereids.properties.PhysicalProperties;
import org.apache.doris.nereids.trees.expressions.Alias;
@@ -462,6 +461,68 @@ public class IcebergDDLAndDMLPlanTest extends
TestWithFeService {
}
}
+ @Test
+ public void testIcebergMergeIntoExchangeIgnoresStrictConsistencySwitch()
throws Exception {
+ useIceberg();
+ boolean previousMergePartitioning =
connectContext.getSessionVariable().enableIcebergMergePartitioning;
+ boolean previousStrictConsistency =
connectContext.getSessionVariable().enableStrictConsistencyDml;
+ connectContext.getSessionVariable().enableIcebergMergePartitioning =
true;
+ connectContext.getSessionVariable().enableStrictConsistencyDml = false;
+ try {
+ String sql = "merge into " + tableName + " t "
+ + "using (select 1 as id, 'name1' as name, 10 as age, 1 as
score, 1.23 as amount) s "
+ + "on t.id = s.id "
+ + "when matched then update set name = s.name";
+ LogicalPlan mergePlan = parseStmt(sql);
+ Plan explainPlan = ((MergeIntoCommand)
mergePlan).getExplainPlan(connectContext);
+ PhysicalPlan physicalPlan =
+ planPhysicalPlan((LogicalPlan) explainPlan,
PhysicalProperties.GATHER, sql);
+
+ PhysicalIcebergMergeSink<?> sink =
+ getSinglePhysicalSink(physicalPlan,
PhysicalIcebergMergeSink.class);
+ Assertions.assertTrue(sink.isRequireMergeCardinalityCheck());
+ Assertions.assertTrue(sink.child() instanceof PhysicalDistribute,
+ "MERGE cardinality requires row-id routing even when
strict consistency is disabled\n"
+ + physicalPlan.treeString());
+ Assertions.assertTrue(
+ ((PhysicalDistribute<?>)
sink.child()).getDistributionSpec() instanceof DistributionSpecMerge,
+ "Missing merge distribution spec\n" +
physicalPlan.treeString());
+ } finally {
+ connectContext.getSessionVariable().enableIcebergMergePartitioning
= previousMergePartitioning;
+ connectContext.getSessionVariable().enableStrictConsistencyDml =
previousStrictConsistency;
+ }
+ }
+
+ @Test
+ public void testIcebergUpdateAvoidsExchangeWhenStrictConsistencyDisabled()
throws Exception {
+ useIceberg();
+ boolean previousMergePartitioning =
connectContext.getSessionVariable().enableIcebergMergePartitioning;
+ boolean previousStrictConsistency =
connectContext.getSessionVariable().enableStrictConsistencyDml;
+ connectContext.getSessionVariable().enableIcebergMergePartitioning =
true;
+ connectContext.getSessionVariable().enableStrictConsistencyDml = false;
+ try {
+ // UPDATE FROM may produce the same target row more than once, but
SQL MERGE's
+ // one-source-row cardinality rule must not be applied to UPDATE
statements.
+ String sql = "update " + tableName + " t set name = s.name "
+ + "from (select 1 as id, 'first' as name union all "
+ + "select 1 as id, 'second' as name) s where t.id = s.id";
+ LogicalPlan updatePlan = parseStmt(sql);
+ Plan explainPlan = ((UpdateCommand)
updatePlan).getExplainPlan(connectContext);
+ PhysicalPlan physicalPlan =
+ planPhysicalPlan((LogicalPlan) explainPlan,
PhysicalProperties.GATHER, sql);
+
+ PhysicalIcebergMergeSink<?> sink =
+ getSinglePhysicalSink(physicalPlan,
PhysicalIcebergMergeSink.class);
+ Assertions.assertFalse(sink.isRequireMergeCardinalityCheck());
+ Assertions.assertFalse(sink.child() instanceof PhysicalDistribute,
+ "Strict-consistency-off UPDATE should not add an Iceberg
merge exchange\n"
+ + physicalPlan.treeString());
+ } finally {
+ connectContext.getSessionVariable().enableIcebergMergePartitioning
= previousMergePartitioning;
+ connectContext.getSessionVariable().enableStrictConsistencyDml =
previousStrictConsistency;
+ }
+ }
+
@Test
public void testIcebergUpdateCastsConstantForSmallintAndDecimal() throws
Exception {
useIceberg();
@@ -477,7 +538,7 @@ public class IcebergDDLAndDMLPlanTest extends
TestWithFeService {
}
@Test
- public void testIcebergUpdateExchangeUsesRowIdOnlyWhenDisabled() throws
Exception {
+ public void
testIcebergUpdateExchangeRoutesOnlyDeletesByRowIdWhenDisabled() throws
Exception {
useIceberg();
boolean previous =
connectContext.getSessionVariable().enableIcebergMergePartitioning;
connectContext.getSessionVariable().enableIcebergMergePartitioning =
false;
@@ -490,19 +551,93 @@ public class IcebergDDLAndDMLPlanTest extends
TestWithFeService {
PhysicalIcebergMergeSink<?> sink =
getSinglePhysicalSink(physicalPlan,
PhysicalIcebergMergeSink.class);
+ ExprId operationExprId =
findOperationExprId(sink.child().getOutput());
ExprId rowIdExprId = findRowIdExprId(sink.child().getOutput());
Assertions.assertTrue(sink.child() instanceof PhysicalDistribute,
- "Missing row_id exchange\n" + physicalPlan.treeString());
+ "Missing operation-aware exchange\n" +
physicalPlan.treeString());
PhysicalDistribute<?> distribute = (PhysicalDistribute<?>)
sink.child();
- Assertions.assertTrue(distribute.getDistributionSpec() instanceof
DistributionSpecHash,
- "Missing row_id hash distribution\n" +
physicalPlan.treeString());
- DistributionSpecHash hash = (DistributionSpecHash)
distribute.getDistributionSpec();
- Assertions.assertEquals(ImmutableList.of(rowIdExprId),
hash.getOrderedShuffledColumns());
+ Assertions.assertTrue(distribute.getDistributionSpec() instanceof
DistributionSpecMerge,
+ "Missing operation-aware merge distribution\n" +
physicalPlan.treeString());
+ DistributionSpecMerge spec = (DistributionSpecMerge)
distribute.getDistributionSpec();
+ Assertions.assertEquals(operationExprId,
spec.getOperationExprId());
+ Assertions.assertTrue(spec.isInsertRandom());
+ Assertions.assertEquals(ImmutableList.of(rowIdExprId),
spec.getDeletePartitionExprIds());
} finally {
connectContext.getSessionVariable().enableIcebergMergePartitioning
= previous;
}
}
+ @Test
+ public void
testIcebergInsertOnlyMergeDoesNotHashNullRowIdsWhenStrictConsistencyDisabled()
+ throws Exception {
+ useIceberg();
+ boolean previousMergePartitioning =
connectContext.getSessionVariable().enableIcebergMergePartitioning;
+ boolean previousStrictConsistency =
connectContext.getSessionVariable().enableStrictConsistencyDml;
+ connectContext.getSessionVariable().enableIcebergMergePartitioning =
false;
+ connectContext.getSessionVariable().enableStrictConsistencyDml = false;
+ try {
+ String sql = "merge into " + tableName + " t "
+ + "using (select 1 as id, 'name1' as name, 10 as age, 1 as
score, 1.23 as amount) s "
+ + "on t.id = s.id "
+ + "when not matched then insert (id, name, age, score,
amount) "
+ + "values (s.id, s.name, s.age, s.score, s.amount)";
+ LogicalPlan mergePlan = parseStmt(sql);
+ Plan explainPlan = ((MergeIntoCommand)
mergePlan).getExplainPlan(connectContext);
+ PhysicalPlan physicalPlan =
+ planPhysicalPlan((LogicalPlan) explainPlan,
PhysicalProperties.GATHER, sql);
+
+ PhysicalIcebergMergeSink<?> sink =
+ getSinglePhysicalSink(physicalPlan,
PhysicalIcebergMergeSink.class);
+ Assertions.assertTrue(sink.child() instanceof PhysicalDistribute,
+ "Missing operation-aware exchange\n" +
physicalPlan.treeString());
+ DistributionSpecMerge spec = (DistributionSpecMerge)
+ ((PhysicalDistribute<?>)
sink.child()).getDistributionSpec();
+ Assertions.assertTrue(spec.isInsertRandom());
+ Assertions.assertTrue(spec.getInsertPartitionExprIds().isEmpty());
+ Assertions.assertEquals(
+
ImmutableList.of(findRowIdExprId(sink.child().getOutput())),
+ spec.getDeletePartitionExprIds());
+ } finally {
+ connectContext.getSessionVariable().enableIcebergMergePartitioning
= previousMergePartitioning;
+ connectContext.getSessionVariable().enableStrictConsistencyDml =
previousStrictConsistency;
+ }
+ }
+
+ @Test
+ public void
testIcebergMixedMergeRoutesInsertsRandomlyWhenStrictConsistencyDisabled()
+ throws Exception {
+ useIceberg();
+ boolean previousMergePartitioning =
connectContext.getSessionVariable().enableIcebergMergePartitioning;
+ boolean previousStrictConsistency =
connectContext.getSessionVariable().enableStrictConsistencyDml;
+ connectContext.getSessionVariable().enableIcebergMergePartitioning =
false;
+ connectContext.getSessionVariable().enableStrictConsistencyDml = false;
+ try {
+ String sql = "merge into " + tableName + " t "
+ + "using (select 1 as id, 'name1' as name, 10 as age, 1 as
score, 1.23 as amount) s "
+ + "on t.id = s.id "
+ + "when matched then update set name = s.name "
+ + "when not matched then insert (id, name, age, score,
amount) "
+ + "values (s.id, s.name, s.age, s.score, s.amount)";
+ LogicalPlan mergePlan = parseStmt(sql);
+ Plan explainPlan = ((MergeIntoCommand)
mergePlan).getExplainPlan(connectContext);
+ PhysicalPlan physicalPlan =
+ planPhysicalPlan((LogicalPlan) explainPlan,
PhysicalProperties.GATHER, sql);
+
+ PhysicalIcebergMergeSink<?> sink =
+ getSinglePhysicalSink(physicalPlan,
PhysicalIcebergMergeSink.class);
+ DistributionSpecMerge spec = (DistributionSpecMerge)
+ ((PhysicalDistribute<?>)
sink.child()).getDistributionSpec();
+ Assertions.assertTrue(spec.isInsertRandom());
+ Assertions.assertTrue(spec.getInsertPartitionExprIds().isEmpty());
+ Assertions.assertEquals(
+
ImmutableList.of(findRowIdExprId(sink.child().getOutput())),
+ spec.getDeletePartitionExprIds());
+ } finally {
+ connectContext.getSessionVariable().enableIcebergMergePartitioning
= previousMergePartitioning;
+ connectContext.getSessionVariable().enableStrictConsistencyDml =
previousStrictConsistency;
+ }
+ }
+
@Test
public void testIcebergUpdateExchangeUsesMergePartitioningWhenEnabled()
throws Exception {
useIceberg();
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/datasource/paimon/PaimonMetadataOpsTest.java
b/fe/fe-core/src/test/java/org/apache/doris/datasource/paimon/PaimonMetadataOpsTest.java
index dda2c3d2344..c4a887c132a 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/datasource/paimon/PaimonMetadataOpsTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/datasource/paimon/PaimonMetadataOpsTest.java
@@ -20,6 +20,8 @@ package org.apache.doris.datasource.paimon;
import org.apache.doris.common.DdlException;
import org.apache.doris.common.UserException;
import org.apache.doris.datasource.CatalogFactory;
+import org.apache.doris.datasource.ExternalCatalog;
+import org.apache.doris.datasource.ExternalDatabase;
import org.apache.doris.nereids.parser.NereidsParser;
import org.apache.doris.nereids.trees.plans.commands.CreateCatalogCommand;
import org.apache.doris.nereids.trees.plans.commands.CreateTableCommand;
@@ -32,6 +34,7 @@ import org.apache.paimon.catalog.Catalog;
import org.apache.paimon.catalog.FileSystemCatalog;
import org.apache.paimon.catalog.Identifier;
import org.apache.paimon.hive.HiveCatalog;
+import org.apache.paimon.schema.Schema;
import org.apache.paimon.table.Table;
import org.apache.paimon.types.BigIntType;
import org.apache.paimon.types.DataField;
@@ -46,12 +49,14 @@ import org.junit.Assert;
import org.junit.BeforeClass;
import org.junit.Test;
import org.junit.jupiter.api.Assertions;
+import org.mockito.Mockito;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
+import java.util.Optional;
import java.util.UUID;
import java.util.stream.Collectors;
@@ -245,13 +250,70 @@ public class PaimonMetadataOpsTest {
Assert.assertEquals("c0", table.options().get("bucket-key"));
}
+ @Test
+ public void testIfNotExistsRefreshesNamesWhenRemoteTableExists() throws
Exception {
+ String tableName = getTableName();
+ Catalog remoteCatalog = Mockito.mock(Catalog.class);
+ ExternalCatalog dorisCatalog = Mockito.mock(ExternalCatalog.class);
+ ExternalDatabase<?> database = Mockito.mock(ExternalDatabase.class);
+ Mockito.doReturn(database).when(dorisCatalog).getDbNullable(dbName);
+
Mockito.doReturn(Optional.of(database)).when(dorisCatalog).getDbForReplay(dbName);
+ Mockito.when(database.getRemoteName()).thenReturn(dbName);
+
+ PaimonMetadataOps existingTableOps = new
PaimonMetadataOps(dorisCatalog, remoteCatalog) {
+ @Override
+ public boolean tableExist(String ignoredDbName, String
ignoredTableName) {
+ return true;
+ }
+ };
+ CreateTableInfo createTableInfo = parseCreateTableInfo(
+ "create table if not exists " + dbName + "." + tableName + "
(id int) engine = paimon");
+
+
Assert.assertTrue(existingTableOps.performCreateTable(createTableInfo));
+ Mockito.verify(database).resetMetaCacheNames();
+ Mockito.verify(remoteCatalog, Mockito.never()).createTable(
+ Mockito.any(Identifier.class), Mockito.any(Schema.class),
Mockito.anyBoolean());
+ }
+
+ @Test
+ public void testIfNotExistsReportsConcurrentWinner() throws Exception {
+ String tableName = getTableName();
+ Identifier identifier = new Identifier(dbName, tableName);
+ Catalog remoteCatalog = Mockito.mock(Catalog.class);
+ ExternalCatalog dorisCatalog = Mockito.mock(ExternalCatalog.class);
+ ExternalDatabase<?> database = Mockito.mock(ExternalDatabase.class);
+ Mockito.doReturn(database).when(dorisCatalog).getDbNullable(dbName);
+
Mockito.doReturn(Optional.of(database)).when(dorisCatalog).getDbForReplay(dbName);
+ Mockito.when(database.getRemoteName()).thenReturn(dbName);
+ Mockito.when(database.getTableNullable(tableName)).thenReturn(null);
+
+ PaimonMetadataOps raceOps = new PaimonMetadataOps(dorisCatalog,
remoteCatalog) {
+ @Override
+ public boolean tableExist(String ignoredDbName, String
ignoredTableName) {
+ return false;
+ }
+ };
+ CreateTableInfo createTableInfo = parseCreateTableInfo(
+ "create table if not exists " + dbName + "." + tableName + "
(id int) engine = paimon");
+ Mockito.doThrow(new Catalog.TableAlreadyExistException(identifier))
+ .when(remoteCatalog).createTable(
+ Mockito.eq(identifier), Mockito.any(Schema.class),
Mockito.eq(false));
+
+ Assert.assertTrue(raceOps.performCreateTable(createTableInfo));
+ Mockito.verify(database).resetMetaCacheNames();
+ }
+
public void createTable(String sql) throws UserException {
+ ops.createTable(parseCreateTableInfo(sql));
+ }
+
+ private CreateTableInfo parseCreateTableInfo(String sql) throws
UserException {
LogicalPlan plan = new NereidsParser().parseSingle(sql);
Assertions.assertTrue(plan instanceof CreateTableCommand);
CreateTableInfo createTableInfo = ((CreateTableCommand)
plan).getCreateTableInfo();
createTableInfo.setIsExternal(true);
createTableInfo.analyzeEngine();
- ops.createTable(createTableInfo);
+ return createTableInfo;
}
public String getTableName() {
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/CreateTableCommandTest.java
b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/CreateTableCommandTest.java
new file mode 100644
index 00000000000..38324e828ae
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/CreateTableCommandTest.java
@@ -0,0 +1,133 @@
+// 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.
+
+package org.apache.doris.nereids.trees.plans.commands;
+
+import org.apache.doris.catalog.Env;
+import org.apache.doris.common.UserException;
+import org.apache.doris.nereids.analyzer.UnboundTableSinkCreator;
+import org.apache.doris.nereids.trees.plans.commands.info.CreateTableInfo;
+import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
+import org.apache.doris.qe.ConnectContext;
+import org.apache.doris.qe.StmtExecutor;
+
+import com.google.common.collect.ImmutableList;
+import org.junit.jupiter.api.Test;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
+
+import java.util.List;
+import java.util.Optional;
+
+class CreateTableCommandTest {
+
+ @Test
+ void ctasIfNotExistsReturnsBeforeUnsupportedSinkValidation() throws
Exception {
+ LogicalPlan query = Mockito.mock(LogicalPlan.class);
+ CreateTableInfo createTableInfo = Mockito.mock(CreateTableInfo.class);
+ ConnectContext context = Mockito.mock(ConnectContext.class);
+ StmtExecutor executor = Mockito.mock(StmtExecutor.class);
+ Env env = Mockito.mock(Env.class);
+ List<String> tableNameParts = ImmutableList.of("catalog", "database",
"target");
+
+
Mockito.when(createTableInfo.getTableNameParts()).thenReturn(tableNameParts);
+ Mockito.when(createTableInfo.isIfNotExists()).thenReturn(true);
+ Mockito.when(env.createTable(createTableInfo)).thenReturn(true);
+ CreateTableCommand command = Mockito.spy(
+ new CreateTableCommand(Optional.of(query), createTableInfo));
+ Mockito.doNothing().when(command).validateCreateTableAsSelect(context,
query);
+
+ try (MockedStatic<Env> envMock = Mockito.mockStatic(Env.class);
+ MockedStatic<UnboundTableSinkCreator> sinkMock =
+ Mockito.mockStatic(UnboundTableSinkCreator.class)) {
+ envMock.when(Env::getCurrentEnv).thenReturn(env);
+
+ org.junit.jupiter.api.Assertions.assertDoesNotThrow(
+ () -> command.run(context, executor));
+
+ sinkMock.verifyNoInteractions();
+ Mockito.verify(env).createTable(createTableInfo);
+ }
+ }
+
+ @Test
+ void ctasValidatesSinkBeforePublishingMetadata() throws Exception {
+ LogicalPlan query = Mockito.mock(LogicalPlan.class);
+ CreateTableInfo createTableInfo = Mockito.mock(CreateTableInfo.class);
+ ConnectContext context = Mockito.mock(ConnectContext.class);
+ StmtExecutor executor = Mockito.mock(StmtExecutor.class);
+ Env env = Mockito.mock(Env.class);
+ List<String> tableNameParts = ImmutableList.of("catalog", "database",
"target");
+
+
Mockito.when(createTableInfo.getTableNameParts()).thenReturn(tableNameParts);
+ CreateTableCommand command = Mockito.spy(
+ new CreateTableCommand(Optional.of(query), createTableInfo));
+ Mockito.doNothing().when(command).validateCreateTableAsSelect(context,
query);
+ Mockito.doReturn(false).when(command).targetTableExists(context);
+
+ try (MockedStatic<Env> envMock = Mockito.mockStatic(Env.class);
+ MockedStatic<UnboundTableSinkCreator> sinkMock =
+ Mockito.mockStatic(UnboundTableSinkCreator.class)) {
+ envMock.when(Env::getCurrentEnv).thenReturn(env);
+ sinkMock.when(() -> UnboundTableSinkCreator.createUnboundTableSink(
+ tableNameParts, ImmutableList.of(), ImmutableList.of(),
ImmutableList.of(), query))
+ .thenThrow(new UserException("unsupported sink"));
+
+ org.junit.jupiter.api.Assertions.assertThrows(
+ Exception.class, () -> command.run(context, executor));
+
+ Mockito.verify(env, Mockito.never()).createTable(createTableInfo);
+ Mockito.verify(env, Mockito.never()).dropTable(
+ Mockito.anyString(), Mockito.anyString(),
Mockito.anyString(),
+ Mockito.anyBoolean(), Mockito.anyBoolean(),
Mockito.anyBoolean(),
+ Mockito.anyBoolean(), Mockito.anyBoolean(),
Mockito.anyBoolean());
+ }
+ }
+
+ @Test
+ void existingCtasTargetPrecedesUnsupportedSinkValidation() throws
Exception {
+ LogicalPlan query = Mockito.mock(LogicalPlan.class);
+ CreateTableInfo createTableInfo = Mockito.mock(CreateTableInfo.class);
+ ConnectContext context = Mockito.mock(ConnectContext.class);
+ StmtExecutor executor = Mockito.mock(StmtExecutor.class);
+ Env env = Mockito.mock(Env.class);
+ List<String> tableNameParts = ImmutableList.of("catalog", "database",
"target");
+
+
Mockito.when(createTableInfo.getTableNameParts()).thenReturn(tableNameParts);
+ Mockito.when(createTableInfo.getTableName()).thenReturn("target");
+ CreateTableCommand command = Mockito.spy(
+ new CreateTableCommand(Optional.of(query), createTableInfo));
+ Mockito.doNothing().when(command).validateCreateTableAsSelect(context,
query);
+ Mockito.doReturn(true).when(command).targetTableExists(context);
+
+ try (MockedStatic<Env> envMock = Mockito.mockStatic(Env.class);
+ MockedStatic<UnboundTableSinkCreator> sinkMock =
+ Mockito.mockStatic(UnboundTableSinkCreator.class)) {
+ envMock.when(Env::getCurrentEnv).thenReturn(env);
+ Exception exception =
org.junit.jupiter.api.Assertions.assertThrows(
+ Exception.class, () -> command.run(context, executor));
+
+
org.junit.jupiter.api.Assertions.assertTrue(exception.getMessage().contains("already
exists"));
+ sinkMock.verifyNoInteractions();
+ Mockito.verify(env, Mockito.never()).createTable(createTableInfo);
+ Mockito.verify(env, Mockito.never()).dropTable(
+ Mockito.anyString(), Mockito.anyString(),
Mockito.anyString(),
+ Mockito.anyBoolean(), Mockito.anyBoolean(),
Mockito.anyBoolean(),
+ Mockito.anyBoolean(), Mockito.anyBoolean(),
Mockito.anyBoolean());
+ }
+ }
+}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/insert/IcebergMergeExecutorTest.java
b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/insert/IcebergMergeExecutorTest.java
index 3a547d06394..f0271b57bb0 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/insert/IcebergMergeExecutorTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/insert/IcebergMergeExecutorTest.java
@@ -99,7 +99,8 @@ public class IcebergMergeExecutorTest {
ctx.setThreadLocalInfo();
IcebergMergeExecutor executor = new IcebergMergeExecutor(ctx, table,
"label", planner, false, -1L);
- IcebergMergeSink sink = new IcebergMergeSink(table, new
org.apache.doris.nereids.trees.plans.commands.delete.DeleteCommandContext());
+ IcebergMergeSink sink = new IcebergMergeSink(table,
+ new
org.apache.doris.nereids.trees.plans.commands.delete.DeleteCommandContext(),
true);
executor.finalizeSinkForMerge(null, sink, null);
TDataSink tDataSink = getTDataSink(sink);
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/planner/IcebergMergeSinkTest.java
b/fe/fe-core/src/test/java/org/apache/doris/planner/IcebergMergeSinkTest.java
index be73aefe43b..3aea3fc6f67 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/planner/IcebergMergeSinkTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/planner/IcebergMergeSinkTest.java
@@ -46,7 +46,8 @@ public class IcebergMergeSinkTest {
@Test
public void
testBindDataSinkIncludesRowLineageSchemaAndRewritableDeleteFileSetsForV3()
throws Exception {
- IcebergMergeSink sink = new
IcebergMergeSink(mockIcebergExternalTable(3), new DeleteCommandContext());
+ IcebergMergeSink sink = new IcebergMergeSink(
+ mockIcebergExternalTable(3), new DeleteCommandContext(), true);
sink.setRewritableDeleteFileSets(Collections.singletonList(buildDeleteFileSet()));
sink.bindDataSink(Optional.empty());
@@ -57,11 +58,14 @@ public class IcebergMergeSinkTest {
Assertions.assertTrue(thriftSink.getSchemaJson().contains(
IcebergUtils.ICEBERG_LAST_UPDATED_SEQUENCE_NUMBER_COL));
Assertions.assertEquals(1,
thriftSink.getRewritableDeleteFileSetsSize());
+ Assertions.assertTrue(thriftSink.isSetRequireMergeCardinalityCheck());
+ Assertions.assertTrue(thriftSink.isRequireMergeCardinalityCheck());
}
@Test
public void
testBindDataSinkSkipsRewritableDeleteFileSetsAndRowLineageSchemaForV2() throws
Exception {
- IcebergMergeSink sink = new
IcebergMergeSink(mockIcebergExternalTable(2), new DeleteCommandContext());
+ IcebergMergeSink sink = new IcebergMergeSink(
+ mockIcebergExternalTable(2), new DeleteCommandContext(),
false);
sink.setRewritableDeleteFileSets(Collections.singletonList(buildDeleteFileSet()));
sink.bindDataSink(Optional.empty());
@@ -72,12 +76,14 @@ public class IcebergMergeSinkTest {
Assertions.assertFalse(thriftSink.getSchemaJson().contains(IcebergUtils.ICEBERG_ROW_ID_COL));
Assertions.assertFalse(thriftSink.getSchemaJson().contains(
IcebergUtils.ICEBERG_LAST_UPDATED_SEQUENCE_NUMBER_COL));
+ Assertions.assertTrue(thriftSink.isSetRequireMergeCardinalityCheck());
+ Assertions.assertFalse(thriftSink.isRequireMergeCardinalityCheck());
}
@Test
public void testBindDataSinkDisablesColumnStatsWhenAllMetricsAreNone()
throws Exception {
IcebergMergeSink sink = new
IcebergMergeSink(mockIcebergExternalTable(2, Map.of(
- TableProperties.DEFAULT_WRITE_METRICS_MODE, "none")), new
DeleteCommandContext());
+ TableProperties.DEFAULT_WRITE_METRICS_MODE, "none")), new
DeleteCommandContext(), false);
sink.bindDataSink(Optional.empty());
@@ -91,7 +97,7 @@ public class IcebergMergeSinkTest {
IcebergMergeSink sink = new
IcebergMergeSink(mockIcebergExternalTable(2, Map.of(
TableProperties.DEFAULT_WRITE_METRICS_MODE, "none",
TableProperties.METRICS_MODE_COLUMN_CONF_PREFIX + "id",
"counts")),
- new DeleteCommandContext());
+ new DeleteCommandContext(), false);
sink.bindDataSink(Optional.empty());
@@ -105,7 +111,7 @@ public class IcebergMergeSinkTest {
IcebergMergeSink sink = new
IcebergMergeSink(mockIcebergExternalTable(3, Map.of(
TableProperties.DEFAULT_WRITE_METRICS_MODE, "counts",
TableProperties.METRICS_MODE_COLUMN_CONF_PREFIX + "id",
"none")),
- new DeleteCommandContext());
+ new DeleteCommandContext(), false);
sink.bindDataSink(Optional.empty());
@@ -122,7 +128,7 @@ public class IcebergMergeSinkTest {
TableProperties.DEFAULT_FILE_FORMAT, "orc",
TableProperties.DEFAULT_WRITE_METRICS_MODE, "none",
TableProperties.METRICS_MODE_COLUMN_CONF_PREFIX + "items",
"counts")),
- new DeleteCommandContext());
+ new DeleteCommandContext(), false);
sink.bindDataSink(Optional.empty());
diff --git a/gensrc/thrift/DataSinks.thrift b/gensrc/thrift/DataSinks.thrift
index 18e152f447d..9efe29a51eb 100644
--- a/gensrc/thrift/DataSinks.thrift
+++ b/gensrc/thrift/DataSinks.thrift
@@ -537,6 +537,8 @@ struct TIcebergMergeSink {
13: optional list<Types.TNetworkAddress> broker_addresses;
// Unset keeps collection enabled for rolling upgrades with older FEs.
14: optional bool collect_column_stats;
+ // Unset preserves old-FE UPDATE behavior; execution version gates SQL
MERGE validation.
+ 15: optional bool require_merge_cardinality_check;
// delete side (position delete only)
20: optional TFileContent delete_type
diff --git
a/regression-test/data/external_table_p0/iceberg/write/test_iceberg_write_merge_duplicate_source_negative.out
b/regression-test/data/external_table_p0/iceberg/write/test_iceberg_write_merge_duplicate_source_negative.out
index 3f8c510addc..5d765df5cac 100644
---
a/regression-test/data/external_table_p0/iceberg/write/test_iceberg_write_merge_duplicate_source_negative.out
+++
b/regression-test/data/external_table_p0/iceberg/write/test_iceberg_write_merge_duplicate_source_negative.out
@@ -1,3 +1,5 @@
-- This file is automatically generated. You should know what you did if you
want to edit this
-- !duplicate_source_atomic_state --
1 A committed
+-- !sibling_close_atomic_state --
+1 A committed
diff --git
a/regression-test/suites/external_table_p0/PAIMON_ICEBERG_READ_WRITE_P0_COVERAGE.md
b/regression-test/suites/external_table_p0/PAIMON_ICEBERG_READ_WRITE_P0_COVERAGE.md
index f3760f4db68..5a551b6d3d8 100644
---
a/regression-test/suites/external_table_p0/PAIMON_ICEBERG_READ_WRITE_P0_COVERAGE.md
+++
b/regression-test/suites/external_table_p0/PAIMON_ICEBERG_READ_WRITE_P0_COVERAGE.md
@@ -43,8 +43,9 @@ boundaries.
Paimon P0 was incomplete. No suite selected a `merge-engine`, and write
rejection was not checked
across all DML shapes with pre/post data and snapshot invariants. PM01-PM03
below close those gaps.
-PM04 exposes one remaining product defect as an opt-in negative regression:
failed Paimon CTAS leaves
-target metadata. Streaming writes, overwrite, delete/update and merge listed
as unsupported by the
+PM04 now runs as an active negative regression: failed Paimon CTAS is rejected
without leaving target
+metadata, while `IF NOT EXISTS` remains a no-op for an existing table.
Streaming writes, overwrite,
+delete/update and merge listed as unsupported by the
Paimon ecosystem matrix are boundary tests rather than positive Doris P0
contracts.
## Risks
@@ -75,7 +76,7 @@ Paimon ecosystem matrix are boundary tests rather than
positive Doris P0 contrac
| Paimon | Deletion vectors, upsert/delete visibility and data/system tables |
Covered | `test_paimon_deletion_vector`, `paimon_data_system_table`,
`paimon_system_table` |
| Paimon | Catalog/database/table create and drop | Covered |
`test_create_paimon_table` |
| Paimon | Doris data write-back | Negative boundary covered |
`test_paimon_write_boundary` |
-| Paimon | Failed CTAS metadata atomicity | Isolated known-bug regression |
`test_paimon_ctas_atomicity_negative` |
+| Paimon | Failed CTAS metadata atomicity | Active negative regression |
`test_paimon_ctas_atomicity_negative` |
| Iceberg | V1/V2/V3, Parquet/ORC, position/equality deletes and deletion
vectors | Covered | `test_iceberg_position_delete`,
`test_iceberg_equality_delete`, `test_iceberg_deletion_vector` |
| Iceberg | Schema, partition and sort-order evolution | Covered |
`test_iceberg_schema_time_travel_matrix`,
`test_iceberg_partition_evolution_format_scanner`, `iceberg_schema_change_ddl` |
| Iceberg | Snapshot/timestamp/tag/branch reads and reference actions |
Covered | `test_iceberg_time_travel`, `iceberg_query_tag_branch`,
`test_iceberg_schema_ref_actions_matrix` |
@@ -95,7 +96,7 @@ Detailed schema/time-travel and Iceberg write combinations
are maintained in
| PM01 | Distinguish all four primary-key merge engines | R01, R02, R04 |
Functional, correctness, compatibility | Paimon Parquet/ORC tables | Duplicate
keys across several commits | Each engine returns its documented merged row
under automatic and forced-JNI routing |
| PM02 | Validate dynamic-bucket cross-partition deduplication | R03, R04 |
Correctness | Primary key excludes partition key, bucket=-1 | Move one key
between partitions | Exactly one current row remains in the new partition |
| PM03 | Preserve the Paimon read-only boundary | R05, R10 | Negative,
atomicity | Existing Paimon PK table | VALUES, SELECT, OVERWRITE, UPDATE,
DELETE, MERGE | Every statement fails before a snapshot or data change |
-| PM04 | Reject or roll back Paimon CTAS atomically | R05, R10 | Isolated
negative, atomicity | Paimon catalog with no target table | CREATE TABLE AS
SELECT | Desired contract: the command fails and no target table remains;
current bug leaves the table |
+| PM04 | Reject or roll back Paimon CTAS atomically | R05, R10 | Negative,
atomicity | Missing and existing Paimon target tables | CREATE TABLE AS SELECT,
including IF NOT EXISTS | Unsupported CTAS leaves no target; an existing target
makes IF NOT EXISTS a successful no-op |
Every P0 risk maps to at least one deterministic positive, boundary, or
isolated known-bug
regression. Catalog authentication and cloud storage permutations stay in
their existing connector
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 7f0ba5b849c..6829fd1d4c4 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
@@ -70,10 +70,10 @@ under the License.
| Doris 源 bucket | HASH 固定 bucket、RANDOM bucket、HASH AUTO bucket | 已覆盖（验证通过） |
`test_iceberg_write_source_models` |
| 分区源类型 | STRING/INT/BIGINT/DATE/DATETIME/DECIMAL 的 bucket 与适用
transform；BOOLEAN identity 与非法 bucket | 已覆盖（验证通过） |
`test_iceberg_write_partition_types_null` |
| NULL 分区 | identity NULL、数值/decimal bucket 与 truncate NULL、time transform
NULL、多列组合 NULL | 已覆盖（验证通过） | `test_iceberg_write_partition_types_null` |
-| nullable STRING truncate | nullable STRING 经过 truncate transform 的 INSERT，以及
UPDATE 产生的 Nullable projection 写入 | 已覆盖（隔离负向） |
`test_iceberg_write_nullable_truncate_negative` |
+| nullable STRING truncate | nullable STRING 经过 truncate transform 的 INSERT，以及
UPDATE 产生的 Nullable projection 写入 | 已覆盖（常规回归） |
`test_iceberg_write_nullable_truncate_negative` |
| MERGE 完整语义 | 条件 MATCHED、DELETE/UPDATE、多个条件 NOT MATCHED、NULL-safe 与普通 NULL
key | 已覆盖（验证通过） | `test_iceberg_write_merge_semantics` |
-| MERGE 基数约束 | 多个源行匹配同一目标行必须整句失败且不发布快照 | 已覆盖（隔离负向） |
`test_iceberg_write_merge_duplicate_source_negative` |
-| MERGE + STRING truncate | required truncate 源列经 MERGE nullable projection 写入
| 已覆盖（隔离负向） | `test_iceberg_write_merge_truncate_negative` |
+| MERGE 基数约束 | 多个源行匹配同一目标行必须整句失败且不发布快照 | 已覆盖（常规回归） |
`test_iceberg_write_merge_duplicate_source_negative` |
+| MERGE + STRING truncate | required truncate 源列经 MERGE nullable projection 写入
| 已覆盖（常规回归） | `test_iceberg_write_merge_truncate_negative` |
| branch/tag 写入边界 | branch INSERT/OVERWRITE 隔离；tag 写入和 branch
DELETE/UPDATE/MERGE 明确拒绝 | 已覆盖（验证通过） | `test_iceberg_write_branch_dml_boundary`
|
| nullable 数据 | 顶层 NULL、ARRAY NULL 元素、MAP NULL value、STRUCT NULL child |
已覆盖并增强 | `test_iceberg_write_insert`、`test_iceberg_write_complex_evolution` |
| required 列正向与 schema change | required 列合法写入、nullable 列写 NULL、增加 required 列与
nullable→required 拒绝 | 已覆盖（验证通过） | `test_iceberg_write_nullability_atomicity` |
@@ -100,9 +100,9 @@ under the License.
| W05 | 验证不同类型与 NULL 的 partition/bucket transform | R02、R03、R11 | 功能、正确性、边界 |
Iceberg v2 | identity/bucket/truncate/time transform 多列组合，包含 NULL | 数据与
`$partitions` 统计一致；NULL 行可过滤且可继续写入 |
| W06 | 验证 required/nullable schema change 与合法写入 | R09、R11 | 异常、正确性 | Iceberg
required 列 | 拒绝增加无默认值 required 列和 nullable→required；执行 VALUES/INSERT SELECT
合法写入 | schema change 失败不产生 snapshot；合法写入与 Spark 结果一致 |
| W07 | 验证 required 列 NULL 拒绝和 statement 原子性 | R09、R11 | 隔离负向、正确性 | 隔离 Iceberg
database | VALUES 写 NULL；多 bucket 源表 INSERT SELECT 混合有效与 NULL 行 |
修复前会错误提交并产生不可读文件；修复后整条语句在 snapshot 发布前拒绝 |
-| W08 | 验证 STRING truncate 的 Nullable block 处理 | R03、R05、R12 | 隔离负向、稳定性 |
可重启的隔离 Doris 集群 | nullable STRING INSERT；partition evolution 后 UPDATE 产生
Nullable block | 修复前 BE FATAL；修复后写入成功并保持 NULL 分区语义 |
+| W08 | 验证 STRING truncate 的 Nullable block 处理 | R03、R05、R12 | 常规回归、稳定性 |
Iceberg v2 | nullable STRING INSERT；partition evolution 后 UPDATE 产生 Nullable
block | 写入成功并保持 NULL 分区语义，BE 不发生 FATAL |
| W09 | 验证 MERGE 条件动作、多个 NOT MATCHED 与 NULL key 语义 | R02、R03、R05 | 功能、正确性 |
Iceberg v2 MOR | identity/bucket 分区间移动、删除、插入、NULL-safe 与普通等值匹配 |
每个源行只选择一个动作，Spark 与 Doris 结果一致 |
-| W10 | 验证 MERGE 多源匹配单目标的基数约束 | R13 | 隔离负向、原子性 | Iceberg v2 MOR |
两个源行同时更新一个目标行 | 修复前错误提交重复行；修复后整句拒绝且无新快照和文件 |
+| W10 | 验证 MERGE 多源匹配单目标的基数约束 | R13 | 常规回归、原子性 | Iceberg v2 MOR |
两个源行同时更新一个目标行 | 整句拒绝且无新快照和文件 |
| W11 | 验证 branch/tag 的写入能力边界 | R04、R14 | 功能、异常、原子性 | 已建立 branch 与 tag |
branch INSERT/OVERWRITE；branch 行级 DML 与 tag 写入 | branch 与 main
隔离；不支持操作明确拒绝且引用不变化 |
| W12 | 验证多次 Partition Evolution 后覆盖写和历史引用 | R02、R04、R10、R15 | 功能、正确性 |
Iceberg v2 | ADD/REPLACE/DROP identity、bucket、truncate、day/hour 后动态覆盖写 | 仅替换当前
spec 命中的分区，tag/branch 和旧 spec 保持可读 |
| W13 | 验证 delete files 与覆盖写、演进的交互 | R02、R05、R15 | 正确性、兼容性 | Iceberg v2 MOR |
DELETE/UPDATE/MERGE 生成 delete files，再在新旧 spec 上覆盖写 | replacement 行不被旧 delete
files 隐藏，历史 tag 不受影响 |
@@ -111,10 +111,10 @@ under the License.
| W16 | 验证 CTAS、复杂类型、格式和失败清理 | R08、R09、R16 | 功能、异常、兼容性 | 内部多 bucket 源表 | CTAS
到 ORC 分区表；严格转换失败；向 Avro 表写入 | ORC 与 Spark 一致；失败不遗留表或快照；Avro 明确拒绝 |
| W17 | 验证 sort order、distribution mode 和多文件 flush | R03、R11、R17 | 正确性、稳定性 | 多
BE Doris | NULL sort key、多列升降序、none/hash/range、低 target file size |
计划包含声明排序，多文件总行数正确，三种分布模式结果一致 |
| W18 | 验证并发 MERGE 与 append 的提交不变量 | R11、R13、R17 | 并发、原子性 | 多 BE Doris |
readiness barrier 保证两个独立会话同时具备 dispatch 条件；同行更新与互不冲突 append | 同行提交可串行化且基数为一；仅接受
Iceberg validation/commit conflict；非冲突写入无丢失或重复 |
-| W19 | 验证 MERGE source projection 进入 truncate transform 的类型安全 | R12、R18 |
隔离负向、稳定性 | 可重启的隔离 Doris 集群 | required STRING truncate 列执行匹配更新与未匹配插入 | 修复前 BE
FATAL；修复后 MERGE 成功且物理分区正确 |
+| W19 | 验证 MERGE source projection 进入 truncate transform 的类型安全 | R12、R18 |
常规回归、稳定性 | Iceberg v2 MOR | required STRING truncate 列执行匹配更新与未匹配插入 | MERGE 接受
nullable 物理投影中的非 NULL 值，并生成正确物理分区 |
## P0 覆盖检查
-R01-R18 均映射到至少一个 P0 regression。十五个正向 suite 在最后一次成功写入后均由 Spark/Doris
交叉校验同表逻辑结果；transform、rollover、CTAS property 和失败原子性另有物理 metadata 或状态
oracle。稳定性或已确认正确性缺陷使用独立 suite、完整预期输出和显式隔离开关保存复现，避免默认 P0 破坏共享集群或固化错误结果。
+R01-R18 均映射到至少一个 P0 regression。十五个正向 suite 在最后一次成功写入后均由 Spark/Doris
交叉校验同表逻辑结果；transform、rollover、CTAS property 和失败原子性另有物理 metadata 或状态
oracle。W08、W10、W19 已作为常规回归运行；仍可能写出不可读 required-null 文件的 W07 保留隔离开关，避免破坏共享集群。
-本矩阵未覆盖项为 0。COW 行级 DML、branch 行级 DML、tag 写入和 Avro
写入属于当前明确能力边界，均以预期拒绝用例固化错误语义与失败原子性；已确认的产品缺陷均有隔离负向 regression。
+本矩阵未覆盖项为 0。COW 行级 DML、branch 行级 DML、tag 写入和 Avro
写入属于当前明确能力边界，均以预期拒绝用例固化错误语义与失败原子性；仅 required-null 已知缺陷继续使用隔离负向 regression。
diff --git
a/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_merge_duplicate_source_negative.groovy
b/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_merge_duplicate_source_negative.groovy
index fbb3be22c44..075bfccad96 100644
---
a/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_merge_duplicate_source_negative.groovy
+++
b/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_merge_duplicate_source_negative.groovy
@@ -16,23 +16,33 @@
// under the License.
suite("test_iceberg_write_merge_duplicate_source_negative",
- "p0,external,iceberg,external_docker,external_docker_iceberg") {
+
"p0,external,iceberg,external_docker,external_docker_iceberg,nonConcurrent") {
String enabled = context.config.otherConfigs.get("enableIcebergTest")
if (enabled == null || !enabled.equalsIgnoreCase("true")) {
logger.info("disable iceberg test")
return
}
- String knownBugEnabled =
context.config.otherConfigs.get("enableIcebergKnownBugTest")
- if (knownBugEnabled == null || !knownBugEnabled.equalsIgnoreCase("true")) {
- logger.info("skip isolated Iceberg known-bug 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_merge_duplicate_source_negative"
String dbName = "iceberg_write_merge_duplicate_source_negative_db"
+ String dockerCommand =
context.config.otherConfigs.get("externalDockerCommand") ?: "docker"
+ String mcImage = "minio/mc:RELEASE.2025-01-17T23-25-50Z"
+
+ def countDataObjects = { String objectPath ->
+ def stdout = new StringBuilder()
+ def stderr = new StringBuilder()
+ def process = new ProcessBuilder("/bin/bash", "-c",
+ "${dockerCommand} run --rm --entrypoint /bin/sh ${mcImage} -c "
+ + "'/usr/bin/mc alias set minio
http://${externalEnvIp}:${minioPort} "
+ + "admin password >/dev/null "
+ + "&& /usr/bin/mc find ${objectPath} --type f | wc
-l'").start()
+ process.consumeProcessOutput(stdout, stderr)
+ process.waitForOrKill(30000)
+ assertEquals(0, process.exitValue(), "Failed to list Iceberg data
objects: ${stderr}")
+ return stdout.toString().trim() as long
+ }
sql """drop catalog if exists ${catalogName}"""
sql """
@@ -68,35 +78,94 @@ suite("test_iceberg_write_merge_duplicate_source_negative",
"""
sql """insert into duplicate_source_target values (1, 'A', 'committed')"""
+ String committedFile = (sql """
+ select file_path from duplicate_source_target\$files order by
file_path limit 1
+ """)[0][0].toString()
+ int dataDirectoryEnd = committedFile.indexOf('/data/') + '/data'.length()
+ assertTrue(dataDirectoryEnd >= '/data'.length(), "Unexpected Iceberg data
path: ${committedFile}")
+ String dataObjectPath = committedFile.substring(0, dataDirectoryEnd)
+ .replaceFirst('^s3a?://warehouse', 'minio/warehouse')
+ long objectsBefore = countDataObjects(dataObjectPath)
+
long snapshotsBefore =
(sql """select count(*) from
duplicate_source_target\$snapshots""")[0][0] as long
long filesBefore =
(sql """select count(*) from
duplicate_source_target\$files""")[0][0] as long
// Negative scenario: Iceberg MERGE cardinality permits only one source row
- // to update a target row. The entire statement must fail before
publishing.
+ // to update a target row. Tiny blocks and files force a valid row to roll
before
+ // the late duplicate, and the entire statement must still leave no data
object behind.
+ sql """set batch_size = 1"""
+ // A single pipeline instance preserves the ordered source sequence needed
to prove late cleanup.
+ sql """set parallel_pipeline_task_num = 1"""
+ sql """set iceberg_write_target_file_size_bytes = 1"""
test {
sql """
merge into duplicate_source_target t
using (
- select 1 as id, 'B' as region, 'first-update' as payload
- union all
- select 1, 'C', 'second-update'
+ select id, region, payload
+ from (
+ select 1 as seq, 2 as id, 'B' as region, 'valid-insert' as
payload
+ union all
+ select 2, 1, 'B', 'first-update'
+ union all
+ select 3, 1, 'C', 'second-update'
+ ) ordered_source
+ order by seq
) s
on t.id = s.id
when matched then update set
region = s.region,
payload = s.payload
+ when not matched then insert (id, region, payload)
+ values (s.id, s.region, s.payload)
"""
- exception "more than one"
+ exception "multiple source rows matched the same target row"
}
assertEquals(snapshotsBefore,
(sql """select count(*) from
duplicate_source_target\$snapshots""")[0][0] as long)
assertEquals(filesBefore,
(sql """select count(*) from
duplicate_source_target\$files""")[0][0] as long)
+ assertEquals(objectsBefore, countDataObjects(dataObjectPath))
order_qt_duplicate_source_atomic_state """
select id, region, payload
from duplicate_source_target
order by id
"""
+
+ // A sibling delete close can fail only after the data side has closed
successfully. The outer
+ // MERGE still owns and must remove those unpublished data objects.
+ long siblingFailureObjectsBefore = countDataObjects(dataObjectPath)
+ try {
+
GetDebugPoint().enableDebugPointForAllBEs("VIcebergDeleteSink.close.inject_failure")
+ test {
+ sql """
+ merge into duplicate_source_target t
+ using (
+ select 1 as id, 'B' as region, 'updated' as payload
+ union all
+ select 3, 'C', 'inserted'
+ ) s
+ on t.id = s.id
+ when matched then update set
+ region = s.region,
+ payload = s.payload
+ when not matched then insert (id, region, payload)
+ values (s.id, s.region, s.payload)
+ """
+ exception "injected Iceberg delete close failure"
+ }
+ } finally {
+
GetDebugPoint().disableDebugPointForAllBEs("VIcebergDeleteSink.close.inject_failure")
+ }
+ assertEquals(snapshotsBefore,
+ (sql """select count(*) from
duplicate_source_target\$snapshots""")[0][0] as long)
+ assertEquals(filesBefore,
+ (sql """select count(*) from
duplicate_source_target\$files""")[0][0] as long)
+ assertEquals(siblingFailureObjectsBefore, countDataObjects(dataObjectPath))
+ order_qt_sibling_close_atomic_state """
+ select id, region, payload
+ from duplicate_source_target
+ order by id
+ """
}
diff --git
a/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_merge_truncate_negative.groovy
b/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_merge_truncate_negative.groovy
index 31bfeb251ab..8a503d879ac 100644
---
a/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_merge_truncate_negative.groovy
+++
b/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_merge_truncate_negative.groovy
@@ -18,10 +18,8 @@
suite("test_iceberg_write_merge_truncate_negative",
"p0,external,iceberg,external_docker,external_docker_iceberg") {
String enabled = context.config.otherConfigs.get("enableIcebergTest")
- String crashTestEnabled =
context.config.otherConfigs.get("enableIcebergCrashTest")
- if (enabled == null || !enabled.equalsIgnoreCase("true")
- || crashTestEnabled == null ||
!crashTestEnabled.equalsIgnoreCase("true")) {
- logger.info("disable iceberg crash test")
+ if (enabled == null || !enabled.equalsIgnoreCase("true")) {
+ logger.info("disable iceberg test")
return
}
@@ -67,9 +65,9 @@ suite("test_iceberg_write_merge_truncate_negative",
"""
sql """insert into merge_truncate_negative values (1, 'alpha', 'before')"""
- // WM03-S01: A MERGE source projection is nullable even when every source
- // value and the Iceberg target column are NOT NULL. The writer must reject
- // an invalid input as a query error and must never terminate a BE.
+ // WM03-S01: A MERGE source projection is physically nullable even when
every source
+ // value and the Iceberg target column are NOT NULL. The writer must
accept those
+ // non-NULL values without terminating a BE or changing truncate partition
routing.
sql """
merge into merge_truncate_negative t
using (
@@ -92,8 +90,9 @@ suite("test_iceberg_write_merge_truncate_negative",
"""
// Logical rows cannot prove the MERGE writer used truncate(2) when routing
// its updated and inserted records.
+ // Iceberg names truncate transform fields with the `_trunc` suffix; the
width is not part of the name.
order_qt_merge_truncate_physical_partitions """
- select distinct hex(struct_element(`partition`,
'partition_value_trunc_2'))
+ select distinct hex(struct_element(`partition`,
'partition_value_trunc'))
from merge_truncate_negative\$partitions
order by 1
"""
diff --git
a/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_nullable_truncate_negative.groovy
b/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_nullable_truncate_negative.groovy
index 25074a0af78..a0358aef5b8 100644
---
a/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_nullable_truncate_negative.groovy
+++
b/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_nullable_truncate_negative.groovy
@@ -23,14 +23,6 @@ suite("test_iceberg_write_nullable_truncate_negative",
return
}
- // This opt-in switch isolates a BE-fatal negative scenario from the
shared P0 cluster.
- // Enable it only in a cluster whose BE processes can be restarted after
the suite.
- String crashTestEnabled =
context.config.otherConfigs.get("enableIcebergCrashTest")
- if (crashTestEnabled == null ||
!crashTestEnabled.equalsIgnoreCase("true")) {
- logger.info("skip isolated Iceberg crash regression")
- 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")
diff --git
a/regression-test/suites/external_table_p0/paimon/test_paimon_ctas_atomicity_negative.groovy
b/regression-test/suites/external_table_p0/paimon/test_paimon_ctas_atomicity_negative.groovy
index 6d449f4af1e..1fd4e27db7b 100644
---
a/regression-test/suites/external_table_p0/paimon/test_paimon_ctas_atomicity_negative.groovy
+++
b/regression-test/suites/external_table_p0/paimon/test_paimon_ctas_atomicity_negative.groovy
@@ -23,14 +23,6 @@ suite("test_paimon_ctas_atomicity_negative",
return
}
- // CTAS currently creates Paimon metadata before discovering that no
Paimon data sink exists.
- // Keep this opt-in until the product rejects or rolls back the statement
atomically.
- String knownBugEnabled =
context.config.otherConfigs.get("enablePaimonKnownBugTest")
- if (knownBugEnabled == null || !knownBugEnabled.equalsIgnoreCase("true")) {
- logger.info("skip isolated Paimon known-bug regression")
- return
- }
-
String minioPort = context.config.otherConfigs.get("iceberg_minio_port")
String externalEnvIp = context.config.otherConfigs.get("externalEnvIp")
String catalogName = "test_paimon_ctas_atomicity_negative"
@@ -67,6 +59,28 @@ suite("test_paimon_ctas_atomicity_negative",
exception "PaimonExternalCatalog"
}
assertEquals(0, (sql """show tables like 'ctas_target'""").size())
+
+ spark_paimon """
+ create table paimon.${dbName}.ctas_target (id int, payload string)
+ using paimon
+ """
+ // IF NOT EXISTS must remain a no-op even though Paimon does not
support the CTAS sink.
+ sql """
+ create table if not exists ctas_target engine=paimon
+ as select cast(1 as int) as id, cast('candidate' as string) as
payload
+ """
+ assertEquals(1, (sql """show tables like 'ctas_target'""").size())
+ assertEquals(0, (sql """select * from ctas_target""").size())
+
+ // An existing non-idempotent target must keep catalog error
precedence; no sink can own it.
+ test {
+ sql """
+ create table ctas_target engine=paimon
+ as select cast(2 as int) as id, cast('replacement' as string)
as payload
+ """
+ exception "already exists"
+ }
+ assertEquals(0, (sql """select * from ctas_target""").size())
} finally {
spark_paimon """drop table if exists paimon.${dbName}.ctas_target"""
sql """drop catalog if exists ${catalogName}"""
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
