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 e088b934b5 [python] Propagate native fork-safety errors (#10328)
e088b934b5 is described below
commit e088b934b5831f95da5bbb35dea19c57ab0dc611
Author: XiaoHongbo <[email protected]>
AuthorDate: Tue Oct 6 23:09:58 2026 +0800
[python] Propagate native fork-safety errors (#10328)
---
docs/docs/pypaimon/pyjindosdk-support.md | 3 +
paimon-python/pypaimon/read/native_plan.py | 16 +++++
.../pypaimon/read/streaming_table_scan.py | 12 ++--
paimon-python/pypaimon/read/table_read.py | 7 +-
paimon-python/pypaimon/read/table_scan.py | 4 +-
paimon-python/pypaimon/tests/native_plan_test.py | 15 ++++
paimon-python/pypaimon/tests/native_read_test.py | 18 +++++
.../pypaimon/tests/streaming_table_scan_test.py | 79 ++++++++++++++++++++++
8 files changed, 148 insertions(+), 6 deletions(-)
diff --git a/docs/docs/pypaimon/pyjindosdk-support.md
b/docs/docs/pypaimon/pyjindosdk-support.md
index 32ca06816b..9537622c28 100644
--- a/docs/docs/pypaimon/pyjindosdk-support.md
+++ b/docs/docs/pypaimon/pyjindosdk-support.md
@@ -24,6 +24,9 @@ pip install pyjindosdk
Once installed, PyPaimon will automatically use PyJindoSDK as the default file
I/O implementation for accessing OSS. No additional configuration is required.
+With multiprocessing, use `spawn` or initialize PyPaimon inside each worker;
+JindoSDK state created before `fork` cannot be reused.
+
### Fallback to Legacy Implementation
Since JindoSDK is a native implementation, pre-built Python packages may not
be available for all OS or platform versions. If you need to fall back to the
legacy PyArrow-based implementation for any reason, there are two ways to do so:
diff --git a/paimon-python/pypaimon/read/native_plan.py
b/paimon-python/pypaimon/read/native_plan.py
index e4eb38cec8..fbad1d8c37 100644
--- a/paimon-python/pypaimon/read/native_plan.py
+++ b/paimon-python/pypaimon/read/native_plan.py
@@ -40,6 +40,22 @@ from pypaimon.read.split_serializer import (
from pypaimon.schema.data_types import DataField, RowType
+def _raise_if_native_fork_safety_error(error):
+ """Propagate native fork errors instead of using inherited SDK state."""
+ try:
+ from pypaimon_rust import ForkSafetyError
+ except ImportError:
+ return
+
+ current = error
+ seen = set()
+ while current is not None and id(current) not in seen:
+ seen.add(id(current))
+ if isinstance(current, ForkSafetyError):
+ raise current
+ current = current.__cause__ or current.__context__
+
+
def native_runtime_available() -> bool:
"""Whether an installed pypaimon-rust exposes the full split-planning API.
diff --git a/paimon-python/pypaimon/read/streaming_table_scan.py
b/paimon-python/pypaimon/read/streaming_table_scan.py
index 2481293a8c..cc904e3961 100644
--- a/paimon-python/pypaimon/read/streaming_table_scan.py
+++ b/paimon-python/pypaimon/read/streaming_table_scan.py
@@ -35,6 +35,7 @@ from pypaimon.consumer.consumer import Consumer
from pypaimon.consumer.consumer_manager import ConsumerManager
from pypaimon.manifest.manifest_file_manager import ManifestFileManager
from pypaimon.manifest.manifest_list_manager import ManifestListManager
+from pypaimon.read.native_plan import _raise_if_native_fork_safety_error
from pypaimon.read.plan import Plan
from pypaimon.read.query_auth_split import wrap_plan_with_auth
from pypaimon.read.scanner.append_table_split_generator import \
@@ -194,7 +195,8 @@ class AsyncStreamingTableScan:
if prefetch_plan is not None:
plan = prefetch_plan
- except Exception:
+ except Exception as error:
+ _raise_if_native_fork_safety_error(error)
# Prefetch failed, fall back to synchronous
prefetch_used = False
finally:
@@ -304,7 +306,8 @@ class AsyncStreamingTableScan:
plan = self._create_follow_up_plan(snapshot)
return (plan, next_id, skipped_count)
- except Exception:
+ except Exception as error:
+ _raise_if_native_fork_safety_error(error)
logging.exception("Prefetch failed for snapshot_id=%d; falling
back to synchronous", start_id)
return None
@@ -372,8 +375,8 @@ class AsyncStreamingTableScan:
An arbitrary Python bucket predicate cannot be represented by the Rust
planner, so sharded stream consumers retain the Python plan. Any native
- setup or planning error is an optimization miss and falls back before
- a plan is returned.
+ setup or planning error other than a fork-safety failure falls back
+ before a plan is returned.
"""
if (not self.table.options.native_plan_enabled()
or self._bucket_filter is not None):
@@ -405,6 +408,7 @@ class AsyncStreamingTableScan:
return None
return plan
except Exception as error:
+ _raise_if_native_fork_safety_error(error)
logging.warning(
"Native streaming plan failed, falling back to Python
planning: %s",
error)
diff --git a/paimon-python/pypaimon/read/table_read.py
b/paimon-python/pypaimon/read/table_read.py
index 2f52c1fb9c..0c572ef921 100644
--- a/paimon-python/pypaimon/read/table_read.py
+++ b/paimon-python/pypaimon/read/table_read.py
@@ -525,7 +525,8 @@ class TableRead:
"requires the Python reader")
try:
from pypaimon.read.native_plan import (
- _prepare_native_read, native_read, native_split_from_python)
+ _prepare_native_read, _raise_if_native_fork_safety_error,
+ native_read, native_split_from_python)
except Exception as e:
return self._native_fallback(str(e), e, log=True)
rust_splits = []
@@ -551,6 +552,7 @@ class TableRead:
try:
read_splits = _prepare_native_read(self.table, **read_kwargs)
except Exception as e:
+ _raise_if_native_fork_safety_error(e)
return self._native_fallback(str(e), e, log=True)
if streaming:
groups = self._native_split_groups(
@@ -571,6 +573,7 @@ class TableRead:
"another reader failed to start",
exc_info=True,
)
+ _raise_if_native_fork_safety_error(e)
return self._native_fallback(str(e), e, log=True)
batches = self._native_batches_parallel_streaming(readers)
return self._convert_native_batches(batches, schema)
@@ -579,11 +582,13 @@ class TableRead:
read_splits, rust_splits, schema, parallelism,
split_weights)
except _NativeReadSetupError as e:
+ _raise_if_native_fork_safety_error(e)
return self._native_fallback(str(e), e, log=True)
try:
read_kwargs = self._native_read_kwargs(blob_parallelism)
batches = native_read(self.table, rust_splits, **read_kwargs)
except Exception as e:
+ _raise_if_native_fork_safety_error(e)
return self._native_fallback(str(e), e, log=True)
return self._convert_native_batches(batches, schema)
diff --git a/paimon-python/pypaimon/read/table_scan.py
b/paimon-python/pypaimon/read/table_scan.py
index 344e1a1703..a2755cf6fb 100755
--- a/paimon-python/pypaimon/read/table_scan.py
+++ b/paimon-python/pypaimon/read/table_scan.py
@@ -171,7 +171,8 @@ class TableScan:
by the reader. Snapshot metadata is preserved even when pruning removes
every split.
"""
- from pypaimon.read.native_plan import native_plan
+ from pypaimon.read.native_plan import (
+ _raise_if_native_fork_safety_error, native_plan)
try:
fs = self.file_scanner
@@ -256,6 +257,7 @@ class TableScan:
) for split in splits]
return Plan(splits, snapshot_id=plan.snapshot_id)
except Exception as e:
+ _raise_if_native_fork_safety_error(e)
# Any native construction/planning/pruning failure -> fall back.
logger.warning(
"Native plan failed, falling back to the Python scanner: %s",
e)
diff --git a/paimon-python/pypaimon/tests/native_plan_test.py
b/paimon-python/pypaimon/tests/native_plan_test.py
index e2dad2f191..7f5bd3386f 100644
--- a/paimon-python/pypaimon/tests/native_plan_test.py
+++ b/paimon-python/pypaimon/tests/native_plan_test.py
@@ -588,6 +588,21 @@ class NativePlanTest(unittest.TestCase):
self.assertIs(scan.plan(), sentinel)
fs.scan.assert_called_once_with()
+ def test_plan_propagates_native_fork_safety_error(self):
+ class ForkSafetyError(RuntimeError):
+ pass
+
+ module = ModuleType('pypaimon_rust')
+ module.ForkSafetyError = ForkSafetyError
+ fs = Mock(partition_key_predicate=None)
+ scan = _scan(native_enabled=True, file_scanner=fs)
+ with patch.dict(sys.modules, {'pypaimon_rust': module}), patch(
+ 'pypaimon.read.native_plan.native_plan',
+ side_effect=ForkSafetyError('cannot reuse Jindo after fork')):
+ with self.assertRaises(ForkSafetyError):
+ scan.plan()
+ fs.scan.assert_not_called()
+
def test_plan_uses_resolved_schema_for_jdbc_catalog_loader(self):
fs = Mock(partition_key_predicate=None)
scan = _scan(native_enabled=True, file_scanner=fs)
diff --git a/paimon-python/pypaimon/tests/native_read_test.py
b/paimon-python/pypaimon/tests/native_read_test.py
index c076fa2666..e01e95ede3 100644
--- a/paimon-python/pypaimon/tests/native_read_test.py
+++ b/paimon-python/pypaimon/tests/native_read_test.py
@@ -15,7 +15,9 @@
# specific language governing permissions and limitations
# under the License.
+import sys
import threading
+from types import ModuleType
from unittest.mock import Mock, patch
import pyarrow as pa
@@ -214,6 +216,22 @@ def
test_native_read_consumes_retained_rust_splits_and_enforces_limit():
)
+def test_native_read_propagates_native_fork_safety_error():
+ class ForkSafetyError(RuntimeError):
+ pass
+
+ module = ModuleType('pypaimon_rust')
+ module.ForkSafetyError = ForkSafetyError
+ read = _table_read()
+ split = _Split()
+ split._native_split = object()
+ with patch.dict(sys.modules, {'pypaimon_rust': module}), patch(
+ 'pypaimon.read.native_plan.native_read',
+ side_effect=ForkSafetyError('cannot reuse Jindo after fork')):
+ with pytest.raises(ForkSafetyError):
+ read._try_native_batches([split], pa.schema([('id', pa.int32())]))
+
+
def test_native_read_flattens_nested_rows_and_map_keys_with_parent_nulls():
read = _table_read()
read.table.fields = [
diff --git a/paimon-python/pypaimon/tests/streaming_table_scan_test.py
b/paimon-python/pypaimon/tests/streaming_table_scan_test.py
index 04ecb48ad3..41c1416a1c 100644
--- a/paimon-python/pypaimon/tests/streaming_table_scan_test.py
+++ b/paimon-python/pypaimon/tests/streaming_table_scan_test.py
@@ -18,7 +18,10 @@
"""Tests for AsyncStreamingTableScan."""
import asyncio
+import sys
import unittest
+from concurrent.futures import Future
+from types import ModuleType
from unittest.mock import Mock, patch
from pypaimon.common.options.core_options import ChangelogProducer
@@ -137,6 +140,82 @@ class AsyncStreamingTableScanTest(unittest.TestCase):
self.assertEqual(
native_plan.call_args.kwargs['projection'], ['payload'])
+ @patch('pypaimon.read.streaming_table_scan.ManifestListManager')
+ @patch('pypaimon.read.streaming_table_scan.ManifestFileManager')
+ @patch('pypaimon.read.native_plan.native_plan')
+ def test_streaming_plans_propagate_native_fork_safety_error(
+ self, native_plan, _manifest_files, manifest_lists):
+ class ForkSafetyError(RuntimeError):
+ pass
+
+ module = ModuleType('pypaimon_rust')
+ module.ForkSafetyError = ForkSafetyError
+ native_plan.side_effect = ForkSafetyError('cannot reuse Jindo after
fork')
+
+ for mode in ('initial', 'delta', 'changelog'):
+ with self.subTest(mode=mode):
+ table, _ = _create_mock_table()
+ table.options.native_plan_enabled.return_value = True
+ table.options.batch_scan_skip_level0.return_value = False
+ scan = AsyncStreamingTableScan(table, prefetch_enabled=False)
+ snapshot = _create_mock_snapshot(5)
+ with patch.dict(sys.modules, {'pypaimon_rust': module}),
patch.object(
+ scan, '_AsyncStreamingTableScan__auth_query',
return_value=None), patch.object(
+ scan,
'_AsyncStreamingTableScan__create_initial_plan_raw') as initial, patch.object(
+ scan, '_create_plan_from_manifests') as from_manifests:
+ with self.assertRaises(ForkSafetyError):
+ getattr(scan, '_create_{}_plan'.format(mode))(snapshot)
+ initial.assert_not_called()
+ from_manifests.assert_not_called()
+ manifest_lists.return_value.read_all.assert_not_called()
+ manifest_lists.return_value.read_delta.assert_not_called()
+ manifest_lists.return_value.read_changelog.assert_not_called()
+
+ @patch('pypaimon.read.streaming_table_scan.ManifestListManager')
+ @patch('pypaimon.read.streaming_table_scan.ManifestFileManager')
+ @patch('pypaimon.read.native_plan.native_plan',
side_effect=RuntimeError('unsupported'))
+ def test_delta_scan_falls_back_after_other_native_error(
+ self, _native_plan, _manifest_files, manifest_lists):
+ table, _ = _create_mock_table()
+ table.options.native_plan_enabled.return_value = True
+ manifest_lists.return_value.read_delta.return_value = []
+ scan = AsyncStreamingTableScan(table, prefetch_enabled=False)
+ snapshot = _create_mock_snapshot(5)
+
+ self.assertEqual(scan._create_delta_plan(snapshot).snapshot_id, 5)
+
manifest_lists.return_value.read_delta.assert_called_once_with(snapshot)
+
+ @patch('pypaimon.read.streaming_table_scan.ManifestListManager')
+ @patch('pypaimon.read.streaming_table_scan.ManifestFileManager')
+ @patch('pypaimon.read.native_plan.native_plan')
+ def test_prefetch_propagates_native_fork_safety_error(
+ self, native_plan, _manifest_files, _manifest_lists):
+ class ForkSafetyError(RuntimeError):
+ pass
+
+ module = ModuleType('pypaimon_rust')
+ module.ForkSafetyError = ForkSafetyError
+ error = ForkSafetyError('cannot reuse Jindo after fork')
+ native_plan.side_effect = error
+ table, _ = _create_mock_table()
+ table.options.native_plan_enabled.return_value = True
+ scan = AsyncStreamingTableScan(table, prefetch_enabled=False)
+ scan._snapshot_manager.find_next_scannable.return_value = (
+ _create_mock_snapshot(5), 6, 0)
+
+ with patch.dict(sys.modules, {'pypaimon_rust': module}):
+ with self.assertRaises(ForkSafetyError):
+ scan._fetch_plan_with_lookahead(5)
+
+ future = Future()
+ future.set_exception(error)
+ scan._prefetch_future = future
+ scan.next_snapshot_id = 5
+ with patch.object(scan, '_should_use_diff_catch_up',
return_value=False):
+ with self.assertRaises(ForkSafetyError):
+ asyncio.run(scan.stream().__anext__())
+ scan._snapshot_manager.find_next_scannable.assert_called_once()
+
@patch('pypaimon.read.streaming_table_scan.ManifestListManager')
@patch('pypaimon.read.streaming_table_scan.ManifestFileManager')
@patch('pypaimon.read.native_plan.native_plan')
