github-actions[bot] commented on code in PR #66112:
URL: https://github.com/apache/doris/pull/66112#discussion_r3664610142
##########
be/src/exec/sink/viceberg_merge_sink.cpp:
##########
@@ -167,6 +180,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(
Review Comment:
[P1] Clean up rolled files when a later duplicate fails
This error can arrive after earlier valid blocks have already made the table
writer roll a data file. Rollover closes that `VIcebergPartitionWriter` with
`Status::OK()`, records commit data, and erases it from
`_partitions_to_writers`; the later error close can delete only writers that
are still open, while `IcebergTransaction.rollback()` does nothing for
insert/update mode. The statement therefore fails without publishing metadata
but leaves the earlier data object orphaned. The active regression's
`$snapshots`/`$files` checks cannot observe uncommitted storage objects. Track
all successfully closed files for error cleanup (or clean the accumulated
commit data on rollback), and cover a multi-block late duplicate with a tiny
target-file size plus an object-store listing.
##########
be/src/exec/sink/viceberg_merge_sink.cpp:
##########
@@ -120,17 +124,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;
}
}
+ // The physical sink hashes matched rows by row_id, so identical targets
reach the same sink.
+ // Exact state retained across blocks therefore enforces MERGE cardinality
for the whole query.
+ RETURN_IF_ERROR(_validate_matched_row_ids(output_block,
delete_filter.data()));
Review Comment:
[P1] Honor the UPDATE cardinality flag in BE
`IcebergUpdateCommand` now passes `requireMergeCardinalityCheck=false`, but
that flag is consumed only by FE property derivation and never reaches
`TIcebergMergeSink`. This is a functional regression for `UPDATE ... FROM`:
`LogicalPlanBuilder.visitUpdate()` joins the target with all `FROM` relations,
so a many-to-one join can emit the same target row ID twice and this new call
rejects the UPDATE with SQL MERGE's multiple-source error even though FE
explicitly marked validation unnecessary. Ordinary UPDATEs also retain a
redundant second file-to-position bitmap. Carry the discriminator through
thrift and honor false here, but preserve old-FE UPDATE behavior when the field
is absent and capability-gate new MERGE enforcement until every executing BE
understands it; add many-to-one UPDATE FROM, mixed-version, and
no-validator-state coverage.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/IcebergMergeCommand.java:
##########
@@ -465,6 +465,7 @@ private LogicalPlan buildMergePlan(ConnectContext ctx,
IcebergExternalTable iceb
icebergTable.getBaseSchema(true),
outputExprs,
deleteCtx,
+ true,
Review Comment:
[P2] Do not hash unmatched INSERTs by NULL row ID
This makes the sink property mandatory even when strict consistency is
disabled. With merge partitioning off, that property is `HASH(row_id)` for the
whole MERGE output, but `buildInsertProjection()` gives every unmatched INSERT
`row_id=NULL`; all of those inserts therefore collapse onto one exchange
channel. Insert-only MERGE sends its entire workload there, and mixed MERGE
hot-spots its unmatched branch as well. Cardinality needs row-ID colocation
only for matched delete/update images, so use operation-aware routing for that
requirement (or omit it when there are no matched actions) without hashing
INSERTs by NULL. Cover both strict-off not-matched-only and mixed MERGE.
##########
regression-test/suites/external_table_p0/paimon/test_paimon_ctas_atomicity_negative.groovy:
##########
@@ -67,6 +59,18 @@ 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())
Review Comment:
[P1] Refresh Paimon table names on the no-op path
The activated external regression already fails here in TeamCity build
1007534: expected one table, got zero. The earlier `SHOW TABLES` primes the
local name cache as empty; Spark then creates the remote target;
`performCreateTable()` observes it and returns `true`, so
`ExternalMetadataOps.createTable()` skips `afterCreateTable()` and Paimon's
only `resetMetaCacheNames()` call. CTAS `IF NOT EXISTS` reports success while
catalog enumeration still hides the target. This is distinct from the existing
no-op-error thread: the command now succeeds but leaves stale metadata.
Invalidate the table-name cache for both remote-existing `true` paths and
retain this assertion.
--
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]
