[python] Reject changing immutable options once a table has snapshots (#8565)

plusplusjiajia · web-flow · commit 8393d28ba2a7 · 2026-07-13T10:18:48.000+08:00
diff --git 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
@@ -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 @@ def _generate_table_schema(
         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
@@ -94,23 +94,75 @@ def get_latest_snapshot(self) -> Optional[Snapshot]:
 
     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
@@ -219,6 +219,138 @@ def test_alter_table(self):
         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
@@ -255,21 +255,19 @@ def test_rejects_multiple_time_travel_keys(self):
             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
