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 e08dc6b732 [python] Simplify native plan and read capability gates
(#10056)
e08dc6b732 is described below
commit e08dc6b7327741e6ba591b6dfd37f61eddbdbe51
Author: Jingsong Lee <[email protected]>
AuthorDate: Mon Sep 21 16:38:17 2026 +0800
[python] Simplify native plan and read capability gates (#10056)
---
paimon-python/pypaimon/read/native_plan.py | 2 -
paimon-python/pypaimon/read/table_read.py | 2 -
paimon-python/pypaimon/read/table_scan.py | 105 +---------
paimon-python/pypaimon/tests/native_plan_test.py | 238 +++++++++--------------
4 files changed, 93 insertions(+), 254 deletions(-)
diff --git a/paimon-python/pypaimon/read/native_plan.py
b/paimon-python/pypaimon/read/native_plan.py
index 4388716154..adae794d29 100644
--- a/paimon-python/pypaimon/read/native_plan.py
+++ b/paimon-python/pypaimon/read/native_plan.py
@@ -69,8 +69,6 @@ def native_split_from_python(split):
Vector scores intentionally stay on the Python IndexedSplit. Rust needs
only its row ranges to perform the physical read.
"""
- if not native_split_bridge_available():
- return None
from pypaimon_rust.datafusion import Split as NativeSplit
return NativeSplit.deserialize(
serialize_split_v1(split, include_scores=False))
diff --git a/paimon-python/pypaimon/read/table_read.py
b/paimon-python/pypaimon/read/table_read.py
index b7b1181381..aa6f395229 100644
--- a/paimon-python/pypaimon/read/table_read.py
+++ b/paimon-python/pypaimon/read/table_read.py
@@ -448,8 +448,6 @@ class TableRead:
"Native split conversion failed, falling back to the "
"Python reader: %s", e)
return None
- if rust_split is None:
- return None
rust_splits.append(rust_split)
split_weights.append(self._native_split_weight(split))
if (parallelism is not None
diff --git a/paimon-python/pypaimon/read/table_scan.py
b/paimon-python/pypaimon/read/table_scan.py
index 71d3577685..344e1a1703 100755
--- a/paimon-python/pypaimon/read/table_scan.py
+++ b/paimon-python/pypaimon/read/table_scan.py
@@ -32,35 +32,6 @@ from pypaimon.read.scanner.file_scanner import FileScanner
logger = logging.getLogger(__name__)
-_NATIVE_FAMILY_SEARCH_MODE_OPTIONS = frozenset({
- CoreOptions.SCALAR_INDEX_SEARCH_MODE.key(),
- CoreOptions.VECTOR_INDEX_SEARCH_MODE.key(),
- CoreOptions.FULL_TEXT_INDEX_SEARCH_MODE.key(),
-})
-_NATIVE_SEARCH_MODE_OPTIONS = _NATIVE_FAMILY_SEARCH_MODE_OPTIONS | {
- CoreOptions.GLOBAL_INDEX_SEARCH_MODE.key(),
-}
-_NATIVE_FORWARDED_OPTIONS = frozenset({
- CoreOptions.SCAN_NATIVE_PLAN_ENABLED.key(),
- CoreOptions.SCAN_MODE.key(),
- CoreOptions.INCREMENTAL_BETWEEN_TIMESTAMP.key(),
- CoreOptions.SOURCE_SPLIT_TARGET_SIZE.key(),
- CoreOptions.SOURCE_SPLIT_OPEN_FILE_COST.key(),
- CoreOptions.DELETION_VECTORS_MERGE_ON_READ.key(),
- CoreOptions.SCAN_VERSION.key(),
- CoreOptions.SCAN_SNAPSHOT_ID.key(),
- CoreOptions.SCAN_TAG_NAME.key(),
- CoreOptions.SCAN_TIMESTAMP.key(),
- CoreOptions.SCAN_TIMESTAMP_MILLIS.key(),
- CoreOptions.SCAN_WATERMARK.key(),
- CoreOptions.BRANCH.key(),
-}) | _NATIVE_SEARCH_MODE_OPTIONS
-_NATIVE_PLAN_INDEPENDENT_OPTIONS = frozenset({
- CoreOptions.BLOB_AS_DESCRIPTOR.key(),
- CoreOptions.READ_NATIVE_ENABLED.key(),
- CoreOptions.READ_BATCH_SIZE.key(),
- CoreOptions.READ_PARALLELISM.key(),
-})
_NATIVE_TIME_TRAVEL_OPTIONS = frozenset({
CoreOptions.SCAN_VERSION.key(),
CoreOptions.SCAN_SNAPSHOT_ID.key(),
@@ -127,52 +98,20 @@ class TableScan:
primary-key global-index results,
a primary-key table whose trimmed PK is empty (PK equals the partition
key; Rust rejects this schema), unsupported time travel selectors,
- and catalog-loaded tables with schema/option overrides Rust cannot
carry,
- query auth, a missing/old
- pypaimon-rust, or a catalog / identifier Rust cannot reconstruct. Keep
- this capability gate in sync when adding scan features."""
+ query auth, a missing pypaimon-rust, or a catalog / identifier Rust
+ cannot reconstruct. Keep this capability gate in sync when adding scan
+ features."""
from pypaimon.read.native_plan import (
- _resolved_schema_file_io_options, native_method_available,
- native_runtime_available, native_version_at_least,
+ _resolved_schema_file_io_options, native_runtime_available,
)
if not native_runtime_available():
return False
fs = self.file_scanner
- if fs.is_streaming and not native_method_available('Split',
'is_streaming'):
- return False
if not self._native_global_index_result_supported():
return False
if getattr(fs, 'chunk_shuffle', None) is not None:
fs._validate_chunk_shuffle_compat()
- if not native_method_available('TableScan', 'with_chunk_shuffle'):
- return False
- # Positional append distribution needs the stable partition/file order
- # introduced in 0.4. Older bindings can assign different rows per call.
- if (not self.table.is_primary_key_table and not fs.data_evolution
- and (fs.idx_of_this_subtask is not None or
fs.start_pos_of_this_subtask is not None)
- and not native_version_at_least(0, 4)):
- return False
- if getattr(fs, 'deletion_vectors_enabled', False):
- # 0.4.0 includes Python-written DV decoding and legacy bucket
paths.
- if not native_version_at_least(0, 4, 0):
- return False
- if (not self.table.is_primary_key_table
- and (getattr(fs, 'idx_of_this_subtask', None) is not None
- or getattr(fs, 'start_pos_of_this_subtask', None) is not
None)):
- if (getattr(fs, 'idx_of_this_subtask', None) is not None
- and not native_method_available('TableScan',
'with_row_position_shard')):
- return False
- if (getattr(fs, 'start_pos_of_this_subtask', None) is not None
- and not native_method_available('TableScan',
'with_row_position_slice')):
- return False
- if (getattr(fs, '_row_ranges', None) is not None
- and not native_method_available('ReadBuilder',
'with_row_ranges')):
- return False
- if (self.table.current_branch() != 'main'
- and not native_method_available('Table', 'branch')):
- return False
- resolved_schema = _resolved_schema_file_io_options(self.table) is not
None
- if not resolved_schema:
+ if _resolved_schema_file_io_options(self.table) is None:
loader = getattr(
getattr(self.table, 'catalog_environment', None),
'catalog_loader',
@@ -194,9 +133,6 @@ class TableScan:
database_name = self.table.identifier.get_database_name()
if not database_name or database_name == UNKNOWN_DATABASE:
return False
- if ('.' in database_name
- and not native_method_available('Table',
'copy_with_resolved_schema')):
- return False
if self.table.options.query_auth_enabled:
return False
# Rust rejects schemas whose primary keys are all partition keys.
@@ -204,38 +140,9 @@ class TableScan:
and not self.table.trimmed_primary_keys:
return False
options = self.table.options.options
- if (options.contains_key(CoreOptions.SCAN_WATERMARK.key())
- and not native_version_at_least(0, 4)):
- return False
- if (any(options.contains_key(key)
- for key in _NATIVE_FAMILY_SEARCH_MODE_OPTIONS)):
- from pypaimon.read.native_plan import
native_family_search_modes_available
- if not native_family_search_modes_available():
- return False
- if not resolved_schema and not native_method_available('Table',
'copy_with_resolved_schema'):
- supported_time_travel = any(
- options.contains_key(key) for key in
_NATIVE_TIME_TRAVEL_OPTIONS)
- # Time travel intentionally carries a historical schema; other
stale
- # table objects must still fall back because Rust reloads the
latest.
- latest_schema = self.table.schema_manager.latest()
- if (not supported_time_travel and latest_schema is not None
- and latest_schema.id != self.table.table_schema.id):
- return False
- # Rust cannot remove an option persisted in the catalog-loaded
schema.
- applied_options = getattr(self.table, '_applied_dynamic_options',
{}) or {}
- allowed_options = (
- _NATIVE_FORWARDED_OPTIONS | _NATIVE_PLAN_INDEPENDENT_OPTIONS)
- if (set(applied_options) - allowed_options
- or any(key in (_NATIVE_TIME_TRAVEL_OPTIONS
- | _NATIVE_SEARCH_MODE_OPTIONS) and value is
None
- for key, value in applied_options.items())):
- return False
from pypaimon.snapshot.time_travel_util import SCAN_KEYS
unsupported_scan_keys = set(SCAN_KEYS) - _NATIVE_TIME_TRAVEL_OPTIONS
- if any(options.contains_key(k) for k in unsupported_scan_keys):
- return False
- return (not options.contains(CoreOptions.INCREMENTAL_BETWEEN_TIMESTAMP)
- or native_method_available('ReadBuilder',
'new_incremental_scan'))
+ return not any(options.contains_key(k) for k in unsupported_scan_keys)
def _native_global_index_result_supported(self) -> bool:
result = self.file_scanner._global_index_result
diff --git a/paimon-python/pypaimon/tests/native_plan_test.py
b/paimon-python/pypaimon/tests/native_plan_test.py
index 0d9f2f581d..2687d55c99 100644
--- a/paimon-python/pypaimon/tests/native_plan_test.py
+++ b/paimon-python/pypaimon/tests/native_plan_test.py
@@ -369,9 +369,9 @@ class NativePlanTest(unittest.TestCase):
np.assert_not_called()
fs.scan.assert_called_once_with()
- def test_plan_falls_back_when_scan_is_not_plain(self):
- # Older native bindings cannot carry DE shard/slice or explicit row
ranges,
- # arbitrary global-index results, or incremental scans.
+ def test_plan_falls_back_for_unsupported_scan_context(self):
+ # These cases cannot be reconstructed by native planning even with the
+ # current Rust bindings.
def check(setup):
fs = Mock(partition_key_predicate=None)
sentinel = object()
@@ -383,25 +383,10 @@ class NativePlanTest(unittest.TestCase):
np.assert_not_called()
fs.scan.assert_called_once_with()
- check(lambda s, fs: (setattr(fs, 'data_evolution', True),
- setattr(fs, 'idx_of_this_subtask', 0)))
- check(lambda s, fs: (setattr(fs, 'data_evolution', True),
- setattr(fs, 'start_pos_of_this_subtask', 0)))
check(lambda s, fs: setattr(fs, '_global_index_result', object()))
- check(lambda s, fs: setattr(fs, '_row_ranges', [object()]))
- check(lambda s, fs: setattr(fs, 'deletion_vectors_enabled', True))
- check(lambda s, fs: setattr(fs, 'is_streaming', True))
check(lambda s, fs: (setattr(s.table, 'is_primary_key_table', True),
setattr(s.table, 'trimmed_primary_keys', [])))
- check(lambda s, fs: setattr(
- s.table, '_applied_dynamic_options', {'scan.snapshot-id': None}))
- check(lambda s, fs:
setattr(s.table.schema_manager.latest.return_value, 'id', 2))
- check(lambda s, fs: s.table.schema_manager.latest.__setattr__(
- 'side_effect', RuntimeError('metadata read failed')))
check(lambda s, fs: setattr(s.table.options, 'query_auth_enabled',
True))
- check(lambda s, fs: s.table.current_branch.__setattr__('return_value',
'b1'))
- check(lambda s, fs: s.table.identifier.get_database_name.__setattr__(
- 'return_value', 'db.name'))
check(lambda s, fs: s.table.identifier.get_database_name.__setattr__(
'return_value', 'unknown'))
check(lambda s, fs: setattr(
@@ -409,8 +394,6 @@ class NativePlanTest(unittest.TestCase):
for attr in ('hadoop_conf', 'prefer_io_loader', 'fallback_io_loader'):
check(lambda s, fs, attr=attr: setattr(
s.table.catalog_environment.catalog_loader.context(), attr,
object()))
- check(lambda s, fs: s.table.options.options.contains.__setattr__(
- 'return_value', True)) # incremental
def test_plan_native_empty_preserves_snapshot_without_fallback(self):
for snapshot_id in (None, 7):
@@ -425,8 +408,7 @@ class NativePlanTest(unittest.TestCase):
fs.scan.assert_not_called()
def test_plan_falls_back_when_rust_unavailable(self):
- # scan.native-plan.enabled but pypaimon-rust missing/old -> fall back,
- # not crash.
+ # scan.native-plan.enabled but pypaimon-rust missing -> fall back.
fs = Mock(partition_key_predicate=None)
sentinel = object()
fs.scan.return_value = sentinel
@@ -438,49 +420,36 @@ class NativePlanTest(unittest.TestCase):
np.assert_not_called()
fs.scan.assert_called_once_with()
- def test_family_search_modes_require_rust_0_4(self):
- for available, expect_native in ((False, False), (True, True)):
- with self.subTest(available=available):
- fs = Mock(partition_key_predicate=None)
- fs.scan.return_value = fallback = object()
- scan = _scan(native_enabled=True, file_scanner=fs)
- scan.table.options.options.contains_key.side_effect = (
- lambda key: key == 'scalar-index.search-mode')
- scan.table._applied_dynamic_options = {
- 'scalar-index.search-mode': 'full',
- }
- split = Mock(partition=Mock(values=[]), snapshot_id=1)
-
- with patch(
- 'pypaimon.read.native_plan.'
- 'native_family_search_modes_available',
- return_value=available), patch(
- 'pypaimon.read.native_plan.native_plan',
- return_value=Plan([split], 1)) as native:
- plan = scan.plan()
-
- if expect_native:
- self.assertEqual(plan.splits(), [split])
- native.assert_called_once()
- fs.scan.assert_not_called()
- else:
- self.assertIs(plan, fallback)
- native.assert_not_called()
- fs.scan.assert_called_once_with()
+ def test_family_search_modes_use_native_plan(self):
+ fs = Mock(partition_key_predicate=None)
+ scan = _scan(native_enabled=True, file_scanner=fs)
+ scan.table.options.options.contains_key.side_effect = (
+ lambda key: key == 'scalar-index.search-mode')
+ scan.table._applied_dynamic_options = {
+ 'scalar-index.search-mode': 'full',
+ }
+ split = Mock(partition=Mock(values=[]), snapshot_id=1)
- def test_removing_search_mode_falls_back(self):
+ with patch('pypaimon.read.native_plan.native_plan',
+ return_value=Plan([split], 1)) as native:
+ self.assertEqual(scan.plan().splits(), [split])
+
+ native.assert_called_once()
+ fs.scan.assert_not_called()
+
+ def test_removing_search_mode_uses_native_plan(self):
fs = Mock(partition_key_predicate=None)
- fs.scan.return_value = fallback = object()
scan = _scan(native_enabled=True, file_scanner=fs)
scan.table._applied_dynamic_options = {
'scalar-index.search-mode': None,
}
- with patch('pypaimon.read.native_plan.native_plan') as native:
- self.assertIs(scan.plan(), fallback)
+ with patch('pypaimon.read.native_plan.native_plan',
+ return_value=Plan([], 1)) as native:
+ self.assertEqual(scan.plan().snapshot_id, 1)
- native.assert_not_called()
- fs.scan.assert_called_once_with()
+ native.assert_called_once()
+ fs.scan.assert_not_called()
def test_dynamic_read_option_uses_native_plan(self):
fs = Mock(partition_key_predicate=None)
@@ -500,19 +469,19 @@ class NativePlanTest(unittest.TestCase):
native.assert_called_once()
fs.scan.assert_not_called()
- def test_unknown_dynamic_option_falls_back(self):
+ def test_unknown_dynamic_option_uses_native_plan(self):
fs = Mock(partition_key_predicate=None)
- fs.scan.return_value = fallback = object()
scan = _scan(native_enabled=True, file_scanner=fs)
scan.table._applied_dynamic_options = {
'future.scan-option': 'value',
}
- with patch('pypaimon.read.native_plan.native_plan') as native:
- self.assertIs(scan.plan(), fallback)
+ with patch('pypaimon.read.native_plan.native_plan',
+ return_value=Plan([], 1)) as native:
+ self.assertEqual(scan.plan().snapshot_id, 1)
- native.assert_not_called()
- fs.scan.assert_called_once_with()
+ native.assert_called_once()
+ fs.scan.assert_not_called()
def test_plan_falls_back_when_native_plan_raises(self):
# A native planning failure (e.g. unsupported scheme) must fall back,
not crash.
@@ -872,94 +841,62 @@ class NativePlanTest(unittest.TestCase):
with self.assertRaisesRegex(RuntimeError, 'requested branch'):
native_plan(table)
- def test_explicit_row_ranges_require_runtime_api(self):
- for available in (False, True):
- for ranges in ([], [Range(1, 2), Range(5, 8)]):
- with self.subTest(available=available, ranges=ranges):
- fs = Mock(partition_key_predicate=None)
- scan = _scan(True, fs)
- fs._row_ranges = ranges
- fs.scan.return_value = fallback = object()
- with
patch('pypaimon.read.native_plan.native_method_available',
- return_value=available), patch(
- 'pypaimon.read.native_plan.native_plan',
- return_value=Plan([], 3)) as native:
- result = scan.plan()
- if available:
- self.assertEqual(result.snapshot_id, 3)
- self.assertEqual(native.call_args[1]['row_ranges'],
- [(r.from_, r.to) for r in ranges])
- fs.scan.assert_not_called()
- else:
- self.assertIs(result, fallback)
- native.assert_not_called()
-
- def test_append_distribution_requires_stable_native_order(self):
- for available in (False, True):
- for selection in ('idx_of_this_subtask',
'start_pos_of_this_subtask'):
- with self.subTest(available=available, selection=selection):
- fs = Mock(partition_key_predicate=None)
- scan = _scan(True, fs)
- setattr(fs, selection, 0)
- fs.scan.return_value = fallback = object()
- with
patch('pypaimon.read.native_plan.native_version_at_least',
- return_value=available), patch(
-
'pypaimon.read.native_plan.native_method_available',
- return_value=True):
- self.assertEqual(scan._native_plan_supported(),
available)
- if not available:
- with
patch('pypaimon.read.native_plan.native_plan') as native:
- self.assertIs(scan.plan(), fallback)
- native.assert_not_called()
-
- def test_watermark_forwarding_requires_current_runtime(self):
- for available in (False, True):
- with self.subTest(available=available):
+ def test_explicit_row_ranges_are_forwarded(self):
+ for ranges in ([], [Range(1, 2), Range(5, 8)]):
+ with self.subTest(ranges=ranges):
fs = Mock(partition_key_predicate=None)
scan = _scan(True, fs)
- scan.table.options.options = Options({'scan.watermark': '200'})
- scan.table._applied_dynamic_options = {'scan.watermark': '200'}
- scan.table.schema_manager.latest.return_value.id = 2
- fs.scan.return_value = fallback = object()
- self.assertEqual(_read_options(scan.table)['scan.watermark'],
'200')
- with patch('pypaimon.read.native_plan.native_version_at_least',
- return_value=available), patch(
- 'pypaimon.read.native_plan.native_plan',
- return_value=Plan([], 1)) as native:
+ fs._row_ranges = ranges
+ with patch('pypaimon.read.native_plan.native_plan',
+ return_value=Plan([], 3)) as native:
result = scan.plan()
- if available:
- self.assertEqual(result.snapshot_id, 1)
- fs.scan.assert_not_called()
- else:
- self.assertIs(result, fallback)
- native.assert_not_called()
+ self.assertEqual(result.snapshot_id, 3)
+ self.assertEqual(native.call_args[1]['row_ranges'],
+ [(r.from_, r.to) for r in ranges])
+ fs.scan.assert_not_called()
- @unittest.skipIf(sys.version_info < (3, 8),
- "importlib.metadata requires Python 3.8")
- def
test_deletion_vectors_require_fixed_version_and_matching_l0_semantics(self):
- for version in ('0.3.0', '0.4.0rc1', '0.4.0', '0.4.1'):
- for bucket_local in (False, True):
- for merge_on_read in (False, True):
- with self.subTest(version=version,
bucket_local=bucket_local,
- merge_on_read=merge_on_read):
- fs = Mock(partition_key_predicate=None)
- scan = _scan(True, fs)
- fs.deletion_vectors_enabled = True
- scan.table.options.options = Options({
- 'index-file-in-data-file-dir':
str(bucket_local).lower(),
- 'deletion-vectors.merge-on-read':
str(merge_on_read).lower(),
- })
- fs.scan.return_value = fallback = object()
- with patch('importlib.metadata.version',
return_value=version), patch(
- 'pypaimon.read.native_plan.native_plan',
- return_value=Plan([], 1)) as native:
- result = scan.plan()
- if version in ('0.4.0', '0.4.1'):
- self.assertEqual(result.snapshot_id, 1)
- fs.scan.assert_not_called()
- else:
- self.assertIs(result, fallback)
- native.assert_not_called()
+ def test_append_distribution_uses_native_order(self):
+ for selection in ('idx_of_this_subtask', 'start_pos_of_this_subtask'):
+ with self.subTest(selection=selection):
+ fs = Mock(partition_key_predicate=None)
+ scan = _scan(True, fs)
+ setattr(fs, selection, 0)
+ with patch('pypaimon.read.native_plan.native_plan',
+ return_value=Plan([], 3)) as native:
+ self.assertEqual(scan.plan().snapshot_id, 3)
+ native.assert_called_once()
+ fs.scan.assert_not_called()
+
+ def test_watermark_forwarding_uses_native_plan(self):
+ fs = Mock(partition_key_predicate=None)
+ scan = _scan(True, fs)
+ scan.table.options.options = Options({'scan.watermark': '200'})
+ scan.table._applied_dynamic_options = {'scan.watermark': '200'}
+ scan.table.schema_manager.latest.return_value.id = 2
+ self.assertEqual(_read_options(scan.table)['scan.watermark'], '200')
+ with patch('pypaimon.read.native_plan.native_plan',
+ return_value=Plan([], 1)) as native:
+ self.assertEqual(scan.plan().snapshot_id, 1)
+ native.assert_called_once()
+ fs.scan.assert_not_called()
+
+ def test_deletion_vectors_use_native_plan_for_bucket_layouts(self):
+ for bucket_local in (False, True):
+ for merge_on_read in (False, True):
+ with self.subTest(bucket_local=bucket_local,
+ merge_on_read=merge_on_read):
+ fs = Mock(partition_key_predicate=None)
+ scan = _scan(True, fs)
+ fs.deletion_vectors_enabled = True
+ scan.table.options.options = Options({
+ 'index-file-in-data-file-dir':
str(bucket_local).lower(),
+ 'deletion-vectors.merge-on-read':
str(merge_on_read).lower(),
+ })
+ with patch('pypaimon.read.native_plan.native_plan',
+ return_value=Plan([], 1)) as native:
+ self.assertEqual(scan.plan().snapshot_id, 1)
+ native.assert_called_once()
+ fs.scan.assert_not_called()
@unittest.skipIf(sys.version_info < (3, 8),
"importlib.metadata requires Python 3.8")
@@ -987,8 +924,8 @@ class NativePlanTest(unittest.TestCase):
scan.table.options.options = Options({'incremental-between-timestamp':
'100,200'})
scan.table._applied_dynamic_options =
{'incremental-between-timestamp': '100,200'}
scan._incremental_snapshot_range = (2, 4)
- with patch('pypaimon.read.native_plan.native_method_available',
return_value=True), patch(
- 'pypaimon.read.native_plan.native_plan', return_value=Plan([],
4)) as native:
+ with patch('pypaimon.read.native_plan.native_plan',
+ return_value=Plan([], 4)) as native:
self.assertEqual(scan.plan().snapshot_id, 4)
self.assertEqual(native.call_args[1]['incremental_range'], (2, 4))
fs.scan.assert_not_called()
@@ -1020,8 +957,7 @@ class NativePlanTest(unittest.TestCase):
scan = _scan(True, fs)
scan.table.options.options = Options({'incremental-between-timestamp':
'100,200'})
scan._incremental_snapshot_range = None
- with patch('pypaimon.read.native_plan.native_method_available',
return_value=True), patch(
- 'pypaimon.read.native_plan.native_plan') as native:
+ with patch('pypaimon.read.native_plan.native_plan') as native:
plan = scan.plan()
self.assertEqual(plan.splits(), [])
self.assertIsNone(plan.snapshot_id)
