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 77c568940e [python] Make multimodal table creation race-safe (#9590)
77c568940e is described below
commit 77c568940e5f1a938bdd005fd9d343812b4fba2c
Author: XiaoHongbo <[email protected]>
AuthorDate: Thu Sep 3 21:05:18 2026 +0800
[python] Make multimodal table creation race-safe (#9590)
---
paimon-python/pypaimon/multimodal/connection.py | 28 ++++-
.../pypaimon/tests/multimodal_table_test.py | 124 +++++++++++++++++++++
2 files changed, 148 insertions(+), 4 deletions(-)
diff --git a/paimon-python/pypaimon/multimodal/connection.py
b/paimon-python/pypaimon/multimodal/connection.py
index 8f51cdaf97..f69a723920 100644
--- a/paimon-python/pypaimon/multimodal/connection.py
+++ b/paimon-python/pypaimon/multimodal/connection.py
@@ -71,18 +71,35 @@ class MultimodalConnection:
"""Create a multimodal table and optionally add initial data."""
identifier = self._identifier(name)
already_exists = _table_exists(self.catalog, identifier)
- paimon_schema = _to_paimon_schema(schema, data, options, partitioned)
+ if already_exists and ignore_if_exists:
+ try:
+ return self.get_table(name)
+ except (DatabaseNotExistException, TableNotExistException):
+ pass
+ try:
+ paimon_schema = _to_paimon_schema(
+ schema, data, options, partitioned)
+ _validate_multimodal_schema(paimon_schema, identifier)
+ except ValueError:
+ if ignore_if_exists:
+ try:
+ return self.get_table(name)
+ except (DatabaseNotExistException, TableNotExistException):
+ pass
+ raise
self._create_database_for(identifier)
+ created = False
try:
self.catalog.create_table(
- identifier, paimon_schema, ignore_if_exists)
+ identifier, paimon_schema, False)
+ created = True
except TableAlreadyExistException:
if not ignore_if_exists:
raise
table = self.get_table(name)
- if data is not None and not already_exists:
+ if data is not None and created:
table.add(data)
return table
@@ -193,7 +210,10 @@ def _table_exists(catalog, identifier: str) -> bool:
def _validate_multimodal_table(table, identifier: str):
- table_schema = table.table_schema
+ _validate_multimodal_schema(table.table_schema, identifier)
+
+
+def _validate_multimodal_schema(table_schema, identifier: str):
options = table_schema.options
if str(options.get("data-evolution.enabled", "false")).lower() != "true":
raise ValueError(
diff --git a/paimon-python/pypaimon/tests/multimodal_table_test.py
b/paimon-python/pypaimon/tests/multimodal_table_test.py
index e99fe19437..50975a3892 100644
--- a/paimon-python/pypaimon/tests/multimodal_table_test.py
+++ b/paimon-python/pypaimon/tests/multimodal_table_test.py
@@ -760,6 +760,130 @@ class MultimodalTableTest(unittest.TestCase):
with self.assertRaisesRegex(ValueError, "primary keys"):
self.conn.get_table("pk")
+ def test_create_table_ignores_invalid_options_when_table_exists(self):
+ schema = _schema({"id": pa.int32()})
+ expected = self.conn.create_table("existing", schema=schema)
+
+ actual = self.conn.create_table(
+ "existing",
+ schema=_schema({
+ "id": pa.int32(),
+ "embedding": _vector(3),
+ }),
+ options={"data-evolution.enabled": "false"},
+ ignore_if_exists=True,
+ )
+
+ self.assertEqual(expected.identifier, actual.identifier)
+ with patch(
+ "pypaimon.multimodal.connection._table_exists",
+ return_value=False):
+ raced = self.conn.create_table(
+ "existing",
+ schema=_schema({
+ "id": pa.int32(),
+ "embedding": _vector(3),
+ }),
+ options={"data-evolution.enabled": "false"},
+ ignore_if_exists=True,
+ )
+ self.assertEqual(expected.identifier, raced.identifier)
+
+ def test_create_table_handles_concurrent_delete_when_ignoring(self):
+ schema = _schema({"id": pa.int32()})
+ self.conn.create_table("deleted", schema=schema)
+ original_get = self.conn.get_table
+ deleted = [False]
+
+ def delete_once(name):
+ if not deleted[0]:
+ deleted[0] = True
+ self.conn.catalog.drop_table("default.deleted", False)
+ return original_get(name)
+
+ with patch.object(
+ self.conn, "get_table", side_effect=delete_once):
+ table = self.conn.create_table(
+ "deleted",
+ data=pa.table({"id": [1, 2, 3]}),
+ schema=schema,
+ ignore_if_exists=True,
+ )
+ self.assertEqual("default.deleted", table.identifier)
+ self.assertEqual(
+ [1, 2, 3], table.scan().to_arrow()["id"].to_pylist())
+
+ self.conn.create_table("fallback_deleted", schema=schema)
+ deleted[0] = False
+
+ def delete_fallback_once(name):
+ if not deleted[0]:
+ deleted[0] = True
+ self.conn.catalog.drop_table(
+ "default.fallback_deleted", False)
+ return original_get(name)
+
+ with patch(
+ "pypaimon.multimodal.connection._table_exists",
+ return_value=False):
+ with patch.object(
+ self.conn,
+ "get_table",
+ side_effect=delete_fallback_once):
+ with self.assertRaisesRegex(
+ ValueError, "data-evolution.enabled"):
+ self.conn.create_table(
+ "fallback_deleted",
+ schema=schema,
+ options={"data-evolution.enabled": "false"},
+ ignore_if_exists=True,
+ )
+
+ def test_create_table_does_not_add_data_to_concurrent_winner(self):
+ schema = _schema({"id": pa.int32()})
+ winner = self.conn.create_table("winner", schema=schema)
+
+ with patch(
+ "pypaimon.multimodal.connection._table_exists",
+ return_value=False):
+ actual = self.conn.create_table(
+ "winner",
+ data=pa.table({"id": [1, 2, 3]}),
+ schema=schema,
+ ignore_if_exists=True,
+ )
+
+ self.assertEqual(winner.identifier, actual.identifier)
+ self.assertEqual([], actual.scan().to_arrow().to_pylist())
+
+ def test_create_table_preserves_unknown_create_error(self):
+ schema = _schema({"id": pa.int32()})
+ create = self.conn.catalog.create_table
+ create_error = TimeoutError("create response lost")
+
+ def create_then_lose_response(*args, **kwargs):
+ create(*args, **kwargs)
+ raise create_error
+
+ with patch.object(
+ self.conn.catalog,
+ "create_table",
+ side_effect=create_then_lose_response):
+ with self.assertRaises(TimeoutError) as context:
+ self.conn.create_table(
+ "failed_create",
+ data=pa.table({"id": [1, 2, 3]}),
+ schema=schema,
+ ignore_if_exists=True,
+ )
+
+ self.assertIs(create_error, context.exception)
+ self.assertEqual(
+ [],
+ self.conn.get_table("failed_create")
+ .scan().to_arrow().to_pylist(),
+ )
+
def test_create_table_can_add_initial_data_and_get_by_short_name(self):
self.conn.create_table(
"users",
