github-actions[bot] commented on code in PR #66825:
URL: https://github.com/apache/doris/pull/66825#discussion_r3800458237
##########
be/src/exec/pipeline/pipeline_fragment_context.cpp:
##########
@@ -2235,14 +2224,40 @@ void
PipelineFragmentContext::_coordinator_callback(const ReportStatusRequest& r
rpc_status = Status::create<false>(res.status);
} catch (apache::thrift::TException& e) {
+ report_outcome_ambiguous = true;
rpc_status = Status::InternalError("ReportExecStatus() to {} failed:
{}",
PrintThriftNetworkAddress(req.coord_addr), e.what());
}
+ // Iceberg requires the explicit new-protocol ACK. Legacy Hive/Paimon
coordinators transfer
+ // ownership through RPC success, which remains valid during a rolling FE
upgrade.
+ const bool requires_external_file_ack =
params.__isset.iceberg_commit_datas;
Review Comment:
[P1] Do not infer legacy ownership acceptance from RPC success
A pre-ACK FE can return `OK` without storing these vectors: its
`reportExecStatus` returns success when the coordinator is already absent and
also catches any `updateFragmentExecStatus` exception (for example a missing
transaction) and still returns success. For Hive/Paimon/MaxCompute this branch
therefore ACK-finalizes the BE cleanup/abort owner even though FE never
acquired the commit data, stranding staged MPUs/files. Please require an
explicit acceptance capability for every ownership-bearing final report (or
retain a legacy transfer protocol that cannot drop the only owner), rather than
limiting the ACK check to Iceberg.
##########
be/src/exec/sort/sorter.cpp:
##########
@@ -184,35 +199,77 @@ 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 {
+ return get_reserve_mem_size_components(state, eos,
std::numeric_limits<size_t>::max());
+}
+
+SorterReserveMemory FullSorter::get_reserve_mem_size_components(RuntimeState*
state, bool eos,
+ size_t
sort_threshold_bytes) 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()),
sort_threshold_bytes);
+}
+
+SorterReserveMemory FullSorter::get_reserve_mem_size_components(RuntimeState*
state, bool eos,
+ size_t
incoming_rows,
+ size_t
incoming_bytes) const {
+ return get_reserve_mem_size_components(state, eos, incoming_rows,
incoming_bytes,
+ std::numeric_limits<size_t>::max());
+}
+
+SorterReserveMemory FullSorter::get_reserve_mem_size_components(RuntimeState*
state, bool eos,
+ size_t
incoming_rows,
+ size_t
incoming_bytes,
+ size_t
sort_threshold_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.
+ // The reservation must mirror every caller-side rollover that
immediately invokes do_sort().
+ auto sort = (eos && new_rows > 0) || new_rows > _buffered_block_size ||
Review Comment:
[P1] Reserve the pre-append capacity rollover too
This predicate still misses `FullSorter::append_block()`'s earlier
`_reach_limit() && !has_enough_capacity()` sort. Once the buffered columns have
64 MiB allocated, a small skewed input can exhaust one column's capacity while
aggregate logical bytes remain below the 85% growth check, 256 MiB, and the
Iceberg target. `append_block()` then calls `do_sort()` before appending,
retaining the source while materializing the sorted destination and permutation
without reserving that transient workspace. Please pass the actual input
block/per-column capacity summary or otherwise mirror that trigger, with a
below-target skewed-capacity rollover test.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java:
##########
@@ -298,6 +310,271 @@ 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, reachableSchemas(icebergTable,
snapshot));
+ }
+
+ 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) {
+ return projectedFieldIds(scanSchema, desc.getSlots());
+ }
+
+ @VisibleForTesting
+ static Set<Integer> projectedFieldIds(Schema scanSchema,
Iterable<SlotDescriptor> slots) {
+ Set<Integer> projected = new HashSet<>();
+ for (SlotDescriptor slot : slots) {
+ 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) {
+ collectProjectedFieldIds(field, slot.getType(), projected);
+ }
+ }
+ return projected;
+ }
+
+ private static void collectProjectedFieldIds(
+ NestedField field, org.apache.doris.catalog.Type projectedType,
+ Set<Integer> projected) {
+ projected.add(field.fieldId());
+ if (projectedType instanceof StructType &&
field.type().isStructType()) {
+ for (StructField projectedChild : ((StructType)
projectedType).getFields()) {
+ NestedField icebergChild =
field.type().asStructType().fields().stream()
+ .filter(child ->
child.name().equalsIgnoreCase(projectedChild.getName()))
+ .findFirst().orElse(null);
+ if (icebergChild != null) {
+ collectProjectedFieldIds(icebergChild,
projectedChild.getType(), projected);
+ }
+ }
+ } else if (projectedType instanceof ArrayType &&
field.type().isListType()) {
+ collectProjectedFieldIds(field.type().asListType().fields().get(0),
+ ((ArrayType) projectedType).getItemType(), projected);
+ } else if (projectedType instanceof MapType &&
field.type().isMapType()) {
+ collectProjectedFieldIds(field.type().asMapType().fields().get(0),
+ ((MapType) projectedType).getKeyType(), projected);
+ collectProjectedFieldIds(field.type().asMapType().fields().get(1),
+ ((MapType) projectedType).getValueType(), projected);
+ }
+ }
+
+ @VisibleForTesting
+ static Iterable<Schema> reachableSchemas(Table table, Snapshot
selectedSnapshot) {
+ Map<Integer, Schema> schemas = table.schemas();
+ List<Schema> reachable = new ArrayList<>();
+ Set<Long> visitedSnapshots = new HashSet<>();
+ Set<Integer> visitedSchemaIds = new HashSet<>();
+ Snapshot snapshot = selectedSnapshot;
+ while (snapshot != null &&
visitedSnapshots.add(snapshot.snapshotId())) {
+ Schema schema = schemas.get(snapshot.schemaId());
+ if (schema != null && visitedSchemaIds.add(snapshot.schemaId())) {
+ reachable.add(schema);
+ }
+ Long parentId = snapshot.parentId();
+ snapshot = parentId == null ? null : table.snapshot(parentId);
Review Comment:
[P1] Do not lose live-file schema history at an expired parent
Snapshot expiration can remove an ancestor record without removing data
files that remain inherited by this selected snapshot. The descendant still
names that expired parent, so `table.snapshot(parentId)` returns null here and
the walk omits the schema under which those live files were written. If a
projected field was optional/missing there but is now required without a
default, `requiresV2` becomes false and a V1 BE can materialize the old file
without the required-field rejection. Please treat a broken parent chain
conservatively or derive schema provenance from the selected snapshot's live
manifests, with an expired-ancestor mixed-version test.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/rewrite/RewriteDataFileExecutor.java:
##########
@@ -61,88 +63,103 @@ public RewriteDataFileExecutor(IcebergExternalTable
dorisTable,
*/
public RewriteResult executeGroupsConcurrently(List<RewriteDataGroup>
groups, long targetFileSizeBytes)
throws UserException {
- // Begin transaction
- long transactionId =
dorisTable.getCatalog().getTransactionManager().begin();
- IcebergTransaction transaction = (IcebergTransaction)
dorisTable.getCatalog().getTransactionManager()
- .getTransaction(transactionId);
- MvccSnapshot targetSnapshot =
dorisTable.loadSnapshot(Optional.empty(), Optional.empty());
- Table targetIcebergTable = ((IcebergMvccSnapshot)
targetSnapshot).getSnapshotCacheValue()
- .getIcebergTable().orElseThrow(
- () -> new UserException("Iceberg rewrite target
metadata is not available"));
- transaction.beginRewrite(dorisTable, targetIcebergTable);
-
- // Register files to delete
- for (RewriteDataGroup group : groups) {
-
transaction.updateRewriteFiles(Lists.newArrayList(group.getDataFiles()));
- }
-
- // Create result collector and tasks
+ TransactionManager transactionManager =
dorisTable.getCatalog().getTransactionManager();
+ long transactionId = transactionManager.begin();
List<RewriteGroupTask> tasks = Lists.newArrayList();
- RewriteResultCollector resultCollector = new
RewriteResultCollector(groups.size(), tasks);
-
- // Get available BE count once before creating tasks
- // This avoids calling getBackendsNumber() in each task during
multi-threaded execution.
- // Use compute group from connect context to align with actual BE
selection for queries.
- int availableBeCount = getAvailableBeCount();
-
- // Create tasks with callbacks
- for (RewriteDataGroup group : groups) {
- RewriteGroupTask task = new RewriteGroupTask(
- group,
- transactionId,
- dorisTable,
- targetSnapshot,
- connectContext,
- targetFileSizeBytes,
- availableBeCount,
- new RewriteGroupTask.RewriteResultCallback() {
- @Override
- public void onTaskCompleted(Long taskId) {
- resultCollector.onTaskCompleted(taskId);
- }
-
- @Override
- public void onTaskFailed(Long taskId, Exception error)
{
- resultCollector.onTaskFailed(taskId, error);
- }
- });
- tasks.add(task);
- }
-
- // Submit tasks to TransientTaskManager
+ boolean committed = false;
try {
- for (TransientTaskExecutor task : tasks) {
-
Env.getCurrentEnv().getTransientTaskManager().addMemoryTask(task);
+ IcebergTransaction transaction = (IcebergTransaction)
transactionManager
+ .getTransaction(transactionId);
+ MvccSnapshot targetSnapshot =
dorisTable.loadSnapshot(Optional.empty(), Optional.empty());
+ Table targetIcebergTable = ((IcebergMvccSnapshot)
targetSnapshot).getSnapshotCacheValue()
+ .getIcebergTable().orElseThrow(
+ () -> new UserException("Iceberg rewrite target
metadata is not available"));
+ transaction.beginRewrite(dorisTable, targetIcebergTable);
+
+ for (RewriteDataGroup group : groups) {
+
transaction.updateRewriteFiles(Lists.newArrayList(group.getDataFiles()));
}
- } catch (JobException e) {
- throw new UserException("Failed to submit rewrite tasks: " +
e.getMessage(), e);
- }
- // Wait for all tasks to complete
- waitForTasksCompletion(resultCollector, groups.size());
-
- // Finish rewrite operation
- transaction.finishRewrite();
-
- // Collect statistics from transaction after all tasks are completed
- int rewrittenDataFilesCount = groups.stream().mapToInt(group ->
group.getDataFiles().size()).sum();
- // this should after finishRewrite
- int addedDataFilesCount = transaction.getFilesToAddCount();
- long rewrittenBytesCount = groups.stream().mapToLong(group ->
group.getTotalSize()).sum();
- int removedDeleteFilesCount = groups.stream().mapToInt(group ->
group.getDeleteFileCount()).sum();
+ RewriteResultCollector resultCollector = new
RewriteResultCollector(groups.size(), tasks);
+ int availableBeCount = getAvailableBeCount();
+ for (RewriteDataGroup group : groups) {
+ RewriteGroupTask task = new RewriteGroupTask(
+ group, transactionId, dorisTable, targetSnapshot,
connectContext,
+ targetFileSizeBytes, availableBeCount,
+ new RewriteGroupTask.RewriteResultCallback() {
+ @Override
+ public void onTaskCompleted(Long taskId) {
+ resultCollector.onTaskCompleted(taskId);
+ }
+
+ @Override
+ public void onTaskFailed(Long taskId, Exception
error) {
+ resultCollector.onTaskFailed(taskId, error);
+ }
+ });
+ tasks.add(task);
+ }
- commitAndInvalidate(transaction);
+ try {
+ for (TransientTaskExecutor task : tasks) {
+
Env.getCurrentEnv().getTransientTaskManager().addMemoryTask(task);
+ }
+ } catch (JobException e) {
+ throw new UserException("Failed to submit rewrite tasks: " +
e.getMessage(), e);
+ }
- return new RewriteResult(rewrittenDataFilesCount, addedDataFilesCount,
- rewrittenBytesCount, removedDeleteFilesCount);
+ waitForTasksCompletion(resultCollector, groups.size());
Review Comment:
[P1] Propagate rewrite-group failures before committing deletions
`executeSingleInsert()` catches every non-replan failure, sets the task
context to `ERR`, and returns normally, so `RewriteGroupTask` calls
`onTaskCompleted` and this wait sees no `firstError`. The parent has already
registered every group's inputs in `filesToDelete`, but only successful final
reports add replacements; `finishRewrite()` can therefore delete a failed
group's source files with nothing to replace them. The happy path also relies
on swallowing `IcebergRewriteExecutor`'s inherited attempt to commit its unused
`INVALID_TXN_ID`. Please give rewrite groups failure-propagating execution and
rewrite-specific completion semantics, and test that one failed group rolls
back instead of committing its source deletions.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/rewrite/RewriteDataFileExecutor.java:
##########
@@ -61,88 +63,103 @@ public RewriteDataFileExecutor(IcebergExternalTable
dorisTable,
*/
public RewriteResult executeGroupsConcurrently(List<RewriteDataGroup>
groups, long targetFileSizeBytes)
throws UserException {
- // Begin transaction
- long transactionId =
dorisTable.getCatalog().getTransactionManager().begin();
- IcebergTransaction transaction = (IcebergTransaction)
dorisTable.getCatalog().getTransactionManager()
- .getTransaction(transactionId);
- MvccSnapshot targetSnapshot =
dorisTable.loadSnapshot(Optional.empty(), Optional.empty());
- Table targetIcebergTable = ((IcebergMvccSnapshot)
targetSnapshot).getSnapshotCacheValue()
- .getIcebergTable().orElseThrow(
- () -> new UserException("Iceberg rewrite target
metadata is not available"));
- transaction.beginRewrite(dorisTable, targetIcebergTable);
-
- // Register files to delete
- for (RewriteDataGroup group : groups) {
-
transaction.updateRewriteFiles(Lists.newArrayList(group.getDataFiles()));
- }
-
- // Create result collector and tasks
+ TransactionManager transactionManager =
dorisTable.getCatalog().getTransactionManager();
+ long transactionId = transactionManager.begin();
List<RewriteGroupTask> tasks = Lists.newArrayList();
- RewriteResultCollector resultCollector = new
RewriteResultCollector(groups.size(), tasks);
-
- // Get available BE count once before creating tasks
- // This avoids calling getBackendsNumber() in each task during
multi-threaded execution.
- // Use compute group from connect context to align with actual BE
selection for queries.
- int availableBeCount = getAvailableBeCount();
-
- // Create tasks with callbacks
- for (RewriteDataGroup group : groups) {
- RewriteGroupTask task = new RewriteGroupTask(
- group,
- transactionId,
- dorisTable,
- targetSnapshot,
- connectContext,
- targetFileSizeBytes,
- availableBeCount,
- new RewriteGroupTask.RewriteResultCallback() {
- @Override
- public void onTaskCompleted(Long taskId) {
- resultCollector.onTaskCompleted(taskId);
- }
-
- @Override
- public void onTaskFailed(Long taskId, Exception error)
{
- resultCollector.onTaskFailed(taskId, error);
- }
- });
- tasks.add(task);
- }
-
- // Submit tasks to TransientTaskManager
+ boolean committed = false;
try {
- for (TransientTaskExecutor task : tasks) {
-
Env.getCurrentEnv().getTransientTaskManager().addMemoryTask(task);
+ IcebergTransaction transaction = (IcebergTransaction)
transactionManager
+ .getTransaction(transactionId);
+ MvccSnapshot targetSnapshot =
dorisTable.loadSnapshot(Optional.empty(), Optional.empty());
+ Table targetIcebergTable = ((IcebergMvccSnapshot)
targetSnapshot).getSnapshotCacheValue()
+ .getIcebergTable().orElseThrow(
+ () -> new UserException("Iceberg rewrite target
metadata is not available"));
+ transaction.beginRewrite(dorisTable, targetIcebergTable);
+
+ for (RewriteDataGroup group : groups) {
+
transaction.updateRewriteFiles(Lists.newArrayList(group.getDataFiles()));
}
- } catch (JobException e) {
- throw new UserException("Failed to submit rewrite tasks: " +
e.getMessage(), e);
- }
- // Wait for all tasks to complete
- waitForTasksCompletion(resultCollector, groups.size());
-
- // Finish rewrite operation
- transaction.finishRewrite();
-
- // Collect statistics from transaction after all tasks are completed
- int rewrittenDataFilesCount = groups.stream().mapToInt(group ->
group.getDataFiles().size()).sum();
- // this should after finishRewrite
- int addedDataFilesCount = transaction.getFilesToAddCount();
- long rewrittenBytesCount = groups.stream().mapToLong(group ->
group.getTotalSize()).sum();
- int removedDeleteFilesCount = groups.stream().mapToInt(group ->
group.getDeleteFileCount()).sum();
+ RewriteResultCollector resultCollector = new
RewriteResultCollector(groups.size(), tasks);
+ int availableBeCount = getAvailableBeCount();
+ for (RewriteDataGroup group : groups) {
+ RewriteGroupTask task = new RewriteGroupTask(
+ group, transactionId, dorisTable, targetSnapshot,
connectContext,
+ targetFileSizeBytes, availableBeCount,
+ new RewriteGroupTask.RewriteResultCallback() {
+ @Override
+ public void onTaskCompleted(Long taskId) {
+ resultCollector.onTaskCompleted(taskId);
+ }
+
+ @Override
+ public void onTaskFailed(Long taskId, Exception
error) {
+ resultCollector.onTaskFailed(taskId, error);
+ }
+ });
+ tasks.add(task);
+ }
- commitAndInvalidate(transaction);
+ try {
+ for (TransientTaskExecutor task : tasks) {
+
Env.getCurrentEnv().getTransientTaskManager().addMemoryTask(task);
+ }
+ } catch (JobException e) {
+ throw new UserException("Failed to submit rewrite tasks: " +
e.getMessage(), e);
+ }
- return new RewriteResult(rewrittenDataFilesCount, addedDataFilesCount,
- rewrittenBytesCount, removedDeleteFilesCount);
+ waitForTasksCompletion(resultCollector, groups.size());
+ transaction.finishRewrite();
+
+ int rewrittenDataFilesCount = groups.stream()
+ .mapToInt(group -> group.getDataFiles().size()).sum();
+ int addedDataFilesCount = transaction.getFilesToAddCount();
+ long rewrittenBytesCount = groups.stream().mapToLong(group ->
group.getTotalSize()).sum();
+ int removedDeleteFilesCount = groups.stream()
+ .mapToInt(group -> group.getDeleteFileCount()).sum();
+
+ commitAndInvalidate(transactionManager, transactionId);
Review Comment:
[P1] Cross the durable boundary before invalidating the cache
`transactionManager.commit()` has already committed the Iceberg snapshot,
released the fence, and removed the transaction before `invalidateTableCache()`
runs. If invalidation throws, `committed` is still false, so the finally block
rolls back a nonexistent transaction and this procedure reports failure for a
durable rewrite; a caller may then retry while the FE cache is still stale.
Please mark the transaction committed immediately after the manager commit and
handle invalidation as a best-effort post-commit refresh (as the insert path
does), with a failure-injection test at this boundary.
--
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]
