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 9895baaa9d [python] Check option changes on tables with snapshots like
Java (#10856)
9895baaa9d is described below
commit 9895baaa9dde96dc12cabeffeaf7fee15aa1e76d
Author: Jiajia Li <[email protected]>
AuthorDate: Thu Oct 29 10:10:18 2026 +0800
[python] Check option changes on tables with snapshots like Java (#10856)
---
.../pypaimon/common/options/core_options.py | 1 +
paimon-python/pypaimon/schema/schema_manager.py | 123 +++++++++++--
.../pypaimon/tests/filesystem_catalog_test.py | 12 ++
.../pypaimon/tests/schema_manager_test.py | 198 ++++++++++++++++++++-
4 files changed, 321 insertions(+), 13 deletions(-)
diff --git a/paimon-python/pypaimon/common/options/core_options.py
b/paimon-python/pypaimon/common/options/core_options.py
index 91ff535c2b..8c5214f52a 100644
--- a/paimon-python/pypaimon/common/options/core_options.py
+++ b/paimon-python/pypaimon/common/options/core_options.py
@@ -123,6 +123,7 @@ class CoreOptions:
"type",
"bucket-key",
"bucket-function.type",
+ "bucket-function.provider",
"data-file.path-directory",
"merge-engine",
"sequence.snapshot-ordering",
diff --git a/paimon-python/pypaimon/schema/schema_manager.py
b/paimon-python/pypaimon/schema/schema_manager.py
index 0d375d49a2..785a86af50 100644
--- a/paimon-python/pypaimon/schema/schema_manager.py
+++ b/paimon-python/pypaimon/schema/schema_manager.py
@@ -23,6 +23,7 @@ from pypaimon.common.file_io import FileIO
from pypaimon.common.identifier import DEFAULT_MAIN_BRANCH
from pypaimon.common.json_util import JSON
from pypaimon.common.options import CoreOptions, Options
+from pypaimon.common.options.options_utils import OptionsUtils
from pypaimon.schema.column_directive_utils import (
apply_add_column_directive, apply_directives,
remove_dropped_directive_options)
@@ -305,18 +306,115 @@ def _is_unchanged_normalized_key(key, old_value,
new_value, old_table_schema) ->
"""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:
+ if new_value is None:
+ return False
+ if key == "type":
+ effective_old = CoreOptions.TYPE.default_value() if old_value is None
else old_value
+ return new_value.lower() == effective_old.lower()
+ if old_value is not 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 _parse_bucket(value: str) -> int:
+ # Match Integer.parseInt's signed 32-bit decimal syntax.
+ if value is None:
+ raise ValueError("null")
+ negative = value.startswith("-")
+ digits = value[1:] if value.startswith(("+", "-")) else value
+ try:
+ if not digits:
+ raise ValueError
+ bucket = 0
+ limit = 2147483648 if negative else 2147483647
+ for c in digits:
+ if not c.isdecimal() or ord(c) > 0xffff:
+ raise ValueError
+ bucket = bucket * 10 + int(c)
+ if bucket > limit:
+ raise ValueError
+ return -bucket if negative else bucket
+ except ValueError:
+ raise ValueError(f'For input string: "{value}"') from None
+
+
+def _option_text(value) -> Optional[str]:
+ # Options may be persisted as native values such as 2, True or enums;
compare them as Options reads them.
+ return None if value is None else OptionsUtils.convert_to_string(value)
+
+
+def _option_bool(value) -> bool:
+ # Same interpretation as the runtime; unparsable values are false, like
Boolean.parseBoolean.
+ try:
+ return value is not None and OptionsUtils.convert_to_boolean(value)
+ except ValueError:
+ return False
+
+
+def _check_alter_table_option(options: dict, key: str, old_value:
Optional[str], new_value: str):
+ # Booleans are read from the raw values, the way PyPaimon reads them.
+ raw_old, raw_new = old_value, new_value
+ old_value, new_value = _option_text(old_value), _option_text(new_value)
+ if key in CoreOptions.IMMUTABLE_OPTIONS:
+ raise NotImplementedError(f"Change '{key}' is not supported yet.")
+
+ if key == "data-evolution.nested-field.enabled":
+ old_enabled = _option_bool(raw_old)
+ new_enabled = _option_bool(raw_new)
+ if old_enabled and not new_enabled:
+ raise NotImplementedError(f"Cannot disable table option '{key}'.")
+
+ if key == "bucket":
+ old_bucket = CoreOptions.BUCKET.default_value() if old_value is None
else _parse_bucket(old_value)
+ new_bucket = _parse_bucket(new_value)
+ if old_bucket == -1:
+ raise NotImplementedError("Cannot change bucket when it is -1.")
+ if new_bucket == -1:
+ raise NotImplementedError("Cannot change bucket to -1.")
+
+ if key == "deletion-vectors.enabled":
+ dv_modifiable =
_option_bool(options.get("deletion-vectors.modifiable"))
+ if not dv_modifiable:
+ old_dv = _option_bool(raw_old)
+ new_dv = _option_bool(raw_new)
+ if old_dv != new_dv:
+ raise NotImplementedError(
+ f"Cannot change deletion vectors mode from
{str(old_dv).lower()} to {str(new_dv).lower()}. "
+ "If modifying table deletion-vectors mode without
full-compaction, "
+ "this may result in data duplication. If you are
confident, you can set table option "
+ "'deletion-vectors.modifiable' = 'true' to allow deletion
vectors modification.")
+
+ if key in ("ignore-delete", "ignore-update-before"):
+ # The writers read ignore-update-before as the literal "true" only.
+ parse = _option_bool if key == "ignore-delete" else (lambda v:
str(v).lower() == "true")
+ old_ignore = parse(raw_old)
+ new_ignore = parse(raw_new)
+ if old_ignore and not new_ignore:
+ raise NotImplementedError(f"Cannot change {key} from true to
false.")
+
+ if key == "clustering.columns" and "pk-clustering-override" in options:
+ raise NotImplementedError("Cannot change clustering.columns when
pk-clustering-override enabled.")
+
+
+def _check_reset_table_option(options: dict, key: str):
+ if key in CoreOptions.IMMUTABLE_OPTIONS:
+ raise NotImplementedError(f"Change '{key}' is not supported yet.")
+
+ if key == "bucket":
+ raise NotImplementedError(f"Cannot reset {key}.")
+
+ if key in ("deletion-vectors.enabled",
"data-evolution.nested-field.enabled",
+ "ignore-delete", "ignore-update-before"):
+ _check_alter_table_option(options, key, options.get(key), "false")
+
+ if key == "clustering.columns" and "pk-clustering-override" in options:
+ raise NotImplementedError(f"Cannot reset {key} when
pk-clustering-override enabled.")
+
+
def _assert_not_updating_partition_keys(
schema: 'TableSchema', field_names: List[str], operation: str):
if len(field_names) > 1:
@@ -754,6 +852,7 @@ class SchemaManager:
def _generate_table_schema(
self, old_table_schema: TableSchema, changes: List[SchemaChange]
) -> TableSchema:
+ old_options = dict(old_table_schema.options)
new_options = dict(old_table_schema.options)
new_fields = []
for field in old_table_schema.fields:
@@ -798,7 +897,7 @@ class SchemaManager:
# 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
+ unchanged = (_option_text(old_value) ==
_option_text(change.value)
or _is_unchanged_normalized_key(
change.key, old_value, change.value,
old_table_schema))
if unchanged:
@@ -807,15 +906,17 @@ class SchemaManager:
# 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.")
+ raise NotImplementedError(f"Change '{change.key}' is not
supported yet.")
+ if _has_snapshots():
+ # Like Java, checks compare with the table's options
before the batch.
+ _check_alter_table_option(
+ old_options, change.key, old_options.get(change.key),
change.value)
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.")
+ raise NotImplementedError(f"Change '{change.key}' is not
supported yet.")
+ if _has_snapshots():
+ _check_reset_table_option(old_options, change.key)
new_options.pop(change.key, None)
elif isinstance(change, UpdateComment):
new_comment = change.comment
diff --git a/paimon-python/pypaimon/tests/filesystem_catalog_test.py
b/paimon-python/pypaimon/tests/filesystem_catalog_test.py
index 4690ffeefe..f30e44a6b0 100644
--- a/paimon-python/pypaimon/tests/filesystem_catalog_test.py
+++ b/paimon-python/pypaimon/tests/filesystem_catalog_test.py
@@ -304,6 +304,18 @@ class FileSystemCatalogTest(unittest.TestCase):
self.assertIn(
"Change 'index-file-in-data-file-dir' is not supported yet.",
str(ctx.exception))
+ for change, message in [
+ (SchemaChange.set_option("bucket", "4"), "Cannot change bucket
when it is -1."),
+ (SchemaChange.remove_option("bucket"), "Cannot reset bucket."),
+ (SchemaChange.set_option("deletion-vectors.enabled", "true"),
"Cannot change deletion vectors mode"),
+ (SchemaChange.set_option("bucket-function.provider", "custom"),
+ "Change 'bucket-function.provider' is not supported yet."),
+ ]:
+ with self.assertRaises(RuntimeError) as ctx:
+ catalog.alter_table(identifier, [change], False)
+ self.assertIn(message, str(ctx.exception))
+ self.assertEqual(catalog.get_table(identifier).table_schema.id,
table.table_schema.id)
+
# setting the default type explicitly is not a change
catalog.alter_table(
identifier, [SchemaChange.set_option("type", "table")], False)
diff --git a/paimon-python/pypaimon/tests/schema_manager_test.py
b/paimon-python/pypaimon/tests/schema_manager_test.py
index dd73d9142d..26354a6334 100644
--- a/paimon-python/pypaimon/tests/schema_manager_test.py
+++ b/paimon-python/pypaimon/tests/schema_manager_test.py
@@ -18,18 +18,20 @@
import os
import tempfile
import unittest
+from unittest.mock import patch
from pypaimon.branch.branch_manager import BranchManager
from pypaimon.common.identifier import DEFAULT_MAIN_BRANCH
from pypaimon.common.file_io import FileIO
-from pypaimon.common.options.core_options import CoreOptions
+from pypaimon.common.options.core_options import CoreOptions, MergeEngine
from pypaimon.common.options.options import Options
from pypaimon.schema.data_types import (ArrayType, AtomicType, DataField,
MapType, RowType)
from pypaimon.schema.schema import Schema
-from pypaimon.schema.schema_change import SetOption
+from pypaimon.schema.schema_change import RemoveOption, SetOption
from pypaimon.schema.schema_manager import SchemaManager
from pypaimon.schema.table_schema import TableSchema
+from pypaimon.snapshot.snapshot_manager import SnapshotManager
class TestSchemaManagerBranch(unittest.TestCase):
@@ -295,6 +297,198 @@ class TestOptionValidation(unittest.TestCase):
self.assertNotIn(key, manager.latest().options)
+class TestAlterTableOptions(unittest.TestCase):
+
+ def setUp(self):
+ self.temp_dir = tempfile.TemporaryDirectory()
+ self.addCleanup(self.temp_dir.cleanup)
+ self.file_io = FileIO.get(self.temp_dir.name, Options({}))
+ self.manager = SchemaManager(self.file_io, self.temp_dir.name)
+
+ def _alter(self, options, changes, has_snapshots=True):
+ schema = TableSchema.from_schema(0, Schema(
+ fields=[DataField(0, "id", AtomicType("INT"))], options=options))
+ with patch.object(SnapshotManager, "get_latest_snapshot",
+ return_value=object() if has_snapshots else None):
+ return self.manager._generate_table_schema(schema, changes).options
+
+ def _assert_rejected(self, options, changes, message):
+ with self.assertRaises(NotImplementedError) as ctx:
+ self._alter(options, changes)
+ self.assertEqual(str(ctx.exception), message)
+
+ def test_immutable_options(self):
+ for key in CoreOptions.IMMUTABLE_OPTIONS |
{"bucket-function.provider"}:
+ with self.subTest(key=key):
+ old_value = "false" if key == "primary-key.nullable" else "old"
+ for options, change in [
+ ({}, SetOption(key, "new")),
+ ({key: old_value}, SetOption(key, "new")),
+ ({}, RemoveOption(key)),
+ ({key: old_value}, RemoveOption(key)),
+ ]:
+ self._assert_rejected(options, [change], f"Change '{key}'
is not supported yet.")
+ self.assertEqual(self._alter({key: old_value}, [SetOption(key,
old_value)]), {key: old_value})
+
+ def test_bucket_changes(self):
+ for old_value, new_value, message in [
+ (None, "4", "Cannot change bucket when it is -1."),
+ (None, "-1", "Cannot change bucket when it is -1."),
+ ("-1", "4", "Cannot change bucket when it is -1."),
+ ("-1", "-01", "Cannot change bucket when it is -1."),
+ ("4", "-1", "Cannot change bucket to -1."),
+ ]:
+ with self.subTest(old=old_value, new=new_value):
+ options = {} if old_value is None else {"bucket": old_value}
+ self._assert_rejected(options, [SetOption("bucket",
new_value)], message)
+ for options in [{}, {"bucket": "-1"}, {"bucket": "4"}]:
+ self._assert_rejected(options, [RemoveOption("bucket")], "Cannot
reset bucket.")
+ for old_value, new_value in [
+ ("-1", "-1"), ("4", "8"), ("4", "+4"), ("-2", "4"),
+ ("4", "2147483647"), ("4", "-2147483648"), ("4", "\uff14"), ("4",
"0" * 5000 + "4"),
+ ]:
+ self.assertEqual(
+ self._alter({"bucket": old_value}, [SetOption("bucket",
new_value)]),
+ {"bucket": new_value})
+
+ def test_bucket_requires_java_integer(self):
+ for value in ["", "+", "invalid", " 4", "4 ", "4_0", "1.0",
"2147483648", "-2147483649", "\U0001d7dc"]:
+ for old_value, new_value in [("4", value), (value, "4")]:
+ with self.subTest(old=old_value, new=new_value):
+ with self.assertRaises(ValueError) as ctx:
+ self._alter({"bucket": old_value},
[SetOption("bucket", new_value)])
+ self.assertEqual(str(ctx.exception), f'For input string:
"{value}"')
+
+ def test_deletion_vectors_changes(self):
+ key = "deletion-vectors.enabled"
+ for old_value, new_value in [(None, "true"), ("false", "true"),
("true", "false"), ("true", None)]:
+ for modifiable in [None, "false", "no"]:
+ options = {} if old_value is None else {key: old_value}
+ if modifiable is not None:
+ options["deletion-vectors.modifiable"] = modifiable
+ message = (
+ f"Cannot change deletion vectors mode from {old_value or
'false'} to {new_value or 'false'}. "
+ "If modifying table deletion-vectors mode without
full-compaction, "
+ "this may result in data duplication. If you are
confident, you can set table option "
+ "'deletion-vectors.modifiable' = 'true' to allow deletion
vectors modification.")
+ changes = [SetOption(key, new_value)]
+ if new_value == "false":
+ changes.append(RemoveOption(key))
+ for change in changes:
+ with self.subTest(old=old_value, new=new_value,
modifiable=modifiable, change=change):
+ self._assert_rejected(options, [change], message)
+ for old_value, new_value in [(None, "false"), ("false", "FALSE"),
("true", "TRUE")]:
+ options = {} if old_value is None else {key: old_value}
+ self.assertEqual(self._alter(options, [SetOption(key,
new_value)])[key], new_value)
+ for options in [{}, {key: "false"}]:
+ self.assertEqual(self._alter(options, [RemoveOption(key)]), {})
+ for change in [SetOption(key, "true"), SetOption(key, "false"),
RemoveOption(key)]:
+ for old_value in ["false", "true"]:
+ options = {key: old_value, "deletion-vectors.modifiable":
"TRUE"}
+ result = self._alter(options, [change])
+ self.assertEqual(result.get(key), getattr(change, "value",
None))
+
+ def test_options_cannot_be_disabled(self):
+ for key, message in [
+ ("ignore-delete", "Cannot change ignore-delete from true to
false."),
+ ("ignore-update-before", "Cannot change ignore-update-before from
true to false."),
+ ("data-evolution.nested-field.enabled",
+ "Cannot disable table option
'data-evolution.nested-field.enabled'."),
+ ]:
+ with self.subTest(key=key):
+ for change in [SetOption(key, "false"), SetOption(key, None),
SetOption(key, "no"),
+ RemoveOption(key)]:
+ self._assert_rejected({key: "TRUE"}, [change], message)
+ for old_value in [None, "false", "true"]:
+ options = {} if old_value is None else {key: old_value}
+ self.assertEqual(self._alter(options, [SetOption(key,
"true")]), {key: "true"})
+ for options in [{}, {key: "false"}]:
+ self.assertEqual(self._alter(options,
[RemoveOption(key)]), {})
+
+ def test_clustering_columns(self):
+ key = "clustering.columns"
+ for override in ["true", "false"]:
+ for old_value in [None, "id"]:
+ options = {"pk-clustering-override": override}
+ if old_value is not None:
+ options[key] = old_value
+ self._assert_rejected(options, [SetOption(key, "name")],
+ "Cannot change clustering.columns when
pk-clustering-override enabled.")
+ self._assert_rejected(options, [RemoveOption(key)],
+ "Cannot reset clustering.columns when
pk-clustering-override enabled.")
+ self.assertEqual(self._alter(options, [SetOption(key, "id")]),
options)
+ self.assertEqual(self._alter({}, [SetOption(key, "id")]), {key: "id"})
+ self.assertEqual(self._alter({key: "id"}, [RemoveOption(key)]), {})
+
+ def test_guards_require_snapshots(self):
+ for options, key, value in [
+ ({"bucket": "-1"}, "bucket", "4"),
+ ({"bucket": "4"}, "bucket", "-1"),
+ ({"deletion-vectors.enabled": "true"}, "deletion-vectors.enabled",
"false"),
+ ({"ignore-delete": "true"}, "ignore-delete", "false"),
+ ({"ignore-update-before": "true"}, "ignore-update-before",
"false"),
+ ({"data-evolution.nested-field.enabled": "true"},
"data-evolution.nested-field.enabled", "false"),
+ ({"pk-clustering-override": "true", "clustering.columns": "id"},
"clustering.columns", "name"),
+ ({"bucket-function.provider": "old"}, "bucket-function.provider",
"new"),
+ ]:
+ with self.subTest(key=key):
+ self.assertEqual(self._alter(options, [SetOption(key, value)],
False)[key], value)
+ self.assertNotIn(key, self._alter(options,
[RemoveOption(key)], False))
+ for change in [SetOption("type", "format-table"),
RemoveOption("type")]:
+ with self.assertRaisesRegex(NotImplementedError, "Change 'type' is
not supported yet."):
+ self._alter({}, [change], False)
+
+ def test_checks_accept_native_option_values(self):
+ self.assertEqual(self._alter({"bucket": 2}, [SetOption("bucket",
"4")]), {"bucket": "4"})
+ self.assertEqual(self._alter({"bucket": -1}, [SetOption("bucket",
"-1")]), {"bucket": -1})
+ self.assertEqual(self._alter({"ignore-delete": True},
[SetOption("ignore-delete", "true")]),
+ {"ignore-delete": "true"})
+ self.assertEqual(self._alter({"partition.default-name": True},
[SetOption("partition.default-name", "true")]),
+ {"partition.default-name": "true"})
+ self.assertEqual(self._alter({"deletion-vectors.enabled": "yes"},
[SetOption("deletion-vectors.enabled", "1")]),
+ {"deletion-vectors.enabled": "1"})
+ with self.assertRaisesRegex(NotImplementedError, "Cannot change
deletion vectors mode"):
+ self._alter({"deletion-vectors.enabled": "false"},
[SetOption("deletion-vectors.enabled", "yes")])
+ dv = "deletion-vectors.enabled"
+ self.assertEqual(self._alter({dv: 1.0}, [SetOption(dv, "true")]), {dv:
"true"})
+ self.assertEqual(self._alter({"ignore-update-before": "yes"},
[SetOption("ignore-update-before", "false")]),
+ {"ignore-update-before": "false"})
+ self.assertEqual(self._alter({"merge-engine": "first-row"},
[SetOption("merge-engine", MergeEngine.FIRST_ROW)]),
+ {"merge-engine": "first-row"})
+ with self.assertRaisesRegex(NotImplementedError, "Cannot change
ignore-delete from true to false"):
+ self._alter({"ignore-delete": True}, [SetOption("ignore-delete",
"false")])
+ options = {"deletion-vectors.modifiable": True}
+ self.assertEqual(self._alter(options,
[SetOption("deletion-vectors.enabled", "true")]),
+ {"deletion-vectors.modifiable": True,
"deletion-vectors.enabled": "true"})
+
+ def test_checks_use_original_options(self):
+ key = "deletion-vectors.enabled"
+ modifiable = "deletion-vectors.modifiable"
+ for changes in [
+ [SetOption(modifiable, "true"), SetOption(key, "true")],
+ [SetOption(modifiable, "true"), RemoveOption(key)],
+ ]:
+ options = {key: "true"} if isinstance(changes[1], RemoveOption)
else {}
+ with self.assertRaisesRegex(NotImplementedError, "Cannot change
deletion vectors mode"):
+ self._alter(options, changes)
+ self.assertEqual(self._alter({modifiable: "true"},
[RemoveOption(modifiable), SetOption(key, "true")]),
+ {key: "true"})
+ self.assertEqual(self._alter({}, [SetOption("ignore-delete", "true"),
SetOption("ignore-delete", "false")]),
+ {"ignore-delete": "false"})
+ self.assertEqual(self._alter({"bucket": "4"}, [SetOption("bucket",
"8"), SetOption("bucket", "4")]),
+ {"bucket": "4"})
+
+ def test_normalized_unchanged_options(self):
+ for options in [{}, {"type": "table"}, {"type": "TABLE"}]:
+ self.assertEqual(self._alter(options, [SetOption("type",
"TaBlE")]), options)
+ schema = TableSchema.from_schema(0, Schema(
+ fields=[DataField(0, "id", AtomicType("INT"))],
primary_keys=["id"], partition_keys=["id"]))
+ with patch.object(SnapshotManager, "get_latest_snapshot",
return_value=object()):
+ result = self.manager._generate_table_schema(schema, [
+ SetOption("primary-key", " id, "), SetOption("partition", "
id, ")])
+ self.assertEqual(result.options, schema.options)
+
+
class TestRowTrackingValidation(unittest.TestCase):
"""Row tracking tables must follow Java
SchemaValidation#validateRowTracking."""
