diff --git a/paimon-python/pypaimon/__init__.py b/paimon-python/pypaimon/__init__.py index aa18f812e790..017dfa53032c 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 @@ 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 @@ "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 b9ef7c9a28b3..377b0a900a07 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 @@ '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 8f51bf5602e5..384c154797c7 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 000000000000..a1ccb428ae63 --- /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()