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 37d1f51798 [python] Support native OSS C++ reads with a separate
Python backend (#10393)
37d1f51798 is described below
commit 37d1f51798dca4129dea09f6ce2d80fca9e851d3
Author: XiaoHongbo <[email protected]>
AuthorDate: Mon Oct 12 14:09:26 2026 +0800
[python] Support native OSS C++ reads with a separate Python backend
(#10393)
---
docs/docs/pypaimon/installation.md | 12 ++++++++
paimon-python/pypaimon/common/options/config.py | 4 ++-
.../filesystem/jindo_file_system_handler.py | 5 ++--
paimon-python/pypaimon/filesystem/pvfs.py | 13 ++++++---
.../pypaimon/filesystem/pyarrow_file_io.py | 14 +++++++--
.../pypaimon/tests/jindo_file_system_test.py | 13 +++++++++
paimon-python/pypaimon/tests/native_plan_test.py | 29 +++++++++++++++++--
.../pypaimon/tests/oss_legacy_mode_test.py | 33 ++++++++++++++++++++++
.../pypaimon/tests/pvfs_oss_filesystem_test.py | 15 ++++++++++
9 files changed, 125 insertions(+), 13 deletions(-)
diff --git a/docs/docs/pypaimon/installation.md
b/docs/docs/pypaimon/installation.md
index d0a6647583..2987e2b315 100644
--- a/docs/docs/pypaimon/installation.md
+++ b/docs/docs/pypaimon/installation.md
@@ -109,6 +109,18 @@ The source archive and wheel are written to `dist/` with
the same version.
Development versions append the Git commit date, for example `2.2.dev20260927`.
Install the wheel with `python -m pip install dist/*.whl`.
+## Optional OSS C++ backend
+
+Native OSS reads require a `pypaimon-rust` build with `storage-oss-cpp` and
+bridge discovery, plus `pypaimon_oss_cpp` on each worker. The library path is
+detected automatically; set `fs.oss.cpp.library.path` only for a manually
+installed bridge.
+
+Set `fs.oss.impl=cpp` in the catalog and enable `scan.native-plan.enabled` and
+`read.native.enabled` on the table. Python I/O uses Jindo by default, or set
+`fs.oss.python.impl=legacy`. The C++ backend is read-only; use `spawn`, not
+`fork`, for workers.
+
## Verify the installation
```shell
diff --git a/paimon-python/pypaimon/common/options/config.py
b/paimon-python/pypaimon/common/options/config.py
index f6d7026789..bac72d4a4c 100644
--- a/paimon-python/pypaimon/common/options/config.py
+++ b/paimon-python/pypaimon/common/options/config.py
@@ -21,7 +21,9 @@ from pypaimon.common.options.config_options import
ConfigOptions
class OssOptions:
OSS_IMPL =
ConfigOptions.key("fs.oss.impl").string_type().default_value("jindo").with_description(
- "OSS filesystem implementation: legacy or jindo")
+ "OSS filesystem implementation: legacy, jindo, or cpp (native reads
only)")
+ OSS_PYTHON_IMPL =
ConfigOptions.key("fs.oss.python.impl").string_type().default_value("jindo").with_description(
+ "Python OSS implementation when fs.oss.impl=cpp: jindo or legacy")
OSS_ACCESS_KEY_ID =
ConfigOptions.key("fs.oss.accessKeyId").string_type().no_default_value().with_description(
"OSS access key ID")
OSS_ACCESS_KEY_SECRET = ConfigOptions.key(
diff --git a/paimon-python/pypaimon/filesystem/jindo_file_system_handler.py
b/paimon-python/pypaimon/filesystem/jindo_file_system_handler.py
index 5a242df8de..688207c33c 100644
--- a/paimon-python/pypaimon/filesystem/jindo_file_system_handler.py
+++ b/paimon-python/pypaimon/filesystem/jindo_file_system_handler.py
@@ -48,7 +48,8 @@ from pypaimon.filesystem import oss_user_agent
_JINDO_CONFIG_PREFIXES = ("fs.", "logger.")
-_PYPAIMON_ONLY_JINDO_CONFIG_KEYS = {OssOptions.OSS_IMPL.key()}
+_PYPAIMON_ONLY_JINDO_CONFIG_KEYS = {
+ OssOptions.OSS_IMPL.key(), OssOptions.OSS_PYTHON_IMPL.key()}
_CASE_SENSITIVE_JINDO_CONFIG_KEYS = {
OssOptions.OSS_ACCESS_KEY_ID.key().lower():
OssOptions.OSS_ACCESS_KEY_ID.key(),
OssOptions.OSS_ACCESS_KEY_SECRET.key().lower():
OssOptions.OSS_ACCESS_KEY_SECRET.key(),
@@ -79,7 +80,7 @@ def build_jindo_config(catalog_options: Options):
if not supported_prefix or value is None:
continue
key = _CASE_SENSITIVE_JINDO_CONFIG_KEYS.get(raw_key.lower(), raw_key)
- if key in _PYPAIMON_ONLY_JINDO_CONFIG_KEYS:
+ if key in _PYPAIMON_ONLY_JINDO_CONFIG_KEYS or
key.startswith("fs.oss.cpp."):
# This option is handled by PyPaimon.
continue
config.set(key, _jindo_config_value(value))
diff --git a/paimon-python/pypaimon/filesystem/pvfs.py
b/paimon-python/pypaimon/filesystem/pvfs.py
index 13030f08b8..823f3ac3e9 100644
--- a/paimon-python/pypaimon/filesystem/pvfs.py
+++ b/paimon-python/pypaimon/filesystem/pvfs.py
@@ -907,15 +907,16 @@ class PaimonVirtualFileSystem(fsspec.AbstractFileSystem):
``legacy`` backs it with ``ossfs``. ``jindo`` is the default; when
pyjindosdk is not installed it falls back to ``ossfs`` -- consistent
with PyArrowFileIO.
+ With ``cpp``, Python I/O uses ``fs.oss.python.impl`` instead.
``ossfs`` writes every object through OSS ``AppendObject``, which can
fail with ``PositionNotEqualToLength`` (409) on the OSS
data-acceleration
endpoint for multi-chunk writes; the jindo backend avoids that path.
"""
- oss_impl = self.options.get(OssOptions.OSS_IMPL)
+ oss_impl = self._python_oss_impl()
if oss_impl not in ("jindo", "legacy"):
raise Exception(
- "Unsupported fs.oss.impl value: '{}'. "
+ "Unsupported fs.oss.impl or fs.oss.python.impl value: '{}'. "
"Supported values are 'jindo' and 'legacy'.".format(oss_impl)
)
if self._use_jindo_oss_backend():
@@ -923,12 +924,16 @@ class PaimonVirtualFileSystem(fsspec.AbstractFileSystem):
return create_jindo_oss_filesystem("oss://{}/".format(bucket),
options)
if oss_impl == "jindo":
logger.warning(
- "fs.oss.impl is 'jindo' but pyjindosdk is not installed. "
+ "Python OSS backend is 'jindo' but pyjindosdk is not
installed. "
"Falling back to the ossfs (OSS AppendObject) implementation. "
"Install pyjindosdk for native multipart upload: pip install
pyjindosdk"
)
return PaimonVirtualFileSystem._get_ossfs_filesystem(options)
+ def _python_oss_impl(self):
+ impl = self.options.get(OssOptions.OSS_IMPL)
+ return self.options.get(OssOptions.OSS_PYTHON_IMPL) if impl == "cpp"
else impl
+
def _use_jindo_oss_backend(self) -> bool:
"""Whether OSS access uses the native jindo backend rather than ossfs.
@@ -941,7 +946,7 @@ class PaimonVirtualFileSystem(fsspec.AbstractFileSystem):
drift apart.
"""
return (
- self.options.get(OssOptions.OSS_IMPL) == "jindo"
+ self._python_oss_impl() == "jindo"
and JINDO_AVAILABLE
and JINDO_OSSFS_AVAILABLE
)
diff --git a/paimon-python/pypaimon/filesystem/pyarrow_file_io.py
b/paimon-python/pypaimon/filesystem/pyarrow_file_io.py
index 1a4362aedb..dcb1c5d5c3 100644
--- a/paimon-python/pypaimon/filesystem/pyarrow_file_io.py
+++ b/paimon-python/pypaimon/filesystem/pyarrow_file_io.py
@@ -82,9 +82,14 @@ class PyArrowFileIO(FileIO):
if self._is_oss:
self._oss_bucket = self._extract_oss_bucket(path)
+ impl_key = OssOptions.OSS_IMPL.key()
+ if _oss_impl == "cpp":
+ # Keep the original properties for the native FileIO bridge.
+ _oss_impl = self.properties.get(OssOptions.OSS_PYTHON_IMPL)
+ impl_key = OssOptions.OSS_PYTHON_IMPL.key()
if _oss_impl not in ("jindo", "legacy"):
raise ValueError(
- f"Unsupported fs.oss.impl value: '{_oss_impl}'. "
+ f"Unsupported {impl_key} value: '{_oss_impl}'. "
f"Supported values are 'jindo' and 'legacy'.")
if _oss_impl == "legacy":
self.filesystem = self._initialize_oss_fs(path)
@@ -92,7 +97,7 @@ class PyArrowFileIO(FileIO):
self.filesystem = self._initialize_jindo_fs(path)
else:
self.logger.info(
- "fs.oss.impl is 'jindo' but pyjindosdk is not installed. "
+ "Python OSS backend is 'jindo' but pyjindosdk is not
installed. "
"Falling back to legacy PyArrow S3FileSystem
implementation. "
"Install pyjindosdk for better performance: pip install
pyjindosdk")
self.filesystem = self._initialize_oss_fs(path)
@@ -483,10 +488,13 @@ class PyArrowFileIO(FileIO):
def list_status(self, path: str):
if self._legacy_oss_mode():
+ impl_key = (OssOptions.OSS_PYTHON_IMPL.key()
+ if self.properties.get(OssOptions.OSS_IMPL) == "cpp"
+ else OssOptions.OSS_IMPL.key())
raise LegacyOssDirectoryListingError(
"Listing OSS directories is not supported with PyArrow < 16 "
"(it parses the first key segment as a bucket). Upgrade to "
- "pyarrow >= 16, or install pyjindosdk and set
fs.oss.impl=jindo.")
+ f"pyarrow >= 16, or install pyjindosdk and set
{impl_key}=jindo.")
path_str = self.to_filesystem_path(path)
selector = pafs.FileSelector(path_str, recursive=False,
allow_not_found=True)
return self.filesystem.get_file_info(selector)
diff --git a/paimon-python/pypaimon/tests/jindo_file_system_test.py
b/paimon-python/pypaimon/tests/jindo_file_system_test.py
index 0c95c9228b..de627738da 100644
--- a/paimon-python/pypaimon/tests/jindo_file_system_test.py
+++ b/paimon-python/pypaimon/tests/jindo_file_system_test.py
@@ -138,6 +138,19 @@ class JindoConfigTest(unittest.TestCase):
self.assertEqual(call[0][2].values["fs.oss.securityToken"],
"test-token")
self.assertEqual(call[0][2].values["fs.jindocache.client.metrics.enable"],
"true")
+ def test_cpp_options_are_not_forwarded_to_jindo(self):
+ options = Options({'fs.oss.impl': 'cpp', 'fs.oss.python.impl': 'jindo',
+ 'fs.oss.cpp.library.path': '/test/libbridge.so',
+ 'fs.oss.cpp.max.concurrent.requests': '4',
+ 'fs.oss.endpoint': 'oss-cn-test.example.com'})
+ with mock.patch.object(jindo_module, 'JINDO_AVAILABLE', True), \
+ mock.patch.object(jindo_module, 'jutil',
types.SimpleNamespace(Config=_RecordingConfig)):
+ config = jindo_module.build_jindo_config(options)
+ for key in options.to_map():
+ if key != 'fs.oss.endpoint':
+ self.assertNotIn(key, config.values)
+ self.assertEqual(config.values['fs.oss.endpoint'],
'oss-cn-test.example.com')
+
def test_forwards_native_options_to_connect(self):
created_config = _RecordingConfig()
config_factory = mock.Mock(return_value=created_config)
diff --git a/paimon-python/pypaimon/tests/native_plan_test.py
b/paimon-python/pypaimon/tests/native_plan_test.py
index 92378ec1fd..d2a522189e 100644
--- a/paimon-python/pypaimon/tests/native_plan_test.py
+++ b/paimon-python/pypaimon/tests/native_plan_test.py
@@ -719,6 +719,26 @@ class NativePlanTest(unittest.TestCase):
'metastore': 'filesystem',
})
+ def test_native_cpp_options_survive_python_backend_selection(self):
+ from pypaimon.filesystem.oss_file_io import OssFileIO
+ from pypaimon.read.native_plan import _resolved_schema_file_io_options
+
+ options = Options({'fs.oss.impl': 'cpp', 'fs.oss.python.impl':
'legacy',
+ 'fs.oss.cpp.library.path': '/test/libbridge.so',
+ 'fs.oss.cpp.max.concurrent.requests': '4'})
+ for loader_class in (FileSystemCatalogLoader, RESTCatalogLoader):
+ table = Mock(table_path='oss://bucket/table')
+ table.catalog_environment.catalog_loader = loader_class(
+ CatalogContext.create_from_options(options))
+ with patch.object(OssFileIO, '_initialize_oss_fs'), patch(
+
'pypaimon.filesystem.jindo_file_system_handler.JINDO_AVAILABLE', True):
+ table.file_io = OssFileIO(table.table_path, options)
+ native_options = _catalog_options(table)
+ for key, value in options.to_map().items():
+ self.assertEqual(native_options[key], value)
+ if loader_class is FileSystemCatalogLoader:
+ self.assertEqual(_resolved_schema_file_io_options(table),
options.to_map())
+
def test_catalog_options_reject_loader_subclass(self):
class RoutedFileSystemLoader(FileSystemCatalogLoader):
pass
@@ -875,13 +895,15 @@ class NativePlanTest(unittest.TestCase):
from pypaimon.common.identifier import Identifier
from pypaimon.read.native_plan import _native_read_builder
- response = json.dumps({'name': 't$branch_dev', 'path': '/warehouse/t',
+ response = json.dumps({'name': 't$branch_dev', 'path':
'oss://bucket/t',
'id': 'uuid', 'isExternal': False})
loader = RESTCatalogLoader(CatalogContext.create_from_options(Options({
- 'uri': 'http://localhost:1', 'warehouse': 'test',
'data-token.enabled': 'true'})))
+ 'uri': 'http://localhost:1', 'warehouse': 'test',
'data-token.enabled': 'true',
+ 'fs.oss.impl': 'cpp', 'fs.oss.python.impl': 'legacy',
+ 'fs.oss.cpp.library.path': '/test/libbridge.so'})))
table = Mock()
table.identifier = Identifier('db', 't', branch='dev')
- table.table_path = '/warehouse/t'
+ table.table_path = 'oss://bucket/t'
table.current_branch.return_value = 'dev'
table.catalog_environment = CatalogEnvironment(
identifier=table.identifier, uuid='uuid', catalog_loader=loader,
@@ -915,6 +937,7 @@ class NativePlanTest(unittest.TestCase):
fake_df.Table.from_rest_response.assert_called_once_with(
response, database='db', table='t$branch_dev',
rest_options=_catalog_options(table))
+
self.assertEqual(fake_df.Table.from_rest_response.call_args[1]['rest_options']['fs.oss.impl'],
'cpp')
self.assertEqual(native_table.copy_with_resolved_schema.call_args_list,
[call(resolved, branch='dev')] * 5)
self.assertEqual(native_table.new_read_builder.call_count, 4)
diff --git a/paimon-python/pypaimon/tests/oss_legacy_mode_test.py
b/paimon-python/pypaimon/tests/oss_legacy_mode_test.py
index dc300cb391..61ea8b5727 100644
--- a/paimon-python/pypaimon/tests/oss_legacy_mode_test.py
+++ b/paimon-python/pypaimon/tests/oss_legacy_mode_test.py
@@ -47,6 +47,27 @@ def _probe_response(status_code, body):
class OssLegacyModeTest(unittest.TestCase):
"""Behavior of OssFileIO when OSS runs on PyArrow < 16."""
+ def test_cpp_keeps_python_backend_and_original_properties(self):
+ for available, backend in ((True, 'jindo'), (False, 'legacy')):
+ for explicit in (None, 'legacy', 'jindo'):
+ options = Options({'fs.oss.impl': 'cpp',
+ 'fs.oss.cpp.library.path':
'/test/libbridge.so'})
+ if explicit is not None:
+ options.set(OssOptions.OSS_PYTHON_IMPL, explicit)
+ expected = 'legacy' if explicit == 'legacy' else backend
+ with
mock.patch('pypaimon.filesystem.pyarrow_file_io.JINDO_AVAILABLE', available), \
+ mock.patch.object(OssFileIO, '_initialize_oss_fs') as
legacy, \
+ mock.patch.object(OssFileIO, '_initialize_jindo_fs')
as jindo:
+ file_io = OssFileIO(TABLE_PATH, options)
+ self.assertEqual(jindo.call_count, int(expected == 'jindo'))
+ self.assertEqual(legacy.call_count, int(expected == 'legacy'))
+ self.assertIs(file_io.properties, options)
+ self.assertEqual(options.get(OssOptions.OSS_IMPL), 'cpp')
+
+ def test_cpp_rejects_invalid_python_backend(self):
+ with self.assertRaisesRegex(ValueError, 'fs.oss.python.impl'):
+ OssFileIO(TABLE_PATH, Options({'fs.oss.impl': 'cpp',
'fs.oss.python.impl': 'cpp'}))
+
def _new_file_io(self, legacy):
options = Options({
OssOptions.OSS_ACCESS_KEY_ID.key(): "ak",
@@ -238,8 +259,20 @@ class OssLegacyModeTest(unittest.TestCase):
with self.assertRaises(LegacyOssDirectoryListingError) as ctx:
file_io.list_status(TABLE_PATH)
self.assertIn("pyarrow >= 16", str(ctx.exception))
+ self.assertIn("set fs.oss.impl=jindo", str(ctx.exception))
file_io.filesystem.get_file_info.assert_not_called()
+ def test_cpp_legacy_list_status_preserves_native_backend_in_error(self):
+ options = Options({'fs.oss.impl': 'cpp', 'fs.oss.python.impl':
'legacy'})
+ with mock.patch.object(OssFileIO, '_initialize_oss_fs',
return_value=mock.Mock()):
+ file_io = OssFileIO(TABLE_PATH, options)
+ file_io._oss_bucket_in_endpoint = True
+
+ with self.assertRaises(LegacyOssDirectoryListingError) as ctx:
+ file_io.list_status(TABLE_PATH)
+ self.assertIn("set fs.oss.python.impl=jindo", str(ctx.exception))
+ self.assertNotIn("set fs.oss.impl=jindo", str(ctx.exception))
+
def test_modern_list_status_uses_selector(self):
file_io = self._new_file_io(legacy=False)
file_io.filesystem.get_file_info.return_value = []
diff --git a/paimon-python/pypaimon/tests/pvfs_oss_filesystem_test.py
b/paimon-python/pypaimon/tests/pvfs_oss_filesystem_test.py
index d35a851299..0dfdc01d4a 100644
--- a/paimon-python/pypaimon/tests/pvfs_oss_filesystem_test.py
+++ b/paimon-python/pypaimon/tests/pvfs_oss_filesystem_test.py
@@ -109,6 +109,21 @@ class GetOssFilesystemDispatchTest(unittest.TestCase):
self.assertIs(fs, ossfs_sentinel)
self.assertEqual(jindo_calls, [])
+ def test_cpp_keeps_python_jindo_or_legacy_backend(self):
+ fs, _, calls = self._dispatch('cpp', jindo_ossfs_available=True)
+ self.assertEqual(fs, 'JINDO_FS')
+ self.assertEqual(len(calls), 1)
+ fs, legacy, calls = self._dispatch('cpp', jindo_ossfs_available=False)
+ self.assertIs(fs, legacy)
+ self.assertEqual(calls, [])
+ pvfs = _make_pvfs({'fs.oss.impl': 'cpp', 'fs.oss.python.impl':
'legacy'})
+ with mock.patch.object(pvfs_module, 'JINDO_AVAILABLE', True), \
+ mock.patch.object(pvfs_module, 'JINDO_OSSFS_AVAILABLE', True),
\
+ mock.patch.object(PaimonVirtualFileSystem,
'_get_ossfs_filesystem') as legacy:
+ self.assertFalse(pvfs._use_jindo_oss_backend())
+ self.assertIs(pvfs._get_oss_filesystem(self.token_options,
self.STORAGE_LOCATION),
+ legacy.return_value)
+
def test_jindo_uses_jindo_when_available(self):
fs, _, jindo_calls = self._dispatch("jindo",
jindo_ossfs_available=True)
self.assertEqual(fs, "JINDO_FS")
