github-actions[bot] commented on code in PR #66246:
URL: https://github.com/apache/doris/pull/66246#discussion_r3676404505
##########
be/src/exec/sink/viceberg_merge_sink.cpp:
##########
@@ -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);
Review Comment:
[P1] Keep MERGE files owned until the query-wide outcome
`result_status` is local to this `VIcebergMergeSink`, so a locally
successful instance clears
both inner ownership lists here even though another exchange destination can
later reject a
duplicate target match. The same premature release occurs before the later
Iceberg optimistic
commit. In both cases FE rollback is a no-op for insert/MERGE files, leaving
the successful
sibling's unpublished data/delete/puffin objects orphaned. Retain paths at
query/transaction
scope (or provide a coordinator abort cleanup) and add a barrier-controlled
multi-instance
failure test.
##########
fe/fe-common/src/main/java/org/apache/doris/common/Config.java:
##########
@@ -2209,7 +2209,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;
Review Comment:
[P1] Negotiate version 11 with every assigned backend
`be_exec_version` now defaults to 11 and is copied query-wide, but no path
derives the minimum
capability of the BEs assigned to the query. An old BE does not reject
version 11 at fragment
startup in a release build, ignores the new optional sink field, and a local
exchange bypasses
the later block-serialization version check. A new-FE/old-BE or mixed-BE
MERGE can therefore
execute with cardinality checking enabled only on new sinks. Fence or
schedule away BEs below
version 11 (or negotiate a query-wide minimum) before setting this
requirement, and test a
mixed-version local-channel plan.
##########
fe/be-java-extensions/paimon-scanner/src/main/java/org/apache/doris/paimon/PaimonColumnValue.java:
##########
@@ -197,11 +197,12 @@ public void unpackMap(List<ColumnValue> keys,
List<ColumnValue> values) {
@Override
public void unpackStruct(List<Integer> structFieldIndex, List<ColumnValue>
values) {
- // todo: support pruned struct fields
- InternalRow row = record.getRow(idx, structFieldIndex.size());
+ RowType rowType = (RowType) dataType;
+ // Projection entries are original child indexes, so the binary row
must keep the full RowType arity.
+ InternalRow row = record.getRow(idx, rowType.getFieldCount());
for (int i : structFieldIndex) {
values.add(new PaimonColumnValue(row, i,
dorisType.getChildTypes().get(i),
Review Comment:
[P1] Resolve pruned struct children to their original Paimon indexes
Using the full Paimon row arity is correct, but production does not pass
original child indexes
here. Nereids rebuilds a pruned Doris `StructType`, and `VectorColumn`
consequently supplies
dense indexes `0..N-1`. Selecting only `s.c` from `struct<a,b,c>` therefore
passes `i == 0`
against the full Paimon row and decodes `a` as the sole Doris child (or
fails on a type
mismatch). The new test manually passes `[2]` with a full Doris struct,
which production cannot
construct. Map dense output children to full-row fields by name/original
index and add a
scanner-shaped pruned test.
##########
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());
+ 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));
Review Comment:
[P1] Register the data file before `open()` can fail
`create_file()` runs before compression and transformer validation, but the
dynamic-partition
caller inserts the writer into `_partitions_to_writers` only after `open()`
succeeds. The new
closed-file callback also runs only from a successful `close()`. A
post-create failure such as
unsupported Parquet GZ therefore destroys the only writer reference; the
HDFS writer destructor
closes the handle but does not delete the physical path. Please transfer
ownership immediately
after creation (or use an open-scope delete guard) and cover this boundary
with a physical
object-count test.
##########
be/src/exec/sink/writer/iceberg/viceberg_table_writer.cpp:
##########
@@ -486,9 +487,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()) {
Review Comment:
[P1] Do not release rolled files on the pre-defer close status
`VIcebergSortWriter::_close_locked()` returns `close_status` while the
`Defer` that assigns the
underlying partition-writer close result is still alive, so C++ materializes
an OK return value
before that close runs. This new success branch then clears the rolled-file
ownership even when
the final Parquet/ORC close or commit-data build failed, allowing only the
earlier rolled subset
to be committed. Run the underlying close before forming the return value
and add a sorted
final-close failure test.
##########
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 engine, `paddingEngineName()` skips catalog lookup and
`checkEngineWithCatalog()` falls through when the catalog is absent.
`targetTableExists()` then
returns false, so this new eager call reaches `UnboundTableSinkCreator`,
where
`curCatalog.getClass()` throws a null dereference instead of the catalog
API's normal
`Unknown catalog` diagnostic. Validate catalog existence before this call
(or make the factory's
null path explicit) and add an explicit-engine missing-catalog CTAS test.
##########
be/src/exec/sink/writer/iceberg/partition_transformers.h:
##########
@@ -167,7 +167,10 @@ class StringTruncatePartitionColumnTransform : public
PartitionColumnTransform {
// 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),
Review Comment:
[P1] Unwrap constant nullable strings before truncate
Iceberg `UPDATE ... SET nullable_truncate_col = CAST(NULL AS STRING)`
reaches the exchange
transform as `ColumnConst(ColumnNullable)`: UPDATE preserves the literal
alias, and
`MergePartitioner` applies this transform before its later const
materialization. The top-level
nullable check misses the wrapper; `SubstringUtil` unwraps the const and
then asserts that the
remaining `ColumnNullable` is a `ColumnString`, causing the update to fail
before the writer.
Materialize before checking nullability (or skip a constant transform for
routing) and add a
`ColumnConst(ColumnNullable)`/literal-UPDATE regression.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/CreateTableCommand.java:
##########
@@ -107,12 +122,16 @@ 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 {
+ 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(),
Review Comment:
[P2] Do not roll back an `IF NOT EXISTS` CTAS by table name
For an absent Paimon target this path creates the table, then performs
unsupported-sink
validation. If validation fails, `handleFallbackFailedCtas()` carries only
`(catalog,database,table)` into a name-based drop. A concurrent client can
drop the empty table
and create a replacement before fallback, causing Doris to delete the
replacement it does not
own. Make validation/creation atomic or carry a connector-issued identity
into a conditional
rollback, and cover the replacement interleaving.
--
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]
