[python] Fix pyjindo advanced config setting in pypaimon

paimon-python/pypaimon/filesystem/jindo_file_system_handler.py

@@ -22,11 +22,7 @@
 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.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,
     )
 

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."""