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 8393d28ba2 [python] Reject changing immutable options once a table has
snapshots (#8565)
8393d28ba2 is described below
commit 8393d28ba2a73770bcb0a641affb67666671114e
Author: Jiajia Li <[email protected]>
AuthorDate: Mon Jul 13 10:18:48 2026 +0800
[python] Reject changing immutable options once a table has snapshots
(#8565)
---
.../pypaimon/common/options/core_options.py | 30 ++++-
paimon-python/pypaimon/schema/schema_manager.py | 54 +++++++++
.../pypaimon/snapshot/snapshot_manager.py | 70 +++++++++--
.../pypaimon/tests/filesystem_catalog_test.py | 132 +++++++++++++++++++++
.../pypaimon/tests/ray_read_by_row_id_test.py | 20 ++--
.../pypaimon/tests/snapshot_manager_test.py | 63 +++++++++-
6 files changed, 347 insertions(+), 22 deletions(-)
diff --git a/paimon-python/pypaimon/common/options/core_options.py
b/paimon-python/pypaimon/common/options/core_options.py
index 32d115a723..aa58e48eb4 100644
--- a/paimon-python/pypaimon/common/options/core_options.py
+++ b/paimon-python/pypaimon/common/options/core_options.py
@@ -99,6 +99,34 @@ class GlobalIndexSearchMode(str, Enum):
class CoreOptions:
"""Core options for Paimon tables."""
+
+ # Options that define the table's structure/identity and cannot be changed
+ # once the table has snapshots. Mirrors the @Immutable annotated options in
+ # the Java org.apache.paimon.CoreOptions (IMMUTABLE_OPTIONS).
+ IMMUTABLE_OPTIONS: frozenset = frozenset([
+ "type",
+ "bucket-key",
+ "bucket-function.type",
+ "data-file.path-directory",
+ "merge-engine",
+ "sequence.snapshot-ordering",
+ "aggregation.remove-record-on-delete",
+ "partial-update.remove-record-on-delete",
+ "partial-update.remove-record-on-sequence-group",
+ "rowkind.field",
+ "primary-key",
+ "partition",
+ "dynamic-bucket.initial-buckets",
+ "force-lookup",
+ "row-tracking.enabled",
+ "data-evolution.enabled",
+ "index-file-in-data-file-dir",
+ "blob-field",
+ "blob-descriptor-field",
+ "blob-view-field",
+ "pk-clustering-override",
+ ])
+
# File format constants
FILE_FORMAT_ORC: str = "orc"
FILE_FORMAT_AVRO: str = "avro"
@@ -127,7 +155,7 @@ class CoreOptions:
TYPE: ConfigOption[str] = (
ConfigOptions.key("type")
.string_type()
- .default_value("primary-key")
+ .default_value("table")
.with_description("Specify what type of table this is.")
)
diff --git a/paimon-python/pypaimon/schema/schema_manager.py
b/paimon-python/pypaimon/schema/schema_manager.py
index 3e25c9b242..0f31b55f34 100644
--- a/paimon-python/pypaimon/schema/schema_manager.py
+++ b/paimon-python/pypaimon/schema/schema_manager.py
@@ -294,6 +294,26 @@ def _get_type_root(data_type) -> str:
return getattr(data_type, 'type', '')
+def _normalize_key_list(value: str) -> List[str]:
+ return [s.strip() for s in value.split(',') if s.strip()]
+
+
+def _is_unchanged_normalized_key(key, old_value, new_value, old_table_schema)
-> bool:
+ """Values whose canonical home is outside the options map (or the implicit
+ default) are not a change when replayed. Mirrors the Java SchemaManager's
+ isUnchangedNormalizedKey."""
+ if old_value is not None or new_value is None:
+ return False
+ if key == "primary-key":
+ return _normalize_key_list(new_value) == old_table_schema.primary_keys
+ if key == "partition":
+ return _normalize_key_list(new_value) ==
old_table_schema.partition_keys
+ if key == "type":
+ # a table created without an explicit type is of the default type
+ return new_value == CoreOptions.TYPE.default_value()
+ return False
+
+
def _assert_not_updating_partition_keys(
schema: 'TableSchema', field_names: List[str], operation: str):
if len(field_names) > 1:
@@ -646,10 +666,44 @@ class SchemaManager:
disable_null_to_not_null = str(old_table_schema.options.get(
'alter-column-null-to-not-null.disabled', 'true')).lower() !=
'false'
+ # Structural options cannot be changed once the table has snapshots
+ # (mirrors the Java SchemaManager checkAlterTableOption /
+ # checkResetTableOption guards). The snapshot check is lazy so that
+ # alters without option changes never pay for it.
+ has_snapshots: Optional[bool] = None
+
+ def _has_snapshots() -> bool:
+ nonlocal has_snapshots
+ if has_snapshots is None:
+ from pypaimon.snapshot.snapshot_manager import SnapshotManager
+ has_snapshots = SnapshotManager(
+ self.file_io, self.table_path, self.branch
+ ).get_latest_snapshot() is not None
+ return has_snapshots
+
for change in changes:
if isinstance(change, SetOption):
+ # compare against the accumulated options so repeated changes
to
+ # the same key in one batch apply in order
+ old_value = new_options.get(change.key)
+ unchanged = (old_value == change.value
+ or _is_unchanged_normalized_key(
+ change.key, old_value, change.value,
old_table_schema))
+ if unchanged:
+ continue
+ # the type decides the table implementation, and table kinds
like
+ # format tables never create Paimon snapshots but can still
hold
+ # data -- reject independently of the snapshot check
+ if change.key == "type":
+ raise ValueError(f"Change '{change.key}' is not supported
yet.")
+ if change.key in CoreOptions.IMMUTABLE_OPTIONS and
_has_snapshots():
+ raise ValueError(f"Change '{change.key}' is not supported
yet.")
new_options[change.key] = change.value
elif isinstance(change, RemoveOption):
+ if change.key == "type":
+ raise ValueError(f"Change '{change.key}' is not supported
yet.")
+ if change.key in CoreOptions.IMMUTABLE_OPTIONS and
_has_snapshots():
+ raise ValueError(f"Change '{change.key}' is not supported
yet.")
new_options.pop(change.key, None)
elif isinstance(change, UpdateComment):
new_comment = change.comment
diff --git a/paimon-python/pypaimon/snapshot/snapshot_manager.py
b/paimon-python/pypaimon/snapshot/snapshot_manager.py
index d404a0dd94..2fc1438f56 100644
--- a/paimon-python/pypaimon/snapshot/snapshot_manager.py
+++ b/paimon-python/pypaimon/snapshot/snapshot_manager.py
@@ -94,23 +94,75 @@ class SnapshotManager:
def _get_latest_snapshot_from_filesystem(self) -> Optional[Snapshot]:
"""
- Get the latest snapshot from filesystem by reading LATEST file.
-
+ Get the latest snapshot from the filesystem. The LATEST file is only a
+ hint: when it is missing, stale, or unreadable, reconcile with a scan
+ of the snapshot files, like the Java SnapshotManager.findLatest.
+
+ If the hint indicates the table has snapshots but none can be resolved
+ (e.g. an unreadable LATEST plus a listing that an IO error truncated to
+ empty), fail closed rather than report the table as empty -- callers
+ such as the immutable-option guard must not mistake a populated table
+ for an empty one.
+
Returns:
- The latest snapshot, or None if not found
+ The latest snapshot, or None if the table genuinely has none
"""
- if not self.file_io.exists(self.latest_file):
+ if not self.file_io.exists(self.snapshot_dir):
return None
- latest_content = self.read_latest_file()
- latest_snapshot_id = int(latest_content.strip())
-
- snapshot_file = f"{self.snapshot_dir}/snapshot-{latest_snapshot_id}"
- if not self.file_io.exists(snapshot_file):
+ hint_id = None
+ hint_indicates_data = False
+ hint_read_error = None
+ if self.file_io.exists(self.latest_file):
+ try:
+ content = self.file_io.read_file_utf8(self.latest_file)
+ if content and content.strip():
+ hint_indicates_data = True
+ hint_id = int(content.strip())
+ except IOError as e:
+ # A present-but-unreadable hint most likely guards existing
+ # snapshots; remember the failure so we fail closed below.
+ hint_indicates_data = True
+ hint_read_error = e
+ except ValueError:
+ # Non-empty but corrupt hint; reconcile with a listing.
+ hint_indicates_data = True
+
+ # Trust the hint only when it points to an existing snapshot and no
+ # newer snapshot exists -- a lagging _commit_latest_hint leaves the
+ # hint stale. Otherwise reconcile with a directory listing (mirrors
+ # Java SnapshotManager.findLatest).
+ latest_snapshot_id = hint_id
+ if (latest_snapshot_id is None
+ or not
self.file_io.exists(self.get_snapshot_path(latest_snapshot_id))
+ or
self.file_io.exists(self.get_snapshot_path(latest_snapshot_id + 1))):
+ latest_snapshot_id = self._max_snapshot_id_from_listing()
+
+ if latest_snapshot_id is None:
+ if hint_indicates_data:
+ raise RuntimeError(
+ f"Snapshot hint under {self.snapshot_dir} indicates
existing "
+ f"snapshots but none could be resolved"
+ + (f": {hint_read_error}" if hint_read_error else ""))
return None
+ snapshot_file = self.get_snapshot_path(latest_snapshot_id)
return JSON.from_json(self.file_io.read_file_utf8(snapshot_file),
Snapshot)
+ def _max_snapshot_id_from_listing(self) -> Optional[int]:
+ """Scan the snapshot directory for the largest snapshot-N file id."""
+ import re
+
+ snapshot_pattern = re.compile(r'^snapshot-(\d+)$')
+ max_snapshot_id = None
+ for file_info in self.file_io.list_status(self.snapshot_dir):
+ match = snapshot_pattern.match(file_info.path.split('/')[-1])
+ if match:
+ snapshot_id = int(match.group(1))
+ if max_snapshot_id is None or snapshot_id > max_snapshot_id:
+ max_snapshot_id = snapshot_id
+ return max_snapshot_id
+
def read_latest_file(self, max_retries: int = 5):
"""
Read the latest snapshot ID from LATEST file with retry mechanism.
diff --git a/paimon-python/pypaimon/tests/filesystem_catalog_test.py
b/paimon-python/pypaimon/tests/filesystem_catalog_test.py
index 0251e7fd88..c22193ea15 100644
--- a/paimon-python/pypaimon/tests/filesystem_catalog_test.py
+++ b/paimon-python/pypaimon/tests/filesystem_catalog_test.py
@@ -219,6 +219,138 @@ class FileSystemCatalogTest(unittest.TestCase):
table = catalog.get_table(identifier)
self.assertEqual(len(table.fields), 2)
+ def test_alter_immutable_options(self):
+ catalog = CatalogFactory.create({"warehouse": self.warehouse})
+ catalog.create_database("test_db", False)
+
+ # while a table has no snapshots, structural options may still be
changed
+ empty_identifier = "test_db.immutable_options_empty"
+ pa_schema = pa.schema([('id', pa.int32()), ('name', pa.string())])
+ catalog.create_table(empty_identifier,
Schema.from_pyarrow_schema(pa_schema), False)
+ catalog.alter_table(
+ empty_identifier,
+ [SchemaChange.set_option("merge-engine", "first-row")],
+ False
+ )
+ # the type is rejected even without snapshots: table kinds like format
+ # tables never create Paimon snapshots but can still hold data
+ with self.assertRaises(RuntimeError) as ctx:
+ catalog.alter_table(
+ empty_identifier,
+ [SchemaChange.set_option("type", "format-table")], False)
+ self.assertIn("Change 'type' is not supported yet.",
str(ctx.exception))
+ with self.assertRaises(RuntimeError):
+ catalog.alter_table(
+ empty_identifier, [SchemaChange.remove_option("type")], False)
+
+ # once the table has a snapshot, immutable options are rejected
+ identifier = "test_db.immutable_options_table"
+ catalog.create_table(identifier,
Schema.from_pyarrow_schema(pa_schema), False)
+ table = catalog.get_table(identifier)
+ write_builder = table.new_batch_write_builder()
+ table_write = write_builder.new_write()
+ table_commit = write_builder.new_commit()
+ table_write.write_arrow(
+ pa.Table.from_pydict({'id': [1, 2], 'name': ['a', 'b']},
schema=pa_schema))
+ table_commit.commit(table_write.prepare_commit())
+ table_write.close()
+ table_commit.close()
+
+ # the catalog wraps alter failures in RuntimeError
+ with self.assertRaises(RuntimeError) as ctx:
+ catalog.alter_table(
+ identifier, [SchemaChange.set_option("type", "format-table")],
False)
+ self.assertIn("Change 'type' is not supported yet.",
str(ctx.exception))
+
+ with self.assertRaises(RuntimeError) as ctx:
+ catalog.alter_table(
+ identifier, [SchemaChange.set_option("merge-engine",
"deduplicate")], False)
+ self.assertIn("Change 'merge-engine' is not supported yet.",
str(ctx.exception))
+
+ with self.assertRaises(RuntimeError) as ctx:
+ catalog.alter_table(
+ identifier, [SchemaChange.remove_option("merge-engine")],
False)
+ self.assertIn("Change 'merge-engine' is not supported yet.",
str(ctx.exception))
+
+ # canonical hyphenated keys are guarded too
+ with self.assertRaises(RuntimeError) as ctx:
+ catalog.alter_table(
+ identifier,
+ [SchemaChange.set_option("index-file-in-data-file-dir",
"true")], False)
+ self.assertIn(
+ "Change 'index-file-in-data-file-dir' is not supported yet.",
str(ctx.exception))
+
+ # setting the default type explicitly is not a change
+ catalog.alter_table(
+ identifier, [SchemaChange.set_option("type", "table")], False)
+ table = catalog.get_table(identifier)
+ self.assertNotIn("type", table.table_schema.options)
+ # ...and the reloaded effective type resolves to the requested "table"
+ # (the default), not some other configured default
+ from pypaimon.common.options import CoreOptions, Options
+ self.assertEqual(
+ CoreOptions(Options(table.table_schema.options)).type(), "table")
+
+ # mutable options can still be changed
+ catalog.alter_table(
+ identifier, [SchemaChange.set_option("target-file-size", "10mb")],
False)
+
+ # repeated changes to the same option in one batch apply in order
+ catalog.alter_table(identifier, [
+ SchemaChange.set_option("target-file-size", "20mb"),
+ SchemaChange.set_option("target-file-size", "10mb")], False)
+ table = catalog.get_table(identifier)
+ self.assertEqual(table.table_schema.options.get("target-file-size"),
"10mb")
+ catalog.alter_table(identifier, [
+ SchemaChange.remove_option("target-file-size"),
+ SchemaChange.set_option("target-file-size", "10mb")], False)
+ table = catalog.get_table(identifier)
+ self.assertEqual(table.table_schema.options.get("target-file-size"),
"10mb")
+
+ # replaying the actual primary/partition keys is not a change: they are
+ # stored in the schema fields, not the options map
+ pk_identifier = "test_db.immutable_options_pk"
+ catalog.create_table(
+ pk_identifier,
+ Schema.from_pyarrow_schema(
+ pa_schema, primary_keys=['id'], options={'bucket': '1'}),
+ False)
+ pk_table = catalog.get_table(pk_identifier)
+ write_builder = pk_table.new_batch_write_builder()
+ table_write = write_builder.new_write()
+ table_commit = write_builder.new_commit()
+ table_write.write_arrow(
+ pa.Table.from_pydict({'id': [1], 'name': ['a']}, schema=pa_schema))
+ table_commit.commit(table_write.prepare_commit())
+ table_write.close()
+ table_commit.close()
+ catalog.alter_table(
+ pk_identifier, [SchemaChange.set_option("primary-key", "id")],
False)
+ pk_table = catalog.get_table(pk_identifier)
+ self.assertNotIn("primary-key", pk_table.table_schema.options)
+ with self.assertRaises(RuntimeError):
+ catalog.alter_table(
+ pk_identifier, [SchemaChange.set_option("primary-key",
"name")], False)
+
+ part_identifier = "test_db.immutable_options_part"
+ self._create_partitioned_table_with_data(
+ catalog, part_identifier, [{'dt': '2026-01-01', 'rows': 1}])
+ catalog.alter_table(
+ part_identifier, [SchemaChange.set_option("partition", "dt")],
False)
+ with self.assertRaises(RuntimeError):
+ catalog.alter_table(
+ part_identifier, [SchemaChange.set_option("partition",
"col1")], False)
+
+ # the LATEST file is only a hint: the guard must still see the
+ # snapshots when it is missing
+ os.remove(os.path.join(
+ self.warehouse, "test_db.db", "immutable_options_table",
+ "snapshot", "LATEST"))
+ with self.assertRaises(RuntimeError):
+ catalog.alter_table(
+ identifier,
+ [SchemaChange.set_option("merge-engine", "deduplicate")],
False)
+
def test_update_column_type_guards_null_to_not_null(self):
catalog = CatalogFactory.create({"warehouse": self.warehouse})
catalog.create_database("test_db_guard", False)
diff --git a/paimon-python/pypaimon/tests/ray_read_by_row_id_test.py
b/paimon-python/pypaimon/tests/ray_read_by_row_id_test.py
index 2fa6cd00c9..c49d359042 100644
--- a/paimon-python/pypaimon/tests/ray_read_by_row_id_test.py
+++ b/paimon-python/pypaimon/tests/ray_read_by_row_id_test.py
@@ -255,21 +255,19 @@ class RayReadByRowIdTest(unittest.TestCase):
read_by_row_id(target, src, self.catalog_options,
projection=["age"],
dynamic_options={"scan.snapshot-id": "1",
"scan.tag-name": "x"})
- def test_time_travel_before_row_tracking_raises(self):
+ def test_row_tracking_cannot_be_enabled_after_data(self):
+ # row-tracking.enabled / data-evolution.enabled are immutable once the
+ # table has snapshots (Java parity), so a "snapshot predates
+ # row-tracking" state can no longer be constructed through ALTER; the
+ # alter itself is rejected instead.
from pypaimon.schema.schema_change import SchemaChange
name = self._create(options={}) # plain table: no data-evolution /
row-tracking
self._write(name, pa.Table.from_pydict(
{"id": [1], "name": ["a"], "age": [1]}, schema=self.pa_schema))
- self.catalog.alter_table(name, [
- SchemaChange.set_option("row-tracking.enabled", "true"),
- SchemaChange.set_option("data-evolution.enabled", "true")])
- self._write(name, pa.Table.from_pydict(
- {"id": [2], "name": ["b"], "age": [2]}, schema=self.pa_schema))
- src = pa.table({"_ROW_ID": [0]}, schema=pa.schema([("_ROW_ID",
pa.int64())]))
- # snapshot 1 predates row-tracking -> clear error, not a silent empty
read
- with self.assertRaisesRegex(ValueError, "row-tracking|data-evolution"):
- read_by_row_id(name, src, self.catalog_options, projection=["age"],
- dynamic_options={"scan.snapshot-id": "1"})
+ with self.assertRaisesRegex(RuntimeError, "not supported yet"):
+ self.catalog.alter_table(name, [
+ SchemaChange.set_option("row-tracking.enabled", "true"),
+ SchemaChange.set_option("data-evolution.enabled", "true")])
def test_pins_base_snapshot(self):
import importlib
diff --git a/paimon-python/pypaimon/tests/snapshot_manager_test.py
b/paimon-python/pypaimon/tests/snapshot_manager_test.py
index c5adcb0537..f09f61fc3a 100644
--- a/paimon-python/pypaimon/tests/snapshot_manager_test.py
+++ b/paimon-python/pypaimon/tests/snapshot_manager_test.py
@@ -20,7 +20,7 @@ Tests for SnapshotManager batch lookahead functionality.
"""
import unittest
-from unittest.mock import Mock
+from unittest.mock import Mock, patch
def _create_mock_snapshot(snapshot_id: int, commit_kind: str = "APPEND"):
@@ -162,5 +162,66 @@ class SnapshotManagerTest(unittest.TestCase):
manager.try_get_earliest_snapshot()
+def _file_info(path):
+ info = Mock()
+ info.path = path
+ return info
+
+
+class SnapshotManagerFilesystemHintTest(unittest.TestCase):
+ """Tests for _get_latest_snapshot_from_filesystem hint reconciliation."""
+
+ SNAP_DIR = "/tmp/test_table/snapshot"
+ LATEST = "/tmp/test_table/snapshot/LATEST"
+
+ def _snap(self, n):
+ return "{}/snapshot-{}".format(self.SNAP_DIR, n)
+
+ def test_ignores_stale_latest_hint(self):
+ # LATEST still points to snapshot 1 but snapshot 2 was committed (a
+ # lagging _commit_latest_hint); the newer snapshot must win.
+ present = {self.SNAP_DIR, self.LATEST, self._snap(1), self._snap(2)}
+ file_io = Mock()
+ file_io.exists.side_effect = lambda p: p in present
+ file_io.read_file_utf8.side_effect = (
+ lambda p: "1" if p == self.LATEST else "content-" + p.rsplit("-",
1)[-1])
+ file_io.list_status.return_value = [
+ _file_info(self._snap(1)), _file_info(self._snap(2))]
+ manager = _build_manager(file_io)
+ with patch(
+ 'pypaimon.snapshot.snapshot_manager.JSON.from_json',
+ side_effect=lambda content, cls: content):
+ result = manager._get_latest_snapshot_from_filesystem()
+ self.assertEqual(result, "content-2")
+
+ def test_fails_closed_when_hint_unreadable_and_listing_empty(self):
+ # LATEST exists (the table is non-empty) but cannot be read, and the
+ # listing came back empty (e.g. a swallowed PermissionError). Returning
+ # None here would let the immutable-option guard fail open.
+ present = {self.SNAP_DIR, self.LATEST}
+ file_io = Mock()
+ file_io.exists.side_effect = lambda p: p in present
+
+ def _read(p):
+ if p == self.LATEST:
+ raise IOError("permission denied")
+ raise AssertionError("unexpected read of " + p)
+
+ file_io.read_file_utf8.side_effect = _read
+ file_io.list_status.return_value = []
+ manager = _build_manager(file_io)
+ with self.assertRaises(RuntimeError):
+ manager._get_latest_snapshot_from_filesystem()
+
+ def test_returns_none_when_table_genuinely_empty(self):
+ # No LATEST and no snapshot files: a real empty table, not an error.
+ present = {self.SNAP_DIR}
+ file_io = Mock()
+ file_io.exists.side_effect = lambda p: p in present
+ file_io.list_status.return_value = []
+ manager = _build_manager(file_io)
+ self.assertIsNone(manager._get_latest_snapshot_from_filesystem())
+
+
if __name__ == '__main__':
unittest.main()
