[branch-4.1](fix) Backport external write correctness and Paimon struct reads - #66246
[branch-4.1](fix) Backport external write correctness and Paimon struct reads#66246Gabriel39 wants to merge 3 commits into
Conversation
- Validate unsupported CTAS sinks before publishing table metadata, so setup failures cannot leave an unusable table or remove a concurrent replacement. - Preserve nullable string wrappers in truncate partition transforms. - Enforce Iceberg MERGE cardinality routing independently of `enable_strict_consistency_dml`. - Detect duplicate target matches with file-path-interned Roaring bitmaps and expose retained validation-state bytes in the sink profile. - Activate the negative regressions for the three retained Jira fixes and align the duplicate-match oracle with the emitted diagnostic. - Full build: `./build.sh --fe --be -j 48` - FE: `CreateTableCommandTest`, `IcebergDDLAndDMLPlanTest` (19 tests) - BE: `VIcebergMergeSinkTest` (8 tests, including 100,000 matched rows) - Regression: `test_paimon_ctas_atomicity_negative` (1 suite, passed) - Iceberg end-to-end SQL: duplicate-source MERGE rejected atomically; nullable/merge truncate writes and logical rows verified; physical partition values verified from the same Iceberg tables through Spark metadata. The Doris-side `$partitions`/`$snapshots` checks in the Iceberg suites currently hit an unrelated JNI scanner initialization abort on the current master test binary; the equivalent table state and metadata assertions above passed. (cherry picked from commit 8305fe7)
## Proposed changes - Preserve the full Paimon `RowType` arity when reading projected struct fields. - Add unit coverage for non-leading, non-contiguous, and complete struct projections backed by a binary row. ## Problem summary Struct projection entries use original child indexes, but the scanner initialized the nested binary row with the number of projected fields. A projection such as the third child therefore created a one-field row and then accessed index two, producing incorrect reads or an out-of-bounds failure. ## Regression coverage - `testUnpackNonLeadingStructField` constructs a three-field binary row and projects only original field index 2. The previous implementation fails with `index (2) should < 1`. - `testUnpackNonContiguousStructFields` projects original field indexes 0 and 2. The previous implementation fails with `index (2) should < 2`. - `testUnpackCompleteStruct` protects the existing full-struct behavior. - With this fix restored, all three tests pass with Maven build caching disabled. ## Validation - Reverted the row-arity fix locally and confirmed that the two pruned-field regression tests fail. - Restored the fix and reran the same test class: 3 tests, 0 failures, 0 errors. - Ran all Paimon scanner test classes with Maven build caching disabled: 15 tests, 0 failures, 0 errors. - Checkstyle passed for the affected reactor modules. (cherry picked from commit 5ede4b6)
|
Thank you for your contribution to Apache Doris. Please clearly describe your PR:
|
|
run buildall |
|
/review |
FE UT Coverage ReportIncrement line coverage |
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
There was a problem hiding this comment.
Requesting changes for eight correctness issues: mixed-version MERGE cardinality can be
inconsistent; MERGE/file cleanup releases or misses ownership at three failure boundaries;
constant-nullable truncate UPDATE can fail; pruned Paimon structs can read the wrong field; and
CTAS rollback/catalog validation remain unsafe. No additional review focus was provided, so the
full PR was reviewed. Review is static per the bundle contract; no builds were run.
| */ | ||
| @ConfField(mutable = false) | ||
| public static int max_be_exec_version = 10; | ||
| public static int max_be_exec_version = 11; |
There was a problem hiding this 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.
| return table_status; | ||
| Status result_status = table_status.ok() ? delete_status : table_status; | ||
| if (_table_writer) { | ||
| _table_writer->finish_deferred_file_cleanup(result_status); |
There was a problem hiding this 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.
| 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)); |
There was a problem hiding this 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.
| COUNTER_SET(_close_timer, _close_ns); | ||
| COUNTER_SET(_write_file_counter, _write_file_count); | ||
| } | ||
| if (!status.ok() || !result_status.ok()) { |
There was a problem hiding this 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.
| 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), |
There was a problem hiding this 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.
| // 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), |
There was a problem hiding this 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.
| 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(), |
There was a problem hiding this 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.
| } | ||
| // 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(), |
There was a problem hiding this 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 UT Coverage ReportIncrement line coverage Increment coverage report
|
BE Regression && UT Coverage ReportIncrement line coverage Increment coverage report
|
FE Regression Coverage ReportIncrement line coverage |
2 participants
Summary
branch-4.1Env.dropTablesignature.Dependency review
No additional prerequisite PR needs to be backported:
_pathcleanup introduced by [fix](iceberg) Enforce external write correctness #66112 supersedes the relevant behavior, so the unrelated Hive changes from [fix](external-write) Delete the actual data file on write failure to avoid orphan files #64678 are not included.branch-4.1through [branch-4-1](fix) Backport selected external scan fixes #65935; itsImmutableMaptest compatibility is preserved in the conflict resolution.branch-4.1API.Validation
CreateTableCommandTest,IcebergDDLAndDMLPlanTest,PaimonMetadataOpsTest,IcebergMergeExecutorTest, andIcebergMergeSinkTest.PaimonColumnValueTesttests passed; Checkstyle passed.VIcebergMergeSinkTestandPartitionTransformersTest.git diff --check upstream/branch-4.1..HEADpassed.