[python] Fix OSSParam to access DLF (#6332)

Author: discivigour

paimon-python/pypaimon/catalog/rest/rest_catalog.py

@@ -19,6 +19,9 @@
 from typing import Any, Callable, Dict, List, Optional, Union
 from urllib.parse import urlparse
 
+import pyarrow
+from packaging.version import parse
+
 from pypaimon.api.api_response import GetTableResponse, PagedList
 from pypaimon.api.options import Options
 from pypaimon.api.rest_api import RESTApi
@@ -200,17 +203,17 @@ def to_table_metadata(self, db: str, response: GetTableResponse) -> TableMetadat
             uuid=response.get_id()
         )
 
-    def file_io_from_options(self, table_path: Path) -> FileIO:
-        return FileIO(str(table_path), self.context.options.data)
+    def file_io_from_options(self, table_path: str) -> FileIO:
+        return FileIO(table_path, self.context.options.data)
 
-    def file_io_for_data(self, table_path: Path, identifier: Identifier):
+    def file_io_for_data(self, table_path: str, identifier: Identifier):
         return RESTTokenFileIO(identifier, table_path, self.context.options.data) \
             if self.data_token_enabled else self.file_io_from_options(table_path)
 
     def load_table(self,
                    identifier: Identifier,
-                   internal_file_io: Callable[[Path], Any],
-                   external_file_io: Callable[[Path], Any],
+                   internal_file_io: Callable[[str], Any],
+                   external_file_io: Callable[[str], Any],
                    metadata_loader: Callable[[Identifier], TableMetadata],
                    ) -> FileStoreTable:
         metadata = metadata_loader(identifier)
@@ -223,9 +226,12 @@ def load_table(self,
             supports_version_management=True  # REST catalogs support version management
         )
         path_parsed = urlparse(schema.options.get(CoreOptions.PATH))
-        path = Path(path_parsed.path) if path_parsed.scheme is None else Path(schema.options.get(CoreOptions.PATH))
-        table_path = path_parsed.netloc + "/" + path_parsed.path \
-            if path_parsed.scheme == "file" else path_parsed.path[1:]
+        path = path_parsed.path if path_parsed.scheme is None else schema.options.get(CoreOptions.PATH)
+        if path_parsed.scheme == "file":
+            table_path = path_parsed.path
+        else:
+            table_path = path_parsed.netloc + path_parsed.path \
+                if parse(pyarrow.__version__) >= parse("7.0.0") else path_parsed.path[1:]
         table = self.create(data_file_io(path),
                             Path(table_path),
                             schema,

paimon-python/pypaimon/catalog/rest/rest_token_file_io.py

@@ -31,20 +31,20 @@
 
 class RESTTokenFileIO(FileIO):
 
-    def __init__(self, identifier: Identifier, path: Path,
+    def __init__(self, identifier: Identifier, path: str,
                  catalog_options: Optional[dict] = None):
         self.identifier = identifier
         self.path = path
         self.token: Optional[RESTToken] = None
         self.api_instance: Optional[RESTApi] = None
         self.lock = threading.Lock()
         self.log = logging.getLogger(__name__)
-        super().__init__(str(path), catalog_options)
+        super().__init__(path, catalog_options)
 
-    def _initialize_oss_fs(self) -> FileSystem:
+    def _initialize_oss_fs(self, path) -> FileSystem:
         self.try_to_refresh_token()
         self.properties.update(self.token.token)
-        return super()._initialize_oss_fs()
+        return super()._initialize_oss_fs(path)
 
     def new_output_stream(self, path: Path):
         return self.filesystem.open_output_stream(str(path))

paimon-python/pypaimon/common/config.py

@@ -21,7 +21,6 @@ class OssOptions:
     OSS_SECURITY_TOKEN = "fs.oss.securityToken"
     OSS_ENDPOINT = "fs.oss.endpoint"
     OSS_REGION = "fs.oss.region"
-    OSS_BUCKET = "fs.oss.bucket"
 
 
 class S3Options:

paimon-python/pypaimon/common/file_io.py

@@ -31,12 +31,12 @@
 
 
 class FileIO:
-    def __init__(self, warehouse: str, catalog_options: dict):
+    def __init__(self, path: str, catalog_options: dict):
         self.properties = catalog_options
         self.logger = logging.getLogger(__name__)
-        scheme, netloc, path = self.parse_location(warehouse)
+        scheme, netloc, _ = self.parse_location(path)
         if scheme in {"oss"}:
-            self.filesystem = self._initialize_oss_fs()
+            self.filesystem = self._initialize_oss_fs(path)
         elif scheme in {"s3", "s3a", "s3n"}:
             self.filesystem = self._initialize_s3_fs()
         elif scheme in {"hdfs", "viewfs"}:
@@ -56,7 +56,29 @@ def parse_location(location: str):
         else:
             return uri.scheme, uri.netloc, f"{uri.netloc}{uri.path}"
 
-    def _initialize_oss_fs(self) -> FileSystem:
+    def _extract_oss_bucket(self, location) -> str:
+        uri = urlparse(location)
+        if uri.scheme and uri.scheme != "oss":
+            raise ValueError("Not an OSS URI: {}".format(location))
+
+        netloc = uri.netloc or ""
+        # parse oss://access_id:secret_key@Endpoint/bucket/path/to/object
+        if (getattr(uri, "username", None) or getattr(uri, "password", None)) or ("@" in netloc):
+            first_segment = uri.path.lstrip("/").split("/", 1)[0]
+            if not first_segment:
+                raise ValueError("Invalid OSS URI without bucket: {}".format(location))
+            return first_segment
+
+        # parse oss://bucket/... or oss://bucket.endpoint/...
+        host = getattr(uri, "hostname", None) or netloc
+        if not host:
+            raise ValueError("Invalid OSS URI without host: {}".format(location))
+        bucket = host.split(".", 1)[0]
+        if not bucket:
+            raise ValueError("Invalid OSS URI without bucket: {}".format(location))
+        return bucket
+
+    def _initialize_oss_fs(self, path) -> FileSystem:
         from pyarrow.fs import S3FileSystem
 
         client_kwargs = {
@@ -71,7 +93,8 @@ def _initialize_oss_fs(self) -> FileSystem:
             client_kwargs['force_virtual_addressing'] = True
             client_kwargs['endpoint_override'] = self.properties.get(OssOptions.OSS_ENDPOINT)
         else:
-            client_kwargs['endpoint_override'] = (self.properties.get(OssOptions.OSS_BUCKET) + "." +
+            oss_bucket = self._extract_oss_bucket(path)
+            client_kwargs['endpoint_override'] = (oss_bucket + "." +
                                                   self.properties.get(OssOptions.OSS_ENDPOINT))
 
         return S3FileSystem(**client_kwargs)

paimon-python/pypaimon/tests/py36/ao_simple_test.py

@@ -15,11 +15,15 @@
 See the License for the specific language governing permissions and
 limitations under the License.
 """
+from unittest.mock import patch
+
 import pyarrow as pa
 
 from pypaimon import Schema
 from pypaimon.catalog.catalog_exception import TableNotExistException, TableAlreadyExistException, \
     DatabaseNotExistException, DatabaseAlreadyExistException
+from pypaimon.common.config import OssOptions
+from pypaimon.common.file_io import FileIO
 from pypaimon.tests.py36.pyarrow_compat import table_sort_by
 from pypaimon.tests.rest.rest_base_test import RESTBaseTest
 
@@ -385,3 +389,33 @@ def test_create_drop_database_table(self):
             self.rest_catalog.drop_database("db1", True)
         except DatabaseNotExistException:
             self.fail("drop_database with ignore_if_exists=True should not raise DatabaseNotExistException")
+
+    def test_initialize_oss_fs_pyarrow_lt_7(self):
+        props = {
+            OssOptions.OSS_ACCESS_KEY_ID: "AKID",
+            OssOptions.OSS_ACCESS_KEY_SECRET: "SECRET",
+            OssOptions.OSS_SECURITY_TOKEN: "TOKEN",
+            OssOptions.OSS_REGION: "cn-hangzhou",
+            OssOptions.OSS_ENDPOINT: "oss-cn-hangzhou.aliyuncs.com",
+        }
+
+        with patch("pypaimon.common.file_io.pyarrow.__version__", "6.0.0"), \
+                patch("pyarrow.fs.S3FileSystem") as mock_s3fs:
+            FileIO("oss://oss-bucket/paimon-database/paimon-table", props)
+            mock_s3fs.assert_called_once_with(access_key="AKID",
+                                              secret_key="SECRET",
+                                              session_token="TOKEN",
+                                              region="cn-hangzhou",
+                                              endpoint_override="oss-bucket." + props[OssOptions.OSS_ENDPOINT])
+            FileIO("oss://oss-bucket.endpoint/paimon-database/paimon-table", props)
+            mock_s3fs.assert_called_with(access_key="AKID",
+                                         secret_key="SECRET",
+                                         session_token="TOKEN",
+                                         region="cn-hangzhou",
+                                         endpoint_override="oss-bucket." + props[OssOptions.OSS_ENDPOINT])
+            FileIO("oss://access_id:secret_key@Endpoint/oss-bucket/paimon-database/paimon-table", props)
+            mock_s3fs.assert_called_with(access_key="AKID",
+                                         secret_key="SECRET",
+                                         session_token="TOKEN",
+                                         region="cn-hangzhou",
+                                         endpoint_override="oss-bucket." + props[OssOptions.OSS_ENDPOINT])