github-actions[bot] commented on code in PR #66246:
URL: https://github.com/apache/doris/pull/66246#discussion_r3679797708
##########
be/src/exec/sink/viceberg_merge_sink.cpp:
##########
@@ -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;
}
}
Review Comment:
[P2] Avoid retaining every matched row twice
Every row inserted into `_matched_row_positions` is then passed to
`_delete_writer->write()`, whose `_file_deletions` retains the same `file_path`
keys and Roaring position sets until close. Production MERGE therefore keeps
two exact copies of its dominant row-ID state; the compact-state tests set
`_skip_io`, so they measure only this new map. Sparse or many-file MERGEs can
nearly double retained memory and hit the query limit unnecessarily. Make the
delete sink's bitmap perform `addChecked()` when cardinality is required (or
otherwise share one state object), and measure the combined production state in
the regression.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/CreateTableCommand.java:
##########
@@ -99,6 +101,19 @@ public void run(ConnectContext ctx, StmtExecutor executor)
throws Exception {
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.
Review Comment:
[P1] Do not activate the stale Paimon CTAS rejection oracle
This eager call does not reject Paimon writes: the factory returns
`UnboundPaimonTableSink`, `BindSink` produces a bound Paimon sink through
`PaimonWriteTarget`, and `InsertIntoTableCommand` runs `PaimonInsertExecutor`.
A valid filesystem-catalog CTAS therefore creates the table and inserts the
selected row, while the now-unskipped `test_paimon_ctas_atomicity_negative`
still expects a `PaimonExternalCatalog` exception and no target. That P0 oracle
will fail deterministically. Either reject CTAS explicitly before metadata
creation if it is intentionally unsupported, or update the suite and coverage
docs to assert the implemented successful CTAS behavior.
##########
be/src/exec/sink/viceberg_merge_sink.cpp:
##########
@@ -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] =
Review Comment:
[P2] Do not rescan growing bitmaps on every input block
For every touched file this calls `getSizeInBytes()` before mutation and
again afterward. In the pinned CRoaring 2.1.2, `Roaring64Map::getSizeInBytes()`
accumulates all inner bitmaps and each inner Roaring's serialized container
sizes; repeated small blocks for one large data file therefore rescan an
ever-growing bitmap twice per block just to update a profile counter. This can
become superlinear in the file's row-position range. Track a cheap estimate,
sample/update it periodically or at close, and add a
repeated-small-block/same-file scale test.
--
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]
