github-actions[bot] commented on code in PR #66112:
URL: https://github.com/apache/doris/pull/66112#discussion_r3662394796
##########
be/src/exec/sink/viceberg_merge_sink.cpp:
##########
@@ -120,17 +122,22 @@ 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;
}
}
+ // The physical sink hashes matched rows by row_id, so identical targets
reach the same sink.
Review Comment:
[P1] Keep MERGE cardinality routing independent of strict-consistency DML
This set is local to one `IcebergMergeSinkLocalState`, but
`RequestPropertyDeriver.visitPhysicalIcebergMergeSink()` requests
`PhysicalProperties.ANY` when `enable_strict_consistency_dml=false`. MERGE
builds the source on the left and target on the right, so a broadcast-target
join can process two source rows matching the same target in different fragment
instances; each reaches a different set and both validations succeed. That
session switch controls replica-consistency distribution, but it must not
disable the SQL cardinality rule. Please require the row-id/MERGE distribution
unconditionally (or use a query-global validator) and add a multi-instance
broadcast case with the switch off.
##########
be/src/exec/sink/viceberg_merge_sink.h:
##########
@@ -73,6 +75,7 @@ class VIcebergMergeSink final : public AsyncResultWriter {
int _operation_idx = -1;
int _row_id_idx = -1;
std::vector<int> _data_column_indices;
+ std::unordered_set<std::string> _matched_row_ids;
Review Comment:
[P1] Avoid retaining one full row-id string per matched row
Every valid matched target now keeps a separately allocated
file-path-plus-position string and hash node until the writer is destroyed. A
large MERGE therefore adds linear query-lifetime memory on top of the delete
sink's existing file-keyed position bitmaps; this state cannot be revoked or
spilled in low-memory mode, and the writer's `MemoryUsage` profile only
accounts queued/free blocks. The task memory hook does charge the query, but
otherwise-valid high-cardinality MERGEs can still hit the limit solely because
of this duplicate representation. Please use a compact exact structure (for
example interned file paths plus position bitmaps, ideally reusing the delete
collection), expose/reserve its retained bytes, and cover a high-cardinality
low-memory case.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/CreateTableCommand.java:
##########
@@ -107,9 +107,11 @@ public void run(ConnectContext ctx, StmtExecutor executor)
throws Exception {
throw new AnalysisException(e.getMessage(), e.getCause());
}
- query =
UnboundTableSinkCreator.createUnboundTableSink(createTableInfo.getTableNameParts(),
- ImmutableList.of(), ImmutableList.of(), ImmutableList.of(),
query);
try {
+ // Once CTAS metadata is visible, every setup or execution failure
must pass through
+ // the same rollback boundary so an unusable destination table is
never left behind.
+ query =
UnboundTableSinkCreator.createUnboundTableSink(createTableInfo.getTableNameParts(),
Review Comment:
[P1] Roll back only metadata created by this CTAS invocation
`createTable(...) == false` is not an ownership proof under concurrent `IF
NOT EXISTS`. Internally, a loser can get `result.second == true` from the
atomic registration while `createOlapTable()` still returns its never-updated
`false`; Paimon and MaxCompute ignore-if-exists paths likewise return `false`
even when another creator won. If sink setup or execution then fails inside
this expanded catch, `handleFallbackFailedCtas()` force-drops whichever table
currently resolves by name, so it can delete the concurrent winner. Please
propagate an unambiguous atomic created/existed result plus the exact table
identity/version, and drop only if that same object is still installed
(validating unsupported sinks before publishing metadata would also close the
immediate setup case). Add a deterministic losing-race test; the new mock test
assumes the disputed `false == owned` invariant.
##########
be/src/exec/sink/viceberg_merge_sink.cpp:
##########
@@ -167,6 +174,80 @@ 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");
+ }
+
+ 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);
+ }
+ std::string exact_row_id = file_paths->get_data_at(i).to_string();
+ exact_row_id.append(reinterpret_cast<const char*>(&row_position),
sizeof(row_position));
+ if (!_matched_row_ids.emplace(std::move(exact_row_id)).second) {
+ return Status::InvalidArgument(
Review Comment:
[P2] Activate the existing statement-level regression for this fix
`test_iceberg_write_merge_duplicate_source_negative.groovy` already
reproduces this exact bug and verifies that snapshots, files, and table
contents remain unchanged, but it still sits behind
`enableIcebergKnownBugTest`. It also expects the exception to contain `"more
than one"`, whereas this new message contains only `"multiple source rows..."`;
`TestAction` uses a case-sensitive `contains`, so enabling the suite still
fails. Please update either the diagnostic or oracle, remove the known-bug gate
for this suite, and run it. The added unit test proves only duplicate detection
inside one writer, not statement-level atomicity or distribution.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
