Zouxxyy commented on code in PR #8826:
URL: https://github.com/apache/paimon/pull/8826#discussion_r3655887919
##########
paimon-python/pypaimon/ray/data_evolution_merge_join.py:
##########
@@ -481,7 +537,10 @@ def distributed_update_apply(
)
sorted_first_row_ids = list(planner.first_row_ids)
if not sorted_first_row_ids:
- return [], 0, []
+ # No target file groups (e.g. an existing but empty snapshot). Match
the
+ # 4-tuple contract so callers that unpack (msgs, num, row_ids, err) --
+ # including merge_into's NOT MATCHED insert path -- don't crash.
+ return [], 0, [], None
Review Comment:
Returning a 4-tuple fixes the unpacking crash, but it does not make an
existing-but-empty snapshot follow the empty-target path. `_build_datasets`
still passes `target_empty=base_snapshot is None`; for an empty snapshot,
`base_snapshot` is non-null, so the NOT MATCHED insert path builds a left-anti
join against a zero-block target. The new test fails on Python 3.10 and 3.11
with `ArrowInvalid`. Please also treat `base_snapshot.total_record_count == 0`
as empty, and skip the matched branches, so this regression passes the
supported CI matrix.
##########
paimon-python/pypaimon/write/file_store_commit.py:
##########
@@ -612,26 +759,84 @@ def _try_commit_once(self, retry_result:
Optional[RetryResult], commit_kind: str
logger.warning(f"Exception occurs when preparing snapshot: {e}",
exc_info=True)
raise RuntimeError(f"Failed to prepare snapshot: {e}")
- # Use SnapshotCommit for atomic commit
+ def clean_up_rejected_commit():
+ try:
+ self._clean_up_reuse_tmp_manifests(
+ delta_manifest_list,
+ changelog_manifest_list_name,
+ new_index_manifest,
+ )
+ self._clean_up_no_reuse_tmp_manifests(
+ base_manifest_list, merge_new_files)
+ except Exception:
+ logger.warning(
+ "Failed to clean up rejected commit manifests.",
+ exc_info=True,
+ )
+
+ # Mirror a ``with`` block but drive it manually so only commit()
decides
+ # the outcome. __enter__ and commit() share one capture, so an
__enter__
+ # failure flows through the classification below instead of bypassing
it.
+ # __exit__ gets the real commit exception triple (for extension
rollback)
+ # but its own failure is only logged -- never allowed to override the
+ # commit outcome (a lost/landed commit must not be misclassified).
+ success = None
+ commit_exc = None
+ entered = False
try:
- with self.snapshot_commit:
- success = self.snapshot_commit.commit(snapshot_data,
statistics)
- if not success:
- commit_time_s = (int(time.time() * 1000) - start_millis) /
1000
+ self.snapshot_commit.__enter__()
+ entered = True
+ success = self.snapshot_commit.commit(snapshot_data, statistics)
+ except Exception as e:
+ commit_exc = e
+ finally:
+ # Never call __exit__ without a successful __enter__ (mirrors
+ # ``with``); the outcome is still classified from commit_exc below.
+ if entered:
+ exit_exc_info = (
+ (type(commit_exc), commit_exc, commit_exc.__traceback__)
+ if commit_exc is not None else (None, None, None)
+ )
+ try:
+ self.snapshot_commit.__exit__(*exit_exc_info)
+ except Exception:
logger.warning(
- "Atomic commit failed for snapshot #%d by user %s "
- "with identifier %s and kind %s after %.0f seconds.
Try again.",
- new_snapshot_id,
- self.commit_user,
- commit_identifier,
- commit_kind,
- commit_time_s,
+ "Failed to close snapshot commit; ignoring because it "
+ "must not override the commit outcome.",
+ exc_info=True,
)
- return RetryResult(latest_snapshot, None,
base_data_files=base_data_files)
- except Exception as e:
+
+ if commit_exc is not None:
+ if _is_deterministic_atomic_commit_failure(commit_exc):
+ logger.warning(
+ "Atomic commit was rejected deterministically; do not
retry.",
+ exc_info=commit_exc,
+ )
+ clean_up_rejected_commit()
+ raise commit_exc
Review Comment:
Once this branch classifies the rejection as deterministic, the staged
`CommitMessage` files are safe to abort. Rethrowing the raw exception loses
that signal: `TableCommit._commit` only aborts on `CommitConflictError`, while
`merge_into` and `RayDataSink` stop aborting after commit starts, so
400/403/404 rejections can leave data, changelog, and index files orphaned.
Please propagate a safe-to-abort rejection type or marker to the layer that
owns the messages, while preserving the original cause, and add a caller-level
test showing that no manual `abort()` is required.
--
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]
