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 49ffa3f9dc [python] Avoid concurrent import failures during split
deserialization (#9250)
49ffa3f9dc is described below
commit 49ffa3f9dc9f324033421cf94018e0762e2d7247
Author: XiaoHongbo <[email protected]>
AuthorDate: Mon Aug 17 12:18:39 2026 +0800
[python] Avoid concurrent import failures during split deserialization
(#9250)
---
paimon-python/pypaimon/__init__.py | 44 ++++-
paimon-python/pypaimon/globalindex/__init__.py | 113 +++++++++----
paimon-python/pypaimon/tag/__init__.py | 32 +++-
.../pypaimon/tests/concurrent_import_test.py | 187 +++++++++++++++++++++
4 files changed, 336 insertions(+), 40 deletions(-)
diff --git a/paimon-python/pypaimon/__init__.py
b/paimon-python/pypaimon/__init__.py
index aa18f812e7..017dfa5303 100644
--- a/paimon-python/pypaimon/__init__.py
+++ b/paimon-python/pypaimon/__init__.py
@@ -15,7 +15,9 @@
# specific language governing permissions and limitations
# under the License.
+import importlib
import sys
+import threading
if sys.version_info[:2] < (3, 8):
try:
@@ -23,11 +25,13 @@ if sys.version_info[:2] < (3, 8):
except ImportError:
pass
-from pypaimon.catalog.catalog_factory import CatalogFactory
-from pypaimon.filesystem.pvfs import PaimonVirtualFileSystem
-from pypaimon.schema.schema import Schema
-from pypaimon.tag.tag import Tag
-from pypaimon.tag.tag_manager import TagManager
+if sys.version_info[:2] < (3, 7):
+ # Module-level __getattr__ is unavailable before Python 3.7.
+ from pypaimon.catalog.catalog_factory import CatalogFactory
+ from pypaimon.filesystem.pvfs import PaimonVirtualFileSystem
+ from pypaimon.schema.schema import Schema
+ from pypaimon.tag.tag import Tag
+ from pypaimon.tag.tag_manager import TagManager
__all__ = [
"PaimonVirtualFileSystem",
@@ -38,9 +42,31 @@ __all__ = [
"SQLContext",
]
+_LAZY_EXPORTS = {
+ "CatalogFactory": ("pypaimon.catalog.catalog_factory", "CatalogFactory"),
+ "PaimonVirtualFileSystem": (
+ "pypaimon.filesystem.pvfs", "PaimonVirtualFileSystem",
+ ),
+ "Schema": ("pypaimon.schema.schema", "Schema"),
+ "Tag": ("pypaimon.tag.tag", "Tag"),
+ "TagManager": ("pypaimon.tag.tag_manager", "TagManager"),
+ "SQLContext": ("pypaimon_rust.datafusion", "SQLContext"),
+}
+
+# Unsynchronized lazy imports from two threads can acquire module locks in
+# opposite orders (pypaimon.tag initializes both tag and tag_manager) and
+# fail with _DeadlockError; the eager imports previously serialized this on
+# the root module lock.
+_LAZY_IMPORT_LOCK = threading.RLock()
+
def __getattr__(name):
- if name == "SQLContext":
- from pypaimon_rust.datafusion import SQLContext
- return SQLContext
- raise AttributeError("module 'pypaimon' has no attribute {}".format(name))
+ target = _LAZY_EXPORTS.get(name)
+ if target is None:
+ raise AttributeError(
+ "module 'pypaimon' has no attribute {}".format(name))
+ with _LAZY_IMPORT_LOCK:
+ if name not in globals():
+ module = importlib.import_module(target[0])
+ globals()[name] = getattr(module, target[1])
+ return globals()[name]
diff --git a/paimon-python/pypaimon/globalindex/__init__.py
b/paimon-python/pypaimon/globalindex/__init__.py
index b9ef7c9a28..377b0a900a 100644
--- a/paimon-python/pypaimon/globalindex/__init__.py
+++ b/paimon-python/pypaimon/globalindex/__init__.py
@@ -15,35 +15,41 @@
# specific language governing permissions and limitations
# under the License.
-from pypaimon.globalindex.global_index_result import GlobalIndexResult
-from pypaimon.globalindex.global_index_reader import GlobalIndexReader,
FieldRef
-from pypaimon.globalindex.vector_search import VectorSearch
-from pypaimon.globalindex.full_text_search import FullTextSearch
-from pypaimon.globalindex.vector_search_result import (
- ScoredGlobalIndexResult,
- DictBasedScoredIndexResult,
- ScoreGetter,
-)
-from pypaimon.globalindex.global_index_meta import GlobalIndexMeta,
GlobalIndexIOMeta
-from pypaimon.globalindex.global_index_evaluator import GlobalIndexEvaluator
-from pypaimon.globalindex.data_evolution_global_index_scanner import (
- DataEvolutionGlobalIndexScanner,
-)
-from pypaimon.globalindex.key_serializer import KeySerializer
-from pypaimon.globalindex.memory_slice_input import MemorySliceInput
-from pypaimon.globalindex.offset_global_index_reader import
OffsetGlobalIndexReader
-from pypaimon.globalindex.sorted_file_global_index_reader import
SortedFileGlobalIndexReader
-from pypaimon.globalindex.sorted_file_meta_selector import
SortedFileMetaSelector
-from pypaimon.globalindex.sorted_index_file_meta import SortedIndexFileMeta
-from pypaimon.globalindex.create_global_index import (
- GlobalIndexBuilder,
- create_global_index,
-)
-from pypaimon.globalindex.drop_global_index import (
- GlobalIndexDropper,
- drop_global_index,
-)
-from pypaimon.utils.range import Range
+import importlib
+import sys
+import threading
+
+if sys.version_info[:2] < (3, 7):
+ # Module-level __getattr__ is unavailable before Python 3.7.
+ from pypaimon.globalindex.global_index_result import GlobalIndexResult
+ from pypaimon.globalindex.global_index_reader import GlobalIndexReader,
FieldRef
+ from pypaimon.globalindex.vector_search import VectorSearch
+ from pypaimon.globalindex.full_text_search import FullTextSearch
+ from pypaimon.globalindex.vector_search_result import (
+ ScoredGlobalIndexResult,
+ DictBasedScoredIndexResult,
+ ScoreGetter,
+ )
+ from pypaimon.globalindex.global_index_meta import GlobalIndexMeta,
GlobalIndexIOMeta
+ from pypaimon.globalindex.global_index_evaluator import
GlobalIndexEvaluator
+ from pypaimon.globalindex.data_evolution_global_index_scanner import (
+ DataEvolutionGlobalIndexScanner,
+ )
+ from pypaimon.globalindex.key_serializer import KeySerializer
+ from pypaimon.globalindex.memory_slice_input import MemorySliceInput
+ from pypaimon.globalindex.offset_global_index_reader import
OffsetGlobalIndexReader
+ from pypaimon.globalindex.sorted_file_global_index_reader import
SortedFileGlobalIndexReader
+ from pypaimon.globalindex.sorted_file_meta_selector import
SortedFileMetaSelector
+ from pypaimon.globalindex.sorted_index_file_meta import SortedIndexFileMeta
+ from pypaimon.globalindex.create_global_index import (
+ GlobalIndexBuilder,
+ create_global_index,
+ )
+ from pypaimon.globalindex.drop_global_index import (
+ GlobalIndexDropper,
+ drop_global_index,
+ )
+ from pypaimon.utils.range import Range
__all__ = [
'GlobalIndexResult',
@@ -70,3 +76,52 @@ __all__ = [
'drop_global_index',
'Range',
]
+
+_MODULE_BY_EXPORT = {
+ 'GlobalIndexResult': 'pypaimon.globalindex.global_index_result',
+ 'GlobalIndexReader': 'pypaimon.globalindex.global_index_reader',
+ 'FieldRef': 'pypaimon.globalindex.global_index_reader',
+ 'VectorSearch': 'pypaimon.globalindex.vector_search',
+ 'FullTextSearch': 'pypaimon.globalindex.full_text_search',
+ 'ScoredGlobalIndexResult': 'pypaimon.globalindex.vector_search_result',
+ 'DictBasedScoredIndexResult': 'pypaimon.globalindex.vector_search_result',
+ 'ScoreGetter': 'pypaimon.globalindex.vector_search_result',
+ 'GlobalIndexMeta': 'pypaimon.globalindex.global_index_meta',
+ 'GlobalIndexIOMeta': 'pypaimon.globalindex.global_index_meta',
+ 'GlobalIndexEvaluator': 'pypaimon.globalindex.global_index_evaluator',
+ 'DataEvolutionGlobalIndexScanner':
+ 'pypaimon.globalindex.data_evolution_global_index_scanner',
+ 'KeySerializer': 'pypaimon.globalindex.key_serializer',
+ 'MemorySliceInput': 'pypaimon.globalindex.memory_slice_input',
+ 'OffsetGlobalIndexReader':
+ 'pypaimon.globalindex.offset_global_index_reader',
+ 'SortedFileGlobalIndexReader':
+ 'pypaimon.globalindex.sorted_file_global_index_reader',
+ 'SortedFileMetaSelector':
+ 'pypaimon.globalindex.sorted_file_meta_selector',
+ 'SortedIndexFileMeta': 'pypaimon.globalindex.sorted_index_file_meta',
+ 'GlobalIndexBuilder': 'pypaimon.globalindex.create_global_index',
+ 'create_global_index': 'pypaimon.globalindex.create_global_index',
+ 'GlobalIndexDropper': 'pypaimon.globalindex.drop_global_index',
+ 'drop_global_index': 'pypaimon.globalindex.drop_global_index',
+ 'Range': 'pypaimon.utils.range',
+}
+
+# Eagerly importing the exports above builds a circular chain: submodules
+# such as index_file_meta initialize this package on first import, while
+# create_global_index and the scanner chain import those submodules back.
+# Lazy resolution keeps this package init trivial; the lock serializes
+# first-time imports racing from multiple threads.
+_LAZY_IMPORT_LOCK = threading.RLock()
+
+
+def __getattr__(name):
+ module_name = _MODULE_BY_EXPORT.get(name)
+ if module_name is None:
+ raise AttributeError(
+ "module 'pypaimon.globalindex' has no attribute {}".format(name))
+ with _LAZY_IMPORT_LOCK:
+ if name not in globals():
+ module = importlib.import_module(module_name)
+ globals()[name] = getattr(module, name)
+ return globals()[name]
diff --git a/paimon-python/pypaimon/tag/__init__.py
b/paimon-python/pypaimon/tag/__init__.py
index 8f51bf5602..384c154797 100644
--- a/paimon-python/pypaimon/tag/__init__.py
+++ b/paimon-python/pypaimon/tag/__init__.py
@@ -15,7 +15,35 @@
# specific language governing permissions and limitations
# under the License.
-from pypaimon.tag.tag import Tag
-from pypaimon.tag.tag_manager import TagManager
+import importlib
+import sys
+import threading
+
+if sys.version_info[:2] < (3, 7):
+ # Module-level __getattr__ is unavailable before Python 3.7.
+ from pypaimon.tag.tag import Tag
+ from pypaimon.tag.tag_manager import TagManager
__all__ = ["Tag", "TagManager"]
+
+_MODULE_BY_EXPORT = {
+ "Tag": "pypaimon.tag.tag",
+ "TagManager": "pypaimon.tag.tag_manager",
+}
+
+# Eagerly importing both siblings here lets two threads acquire the module
+# locks of tag and tag_manager in opposite orders and fail with
+# _DeadlockError; lazy resolution keeps this package init trivial.
+_LAZY_IMPORT_LOCK = threading.RLock()
+
+
+def __getattr__(name):
+ module_name = _MODULE_BY_EXPORT.get(name)
+ if module_name is None:
+ raise AttributeError(
+ "module 'pypaimon.tag' has no attribute {}".format(name))
+ with _LAZY_IMPORT_LOCK:
+ if name not in globals():
+ module = importlib.import_module(module_name)
+ globals()[name] = getattr(module, name)
+ return globals()[name]
diff --git a/paimon-python/pypaimon/tests/concurrent_import_test.py
b/paimon-python/pypaimon/tests/concurrent_import_test.py
new file mode 100644
index 0000000000..a1ccb428ae
--- /dev/null
+++ b/paimon-python/pypaimon/tests/concurrent_import_test.py
@@ -0,0 +1,187 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import base64
+import os
+import pickle
+import subprocess
+import sys
+import unittest
+
+from pypaimon.manifest.schema.data_file_meta import DataFileMeta
+from pypaimon.manifest.schema.simple_stats import SimpleStats
+from pypaimon.read.query_auth_split import QueryAuthSplit
+from pypaimon.read.split import DataSplit
+from pypaimon.table.row.generic_row import GenericRow
+
+
+class ConcurrentImportTest(unittest.TestCase):
+
+ @unittest.skipIf(
+ sys.version_info[:2] < (3, 7),
+ "module-level lazy attributes require Python 3.7+",
+ )
+ def test_concurrent_query_auth_split_deserialization(self):
+ empty = GenericRow([], [])
+ file_meta = DataFileMeta.create(
+ file_name="data.parquet",
+ file_size=1,
+ row_count=1,
+ min_key=empty,
+ max_key=empty,
+ key_stats=SimpleStats.empty_stats(),
+ value_stats=SimpleStats.empty_stats(),
+ min_sequence_number=0,
+ max_sequence_number=0,
+ schema_id=0,
+ level=0,
+ extra_files=[],
+ first_row_id=0,
+ )
+ payload = base64.b64encode(pickle.dumps(QueryAuthSplit(
+ DataSplit([file_meta], empty, 0, snapshot_id=1), None,
+ ))).decode("ascii")
+ script = r"""
+import base64
+import importlib
+import pickle
+import sys
+import threading
+
+payload = base64.b64decode(sys.argv[1])
+barrier = threading.Barrier(24)
+errors = []
+modules = [
+ "pypaimon.manifest.schema.data_file_meta",
+ "pypaimon.manifest.schema.simple_stats",
+ "pypaimon.read.query_auth_split",
+]
+
+
+def deserialize(index):
+ try:
+ barrier.wait()
+ if index < len(modules):
+ importlib.import_module(modules[index])
+ for _ in range(20):
+ pickle.loads(payload)
+ except BaseException as error:
+ errors.append(repr(error))
+
+
+threads = [
+ threading.Thread(target=deserialize, args=(index,))
+ for index in range(24)
+]
+for thread in threads:
+ thread.start()
+for thread in threads:
+ thread.join()
+if errors:
+ print("\n".join(errors))
+ raise SystemExit(1)
+"""
+ result = subprocess.run(
+ [sys.executable, "-c", script, payload],
+ env=os.environ.copy(),
+ stdout=subprocess.PIPE,
+ stderr=subprocess.STDOUT,
+ universal_newlines=True,
+ timeout=30,
+ )
+ self.assertEqual(0, result.returncode, result.stdout)
+
+ @unittest.skipIf(
+ sys.version_info[:2] < (3, 7),
+ "module-level lazy attributes require Python 3.7+",
+ )
+ def test_concurrent_top_level_lazy_exports(self):
+ script = r"""
+import sys
+import threading
+
+sys.setswitchinterval(1e-6)
+thread_count = 16
+barrier = threading.Barrier(thread_count)
+errors = []
+statements = ["from pypaimon import Tag", "from pypaimon import TagManager"]
+
+
+def resolve(statement):
+ try:
+ barrier.wait()
+ exec(statement)
+ except BaseException as error:
+ errors.append(repr(error))
+
+
+threads = [
+ threading.Thread(target=resolve, args=(statements[index % 2],))
+ for index in range(thread_count)
+]
+for thread in threads:
+ thread.start()
+for thread in threads:
+ thread.join(15)
+if errors:
+ print("\n".join(sorted(set(errors))))
+ raise SystemExit(1)
+"""
+ # The lazy names resolve once per process, so each attempt needs a
+ # fresh interpreter.
+ for attempt in range(8):
+ result = subprocess.run(
+ [sys.executable, "-c", script],
+ env=os.environ.copy(),
+ stdout=subprocess.PIPE,
+ stderr=subprocess.STDOUT,
+ universal_newlines=True,
+ timeout=30,
+ )
+ self.assertEqual(
+ 0, result.returncode,
+ "attempt {}: {}".format(attempt, result.stdout),
+ )
+
+ def test_fresh_import_of_cycle_prone_leaf_modules(self):
+ # Each module must be importable as the very first pypaimon import
+ # of a process. Heavy package initializers used to pull circular
+ # chains (index_file_meta -> globalindex -> create_global_index)
+ # that the eager root import happened to mask.
+ for module in [
+ "pypaimon.index.index_file_meta",
+ "pypaimon.manifest.index_manifest_entry",
+ "pypaimon.read.scanner.bucket_select_converter",
+ "pypaimon.table.data_evolution_merge_into",
+ "pypaimon.write.table_delete",
+ "pypaimon.tag.tag_manager",
+ ]:
+ result = subprocess.run(
+ [sys.executable, "-c", "import " + module],
+ env=os.environ.copy(),
+ stdout=subprocess.PIPE,
+ stderr=subprocess.STDOUT,
+ universal_newlines=True,
+ timeout=60,
+ )
+ self.assertEqual(
+ 0, result.returncode,
+ "{}: {}".format(module, result.stdout),
+ )
+
+
+if __name__ == "__main__":
+ unittest.main()
