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 81a3f27162 [python] Fix pyjindo advanced config setting in pypaimon
(#9206)
81a3f27162 is described below
commit 81a3f2716275c0b8d805a42280eee8227b27cd01
Author: timmyyao <[email protected]>
AuthorDate: Thu Aug 13 21:51:08 2026 +0800
[python] Fix pyjindo advanced config setting in pypaimon (#9206)
---
.../filesystem/jindo_file_system_handler.py | 64 +++++-----
.../pypaimon/tests/jindo_file_system_test.py | 136 +++++++++++++++++++++
2 files changed, 170 insertions(+), 30 deletions(-)
diff --git a/paimon-python/pypaimon/filesystem/jindo_file_system_handler.py
b/paimon-python/pypaimon/filesystem/jindo_file_system_handler.py
index efbfe7f189..73ad09077f 100644
--- a/paimon-python/pypaimon/filesystem/jindo_file_system_handler.py
+++ b/paimon-python/pypaimon/filesystem/jindo_file_system_handler.py
@@ -22,11 +22,7 @@ from pyarrow import PythonFile
from pyarrow._fs import FileSystemHandler
from pyarrow.fs import FileInfo, FileSelector, FileType
-# `JindoFileSystemHandler` (the PyArrow FileIO path) only needs `pyjindo.fs`
-# and `pyjindo.util`. The PVFS jindo backend (`create_jindo_oss_filesystem`)
-# additionally needs `pyjindo.ossfs`. Track the two surfaces independently so
-# that a pyjindosdk build without `pyjindo.ossfs` does not silently disable
-# the previously-working PyArrow path.
+# The PyArrow and PVFS paths use separate pyjindo modules.
try:
import pyjindo.fs as jfs
import pyjindo.util as jutil
@@ -47,18 +43,42 @@ from pypaimon.common.options import Options
from pypaimon.common.options.config import OssOptions
-def build_jindo_config(catalog_options: Options):
- """Build a pyjindo ``Config`` from OSS catalog options.
+_JINDO_CONFIG_PREFIXES = ("fs.", "logger.")
+_PYPAIMON_ONLY_JINDO_CONFIG_KEYS = {OssOptions.OSS_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(),
+ OssOptions.OSS_SECURITY_TOKEN.key().lower():
OssOptions.OSS_SECURITY_TOKEN.key(),
+}
+
+
+def _jindo_config_value(value) -> str:
+ if isinstance(value, bool):
+ return str(value).lower()
+ return str(value)
- Shared by ``JindoFileSystemHandler`` (the PyArrow FileIO path) and
- ``create_jindo_oss_filesystem`` (the PVFS fsspec path) so both jindo entry
- points consume exactly the same credential / endpoint options.
- """
+
+def build_jindo_config(catalog_options: Options):
+ """Build a pyjindo ``Config`` from catalog options."""
if not JINDO_AVAILABLE:
raise ImportError("Module pyjindo is not available. Please install
pyjindosdk.")
+ # Use catalog options as the complete configuration source.
config = jutil.Config()
+ # Forward supported filesystem and logger options.
+ for raw_key, value in catalog_options.to_map().items():
+ supported_prefix = isinstance(raw_key, str)
+ supported_prefix = supported_prefix and raw_key.startswith(
+ _JINDO_CONFIG_PREFIXES)
+ 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:
+ # This option is handled by PyPaimon.
+ continue
+ config.set(key, _jindo_config_value(value))
+
access_key_id = catalog_options.get(OssOptions.OSS_ACCESS_KEY_ID)
access_key_secret = catalog_options.get(OssOptions.OSS_ACCESS_KEY_SECRET)
security_token = catalog_options.get(OssOptions.OSS_SECURITY_TOKEN)
@@ -81,18 +101,7 @@ def build_jindo_config(catalog_options: Options):
def create_jindo_oss_filesystem(root_uri: str, catalog_options: Options):
- """Create an fsspec-compatible ``JindoOssFileSystem`` for an OSS bucket.
-
- ``PaimonVirtualFileSystem`` uses this to back OSS reads/writes with the
- native JindoSDK instead of ``ossfs``. JindoSDK writes objects via
- PutObject / multipart upload, so it never issues OSS ``AppendObject`` --
- the call that fails with ``PositionNotEqualToLength`` (409) on the OSS
- data-acceleration endpoint when ``ossfs`` flushes a multi-chunk write.
-
- ``root_uri`` is the bucket root, e.g. ``oss://my-bucket/``; it must carry
- the bucket so ``JindoOssFileSystem`` can re-attach the ``oss://`` scheme to
- the bucket-relative paths that ``PaimonVirtualFileSystem`` passes in.
- """
+ """Create a Jindo OSS filesystem for ``PaimonVirtualFileSystem``."""
if not (JINDO_AVAILABLE and JINDO_OSSFS_AVAILABLE):
raise ImportError(
"pyjindo.ossfs is not available. Please install
pyjindosdk>=6.10.4."
@@ -101,14 +110,9 @@ def create_jindo_oss_filesystem(root_uri: str,
catalog_options: Options):
return jossfs.JindoOssFileSystem(
uri=root_uri,
config=build_jindo_config(catalog_options),
- # PaimonVirtualFileSystem owns directory semantics for the virtual FS;
- # the backing object-store fs must not auto-create dir-marker objects.
+ # PaimonVirtualFileSystem manages directory semantics.
auto_mkdir=False,
- # Bypass fsspec's _Cached metaclass instance cache, so the only
- # reference to this filesystem -- and to its underlying native jindo
- # connection -- is the PaimonRealStorage cache in PVFS. On token
- # refresh PVFS replaces that entry and the native resources can be
- # released, instead of being pinned forever by fsspec's global cache.
+ # PaimonVirtualFileSystem manages filesystem instances.
skip_instance_cache=True,
)
diff --git a/paimon-python/pypaimon/tests/jindo_file_system_test.py
b/paimon-python/pypaimon/tests/jindo_file_system_test.py
index 9bd45aedda..b828fd8b83 100644
--- a/paimon-python/pypaimon/tests/jindo_file_system_test.py
+++ b/paimon-python/pypaimon/tests/jindo_file_system_test.py
@@ -16,17 +16,153 @@
# under the License.
import os
+import types
import unittest
import uuid
+from unittest import mock
import pyarrow.fs as pafs
from pyarrow.fs import PyFileSystem
from pypaimon.common.options import Options
from pypaimon.common.options.config import OssOptions
+from pypaimon.filesystem import jindo_file_system_handler as jindo_module
from pypaimon.filesystem.jindo_file_system_handler import
JindoFileSystemHandler, JINDO_AVAILABLE
+class _RecordingConfig:
+ def __init__(self, values=None):
+ self.values = dict(values or {})
+
+ def set(self, key, value):
+ self.values[key] = value
+
+
+class JindoConfigTest(unittest.TestCase):
+
+ def test_forwards_native_options_to_connect(self):
+ created_config = _RecordingConfig()
+ config_factory = mock.Mock(return_value=created_config)
+ read_config = mock.Mock()
+ fake_jutil = types.SimpleNamespace(
+ Config=config_factory,
+ read_config=read_config,
+ )
+ options = Options({
+ "fs.oss.accesskeyid": "ak",
+ "fs.oss.accesskeysecret": "sk",
+ "fs.oss.securitytoken": "token",
+ OssOptions.OSS_ENDPOINT.key(): "https://cache-seed:80";,
+ OssOptions.OSS_IMPL.key(): "jindo",
+ "fs.oss.dlf-cache.consistent-hash.enabled": True,
+ "fs.oss.dlf-cache.server.address": "http://cache-server:18101";,
+ "fs.oss.https.enable": False,
+ "fs.oss.second.level.domain.enable": "true",
+ "fs.jindocache.client.metrics.enable": True,
+ "logger.dir": "/tmp/jindo-log",
+ "logger.verbose": 3,
+ "logger.console.log.enable": False,
+ "fs.oss.unset.option": None,
+ "metastore": "rest",
+ })
+ connect = mock.Mock(return_value=mock.sentinel.jindo_fs)
+ fake_jfs = types.SimpleNamespace(connect=connect)
+
+ with mock.patch.object(jindo_module, "JINDO_AVAILABLE", True), \
+ mock.patch.object(jindo_module, "jfs", fake_jfs), \
+ mock.patch.object(jindo_module, "jutil", fake_jutil):
+ handler = JindoFileSystemHandler("oss://bucket/", options)
+
+ config_factory.assert_called_once_with()
+ read_config.assert_not_called()
+ connect.assert_called_once_with("oss://bucket/", "root", mock.ANY)
+ self.assertIs(handler._jindo_fs, mock.sentinel.jindo_fs)
+ config = connect.call_args[0][2]
+ self.assertIs(config, created_config)
+ self.assertEqual(config.values["fs.oss.accessKeyId"], "ak")
+ self.assertEqual(config.values["fs.oss.accessKeySecret"], "sk")
+ self.assertEqual(config.values["fs.oss.securityToken"], "token")
+ self.assertEqual(config.values["fs.oss.endpoint"], "cache-seed:80")
+ self.assertEqual(
+ config.values["fs.oss.dlf-cache.consistent-hash.enabled"], "true")
+ self.assertEqual(
+ config.values["fs.oss.dlf-cache.server.address"],
+ "http://cache-server:18101";)
+ self.assertEqual(config.values["fs.oss.https.enable"], "false")
+ self.assertEqual(
+ config.values["fs.oss.second.level.domain.enable"], "true")
+ self.assertEqual(
+ config.values["fs.jindocache.client.metrics.enable"], "true")
+ self.assertEqual(config.values["logger.dir"], "/tmp/jindo-log")
+ self.assertEqual(config.values["logger.verbose"], "3")
+ self.assertEqual(config.values["logger.console.log.enable"], "false")
+ self.assertEqual(config.values["fs.oss.user.agent.features"],
"pypaimon")
+ self.assertNotIn(OssOptions.OSS_IMPL.key(), config.values)
+ self.assertNotIn("fs.oss.unset.option", config.values)
+ self.assertNotIn("metastore", config.values)
+
+ def test_does_not_load_external_config(self):
+ created_config = _RecordingConfig()
+ config_factory = mock.Mock(return_value=created_config)
+ read_config = mock.Mock(
+ side_effect=AssertionError("external config must not be loaded"))
+ fake_jutil = types.SimpleNamespace(
+ Config=config_factory,
+ read_config=read_config,
+ )
+ options = Options({
+ OssOptions.OSS_ENDPOINT.key(): "http://127.0.0.1:80";,
+ "logger.verbose": "3",
+ })
+
+ with mock.patch.object(jindo_module, "JINDO_AVAILABLE", True), \
+ mock.patch.object(jindo_module, "jutil", fake_jutil):
+ config = jindo_module.build_jindo_config(options)
+
+ config_factory.assert_called_once_with()
+ read_config.assert_not_called()
+ self.assertIs(config, created_config)
+ self.assertEqual(config.values["fs.oss.endpoint"], "127.0.0.1:80")
+ self.assertEqual(config.values["logger.verbose"], "3")
+ self.assertNotIn("fs.oss.provider.endpoint", config.values)
+ self.assertNotIn("fs.oss.provider.format", config.values)
+
+ def test_forwards_native_options_to_jindo_oss_filesystem(self):
+ created_config = _RecordingConfig()
+ config_factory = mock.Mock(return_value=created_config)
+ fake_jutil = types.SimpleNamespace(Config=config_factory)
+ jindo_oss_filesystem =
mock.Mock(return_value=mock.sentinel.jindo_oss_fs)
+ fake_jossfs =
types.SimpleNamespace(JindoOssFileSystem=jindo_oss_filesystem)
+ options = Options({
+ "fs.oss.dlf-cache.consistent-hash.enabled": "true",
+ "fs.oss.dlf-cache.server.address": "http://cache-server:18101";,
+ "logger.verbose": 3,
+ })
+
+ with mock.patch.object(jindo_module, "JINDO_AVAILABLE", True), \
+ mock.patch.object(jindo_module, "JINDO_OSSFS_AVAILABLE", True), \
+ mock.patch.object(jindo_module, "jutil", fake_jutil), \
+ mock.patch.object(jindo_module, "jossfs", fake_jossfs):
+ filesystem = jindo_module.create_jindo_oss_filesystem(
+ "oss://bucket/", options)
+
+ self.assertIs(filesystem, mock.sentinel.jindo_oss_fs)
+ config_factory.assert_called_once_with()
+ jindo_oss_filesystem.assert_called_once_with(
+ uri="oss://bucket/",
+ config=created_config,
+ auto_mkdir=False,
+ skip_instance_cache=True,
+ )
+ self.assertEqual(
+ created_config.values["fs.oss.dlf-cache.consistent-hash.enabled"],
+ "true")
+ self.assertEqual(
+ created_config.values["fs.oss.dlf-cache.server.address"],
+ "http://cache-server:18101";)
+ self.assertEqual(created_config.values["logger.verbose"], "3")
+
+
class JindoFileSystemTest(unittest.TestCase):
"""Test cases for JindoFileSystem."""
