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 b17bbde9b6 [python] Align row tracking and data evolution schema
validation with… (#10527)
b17bbde9b6 is described below
commit b17bbde9b6351b97b95d339844279d5f1e694b0f
Author: zhuanshenbsj1 <[email protected]>
AuthorDate: Wed Oct 14 09:58:13 2026 +0800
[python] Align row tracking and data evolution schema validation with…
(#10527)
---
paimon-python/pypaimon/schema/schema_manager.py | 20 +++++
.../pypaimon/tests/daft/daft_sink_test.py | 3 -
.../pypaimon/tests/global_index_build_test.py | 11 +--
.../tests/manifest/manifest_sidecar_test.py | 1 +
.../pypaimon/tests/multimodal_table_test.py | 17 ++--
.../tests/ray_data_evolution_merge_into_test.py | 14 ++--
.../pypaimon/tests/schema_manager_test.py | 91 ++++++++++++++++++++++
7 files changed, 131 insertions(+), 26 deletions(-)
diff --git a/paimon-python/pypaimon/schema/schema_manager.py
b/paimon-python/pypaimon/schema/schema_manager.py
index 950a2e308b..17953fd33a 100644
--- a/paimon-python/pypaimon/schema/schema_manager.py
+++ b/paimon-python/pypaimon/schema/schema_manager.py
@@ -483,6 +483,25 @@ def _validate_options(options: dict):
)
+def _validate_row_tracking(options: dict, primary_keys: List[str]):
+ """Mirror the row tracking checks of Java
SchemaValidation#validateRowTracking."""
+ core_options = CoreOptions(Options(options or {}))
+ row_tracking_enabled = core_options.row_tracking_enabled()
+ if row_tracking_enabled:
+ if primary_keys:
+ raise ValueError("Cannot define primary-key for row tracking
table.")
+ if core_options.bucket() != -1:
+ raise ValueError(
+ f"Cannot define {CoreOptions.BUCKET.key()} for row tracking
table, "
+ f"it only support bucket = -1"
+ )
+ if core_options.data_evolution_enabled() and not row_tracking_enabled:
+ raise ValueError(
+ f"Data evolution config must enabled with "
+ f"{CoreOptions.ROW_TRACKING_ENABLED.key()}"
+ )
+
+
def _contains_blob_type(data_type: DataType) -> bool:
if is_blob_type(data_type):
return True
@@ -674,6 +693,7 @@ class SchemaManager:
new_schema.primary_keys,
new_schema.partition_keys,
)
+ _validate_row_tracking(new_schema.options, new_schema.primary_keys)
schema_path = self._to_schema_path(new_schema.id)
try:
result = self.file_io.try_to_write_atomic(schema_path,
JSON.to_json(new_schema, indent=2))
diff --git a/paimon-python/pypaimon/tests/daft/daft_sink_test.py
b/paimon-python/pypaimon/tests/daft/daft_sink_test.py
index bf14da8a28..326264e7cf 100644
--- a/paimon-python/pypaimon/tests/daft/daft_sink_test.py
+++ b/paimon-python/pypaimon/tests/daft/daft_sink_test.py
@@ -548,7 +548,6 @@ class TestBlobType:
paimon_schema = pypaimon.Schema.from_pyarrow_schema(
pa_schema,
options={
- "bucket": "1",
"file.format": "parquet",
"row-tracking.enabled": "true",
"data-evolution.enabled": "true",
@@ -588,7 +587,6 @@ class TestBlobType:
paimon_schema = pypaimon.Schema.from_pyarrow_schema(
pa_schema,
options={
- "bucket": "1",
"file.format": "parquet",
"row-tracking.enabled": "true",
"data-evolution.enabled": "true",
@@ -641,7 +639,6 @@ class TestBlobType:
paimon_schema = pypaimon.Schema.from_pyarrow_schema(
pa_schema,
options={
- "bucket": "1",
"file.format": "parquet",
"row-tracking.enabled": "true",
"data-evolution.enabled": "true",
diff --git a/paimon-python/pypaimon/tests/global_index_build_test.py
b/paimon-python/pypaimon/tests/global_index_build_test.py
index cd6c21bb55..687f45fcd9 100644
--- a/paimon-python/pypaimon/tests/global_index_build_test.py
+++ b/paimon-python/pypaimon/tests/global_index_build_test.py
@@ -1075,15 +1075,12 @@ class GlobalIndexBuildTest(
('id', pa.int32()),
('embedding', pa.list_(pa.float32())),
])
+ # Row tracking tables with a fixed bucket are rejected at schema
validation,
+ # so they can no longer reach the global index builder.
bucket_options = dict(self.table_options)
bucket_options['bucket'] = '1'
- bucket_table = self._create_table(pa_schema=schema,
options=bucket_options)
- with self.assertRaisesRegex(ValueError, 'unaware-bucket'):
- bucket_table.create_global_index(
- 'embedding',
- index_type='ivf-flat',
- options={'ivf-flat.dimension': '2'},
- )
+ with self.assertRaisesRegex(ValueError, 'it only support bucket = -1'):
+ self._create_table(pa_schema=schema, options=bucket_options)
dv_options = dict(self.table_options)
dv_options['deletion-vectors.enabled'] = 'true'
diff --git a/paimon-python/pypaimon/tests/manifest/manifest_sidecar_test.py
b/paimon-python/pypaimon/tests/manifest/manifest_sidecar_test.py
index 868b1518cd..cc8da04d42 100644
--- a/paimon-python/pypaimon/tests/manifest/manifest_sidecar_test.py
+++ b/paimon-python/pypaimon/tests/manifest/manifest_sidecar_test.py
@@ -568,6 +568,7 @@ class
ManifestSidecarScanTest(existing.ManifestEntryIdentifierTest):
pa.schema([('p', pa.int32()), ('q', pa.string()), ('value',
pa.string())]),
partition_keys=['p', 'q'],
options={'manifest.sidecar.enabled': 'true',
+ 'row-tracking.enabled': 'true',
'data-evolution.enabled': 'true'})
self.catalog.create_table('default.explain_sidecar', schema, False)
self.table = self.catalog.get_table('default.explain_sidecar')
diff --git a/paimon-python/pypaimon/tests/multimodal_table_test.py
b/paimon-python/pypaimon/tests/multimodal_table_test.py
index 13d306269c..629fffbca0 100644
--- a/paimon-python/pypaimon/tests/multimodal_table_test.py
+++ b/paimon-python/pypaimon/tests/multimodal_table_test.py
@@ -29,6 +29,7 @@ import pyarrow as pa
import pytest
import pypaimon.multimodal as pmm
from pypaimon.multimodal import source_col
+from pypaimon.common.json_util import JSON
from pypaimon.common.predicate_builder import PredicateBuilder
from pypaimon import Schema as PaimonSchema
from pypaimon.globalindex.global_index_result import GlobalIndexResult
@@ -1189,13 +1190,15 @@ class MultimodalTableTest(unittest.TestCase):
def test_get_table_rejects_primary_key_table(self):
self.conn.catalog.create_database("default", ignore_if_exists=True)
self.conn.catalog.create_table(
- "default.pk",
- _raw_schema(
- options=dict(_PARQUET_OPTIONS, **{"bucket": "1"}),
- primary_keys=["id"],
- ),
- False,
- )
+ "default.pk", _raw_schema(options=_PARQUET_OPTIONS), False)
+ # Row tracking tables cannot define primary keys, so such a table can
+ # only come from writers that skipped schema validation. Simulate it
+ # by rewriting the committed schema file.
+ table = self.conn.catalog.get_table("default.pk")
+ legacy_schema = replace(table.table_schema, primary_keys=["id"])
+ table.file_io.overwrite_file_utf8(
+ table.schema_manager._to_schema_path(legacy_schema.id),
+ JSON.to_json(legacy_schema, indent=2))
with self.assertRaisesRegex(ValueError, "primary keys"):
self.conn.get_table("pk")
diff --git a/paimon-python/pypaimon/tests/ray_data_evolution_merge_into_test.py
b/paimon-python/pypaimon/tests/ray_data_evolution_merge_into_test.py
index 84fdbd0d89..461408495b 100644
--- a/paimon-python/pypaimon/tests/ray_data_evolution_merge_into_test.py
+++ b/paimon-python/pypaimon/tests/ray_data_evolution_merge_into_test.py
@@ -441,16 +441,12 @@ class RayDataEvolutionMergeIntoTest(unittest.TestCase):
self.assertIn('data-evolution.enabled', str(ctx.exception))
def test_no_row_tracking_rejected(self):
- target = self._create_table(options={'data-evolution.enabled': 'true'})
+ # Data evolution requires row tracking. Schema validation now rejects
+ # this combination at table creation (aligned with Java
+ # SchemaValidation#validateRowTracking), so the table can no longer be
+ # created and merge_into never gets a chance to run.
with self.assertRaises(ValueError) as ctx:
- merge_into(
- target=target,
- source=self._source(),
- catalog_options=self.catalog_options,
- on=['id'],
- when_matched=[WhenMatched.update('*')],
- num_partitions=_TEST_NUM_PARTITIONS,
- )
+ self._create_table(options={'data-evolution.enabled': 'true'})
self.assertIn('row-tracking.enabled', str(ctx.exception))
def test_source_missing_on_col_raises(self):
diff --git a/paimon-python/pypaimon/tests/schema_manager_test.py
b/paimon-python/pypaimon/tests/schema_manager_test.py
index f58d8876c5..dd73d9142d 100644
--- a/paimon-python/pypaimon/tests/schema_manager_test.py
+++ b/paimon-python/pypaimon/tests/schema_manager_test.py
@@ -295,5 +295,96 @@ class TestOptionValidation(unittest.TestCase):
self.assertNotIn(key, manager.latest().options)
+class TestRowTrackingValidation(unittest.TestCase):
+ """Row tracking tables must follow Java
SchemaValidation#validateRowTracking."""
+
+ def setUp(self):
+ self.temp_dir = tempfile.mkdtemp()
+ self.file_io = FileIO.get(self.temp_dir, Options({}))
+ self.table_path = f"{self.temp_dir}/test_db.db/test_table"
+
+ def tearDown(self):
+ import shutil
+ shutil.rmtree(self.temp_dir, ignore_errors=True)
+
+ @staticmethod
+ def _schema(options=None, primary_keys=None):
+ return Schema(
+ fields=[
+ DataField(0, "id", AtomicType("INT", False)),
+ DataField(1, "name", AtomicType("STRING")),
+ ],
+ primary_keys=primary_keys or [],
+ options=options or {},
+ )
+
+ def test_create_rejects_row_tracking_with_fixed_or_postpone_bucket(self):
+ manager = SchemaManager(self.file_io, self.table_path)
+ for bucket in ["1", "16", "-2"]:
+ with self.subTest(bucket=bucket):
+ with self.assertRaisesRegex(
+ ValueError,
+ "Cannot define bucket for row tracking table, "
+ "it only support bucket = -1",
+ ):
+ manager.create_table(self._schema({
+ "row-tracking.enabled": "true",
+ "bucket": bucket,
+ }))
+ self.assertIsNone(manager.latest())
+
+ def test_create_rejects_row_tracking_with_primary_key(self):
+ manager = SchemaManager(self.file_io, self.table_path)
+ with self.assertRaisesRegex(
+ ValueError, "Cannot define primary-key for row tracking table"
+ ):
+ manager.create_table(self._schema(
+ {"row-tracking.enabled": "true"}, primary_keys=["id"]))
+ self.assertIsNone(manager.latest())
+
+ def test_create_allows_row_tracking_with_unaware_bucket(self):
+ for options in [
+ {"row-tracking.enabled": "true"},
+ {"row-tracking.enabled": "true", "bucket": "-1"},
+ {"row-tracking.enabled": "true", "data-evolution.enabled": "true"},
+ ]:
+ with self.subTest(options=options):
+ temp_dir = tempfile.mkdtemp(dir=self.temp_dir)
+ manager = SchemaManager(self.file_io, f"{temp_dir}/t")
+ manager.create_table(self._schema(options))
+ self.assertEqual(manager.latest().id, 0)
+
+ def test_create_rejects_data_evolution_without_row_tracking(self):
+ manager = SchemaManager(self.file_io, self.table_path)
+ for options in [
+ {"data-evolution.enabled": "true"},
+ {"data-evolution.enabled": "true", "row-tracking.enabled":
"false"},
+ ]:
+ with self.subTest(options=options):
+ with self.assertRaisesRegex(
+ ValueError,
+ "Data evolution config must enabled with
row-tracking.enabled",
+ ):
+ manager.create_table(self._schema(options))
+ self.assertIsNone(manager.latest())
+
+ def test_create_allows_fixed_bucket_without_row_tracking(self):
+ manager = SchemaManager(self.file_io, self.table_path)
+ manager.create_table(self._schema({"bucket": "4"},
primary_keys=["id"]))
+ self.assertEqual(manager.latest().id, 0)
+
+ def test_alter_rejects_enabling_row_tracking_on_fixed_bucket_table(self):
+ manager = SchemaManager(self.file_io, self.table_path)
+ manager.create_table(self._schema({"bucket": "4"},
primary_keys=["id"]))
+
+ with self.assertRaisesRegex(
+ RuntimeError, "Cannot define primary-key for row tracking table"
+ ):
+ manager.commit_changes([SetOption("row-tracking.enabled", "true")])
+
+ self.assertEqual(manager.latest().id, 0)
+ self.assertNotIn("row-tracking.enabled", manager.latest().options)
+
+
if __name__ == '__main__':
unittest.main()
