github-actions[bot] commented on code in PR #66112:
URL: https://github.com/apache/doris/pull/66112#discussion_r3670711976
##########
be/src/exec/sink/writer/iceberg/viceberg_partition_writer.cpp:
##########
@@ -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());
Review Comment:
[P1] Register the data file before `open()` can fail
`_path` becomes a physical file before compression validation and
transformer initialization, but the dynamic-partition caller adds this writer
to `_partitions_to_writers` only after `open()` succeeds. For example, an
Iceberg Parquet table with `write.parquet.compression-codec=gzip` reaches
`create_file()` with `TFileCompressType::GZ` and then returns `Unsupported
compress type`; the local writer is destroyed before either the active-writer
cleanup or this new closed-file callback owns the path. `HdfsFileWriter` only
closes that empty file in its destructor, so a failed MERGE/UPDATE leaves it
orphaned. Transfer cleanup ownership immediately after creation, or
register/close the dynamic writer on open failure, and cover this post-create
failure path with an object-store assertion.
##########
regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_merge_duplicate_source_negative.groovy:
##########
@@ -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.
Review Comment:
[P2] Make the late-rollover sequence deterministic
`parallel_pipeline_task_num=1` does not make this a single global sink. The
new `MERGE_PARTITIONED` path sends the unmatched INSERT by its Iceberg
partition while both matched UPDATE rows use the row-id channel, so the
duplicate can fail in one `VIcebergMergeSink` before the insert channel has
closed any data file. In that execution the object count still stays unchanged
and this passes without exercising `_closed_files` cleanup. This is a distinct
follow-up to the existing rollover thread: add a barrier/failpoint that
confirms a data file has closed and been recorded before releasing the
duplicate row, then keep the object-store assertion.
##########
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.
+ sinkQuery =
UnboundTableSinkCreator.createUnboundTableSink(createTableInfo.getTableNameParts(),
Review Comment:
[P2] Validate the catalog before eager sink construction
With an explicit recognized engine, a missing catalog reaches this line:
`paddingEngineName()` only checks catalog existence when the engine is omitted,
`checkEngineWithCatalog()` falls through on a null catalog, and
`targetTableExists()` returns false. `createUnboundTableSink()` then evaluates
`curCatalog.getClass()` and throws a `NullPointerException`, whereas the former
`Env.createTable()` ordering returned `Unknown catalog`. This is distinct from
the existing Paimon-target diagnostic thread because neither the catalog nor
the table exists here. Resolve the catalog before sink prevalidation (or make
the factory report the standard error) and cover explicit-engine CTAS against
an unknown catalog.
--
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]
