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 d50becf97d [python] Reject or drop stale index when updating globally
indexed columns (#8045)
d50becf97d is described below
commit d50becf97d0ef1e9633978cdf4cae2c5c1387120
Author: XiaoHongbo <[email protected]>
AuthorDate: Sun May 31 21:23:47 2026 +0800
[python] Reject or drop stale index when updating globally indexed columns
(#8045)
---
.../pypaimon/common/options/core_options.py | 14 +++
.../pypaimon/manifest/index_manifest_file.py | 119 +++++++++++++++++++-
.../pypaimon/tests/e2e/java_py_read_write_test.py | 70 +++++++++++-
.../pypaimon/tests/index_manifest_write_test.py | 121 +++++++++++++++++++++
paimon-python/pypaimon/write/commit_message.py | 10 +-
paimon-python/pypaimon/write/file_store_commit.py | 56 ++++++++--
.../pypaimon/write/global_index_update_checker.py | 82 ++++++++++++++
paimon-python/pypaimon/write/table_update.py | 6 -
8 files changed, 459 insertions(+), 19 deletions(-)
diff --git a/paimon-python/pypaimon/common/options/core_options.py
b/paimon-python/pypaimon/common/options/core_options.py
index 1cb9e831d3..bb157bb6e3 100644
--- a/paimon-python/pypaimon/common/options/core_options.py
+++ b/paimon-python/pypaimon/common/options/core_options.py
@@ -56,6 +56,11 @@ class MergeEngine(str, Enum):
FIRST_ROW = "first-row"
+class GlobalIndexColumnUpdateAction(str, Enum):
+ THROW_ERROR = "THROW_ERROR"
+ DROP_PARTITION_INDEX = "DROP_PARTITION_INDEX"
+
+
class CoreOptions:
"""Core options for Paimon tables."""
# File format constants
@@ -495,6 +500,12 @@ class CoreOptions:
)
)
+ GLOBAL_INDEX_COLUMN_UPDATE_ACTION:
ConfigOption[GlobalIndexColumnUpdateAction] = (
+ ConfigOptions.key("global-index.column-update-action")
+ .enum_type(GlobalIndexColumnUpdateAction)
+ .default_value(GlobalIndexColumnUpdateAction.THROW_ERROR)
+ )
+
LOCAL_CACHE_ENABLED: ConfigOption[bool] = (
ConfigOptions.key("local-cache.enabled")
.boolean_type()
@@ -785,6 +796,9 @@ class CoreOptions:
def data_evolution_enabled(self, default=None):
return self.options.get(CoreOptions.DATA_EVOLUTION_ENABLED, default)
+ def global_index_column_update_action(self, default=None):
+ return self.options.get(CoreOptions.GLOBAL_INDEX_COLUMN_UPDATE_ACTION,
default)
+
def deletion_vectors_enabled(self, default=None):
return self.options.get(CoreOptions.DELETION_VECTORS_ENABLED, default)
diff --git a/paimon-python/pypaimon/manifest/index_manifest_file.py
b/paimon-python/pypaimon/manifest/index_manifest_file.py
index 4e65e95e0c..a7d82d4b12 100644
--- a/paimon-python/pypaimon/manifest/index_manifest_file.py
+++ b/paimon-python/pypaimon/manifest/index_manifest_file.py
@@ -15,6 +15,7 @@
# specific language governing permissions and limitations
# under the License.
+import uuid
from io import BytesIO
from typing import List, Optional
@@ -24,11 +25,61 @@ from pypaimon.globalindex.global_index_meta import
GlobalIndexMeta
from pypaimon.index.deletion_vector_meta import DeletionVectorMeta
from pypaimon.index.index_file_meta import IndexFileMeta
from pypaimon.manifest.index_manifest_entry import IndexManifestEntry
-from pypaimon.table.row.generic_row import GenericRowDeserializer
+from pypaimon.table.row.generic_row import (GenericRowDeserializer,
+ GenericRowSerializer)
+from pypaimon.utils.file_store_path_factory import FileStorePathFactory
+
+# DV and global-index sub-schemas required by INDEX_MANIFEST_ENTRY_SCHEMA for
+# Avro compatibility with Java; values are always null in data-evolution
tables.
+_DELETION_VECTOR_META_SCHEMA = {
+ "type": "record",
+ "name": "DeletionVectorMeta",
+ "fields": [
+ {"name": "f0", "type": "string"},
+ {"name": "f1", "type": "long"},
+ {"name": "f2", "type": "int"},
+ {"name": "_CARDINALITY", "type": ["null", "long"], "default": None},
+ ],
+}
+
+_GLOBAL_INDEX_META_SCHEMA = {
+ "type": "record",
+ "name": "GlobalIndexMeta",
+ "fields": [
+ {"name": "_ROW_RANGE_START", "type": "long"},
+ {"name": "_ROW_RANGE_END", "type": "long"},
+ {"name": "_INDEX_FIELD_ID", "type": "int"},
+ {"name": "_EXTRA_FIELD_IDS",
+ "type": ["null", {"type": "array", "items": "int"}], "default": None},
+ {"name": "_INDEX_META", "type": ["null", "bytes"], "default": None},
+ ],
+}
+
+INDEX_MANIFEST_ENTRY_SCHEMA = {
+ "type": "record",
+ "name": "IndexManifestEntry",
+ "fields": [
+ {"name": "_VERSION", "type": "int"},
+ {"name": "_KIND", "type": "int"},
+ {"name": "_PARTITION", "type": "bytes"},
+ {"name": "_BUCKET", "type": "int"},
+ {"name": "_INDEX_TYPE", "type": "string"},
+ {"name": "_FILE_NAME", "type": "string"},
+ {"name": "_FILE_SIZE", "type": "long"},
+ {"name": "_ROW_COUNT", "type": "long"},
+ {"name": "_DELETIONS_VECTORS_RANGES",
+ "type": ["null", {"type": "array", "items":
_DELETION_VECTOR_META_SCHEMA}],
+ "default": None},
+ {"name": "_EXTERNAL_PATH", "type": ["null", "string"], "default":
None},
+ {"name": "_GLOBAL_INDEX",
+ "type": ["null", _GLOBAL_INDEX_META_SCHEMA], "default": None},
+ ],
+}
+
+_INDEX_ENTRY_VERSION = 1
class IndexManifestFile:
- """Index manifest file reader for reading index manifest entries."""
DELETION_VECTORS_INDEX = "DELETION_VECTORS"
@@ -172,5 +223,69 @@ class IndexManifestFile:
row_range_start=global_index_record.get('_ROW_RANGE_START', 0),
row_range_end=global_index_record.get('_ROW_RANGE_END', 0),
index_field_id=global_index_record.get('_INDEX_FIELD_ID', 0),
+ extra_field_ids=global_index_record.get('_EXTRA_FIELD_IDS'),
index_meta=global_index_record.get('_INDEX_META')
)
+
+ def combine_deletes(
+ self,
+ previous_name: Optional[str],
+ deletes: List[IndexManifestEntry],
+ ) -> Optional[str]:
+ if not deletes:
+ return previous_name
+ previous = self.read(previous_name) if previous_name else []
+ delete_names = {e.index_file.file_name for e in deletes}
+ survivors = [e for e in previous if e.index_file.file_name not in
delete_names]
+ if not survivors:
+ return None
+ return self.write(survivors)
+
+ def write(self, entries: List[IndexManifestEntry]) -> str:
+ file_name =
f"{FileStorePathFactory.INDEX_MANIFEST_PREFIX}{uuid.uuid4()}"
+ path = f"{self.manifest_path}/{file_name}"
+ records = [self._to_avro_record(e) for e in entries]
+ try:
+ buffer = BytesIO()
+ fastavro.writer(buffer, INDEX_MANIFEST_ENTRY_SCHEMA, records)
+ with self.file_io.new_output_stream(path) as output_stream:
+ output_stream.write(buffer.getvalue())
+ except Exception as e:
+ self.file_io.delete_quietly(path)
+ raise RuntimeError(
+ f"Exception occurs when writing records to {path}. Clean up."
+ ) from e
+ return file_name
+
+ def _to_avro_record(self, entry: IndexManifestEntry) -> dict:
+ index_file = entry.index_file
+ dv_ranges = None
+ if index_file.dv_ranges:
+ dv_ranges = [
+ {"f0": dv.data_file_name, "f1": dv.offset, "f2": dv.length,
+ "_CARDINALITY": dv.cardinality}
+ for dv in index_file.dv_ranges.values()
+ ]
+ global_index = None
+ if index_file.global_index_meta is not None:
+ gim = index_file.global_index_meta
+ global_index = {
+ "_ROW_RANGE_START": gim.row_range_start,
+ "_ROW_RANGE_END": gim.row_range_end,
+ "_INDEX_FIELD_ID": gim.index_field_id,
+ "_EXTRA_FIELD_IDS": gim.extra_field_ids,
+ "_INDEX_META": gim.index_meta,
+ }
+ return {
+ "_VERSION": _INDEX_ENTRY_VERSION,
+ "_KIND": entry.kind,
+ "_PARTITION": GenericRowSerializer.to_bytes(entry.partition),
+ "_BUCKET": entry.bucket,
+ "_INDEX_TYPE": index_file.index_type,
+ "_FILE_NAME": index_file.file_name,
+ "_FILE_SIZE": index_file.file_size,
+ "_ROW_COUNT": index_file.row_count,
+ "_DELETIONS_VECTORS_RANGES": dv_ranges,
+ "_EXTERNAL_PATH": index_file.external_path,
+ "_GLOBAL_INDEX": global_index,
+ }
diff --git a/paimon-python/pypaimon/tests/e2e/java_py_read_write_test.py
b/paimon-python/pypaimon/tests/e2e/java_py_read_write_test.py
index d01adb26ba..1016f8135b 100644
--- a/paimon-python/pypaimon/tests/e2e/java_py_read_write_test.py
+++ b/paimon-python/pypaimon/tests/e2e/java_py_read_write_test.py
@@ -443,7 +443,9 @@ class JavaPyReadWriteTest(unittest.TestCase):
self._test_read_btree_index_generic("test_btree_index_bigint", 2000,
pa.int64())
self._test_read_btree_index_large()
self._test_read_btree_index_null()
- self._test_index_manifest_inherited_after_write()
+ self._test_partial_append_does_not_trigger_index_action()
+ if sys.version_info[:2] >= (3, 7):
+ self._test_index_manifest_inherited_after_write()
def _test_read_btree_index_generic(self, table_name: str, k, k_type):
table = self.catalog.get_table('default.' + table_name)
@@ -565,6 +567,72 @@ class JavaPyReadWriteTest(unittest.TestCase):
"index_manifest lost after Python data write - indexes become
invisible"
)
+ read_builder = table.new_read_builder()
+ predicate_builder = read_builder.new_predicate_builder()
+ read_builder.with_filter(predicate_builder.equal('k', 'k2'))
+ read_builder.with_projection(['k', '_ROW_ID'])
+ splits = read_builder.new_scan().plan().splits()
+ row_ids =
read_builder.new_read().to_arrow(splits)['_ROW_ID'].to_pylist()
+ self.assertTrue(len(row_ids) > 0, "k2 should exist before update")
+
+ wb = table.new_batch_write_builder()
+ tu = wb.new_update().with_update_type(['k'])
+ update_data = pa.table({
+ '_ROW_ID': pa.array(row_ids, type=pa.int64()),
+ 'k': ['k_updated'] * len(row_ids),
+ })
+ msgs = tu.update_by_arrow_with_row_id(update_data)
+ with self.assertRaises(RuntimeError) as cm:
+ wb.new_commit().commit(msgs)
+ self.assertIn("'k'", str(cm.exception))
+ self.assertIn("Conflicted columns", str(cm.exception))
+
+ table_drop = table.copy(
+ {'global-index.column-update-action': 'DROP_PARTITION_INDEX'}
+ )
+ wb_drop = table_drop.new_batch_write_builder()
+ tu_drop = wb_drop.new_update().with_update_type(['k'])
+
wb_drop.new_commit().commit(tu_drop.update_by_arrow_with_row_id(update_data))
+
+ table_after = self.catalog.get_table('default.test_btree_index_string')
+ rb = table_after.new_read_builder()
+ rb.with_filter(rb.new_predicate_builder().equal('k', 'k_updated'))
+ rows_new = rb.new_read().to_arrow(rb.new_scan().plan().splits())
+ self.assertGreater(len(rows_new), 0,
+ "after DROP_PARTITION_INDEX, new value should read")
+
+ from pypaimon.manifest.index_manifest_file import IndexManifestFile
+ snap = table_after.snapshot_manager().get_latest_snapshot()
+ entries = (IndexManifestFile(table_after).read(snap.index_manifest)
+ if snap.index_manifest else [])
+ field_by_id = {f.id: f.name for f in table_after.fields}
+ remaining = [e for e in entries
+ if e.index_file.global_index_meta is not None
+ and field_by_id.get(
+ e.index_file.global_index_meta.index_field_id) == 'k']
+ self.assertEqual(remaining, [],
+ "btree index entries for 'k' should be dropped")
+
+ def _test_partial_append_does_not_trigger_index_action(self):
+ table = self.catalog.get_table('default.test_btree_index_string')
+ snap_before = table.snapshot_manager().get_latest_snapshot()
+
+ wb = table.new_batch_write_builder()
+ tw = wb.new_write()
+ tw.with_write_type(['k'])
+ tw.write_arrow(pa.table({'k': ['k_new']}))
+ tc = wb.new_commit()
+ tc.commit(tw.prepare_commit())
+ tw.close()
+ tc.close()
+
+ snap_after = table.snapshot_manager().get_latest_snapshot()
+ self.assertGreater(snap_after.id, snap_before.id)
+ self.assertIsNotNone(
+ snap_after.index_manifest,
+ "partial append should not drop index manifest"
+ )
+
@parameterized.expand([('json',), ('csv',)])
def test_read_compressed_text_append_table(self, file_format):
table = self.catalog.get_table(
diff --git a/paimon-python/pypaimon/tests/index_manifest_write_test.py
b/paimon-python/pypaimon/tests/index_manifest_write_test.py
new file mode 100644
index 0000000000..ff61750407
--- /dev/null
+++ b/paimon-python/pypaimon/tests/index_manifest_write_test.py
@@ -0,0 +1,121 @@
+# 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.
+
+import os
+import shutil
+import tempfile
+import unittest
+import uuid
+
+import pyarrow as pa
+
+from pypaimon import CatalogFactory, Schema
+from pypaimon.globalindex.global_index_meta import GlobalIndexMeta
+from pypaimon.index.index_file_meta import IndexFileMeta
+from pypaimon.manifest.index_manifest_entry import IndexManifestEntry
+from pypaimon.manifest.index_manifest_file import IndexManifestFile
+from pypaimon.table.row.generic_row import GenericRow
+
+
+class IndexManifestWriteTest(unittest.TestCase):
+
+ pa_schema = pa.schema([
+ ('id', pa.int32()),
+ ('vec', pa.string()),
+ ])
+
+ @classmethod
+ def setUpClass(cls):
+ cls.tempdir = tempfile.mkdtemp()
+ cls.warehouse = os.path.join(cls.tempdir, 'warehouse')
+ cls.catalog = CatalogFactory.create({'warehouse': cls.warehouse})
+ cls.catalog.create_database('default', True)
+
+ @classmethod
+ def tearDownClass(cls):
+ shutil.rmtree(cls.tempdir, ignore_errors=True)
+
+ def _table(self):
+ name = f'default.idx_{uuid.uuid4().hex[:8]}'
+ s = Schema.from_pyarrow_schema(self.pa_schema)
+ self.catalog.create_table(name, s, False)
+ return self.catalog.get_table(name)
+
+ def _entry(self, file_name, field_id, meta=b'm'):
+ partition = GenericRow([], [])
+ index_file = IndexFileMeta(
+ index_type='BTREE',
+ file_name=file_name,
+ file_size=123,
+ row_count=10,
+ global_index_meta=GlobalIndexMeta(
+ row_range_start=0,
+ row_range_end=10,
+ index_field_id=field_id,
+ extra_field_ids=[field_id + 1],
+ index_meta=meta,
+ ),
+ )
+ return IndexManifestEntry(kind=0, partition=partition, bucket=0,
index_file=index_file)
+
+ def test_write_read_roundtrip(self):
+ imf = IndexManifestFile(self._table())
+ name = imf.write([self._entry('idx-a', 1), self._entry('idx-b', 2)])
+ out = imf.read(name)
+ self.assertEqual(2, len(out))
+ by_name = {e.index_file.file_name: e for e in out}
+ a = by_name['idx-a']
+ self.assertEqual('BTREE', a.index_file.index_type)
+ self.assertEqual(123, a.index_file.file_size)
+ self.assertEqual(10, a.index_file.row_count)
+ self.assertEqual(0, a.kind)
+ gim = a.index_file.global_index_meta
+ self.assertEqual(1, gim.index_field_id)
+ self.assertEqual(0, gim.row_range_start)
+ self.assertEqual(10, gim.row_range_end)
+ self.assertEqual([2], gim.extra_field_ids)
+ self.assertEqual(b'm', bytes(gim.index_meta))
+
+ def test_combine_drops_named_files(self):
+ imf = IndexManifestFile(self._table())
+ previous = imf.write([self._entry('idx-a', 1), self._entry('idx-b',
2)])
+ deletes = [self._entry('idx-a', 1)]
+ new_name = imf.combine_deletes(previous, deletes)
+ self.assertNotEqual(previous, new_name)
+ survivors = {e.index_file.file_name for e in imf.read(new_name)}
+ self.assertEqual({'idx-b'}, survivors)
+
+ def test_combine_unknown_delete_is_noop_on_content(self):
+ imf = IndexManifestFile(self._table())
+ previous = imf.write([self._entry('idx-a', 1)])
+ new_name = imf.combine_deletes(previous, [self._entry('idx-zzz', 9)])
+ survivors = {e.index_file.file_name for e in imf.read(new_name)}
+ self.assertEqual({'idx-a'}, survivors)
+
+ def test_combine_empty_deletes_returns_previous(self):
+ imf = IndexManifestFile(self._table())
+ previous = imf.write([self._entry('idx-a', 1)])
+ self.assertEqual(previous, imf.combine_deletes(previous, []))
+
+ def test_combine_all_deleted_returns_none(self):
+ imf = IndexManifestFile(self._table())
+ previous = imf.write([self._entry('idx-a', 1)])
+ self.assertIsNone(imf.combine_deletes(previous, [self._entry('idx-a',
1)]))
+
+
+if __name__ == '__main__':
+ unittest.main()
diff --git a/paimon-python/pypaimon/write/commit_message.py
b/paimon-python/pypaimon/write/commit_message.py
index 7bce06d8ab..552df0000f 100644
--- a/paimon-python/pypaimon/write/commit_message.py
+++ b/paimon-python/pypaimon/write/commit_message.py
@@ -15,11 +15,14 @@
# specific language governing permissions and limitations
# under the License.
-from dataclasses import dataclass
-from typing import List, Tuple, Optional
+from dataclasses import dataclass, field
+from typing import List, Tuple, Optional, TYPE_CHECKING
from pypaimon.manifest.schema.data_file_meta import DataFileMeta
+if TYPE_CHECKING:
+ from pypaimon.manifest.index_manifest_entry import IndexManifestEntry
+
@dataclass
class CommitMessage:
@@ -27,6 +30,7 @@ class CommitMessage:
bucket: int
new_files: List[DataFileMeta]
check_from_snapshot: Optional[int] = -1
+ index_deletes: List['IndexManifestEntry'] = field(default_factory=list)
def is_empty(self):
- return not self.new_files
+ return not self.new_files and not self.index_deletes
diff --git a/paimon-python/pypaimon/write/file_store_commit.py
b/paimon-python/pypaimon/write/file_store_commit.py
index b5c9976aad..41440ae862 100644
--- a/paimon-python/pypaimon/write/file_store_commit.py
+++ b/paimon-python/pypaimon/write/file_store_commit.py
@@ -143,6 +143,31 @@ class FileStoreCommit:
logger.info("Finished collecting changes, including: %d entries",
len(commit_entries))
+ index_deletes = []
+ for msg in commit_messages:
+ index_deletes.extend(msg.index_deletes)
+
+ if not index_deletes:
+ from pypaimon.write.global_index_update_checker import (
+ apply_global_index_update_action,
+ )
+ updated_cols = set()
+ written_partitions = set()
+ for msg in commit_messages:
+ if msg.check_from_snapshot == -1:
+ continue
+ for f in msg.new_files:
+ if f.write_cols:
+ updated_cols.update(f.write_cols)
+ written_partitions.add(msg.partition)
+ if updated_cols:
+ snapshot = self.snapshot_manager.get_latest_snapshot()
+ index_msgs = apply_global_index_update_action(
+ self.table, snapshot, list(updated_cols),
written_partitions,
+ )
+ for m in index_msgs:
+ index_deletes.extend(m.index_deletes)
+
commit_kind = "APPEND"
detect_conflicts = False
allow_rollback = False
@@ -158,7 +183,8 @@ class FileStoreCommit:
commit_identifier=commit_identifier,
commit_entries_plan=lambda snapshot: commit_entries,
detect_conflicts=detect_conflicts,
- allow_rollback=allow_rollback)
+ allow_rollback=allow_rollback,
+ index_deletes=index_deletes)
def overwrite(self, overwrite_partition, commit_messages:
List[CommitMessage], commit_identifier: int):
"""Commit the given commit messages in overwrite mode."""
@@ -244,7 +270,7 @@ class FileStoreCommit:
)
def _try_commit(self, commit_kind, commit_identifier, commit_entries_plan,
- detect_conflicts=False, allow_rollback=False):
+ detect_conflicts=False, allow_rollback=False,
index_deletes=None):
retry_count = 0
retry_result = None
@@ -255,7 +281,7 @@ class FileStoreCommit:
# No entries to commit (e.g. drop_partitions with no matching
data): skip commit
# to avoid creating manifest/snapshot with empty partition_stats
(causes read errors).
- if not commit_entries:
+ if not commit_entries and not index_deletes:
break
result = self._try_commit_once(
@@ -266,6 +292,7 @@ class FileStoreCommit:
latest_snapshot=latest_snapshot,
detect_conflicts=detect_conflicts,
allow_rollback=allow_rollback,
+ index_deletes=index_deletes,
)
if result.is_success():
@@ -317,7 +344,8 @@ class FileStoreCommit:
commit_entries: List[ManifestEntry],
commit_identifier: int,
latest_snapshot: Optional[Snapshot],
detect_conflicts: bool = False,
- allow_rollback: bool = False) -> CommitResult:
+ allow_rollback: bool = False,
+ index_deletes=None) -> CommitResult:
start_millis = int(time.time() * 1000)
if self._is_duplicate_commit(retry_result, latest_snapshot,
commit_identifier, commit_kind):
return SuccessResult()
@@ -328,6 +356,7 @@ class FileStoreCommit:
# process new_manifest
new_manifest_file = f"manifest-{str(uuid.uuid4())}-0"
+ new_index_manifest = None
# process snapshot
new_snapshot_id = latest_snapshot.id + 1 if latest_snapshot else 1
@@ -378,6 +407,13 @@ class FileStoreCommit:
index_manifest = None
if latest_snapshot and commit_kind == "APPEND":
index_manifest = latest_snapshot.index_manifest
+ if index_deletes:
+ from pypaimon.manifest.index_manifest_file import
IndexManifestFile
+ previous_index_manifest = index_manifest
+ index_manifest = IndexManifestFile(self.table).combine_deletes(
+ previous_index_manifest, index_deletes)
+ if index_manifest != previous_index_manifest:
+ new_index_manifest = index_manifest
snapshot_data = Snapshot(
version=3,
@@ -397,7 +433,8 @@ class FileStoreCommit:
# Generate partition statistics for the commit
statistics = self._generate_partition_statistics(commit_entries)
except Exception as e:
- self._cleanup_preparation_failure(delta_manifest_list,
base_manifest_list)
+ self._cleanup_preparation_failure(delta_manifest_list,
base_manifest_list,
+ new_index_manifest)
logger.warning(f"Exception occurs when preparing snapshot: {e}",
exc_info=True)
raise RuntimeError(f"Failed to prepare snapshot: {e}")
@@ -417,7 +454,8 @@ class FileStoreCommit:
commit_kind,
commit_time_s,
)
- self._cleanup_preparation_failure(delta_manifest_list,
base_manifest_list)
+ self._cleanup_preparation_failure(delta_manifest_list,
base_manifest_list,
+ new_index_manifest)
return RetryResult(latest_snapshot, None)
except Exception as e:
# Commit exception, not sure about the situation and should not
clean up the files
@@ -598,10 +636,14 @@ class FileStoreCommit:
def _cleanup_preparation_failure(self,
delta_manifest_list: Optional[str],
- base_manifest_list: Optional[str]):
+ base_manifest_list: Optional[str],
+ index_manifest: Optional[str] = None):
try:
manifest_path = self.manifest_list_manager.manifest_path
+ if index_manifest:
+
self.table.file_io.delete_quietly(f"{manifest_path}/{index_manifest}")
+
if delta_manifest_list:
manifest_files =
self.manifest_list_manager.read(delta_manifest_list)
for manifest_meta in manifest_files:
diff --git a/paimon-python/pypaimon/write/global_index_update_checker.py
b/paimon-python/pypaimon/write/global_index_update_checker.py
new file mode 100644
index 0000000000..ac405fc4bc
--- /dev/null
+++ b/paimon-python/pypaimon/write/global_index_update_checker.py
@@ -0,0 +1,82 @@
+# 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 Sequence, Set, Tuple
+
+from pypaimon.common.options.core_options import GlobalIndexColumnUpdateAction
+
+
+def scan_global_index_entries(table, snapshot):
+ from pypaimon.index.index_file_handler import IndexFileHandler
+
+ handler = IndexFileHandler(table=table)
+ return handler.scan(
+ snapshot, lambda e: e.index_file.global_index_meta is not None
+ )
+
+
+def build_index_delete_msgs(entries) -> list:
+ from pypaimon.manifest.index_manifest_entry import IndexManifestEntry
+ from pypaimon.write.commit_message import CommitMessage
+
+ by_partition = {}
+ for e in entries:
+ key = tuple(e.partition.values)
+ by_partition.setdefault(key, []).append(
+ IndexManifestEntry(
+ kind=1, partition=e.partition, bucket=e.bucket,
index_file=e.index_file
+ )
+ )
+ return [
+ CommitMessage(partition=key, bucket=0, new_files=[],
index_deletes=dels)
+ for key, dels in by_partition.items()
+ ]
+
+
+def apply_global_index_update_action(
+ table,
+ snapshot,
+ updated_cols: Sequence[str],
+ written_partitions: Set[Tuple],
+) -> list:
+ if snapshot is None or not updated_cols or not written_partitions:
+ return []
+ entries = scan_global_index_entries(table, snapshot)
+ if not entries:
+ return []
+ field_by_id = {f.id: f.name for f in table.fields}
+ update_set = set(updated_cols)
+ affected = [
+ e for e in entries
+ if field_by_id.get(e.index_file.global_index_meta.index_field_id) in
update_set
+ and tuple(e.partition.values) in written_partitions
+ ]
+ if not affected:
+ return []
+ action = table.options.global_index_column_update_action()
+ if action is None:
+ action = GlobalIndexColumnUpdateAction.THROW_ERROR
+ if action == GlobalIndexColumnUpdateAction.DROP_PARTITION_INDEX:
+ return build_index_delete_msgs(affected)
+ conflicted = sorted(
+ {field_by_id.get(e.index_file.global_index_meta.index_field_id) for e
in affected}
+ )
+ raise RuntimeError(
+ f"Update columns contain globally indexed columns, not supported
now.\n"
+ f"Updated columns: {sorted(update_set)}\n"
+ f"Conflicted columns: {conflicted}"
+ )
diff --git a/paimon-python/pypaimon/write/table_update.py
b/paimon-python/pypaimon/write/table_update.py
index fe2fb9a64b..0882bda8c6 100644
--- a/paimon-python/pypaimon/write/table_update.py
+++ b/paimon-python/pypaimon/write/table_update.py
@@ -136,12 +136,6 @@ class TableUpdate:
def _update_by_arrow_with_row_id(
self, table: pa.Table, commit_identifier: int
) -> List[CommitMessage]:
- """Shared implementation for ``update_by_arrow_with_row_id``.
-
- The public method lives on the concrete subclasses so each can
- expose the signature appropriate to its mode (batch vs stream).
- Produced commit messages are tagged with ``commit_identifier``.
- """
return TableUpdateByRowId(
self.table, self.commit_user, commit_identifier,
).update_columns(table, self.update_cols)
