github-actions[bot] commented on code in PR #66825:
URL: https://github.com/apache/doris/pull/66825#discussion_r3795908625
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergTransaction.java:
##########
@@ -164,6 +176,7 @@ public void beginRewrite(ExternalTable dorisTable, Table
targetTable) throws Use
// For rewrite operations, we work directly on the main table
this.branchName = null;
this.isRewriteMode = true;
+ acquireCommitFence(targetTable);
Review Comment:
[P1] Release the rewrite fence on every failed execution
After this read stamp is acquired,
`RewriteDataFileExecutor.executeGroupsConcurrently()` can still throw while
discovering backends, submitting only part of the task set, waiting for tasks
(timeout/interruption/worker error), or running `finishRewrite()`. That caller
has no rollback/finally path; its only `transaction.commit()` is after all of
those operations. Because the transaction remains registered and this guard is
released only by `commit()`/`rollback()`, any such rewrite failure permanently
holds the stripe's read lock, so a later `remove_orphan_files` blocks forever
acquiring the maintenance write lock (and a hash-colliding table can be blocked
too). Please put the rewrite transaction under a success/finally scope, cancel
or quiesce any submitted rewrite tasks, and roll it back through the
transaction manager on every non-committed exit.
##########
be/src/format_v2/table_reader.h:
##########
@@ -1658,9 +1867,10 @@ class TableReader {
// storage invariant, so add it only at the materialization boundary.
element_mapping.table_type = make_nullable(element_mapping.table_type);
NullMap descendant_parent_null_map;
- const NullMap* descendant_parent_null_map_ptr =
_project_collection_parent_null_map(
- parent_null_map, nullable_parent_null_map, rows,
file_array->get_offsets(),
- nested_column->size(), &descendant_parent_null_map);
+ const NullMap* descendant_parent_null_map_ptr =
+ _project_collection_parent_null_map_for_hidden_entries(
Review Comment:
[P1] Probe consumers before allocating the collection mask
The new fast path in `_align_column_nullability()` uses
`_requires_collection_parent_null_map()` before projecting an entry-coordinate
mask, but this nontrivial ARRAY path (and the MAP path below) still projects
unconditionally. If one ancestor-hidden row has even one physical entry, the
helper allocates and zero-fills `child_rows` bytes, even when all mapped
descendants are optional or already protected by nearer nullable wrappers and
cannot consume that mask. A valid evolved collection can have a tiny hidden
span plus millions of visible entries, so this remains an avoidable unbounded
transient allocation/OOM path. Please add a mapping-aware consumer check for
the present element/key/value descendants before projecting (the existing probe
cannot be applied blindly to a differently shaped evolved STRUCT), and add a
nontrivial large-collection test.
##########
be/src/exec/sink/writer/paimon/paimon_table_writer.cpp:
##########
@@ -24,6 +24,39 @@
namespace doris {
+PaimonPreparedCommitOwner::PaimonPreparedCommitOwner(std::unique_ptr<IPaimonWriter>
writer,
+
std::unique_ptr<IPaimonWriteBackend> backend)
+ : _writer(std::move(writer)), _backend(std::move(backend)) {}
+
+PaimonPreparedCommitOwner::~PaimonPreparedCommitOwner() {
+ _close();
+}
+
+void PaimonPreparedCommitOwner::finalize(ExternalFileReportOutcome outcome) {
+ if (_finalized || outcome == ExternalFileReportOutcome::AMBIGUOUS) {
+ return;
+ }
+ _finalized = true;
+ if (outcome == ExternalFileReportOutcome::REJECTED && _writer) {
+ Status abort_status = _writer->abort();
+ if (!abort_status.ok()) {
+ LOG(WARNING) << "Paimon prepared writer abort failed: " <<
abort_status.to_string();
+ }
+ }
+ _close();
+}
+
+void PaimonPreparedCommitOwner::_close() {
+ _writer.reset();
+ if (_backend) {
+ Status close_status = _backend->close();
Review Comment:
[P1] Do not discard the prepared backend's close failure after ACK
This close now runs only from the report finalizer. On the ACK path FE has
already stored the Paimon payloads and replied that ownership transferred, so
reducing a failure here to a warning leaves no way to fail or roll back the
transaction. That is a regression from the prior close path, which treated
Java/backend close as the authoritative SDK shutdown boundary and published
messages only after it succeeded; `JniPaimonWriteBackend::close()` can fail
while SDK users may still be active and deliberately retains native memory in
that case. The transaction can now commit those payloads anyway. Please make
successful backend shutdown part of the acceptance protocol (while retaining an
abort-capable owner for rejection), or otherwise propagate a post-accept close
failure to the coordinator so the transaction cannot commit, and cover a
failing ACK-path close.
##########
be/src/exec/sort/sorter.cpp:
##########
@@ -184,35 +199,62 @@ bool FullSorter::has_enough_capacity(Block* input_block,
Block* unsorted_block)
}
size_t FullSorter::get_reserve_mem_size(RuntimeState* state, bool eos) const {
- size_t size_to_reserve = 0;
+ return get_reserve_mem_size_components(state, eos).total();
+}
+
+SorterReserveMemory FullSorter::get_reserve_mem_size_components(RuntimeState*
state,
+ bool eos)
const {
+ const auto rows = _state->unsorted_block()->rows();
+ const auto bytes = _state->unsorted_block()->bytes();
+ const auto bytes_per_row = rows == 0 ? 0 : bytes / rows;
+ return get_reserve_mem_size_components(
+ state, eos, state->batch_size(),
+ saturating_multiply_size(bytes_per_row, state->batch_size()));
+}
+
+SorterReserveMemory FullSorter::get_reserve_mem_size_components(RuntimeState*
state, bool eos,
+ size_t
incoming_rows,
+ size_t
incoming_bytes) const {
+ SorterReserveMemory reserve;
const auto rows = _state->unsorted_block()->rows();
if (rows != 0) {
const auto bytes = _state->unsorted_block()->bytes();
const auto allocated_bytes =
_state->unsorted_block()->allocated_bytes();
- const auto bytes_per_row = bytes / rows;
- const auto estimated_size_of_next_block = bytes_per_row *
state->batch_size();
- auto new_block_bytes = estimated_size_of_next_block + bytes;
- auto new_rows = rows + state->batch_size();
+ auto new_block_bytes = saturating_add_size(bytes, incoming_bytes);
+ auto new_rows = saturating_add_size(rows, incoming_rows);
// If the new size is greater than 85% of allocalted bytes, it maybe
need to realloc.
- if ((new_block_bytes * 100 / allocated_bytes) >= 85) {
- size_to_reserve += (size_t)(allocated_bytes * 1.15);
+ const auto growth_threshold = static_cast<size_t>(
+ (static_cast<unsigned __int128>(allocated_bytes) * 85 + 99) /
100);
+ const size_t growth_trigger_bytes = growth_threshold > bytes ?
growth_threshold - bytes : 0;
+ if (incoming_rows > 0 && growth_trigger_bytes <= incoming_bytes) {
+ reserve.retained_growth = static_cast<size_t>(std::min<unsigned
__int128>(
+ (static_cast<unsigned __int128>(allocated_bytes) * 115 +
99) / 100,
+ std::numeric_limits<size_t>::max()));
+ reserve.retained_growth_trigger_bytes = growth_trigger_bytes;
}
- auto sort = new_rows > _buffered_block_size || new_block_bytes >
_buffered_block_bytes;
+ // Iceberg close forces every nonempty pending run to sort at EOS,
even when the generic
+ // append thresholds are not reached, so admission must cover that
final allocation too.
+ auto sort = (eos && new_rows > 0) || new_rows > _buffered_block_size ||
Review Comment:
[P1] Reserve for target-size-triggered sorts too
This predicate misses another immediate `do_sort()` path:
`VIcebergSortWriter::write()` calls `_flush_to_file()` as soon as `data_size()`
reaches the user-supplied target file size, and that target accepts any
positive value. For example, a run below 4M rows can cross a 128 MiB target
while remaining below this 256 MiB condition; the estimator then admits only
append growth, but `_flush_to_file()` immediately materializes the full sorted
destination plus permutation while the source stays live. Under memory pressure
that allocation escapes the reservation and can hit the hard limit/OOM. Please
include the same `current data_size + incoming >= target` rollover predicate in
the reservation (or pass the target into this estimator), with a below-256-MiB
test.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java:
##########
@@ -298,6 +306,224 @@ void
checkVariantBackendCompatibilityForCurrentScan(Iterable<Backend> backends)
checkVariantBackendCompatibility(projectsVariant, backends);
}
+ private boolean requiresIcebergScanSemanticsV2() throws UserException {
+ if (isSystemTable) {
+ // position_deletes already has its stricter dedicated
mixed-version gate.
+ return false;
+ }
+ TableScan scan = createTableScan();
+ Snapshot snapshot = scan.snapshot();
+ if (snapshot == null) {
+ return false;
+ }
+ if (hasApplicableEqualityDeletes(scan)) {
+ return true;
+ }
+ Schema scanSchema = scan.schema();
+ Set<Integer> projectedFieldIds = projectedFieldIds(scanSchema);
+ Set<Integer> topLevelIds = new HashSet<>();
+ for (NestedField field : scanSchema.columns()) {
+ topLevelIds.add(field.fieldId());
+ }
+ Map<Integer, NestedField> fieldsById =
TypeUtil.indexById(scanSchema.asStruct());
+ for (Integer fieldId : projectedFieldIds) {
+ NestedField field = fieldsById.get(fieldId);
+ if (field.initialDefault() != null
+ && (!topLevelIds.contains(field.fieldId()) ||
field.type().isNestedType())) {
+ return true;
+ }
+ }
+ if (hasProjectedNameAliasCollision(scanSchema, projectedFieldIds,
extractNameMapping())) {
+ return true;
+ }
+ return schemaHistoryRequiresMissingRequiredFieldRejection(
Review Comment:
[P2] Limit the V2 gate to schemas reachable by this scan
`createTableScan()` projects the schema bound to the selected snapshot/ref,
but this call checks it against every schema in the table metadata. The helper
returns true if a projected required field is absent or optional in any of
them, including a schema created only after a time-traveled snapshot or only on
another branch. Those schemas cannot have produced a file reachable by this
scan, so V1 semantics are safe, yet the query is rejected for the whole
rolling-upgrade window whenever an old BE is present. Please derive the
relevant schema IDs from the selected snapshot/ref's reachable
history/manifests instead of using the table-wide schema map, and cover a
later/unrelated schema in the upgrade test.
##########
be/src/exec/operator/spill_iceberg_table_sink_operator.cpp:
##########
@@ -26,12 +28,42 @@
namespace doris {
#include "common/compile_check_begin.h"
+size_t iceberg_cold_writer_reserve_size(const Block& block, size_t
writer_workspace_bytes) {
+ const size_t block_bytes = block.allocated_bytes();
+ const size_t row_index_bytes =
+ std::min(std::numeric_limits<size_t>::max() / sizeof(size_t),
block.rows()) *
+ sizeof(size_t);
+ const size_t dispatch_copies = block_bytes >
std::numeric_limits<size_t>::max() / 4
+ ? std::numeric_limits<size_t>::max()
+ : block_bytes * 4;
+ size_t reserve = iceberg_saturating_add(writer_workspace_bytes,
dispatch_copies);
+ if (block.rows() > 0) {
+ size_t minimum_selected_block_bytes = 0;
+ for (const auto& column : block.get_columns_with_type_and_name()) {
+ minimum_selected_block_bytes =
+ iceberg_saturating_add(minimum_selected_block_bytes,
+
column.column->clone_resized(1)->allocated_bytes());
Review Comment:
[P1] Keep reserve estimation allocation-free
`PipelineTask::execute()` calls this estimator before
`_try_to_reserve_memory()`, but `clone_resized(1)` copies the complete first
value: STRING copies all bytes through its first offset, ARRAY copies every
element in the first array, and MAP copies every first-row key/value. A legal
block with a tens- or hundreds-of-MiB first value can therefore allocate and
copy that payload under pressure before the returned reserve can
pause/revoke/admit the sink, defeating the admission boundary and potentially
hitting the hard limit/OOM. The full-block `dispatch_copies` term already
accounts for the actual input payload. Please derive the minimum
selected-column capacity without cloning real data (or otherwise make this
estimator allocation-free), and test a huge first STRING/ARRAY/MAP value under
memory pressure.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java:
##########
@@ -298,6 +306,224 @@ void
checkVariantBackendCompatibilityForCurrentScan(Iterable<Backend> backends)
checkVariantBackendCompatibility(projectsVariant, backends);
}
+ private boolean requiresIcebergScanSemanticsV2() throws UserException {
+ if (isSystemTable) {
+ // position_deletes already has its stricter dedicated
mixed-version gate.
+ return false;
+ }
+ TableScan scan = createTableScan();
+ Snapshot snapshot = scan.snapshot();
+ if (snapshot == null) {
+ return false;
+ }
+ if (hasApplicableEqualityDeletes(scan)) {
+ return true;
+ }
+ Schema scanSchema = scan.schema();
+ Set<Integer> projectedFieldIds = projectedFieldIds(scanSchema);
+ Set<Integer> topLevelIds = new HashSet<>();
+ for (NestedField field : scanSchema.columns()) {
+ topLevelIds.add(field.fieldId());
+ }
+ Map<Integer, NestedField> fieldsById =
TypeUtil.indexById(scanSchema.asStruct());
+ for (Integer fieldId : projectedFieldIds) {
+ NestedField field = fieldsById.get(fieldId);
+ if (field.initialDefault() != null
+ && (!topLevelIds.contains(field.fieldId()) ||
field.type().isNestedType())) {
+ return true;
+ }
+ }
+ if (hasProjectedNameAliasCollision(scanSchema, projectedFieldIds,
extractNameMapping())) {
+ return true;
+ }
+ return schemaHistoryRequiresMissingRequiredFieldRejection(
+ scanSchema, projectedFieldIds,
icebergTable.schemas().values());
+ }
+
+ private boolean hasApplicableEqualityDeletes(TableScan scan) throws
UserException {
+ Snapshot snapshot = scan.snapshot();
+ if (snapshot == null) {
+ return false;
+ }
+ String equalityDeleteCount =
snapshot.summary().get("total-equality-deletes");
+ if (equalityDeleteCount != null) {
+ try {
+ // A positive snapshot total proves V2 semantics are required
without opening every
+ // delete manifest; only old summaries that omit the counter
need the fallback.
+ return Long.parseLong(equalityDeleteCount) > 0;
+ } catch (NumberFormatException ignored) {
+ // Fall through for non-standard summaries instead of
weakening compatibility.
+ }
+ }
+ // Inspect only delete manifests, not data tasks: equality-delete
semantics are snapshot-wide
+ // compatibility state even when the current predicate happens to
prune their partitions.
+ for (ManifestFile manifest :
snapshot.deleteManifests(icebergTable.io())) {
+ if (!manifest.hasAddedFiles() && !manifest.hasExistingFiles()) {
+ continue;
+ }
+ try (ManifestReader<DeleteFile> deletes =
ManifestFiles.readDeleteManifest(
+ manifest, icebergTable.io(), icebergTable.specs())) {
+ for (DeleteFile delete : deletes) {
+ if (delete.content() == FileContent.EQUALITY_DELETES) {
+ return true;
+ }
+ }
+ } catch (IOException e) {
+ throw new UserException(
+ "Failed to inspect Iceberg delete manifest " +
manifest.path(), e);
+ }
+ }
+ return false;
+ }
+
+ private static boolean hasSmoothUpgradeSource(Iterable<Backend> backends) {
+ for (Backend backend : backends) {
+ if (backend.isSmoothUpgradeSrc()) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ private Set<Integer> projectedFieldIds(Schema scanSchema) {
+ Set<Integer> projected = new HashSet<>();
+ for (SlotDescriptor slot : desc.getSlots()) {
+ int fieldId = slot.getColumn().getUniqueId();
+ // Stable Iceberg IDs prevent a dropped-and-readded name from
selecting the wrong history.
+ NestedField field = fieldId >= 0 ? scanSchema.findField(fieldId)
+ :
scanSchema.caseInsensitiveFindField(slot.getColumn().getName());
+ if (field != null) {
+ projected.addAll(TypeUtil.indexById(
Review Comment:
[P2] Gate only on descendants this scan actually projects
Nested-column pruning replaces the effective slot type and retains the
actual access paths, but this rebuilds IDs from the full Iceberg root field and
therefore adds every schema descendant. For example, `SELECT payload.keep` can
be rejected during rolling upgrade solely because an unrequested sibling such
as `payload.added` has a nested initial default, alias collision, or
requiredness history: `requiresV2` becomes true even though BE will not
materialize that sibling. This is independent of limiting history to schemas
reachable by the selected snapshot/ref. Please derive descendant IDs from the
slot's pruned type/access paths while retaining the stable Iceberg root ID, and
add a mixed-version test where only the unrequested sibling requires V2
semantics.
--
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]
