This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 2fc0e7e137 [python] Reduce file store scan for overwriting on commit
retries (#8359)
2fc0e7e137 is described below
commit 2fc0e7e13753429d32570009016c229c05245873
Author: XiaoHongbo <[email protected]>
AuthorDate: Sat Jun 27 22:41:47 2026 +0800
[python] Reduce file store scan for overwriting on commit retries (#8359)
---
.../pypaimon/tests/overwrite_changes_cache_test.py | 318 +++++++++++++++++++++
.../pypaimon/tests/partition_predicate_test.py | 2 +-
.../write/commit/overwrite_changes_provider.py | 136 +++++++++
paimon-python/pypaimon/write/file_store_commit.py | 45 ++-
4 files changed, 473 insertions(+), 28 deletions(-)
diff --git a/paimon-python/pypaimon/tests/overwrite_changes_cache_test.py
b/paimon-python/pypaimon/tests/overwrite_changes_cache_test.py
new file mode 100644
index 0000000000..c7fb16e628
--- /dev/null
+++ b/paimon-python/pypaimon/tests/overwrite_changes_cache_test.py
@@ -0,0 +1,318 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""Regression test for caching of OVERWRITE changes across commit retries.
+
+On every OVERWRITE retry, pypaimon re-scanned the full target partitions to
+recompute the files to delete. Under concurrent writers this made each retry as
+expensive as the first attempt. OverwriteChangesProvider caches the existing
+files of the target partitions and, on retry, reuses them when the snapshots in
+between are all APPEND and have not touched the target partitions (verified by
a
+cheap DELTA probe), instead of a full re-scan.
+
+The test deterministically forces ``K`` conflicts, each advancing the latest
+snapshot with an append to an unrelated partition, and asserts the full scan
+runs once (not ``K + 1``) and the cache is advanced by a delta probe per retry.
+"""
+
+import os
+import shutil
+import tempfile
+import unittest
+
+import pandas as pd
+import pyarrow as pa
+
+from pypaimon import CatalogFactory, Schema
+from pypaimon.write.commit.overwrite_changes_provider import
OverwriteChangesProvider
+
+
+class OverwriteChangesCacheTest(unittest.TestCase):
+
+ def setUp(self):
+ self.temp_dir = tempfile.mkdtemp(prefix="ow_cache_")
+ self.warehouse = os.path.join(self.temp_dir, 'wh')
+ self.catalog = CatalogFactory.create({"warehouse": self.warehouse})
+ self.catalog.create_database("test_db", True)
+
+ pa_schema = pa.schema([('f0', pa.int32()), ('f1', pa.string())])
+ schema = Schema.from_pyarrow_schema(
+ pa_schema,
+ partition_keys=['f0'],
+ options={'dynamic-partition-overwrite': 'false'},
+ )
+ self.catalog.create_table('test_db.t', schema, False)
+ self.table = self.catalog.get_table('test_db.t')
+
+ # Seed: f0=1 is the overwrite target, f0=2 untouched.
+ self._append(pd.DataFrame({'f0': [1, 1, 2], 'f1': ['a', 'b', 'c']}))
+
+ def tearDown(self):
+ shutil.rmtree(self.temp_dir, ignore_errors=True)
+
+ def _append(self, df):
+ wb = self.table.new_batch_write_builder()
+ w = wb.new_write()
+ c = wb.new_commit()
+ w.write_pandas(df)
+ c.commit(w.prepare_commit())
+ w.close()
+ c.close()
+
+ def test_overwrite_scan_runs_once_not_once_per_retry(self):
+ K = 3 # force 3 conflicts; the 4th attempt wins
+
+ wb = self.table.new_batch_write_builder().overwrite({'f0': 1})
+ w = wb.new_write()
+ c = wb.new_commit()
+ w.write_pandas(pd.DataFrame({'f0': [1], 'f1': ['new']}))
+ messages = w.prepare_commit()
+
+ fsc = c.file_store_commit
+
+ # --- count provider full scans and delta probes (class-level spies)
---
+ counts = {'full_scan': 0, 'probe': 0}
+ orig_full = OverwriteChangesProvider._full_scan
+ orig_probe = OverwriteChangesProvider._delta_touches_target
+
+ def spy_full(self, *a, **k):
+ counts['full_scan'] += 1
+ return orig_full(self, *a, **k)
+
+ def spy_probe(self, *a, **k):
+ counts['probe'] += 1
+ return orig_probe(self, *a, **k)
+
+ # --- inject K CAS conflicts; each advances latest via an APPEND to an
+ # unrelated partition (f0=99) so the delta probe finds the target
+ # (f0=1) untouched and the cache is reused.
+ orig_cas = fsc.snapshot_commit.commit
+ cas = {'fails': 0}
+
+ def patched_cas(snapshot, statistics):
+ if snapshot.commit_kind == "OVERWRITE" and cas['fails'] < K:
+ cas['fails'] += 1
+ self._append(pd.DataFrame({'f0': [99], 'f1':
[f'x{cas["fails"]}']}))
+ return False
+ return orig_cas(snapshot, statistics)
+
+ fsc.snapshot_commit.commit = patched_cas
+ OverwriteChangesProvider._full_scan = spy_full
+ OverwriteChangesProvider._delta_touches_target = spy_probe
+ try:
+ c.commit(messages)
+ c.close()
+ finally:
+ OverwriteChangesProvider._full_scan = orig_full
+ OverwriteChangesProvider._delta_touches_target = orig_probe
+
+ # Harness sanity: we really did force K conflicts and then converged.
+ self.assertEqual(cas['fails'], K, "expected exactly K forced
conflicts")
+
+ print(f"\n[overwrite-cache] K={K} conflicts -> "
+ f"full_scan={counts['full_scan']}, probe={counts['probe']} "
+ f"(target: full_scan=1, probe=K)")
+
+ # #7894: the full overwrite scan runs once; each retry reuses the cache
+ # after a cheap delta probe of the (unrelated) intervening append.
+ self.assertEqual(
+ counts['full_scan'], 1,
+ f"full overwrite scan ran {counts['full_scan']}x; should run once
and "
+ f"reuse the cache on retries")
+ self.assertEqual(
+ counts['probe'], K,
+ f"delta probe ran {counts['probe']}x; should probe once per retry "
+ f"(= K = {K})")
+
+ # Sanity: f0=1 overwritten, f0=2 preserved, f0=99 appended K times.
+ read_builder = self.table.new_read_builder()
+ actual = read_builder.new_read().to_pandas(
+ read_builder.new_scan().plan().splits())
+ self.assertEqual(sorted(actual[actual['f0'] == 1]['f1'].tolist()),
['new'])
+ self.assertEqual(sorted(actual[actual['f0'] == 2]['f1'].tolist()),
['c'])
+ self.assertEqual(len(actual[actual['f0'] == 99]), K)
+
+ def test_cache_rebuilt_when_concurrent_append_hits_target_partition(self):
+ # Concurrent appends hit the overwrite target; probe sees it touched
-> rebuild.
+ K = 3
+
+ wb = self.table.new_batch_write_builder().overwrite({'f0': 1})
+ w = wb.new_write()
+ c = wb.new_commit()
+ w.write_pandas(pd.DataFrame({'f0': [1], 'f1': ['new']}))
+ messages = w.prepare_commit()
+
+ fsc = c.file_store_commit
+
+ counts = {'full_scan': 0, 'probe': 0}
+ orig_full = OverwriteChangesProvider._full_scan
+ orig_probe = OverwriteChangesProvider._delta_touches_target
+
+ def spy_full(self, *a, **k):
+ counts['full_scan'] += 1
+ return orig_full(self, *a, **k)
+
+ def spy_probe(self, *a, **k):
+ counts['probe'] += 1
+ return orig_probe(self, *a, **k)
+
+ orig_cas = fsc.snapshot_commit.commit
+ cas = {'fails': 0}
+
+ def patched_cas(snapshot, statistics):
+ if snapshot.commit_kind == "OVERWRITE" and cas['fails'] < K:
+ cas['fails'] += 1
+ self._append(pd.DataFrame({'f0': [1], 'f1':
[f'y{cas["fails"]}']}))
+ return False
+ return orig_cas(snapshot, statistics)
+
+ fsc.snapshot_commit.commit = patched_cas
+ OverwriteChangesProvider._full_scan = spy_full
+ OverwriteChangesProvider._delta_touches_target = spy_probe
+ try:
+ c.commit(messages)
+ c.close()
+ finally:
+ OverwriteChangesProvider._full_scan = orig_full
+ OverwriteChangesProvider._delta_touches_target = orig_probe
+
+ self.assertEqual(cas['fails'], K, "expected exactly K forced
conflicts")
+
+ # Target touched each retry => cache rebuilds; full scan runs every
attempt.
+ self.assertEqual(counts['full_scan'], K + 1,
+ f"full scan ran {counts['full_scan']}x; cache must
rebuild "
+ f"when the target partition is touched")
+ self.assertEqual(counts['probe'], K,
+ f"delta probe ran {counts['probe']}x; once per retry
(= K)")
+
+ # Overwrite wins: f0=1 is just 'new', f0=2 untouched.
+ read_builder = self.table.new_read_builder()
+ actual = read_builder.new_read().to_pandas(
+ read_builder.new_scan().plan().splits())
+ self.assertEqual(sorted(actual[actual['f0'] == 1]['f1'].tolist()),
['new'])
+ self.assertEqual(sorted(actual[actual['f0'] == 2]['f1'].tolist()),
['c'])
+
+ def _overwrite_partition(self, part_val, f1_val):
+ wb = self.table.new_batch_write_builder().overwrite({'f0': part_val})
+ w = wb.new_write()
+ c = wb.new_commit()
+ w.write_pandas(pd.DataFrame({'f0': [part_val], 'f1': [f1_val]}))
+ c.commit(w.prepare_commit())
+ w.close()
+ c.close()
+
+ def _run_with_conflicts(self, c, messages, K, concurrent_fn):
+ # Run an overwrite commit, forcing K CAS conflicts (each calls
+ # concurrent_fn(i) to advance the latest snapshot). Returns this
commit's
+ # own OverwriteChangesProvider so the caller can read its counters
+ # (only this provider, not any concurrent writer's).
+ fsc = c.file_store_commit
+ captured = {}
+ orig_factory = fsc._overwrite_changes_provider
+
+ def capturing_factory(*a, **k):
+ captured['provider'] = orig_factory(*a, **k)
+ return captured['provider']
+
+ fsc._overwrite_changes_provider = capturing_factory
+
+ orig_cas = fsc.snapshot_commit.commit
+ cas = {'fails': 0}
+
+ def patched_cas(snapshot, statistics):
+ if snapshot.commit_kind == "OVERWRITE" and cas['fails'] < K:
+ cas['fails'] += 1
+ concurrent_fn(cas['fails'])
+ return False
+ return orig_cas(snapshot, statistics)
+
+ fsc.snapshot_commit.commit = patched_cas
+
+ c.commit(messages)
+ c.close()
+
+ self.assertEqual(cas['fails'], K, "expected exactly K forced
conflicts")
+ return captured['provider']
+
+ def test_cache_rebuilt_on_non_append_snapshot(self):
+ # A non-APPEND (OVERWRITE) snapshot between retries forces a rebuild
even
+ # though it only touches an unrelated partition.
+ K = 2
+ wb = self.table.new_batch_write_builder().overwrite({'f0': 1})
+ w = wb.new_write()
+ c = wb.new_commit()
+ w.write_pandas(pd.DataFrame({'f0': [1], 'f1': ['new']}))
+ provider = self._run_with_conflicts(
+ c, w.prepare_commit(), K,
+ lambda i: self._overwrite_partition(99, f'z{i}'))
+
+ self.assertEqual(provider.full_scan_count, K + 1) # rebuilt every
retry
+ self.assertEqual(provider.delta_probe_count, K) # probed, bailed
at kind check
+
+ def test_whole_table_overwrite_always_full_scans(self):
+ # Whole-table overwrite (no partition filter) can never reuse the
cache.
+ K = 2
+ wb = self.table.new_batch_write_builder().overwrite()
+ w = wb.new_write()
+ c = wb.new_commit()
+ w.write_pandas(pd.DataFrame({'f0': [1], 'f1': ['new']}))
+ provider = self._run_with_conflicts(
+ c, w.prepare_commit(), K,
+ lambda i: self._append(pd.DataFrame({'f0': [99], 'f1':
[f'x{i}']})))
+
+ self.assertEqual(provider.full_scan_count, K + 1) # null filter ->
always full scan
+ self.assertEqual(provider.delta_probe_count, 0) # never enters the
probe loop
+
+ read_builder = self.table.new_read_builder()
+ actual = read_builder.new_read().to_pandas(
+ read_builder.new_scan().plan().splits())
+ self.assertEqual(sorted(actual['f1'].tolist()), ['new'])
+
+ def test_dynamic_partition_overwrite_reuses_cache(self):
+ # Dynamic-partition overwrite is scoped to the data's partitions, so an
+ # unrelated concurrent append lets the cache be reused.
+ pa_schema = pa.schema([('f0', pa.int32()), ('f1', pa.string())])
+ schema = Schema.from_pyarrow_schema(pa_schema, partition_keys=['f0'])
+ self.catalog.create_table('test_db.t_dyn', schema, False)
+ table = self.catalog.get_table('test_db.t_dyn')
+
+ def append(df):
+ wb = table.new_batch_write_builder()
+ w = wb.new_write()
+ c = wb.new_commit()
+ w.write_pandas(df)
+ c.commit(w.prepare_commit())
+ w.close()
+ c.close()
+
+ append(pd.DataFrame({'f0': [1, 1, 2], 'f1': ['a', 'b', 'c']}))
+
+ K = 3
+ wb = table.new_batch_write_builder().overwrite() # dynamic: filter
from data (f0=1)
+ w = wb.new_write()
+ c = wb.new_commit()
+ w.write_pandas(pd.DataFrame({'f0': [1], 'f1': ['new']}))
+ provider = self._run_with_conflicts(
+ c, w.prepare_commit(), K,
+ lambda i: append(pd.DataFrame({'f0': [99], 'f1': [f'x{i}']})))
+
+ self.assertEqual(provider.full_scan_count, 1) # target f0=1
untouched -> reuse
+ self.assertEqual(provider.delta_probe_count, K)
+
+
+if __name__ == '__main__':
+ unittest.main()
diff --git a/paimon-python/pypaimon/tests/partition_predicate_test.py
b/paimon-python/pypaimon/tests/partition_predicate_test.py
index fdea8b17e6..9e47757e82 100644
--- a/paimon-python/pypaimon/tests/partition_predicate_test.py
+++ b/paimon-python/pypaimon/tests/partition_predicate_test.py
@@ -174,7 +174,7 @@ class TestOverwritePartitionPredicate(unittest.TestCase):
def _extract_partition_predicate(self, commit):
entries_plan = commit._try_commit.call_args[1]['commit_entries_plan']
- with patch('pypaimon.write.file_store_commit.FileScanner') as mock_cls:
+ with
patch('pypaimon.write.commit.overwrite_changes_provider.FileScanner') as
mock_cls:
mock_cls.return_value.read_manifest_entries.return_value = []
commit.manifest_list_manager.read_all.return_value = []
entries_plan(Mock(id=1))
diff --git a/paimon-python/pypaimon/write/commit/overwrite_changes_provider.py
b/paimon-python/pypaimon/write/commit/overwrite_changes_provider.py
new file mode 100644
index 0000000000..a76d5936f1
--- /dev/null
+++ b/paimon-python/pypaimon/write/commit/overwrite_changes_provider.py
@@ -0,0 +1,136 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+from typing import List, Optional
+
+from pypaimon.manifest.schema.manifest_entry import ManifestEntry
+from pypaimon.read.scanner.file_scanner import FileScanner
+from pypaimon.snapshot.snapshot import Snapshot
+from pypaimon.table.row.generic_row import GenericRow
+
+
+class OverwriteChangesProvider:
+ """Builds the commit entries (DELETE existing + ADD new) for an OVERWRITE,
+ caching the existing files of the target partitions across commit retries
+ to avoid repeated full scans.
+
+ On retry, if the latest snapshot advanced, the cache is reused only when
the
+ snapshots in between are all APPEND and have not touched the target
+ partitions; otherwise it is rebuilt by a full scan. A whole-table overwrite
+ (``partition_filter is None``) always rebuilds. Mirrors Java
+ ``OverwriteChangesProvider`` (#7894).
+ """
+
+ def __init__(self, table, manifest_list_manager, snapshot_manager,
+ partition_filter, commit_messages):
+ self.table = table
+ self.manifest_list_manager = manifest_list_manager
+ self.snapshot_manager = snapshot_manager
+ self.partition_filter = partition_filter
+ self.commit_messages = commit_messages
+
+ self._cached_snapshot: Optional[Snapshot] = None
+ self._cached_entries: List[ManifestEntry] = []
+
+ # Counters for tests / observability (mirrors Java @VisibleForTesting).
+ self.full_scan_count = 0
+ self.delta_probe_count = 0
+
+ def provide(self, latest_snapshot: Optional[Snapshot]) ->
List[ManifestEntry]:
+ if latest_snapshot is None:
+ # Empty table: nothing existing to delete, just add the new files.
+ return self._build_result([])
+
+ if self._cached_snapshot is None:
+ self._cached_entries = self._full_scan(latest_snapshot)
+ self._cached_snapshot = latest_snapshot
+ elif self._cached_snapshot.id > latest_snapshot.id:
+ raise RuntimeError(
+ f"Cached snapshot id {self._cached_snapshot.id} is greater
than "
+ f"latest snapshot id {latest_snapshot.id}")
+ elif self._cached_snapshot.id < latest_snapshot.id:
+ if not self._can_use_cache(latest_snapshot):
+ self._cached_entries = self._full_scan(latest_snapshot)
+ self._cached_snapshot = latest_snapshot
+ # cached_snapshot.id == latest_snapshot.id -> reuse cache as-is
+
+ return self._build_result(self._cached_entries)
+
+ def _full_scan(self, latest_snapshot: Snapshot) -> List[ManifestEntry]:
+ self.full_scan_count += 1
+ return (FileScanner(self.table, lambda: ([], None),
+ partition_predicate=self.partition_filter)
+
.read_manifest_entries(self.manifest_list_manager.read_all(latest_snapshot)))
+
+ def _can_use_cache(self, latest_snapshot: Snapshot) -> bool:
+ if self.partition_filter is None:
+ # Whole-table overwrite: any concurrent commit touches the target,
+ # so skip the delta probe and force a full scan.
+ return False
+ for snapshot_id in range(self._cached_snapshot.id + 1,
latest_snapshot.id + 1):
+ self.delta_probe_count += 1
+ try:
+ snapshot =
self.snapshot_manager.get_snapshot_by_id(snapshot_id)
+ if snapshot is None:
+ return False
+ if snapshot.commit_kind != "APPEND":
+ # Only APPEND snapshots produce a reliable DELTA manifest
for
+ # probing; other kinds may rewrite/reorganize manifests.
+ return False
+ if self._delta_touches_target(snapshot):
+ return False
+ except Exception:
+ # e.g. the snapshot is being expired; a full scan is always
safe.
+ return False
+ return True
+
+ def _delta_touches_target(self, snapshot: Snapshot) -> bool:
+ delta_manifests = self.manifest_list_manager.read_delta(snapshot)
+ if not delta_manifests:
+ return False
+ # Only APPEND snapshots are probed (see _can_use_cache), so the delta
has
+ # no standalone DELETEs; FileScanner's partition predicate prunes at
the
+ # manifest-file level before reading entries.
+ entries = (FileScanner(self.table, lambda: ([], None),
+ partition_predicate=self.partition_filter)
+ .read_manifest_entries(delta_manifests))
+ return len(entries) > 0
+
+ def _build_result(self, existing_entries: List[ManifestEntry]) ->
List[ManifestEntry]:
+ entries = []
+ # Existing files of the target partitions become DELETE entries. Build
+ # fresh entries so the cached (kind=0) entries are never mutated.
+ for entry in existing_entries:
+ entries.append(ManifestEntry(
+ kind=1,
+ partition=entry.partition,
+ bucket=entry.bucket,
+ total_buckets=entry.total_buckets,
+ file=entry.file,
+ ))
+ # New files being written by this overwrite.
+ for msg in self.commit_messages:
+ partition = GenericRow(list(msg.partition),
self.table.partition_keys_fields)
+ for file in msg.new_files:
+ entries.append(ManifestEntry(
+ kind=0,
+ partition=partition,
+ bucket=msg.bucket,
+ total_buckets=self.table.total_buckets,
+ file=file,
+ ))
+ return entries
diff --git a/paimon-python/pypaimon/write/file_store_commit.py
b/paimon-python/pypaimon/write/file_store_commit.py
index a2e41a1217..3149a3678d 100644
--- a/paimon-python/pypaimon/write/file_store_commit.py
+++ b/paimon-python/pypaimon/write/file_store_commit.py
@@ -30,7 +30,6 @@ from pypaimon.manifest.schema.data_file_meta import
DataFileMeta
from pypaimon.manifest.schema.manifest_entry import ManifestEntry
from pypaimon.manifest.schema.manifest_file_meta import ManifestFileMeta
-from pypaimon.read.scanner.file_scanner import FileScanner
from pypaimon.snapshot.snapshot import Snapshot
from pypaimon.snapshot.snapshot_commit import (PartitionStatistics,
SnapshotCommit)
@@ -39,6 +38,7 @@ from pypaimon.table.row.offset_row import OffsetRow
from pypaimon.write.commit.commit_rollback import CommitRollback
from pypaimon.write.commit.commit_scanner import CommitScanner
from pypaimon.write.commit.conflict_detection import ConflictDetection
+from pypaimon.write.commit.overwrite_changes_provider import
OverwriteChangesProvider
from pypaimon.table.special_fields import SpecialFields
from pypaimon.write.commit_callback import CommitCallback,
CommitCallbackContext
from pypaimon.write.commit_message import CommitMessage
@@ -217,11 +217,11 @@ class FileStoreCommit:
changelog_entries = self._collect_changelog_entries(commit_messages)
if not skip_overwrite:
+ provider = self._overwrite_changes_provider(partition_filter,
commit_messages)
self._try_commit(
commit_kind="OVERWRITE",
commit_identifier=commit_identifier,
- commit_entries_plan=lambda snapshot:
self._generate_overwrite_entries(
- snapshot, partition_filter, commit_messages),
+ commit_entries_plan=provider.provide,
changelog_entries=changelog_entries,
detect_conflicts=True,
allow_rollback=False,
@@ -260,22 +260,22 @@ class FileStoreCommit:
partition_filter =
predicate_builder.or_predicates(partition_predicates)
+ provider = self._overwrite_changes_provider(partition_filter, [])
self._try_commit(
commit_kind="OVERWRITE",
commit_identifier=commit_identifier,
- commit_entries_plan=lambda snapshot:
self._generate_overwrite_entries(
- snapshot, partition_filter, []),
+ commit_entries_plan=provider.provide,
detect_conflicts=True,
allow_rollback=False,
)
def truncate_table(self, commit_identifier: int) -> None:
"""Truncate the entire table, deleting all data."""
+ provider = self._overwrite_changes_provider(None, [])
self._try_commit(
commit_kind="OVERWRITE",
commit_identifier=commit_identifier,
- commit_entries_plan=lambda snapshot:
self._generate_overwrite_entries(
- snapshot, None, []),
+ commit_entries_plan=provider.provide,
detect_conflicts=True,
allow_rollback=False,
)
@@ -579,26 +579,17 @@ class FileStoreCommit:
f"in {msg.partition} does not belong to
this partition")
return partition_filter
- def _generate_overwrite_entries(self, latest_snapshot, partition_filter,
commit_messages):
- """Generate commit entries for OVERWRITE mode based on latest
snapshot."""
- entries = []
- current_entries = [] if latest_snapshot is None \
- else (FileScanner(self.table, lambda: ([], None),
partition_predicate=partition_filter).
-
read_manifest_entries(self.manifest_list_manager.read_all(latest_snapshot)))
- for entry in current_entries:
- entry.kind = 1 # DELETE
- entries.append(entry)
- for msg in commit_messages:
- partition = GenericRow(list(msg.partition),
self.table.partition_keys_fields)
- for file in msg.new_files:
- entries.append(ManifestEntry(
- kind=0, # ADD
- partition=partition,
- bucket=msg.bucket,
- total_buckets=self.table.total_buckets,
- file=file
- ))
- return entries
+ def _overwrite_changes_provider(self, partition_filter, commit_messages):
+ """Build a stateful provider of OVERWRITE commit entries that caches
the
+ existing files of the target partitions across retries (see
+ OverwriteChangesProvider). One instance per overwrite operation."""
+ return OverwriteChangesProvider(
+ self.table,
+ self.manifest_list_manager,
+ self.snapshot_manager,
+ partition_filter,
+ commit_messages,
+ )
def _commit_retry_wait(self, retry_count: int):
