| # Licensed to the Apache Software Foundation (ASF) under one |
| # or more contributor license agreements. See the NOTICE file |
| # distributed with this work for additional information |
| # regarding copyright ownership. The ASF licenses this file |
| # to you under the Apache License, Version 2.0 (the |
| # "License"); you may not use this file except in compliance |
| # with the License. You may obtain a copy of the License at |
| # |
| # http://www.apache.org/licenses/LICENSE-2.0 |
| # |
| # Unless required by applicable law or agreed to in writing, |
| # software distributed under the License is distributed on an |
| # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| # KIND, either express or implied. See the License for the |
| # specific language governing permissions and limitations |
| # under the License. |
| |
| import datetime |
| import importlib |
| import logging |
| import posixpath |
| import time |
| from abc import ABC |
| from dataclasses import dataclass |
| from enum import Enum |
| from typing import Any, Dict, Optional, Tuple, Union |
| |
| import fsspec |
| from cachetools import LRUCache, TTLCache |
| from fsspec import AbstractFileSystem |
| from fsspec.implementations.local import LocalFileSystem |
| from readerwriterlock import rwlock |
| |
| from pypaimon.api.api_response import GetTableResponse, GetTableTokenResponse |
| from pypaimon.api.client import AlreadyExistsException, NoSuchResourceException |
| from pypaimon.api.rest_api import RESTApi |
| from pypaimon.common.options import Options |
| from pypaimon.common.options.config import CatalogOptions, OssOptions, PVFSOptions |
| from pypaimon.common.identifier import Identifier |
| from pypaimon.filesystem.jindo_file_system_handler import ( |
| JINDO_AVAILABLE, |
| JINDO_OSSFS_AVAILABLE, |
| create_jindo_oss_filesystem, |
| ) |
| from pypaimon.schema.schema import Schema |
| |
| logger = logging.getLogger(__name__) |
| |
| PROTOCOL_NAME = "pvfs" |
| |
| |
| class StorageType(Enum): |
| LOCAL = "file" |
| OSS = "oss" |
| |
| |
| class PVFSIdentifier(ABC): |
| catalog: str |
| endpoint: str |
| |
| def get_cache_key(self) -> str: |
| return "{}.{}".format(self.catalog, self.__remove_endpoint_schema(self.endpoint)) |
| |
| @staticmethod |
| def __remove_endpoint_schema(url): |
| if url.startswith('https://'): |
| return url[8:] |
| elif url.startswith('http://'): |
| return url[7:] |
| return url |
| |
| |
| @dataclass |
| class PVFSCatalogIdentifier(PVFSIdentifier): |
| catalog: str |
| endpoint: str |
| |
| |
| @dataclass |
| class PVFSDatabaseIdentifier(PVFSIdentifier): |
| catalog: str |
| endpoint: str |
| database: str |
| |
| |
| @dataclass |
| class PVFSTableIdentifier(PVFSIdentifier): |
| catalog: str |
| endpoint: str |
| database: str |
| table: str |
| sub_path: str = None |
| |
| def __hash__(self) -> int: |
| return hash((self.catalog, self.database, self.table)) |
| |
| def __eq__(self, __value: Any) -> bool: |
| if isinstance(__value, PVFSTableIdentifier): |
| return self.catalog == __value.catalog and self.database == __value.database and self.table == __value.table |
| return False |
| |
| def get_actual_path(self, storage_location: str): |
| if self.sub_path: |
| normalized_sub = posixpath.normpath(self.sub_path) |
| if normalized_sub == ".." or normalized_sub.startswith("../") or normalized_sub.startswith("/"): |
| raise ValueError( |
| "Path traversal detected: resolved path escapes table storage boundary" |
| ) |
| return '{}/{}'.format(storage_location.rstrip("/"), normalized_sub) |
| return storage_location |
| |
| def get_virtual_location(self): |
| return ('{}://{}'.format(PROTOCOL_NAME, self.catalog) + |
| '/{}/{}'.format(self.database, self.table)) |
| |
| def get_identifier(self): |
| return Identifier.create(self.database, self.table) |
| |
| def name(self): |
| return '{}.{}.{}'.format(self.catalog, self.database, self.table) |
| |
| |
| @dataclass |
| class PaimonRealStorage: |
| TOKEN_EXPIRATION_SAFE_TIME_MILLIS = 3_600_000 |
| |
| token: Dict[str, str] |
| expires_at_millis: Optional[int] |
| file_system: AbstractFileSystem |
| |
| def need_refresh(self) -> bool: |
| if self.expires_at_millis is not None: |
| return self.expires_at_millis - int(time.time() * 1000) < self.TOKEN_EXPIRATION_SAFE_TIME_MILLIS |
| return False |
| |
| |
| @dataclass |
| class TableStore: |
| path: str |
| created: datetime |
| modified: datetime |
| |
| |
| class PaimonVirtualFileSystem(fsspec.AbstractFileSystem): |
| options: Options |
| |
| protocol = PROTOCOL_NAME |
| |
| def __init__(self, options: Union[Options, Dict[str, str]] = None, **kwargs): |
| if isinstance(options, dict): |
| options = Options(options) |
| options.set(CatalogOptions.HTTP_USER_AGENT_HEADER, 'PythonPVFS') |
| self.options = options |
| self.warehouse = options.get(CatalogOptions.WAREHOUSE) |
| cache_expired_time = ( |
| PVFSOptions.DEFAULT_TABLE_CACHE_TTL |
| if options is None |
| else options.get( |
| PVFSOptions.TABLE_CACHE_TTL, PVFSOptions.DEFAULT_TABLE_CACHE_TTL |
| ) |
| ) |
| self._cache_enable = options.get(PVFSOptions.CACHE_ENABLED, True) |
| if self._cache_enable: |
| self._table_cache = TTLCache(maxsize=PVFSOptions.DEFAULT_CACHE_SIZE, ttl=cache_expired_time) |
| self._table_cache_lock = rwlock.RWLockFair() |
| self._rest_api_cache = LRUCache(PVFSOptions.DEFAULT_CACHE_SIZE) |
| self._fs_cache = LRUCache(maxsize=PVFSOptions.DEFAULT_CACHE_SIZE) |
| self._table_cache_lock = rwlock.RWLockFair() |
| self._rest_api_cache_lock = rwlock.RWLockFair() |
| self._fs_cache_lock = rwlock.RWLockFair() |
| super().__init__(**kwargs) |
| |
| @property |
| def fsid(self): |
| return PROTOCOL_NAME |
| |
| def sign(self, path, expiration=None, **kwargs): |
| """We do not support to create a signed URL representing the given path in pvfs.""" |
| raise Exception( |
| "Sign is not implemented for Paimon Virtual FileSystem." |
| ) |
| |
| def ls(self, path, detail=True, **kwargs): |
| pvfs_identifier = self._extract_pvfs_identifier(path) |
| rest_api = self.__rest_api(pvfs_identifier) |
| if isinstance(pvfs_identifier, PVFSCatalogIdentifier): |
| databases = rest_api.list_databases() |
| if detail: |
| return [ |
| self._create_dir_detail( |
| self._convert_database_virtual_path(pvfs_identifier.catalog, database) |
| ) |
| for database in databases |
| ] |
| return [ |
| self._convert_database_virtual_path(pvfs_identifier.catalog, database) |
| for database in databases |
| ] |
| elif isinstance(pvfs_identifier, PVFSDatabaseIdentifier): |
| tables = rest_api.list_tables(pvfs_identifier.database) |
| if detail: |
| return [ |
| self._create_dir_detail( |
| self._convert_table_virtual_path(pvfs_identifier.catalog, pvfs_identifier.database, table) |
| ) |
| for table in tables |
| ] |
| return [ |
| self._convert_table_virtual_path(pvfs_identifier.catalog, pvfs_identifier.database, table) |
| for table in tables |
| ] |
| elif isinstance(pvfs_identifier, PVFSTableIdentifier): |
| table_path = self._get_table_store(rest_api, pvfs_identifier).path |
| storage_type = self._get_storage_type(table_path) |
| storage_location = table_path |
| actual_path = pvfs_identifier.get_actual_path(storage_location) |
| virtual_location = pvfs_identifier.get_virtual_location() |
| fs = self._get_filesystem(pvfs_identifier, storage_type, storage_location) |
| entries = fs.ls(actual_path, detail=detail, **kwargs) |
| if detail: |
| virtual_entities = [ |
| self._convert_actual_info(entry, storage_type, storage_location, virtual_location) |
| for entry in entries |
| ] |
| return virtual_entities |
| else: |
| virtual_entry_paths = [ |
| self._convert_actual_path( |
| storage_type, entry_path, storage_location, virtual_location |
| ) |
| for entry_path in entries |
| ] |
| return virtual_entry_paths |
| |
| def info(self, path, **kwargs): |
| pvfs_identifier = self._extract_pvfs_identifier(path) |
| if isinstance(pvfs_identifier, PVFSCatalogIdentifier): |
| return self._create_dir_detail('{}://{}'.format(PROTOCOL_NAME, pvfs_identifier.catalog)) |
| elif isinstance(pvfs_identifier, PVFSDatabaseIdentifier): |
| return self._create_dir_detail( |
| self._convert_database_virtual_path(pvfs_identifier.catalog, pvfs_identifier.database) |
| ) |
| elif isinstance(pvfs_identifier, PVFSTableIdentifier): |
| rest_api = self.__rest_api(pvfs_identifier) |
| table_path = self._get_table_store(rest_api, pvfs_identifier).path |
| storage_type = self._get_storage_type(table_path) |
| storage_location = table_path |
| actual_path = pvfs_identifier.get_actual_path(storage_location) |
| virtual_location = pvfs_identifier.get_virtual_location() |
| fs = self._get_filesystem(pvfs_identifier, storage_type, storage_location) |
| entry = fs.info(actual_path) |
| return self._convert_actual_info(entry, storage_type, storage_location, virtual_location) |
| |
| def exists(self, path, **kwargs): |
| pvfs_identifier = self._extract_pvfs_identifier(path) |
| if isinstance(pvfs_identifier, PVFSCatalogIdentifier): |
| return True |
| elif isinstance(pvfs_identifier, PVFSDatabaseIdentifier): |
| rest_api = self.__rest_api(pvfs_identifier) |
| try: |
| rest_api.get_database(pvfs_identifier.database) |
| return True |
| except NoSuchResourceException: |
| return False |
| elif isinstance(pvfs_identifier, PVFSTableIdentifier): |
| rest_api = self.__rest_api(pvfs_identifier) |
| try: |
| table_path = self._get_table_store(rest_api, pvfs_identifier).path |
| if table_path is not None and pvfs_identifier.sub_path is None: |
| return True |
| storage_type = self._get_storage_type(table_path) |
| storage_location = table_path |
| actual_path = pvfs_identifier.get_actual_path(storage_location) |
| fs = self._get_filesystem(pvfs_identifier, storage_type, storage_location) |
| return fs.exists(actual_path) |
| except NoSuchResourceException: |
| return False |
| |
| def cp_file(self, path1, path2, **kwargs): |
| source = self._extract_pvfs_identifier(path1) |
| target = self._extract_pvfs_identifier(path2) |
| if ((isinstance(source, PVFSTableIdentifier) |
| and isinstance(target, PVFSTableIdentifier)) |
| and target.sub_path is not None |
| and source.sub_path is not None |
| and source == target): |
| rest_api = self.__rest_api(source) |
| table_path = self._get_table_store(rest_api, source).path |
| storage_type = self._get_storage_type(table_path) |
| storage_location = table_path |
| source_actual_path = source.get_actual_path(storage_location) |
| target_actual_path = target.get_actual_path(storage_location) |
| fs = self._get_filesystem(source, storage_type, storage_location) |
| fs.cp_file( |
| self._strip_storage_protocol(storage_type, source_actual_path), |
| self._strip_storage_protocol(storage_type, target_actual_path), |
| ) |
| return None |
| raise Exception("cp is not supported for path: {} to path: {}".format(path1, path2)) |
| |
| def mv(self, path1, path2, recursive=False, maxdepth=None, **kwargs): |
| source = self._extract_pvfs_identifier(path1) |
| target = self._extract_pvfs_identifier(path2) |
| if (isinstance(source, PVFSTableIdentifier) and |
| isinstance(target, PVFSTableIdentifier) and |
| target.catalog == source.catalog): |
| rest_api = self.__rest_api(source) |
| if target.sub_path is None and source.sub_path is None: |
| source_identifier = Identifier.create(source.database, source.table) |
| target_identifier = Identifier.create(target.database, target.table) |
| rest_api.rename_table(source_identifier, target_identifier) |
| return None |
| elif target.sub_path is not None and source.sub_path is not None and target == source: |
| table_path = self._get_table_store(rest_api, source).path |
| storage_type = self._get_storage_type(table_path) |
| storage_location = table_path |
| source_actual_path = source.get_actual_path(storage_location) |
| target_actual_path = target.get_actual_path(storage_location) |
| fs = self._get_filesystem(source, storage_type, storage_location) |
| if storage_type == StorageType.LOCAL: |
| fs.mv( |
| self._strip_storage_protocol(storage_type, source_actual_path), |
| self._strip_storage_protocol(storage_type, target_actual_path), |
| recursive=recursive, |
| maxdepth=maxdepth |
| ) |
| else: |
| fs.mv( |
| self._strip_storage_protocol(storage_type, source_actual_path), |
| self._strip_storage_protocol(storage_type, target_actual_path), |
| ) |
| return None |
| raise Exception("Mv is not supported for path: {} to path: {}".format(path1, path2)) |
| |
| def rm(self, path, recursive=False, maxdepth=None): |
| pvfs_identifier = self._extract_pvfs_identifier(path) |
| rest_api = self.__rest_api(pvfs_identifier) |
| if isinstance(pvfs_identifier, PVFSDatabaseIdentifier): |
| database_name = pvfs_identifier.database |
| if not recursive and len(rest_api.list_tables(database_name)) > 0: |
| raise Exception('Recursive is False but database is not empty') |
| rest_api.drop_database(database_name) |
| if self._table_cache: |
| for table in rest_api.list_tables(database_name): |
| table_pvfs_identifier = PVFSTableIdentifier( |
| pvfs_identifier.catalog, database_name, table |
| ).name() |
| self._table_cache.pop(table_pvfs_identifier) |
| return True |
| elif isinstance(pvfs_identifier, PVFSTableIdentifier): |
| table_identifier = pvfs_identifier.get_identifier() |
| table_path = self._get_table_store(rest_api, pvfs_identifier).path |
| if pvfs_identifier.sub_path is None: |
| rest_api.drop_table(table_identifier) |
| self._table_cache.pop(table_identifier.name()) |
| return True |
| storage_type = self._get_storage_type(table_path) |
| storage_location = table_path |
| actual_path = pvfs_identifier.get_actual_path(storage_location) |
| fs = self._get_filesystem(pvfs_identifier, storage_type, storage_location) |
| return fs.rm( |
| self._strip_storage_protocol(storage_type, actual_path), |
| recursive, |
| maxdepth, |
| ) |
| raise Exception( |
| "Rm is not supported for path: {}.".format(path) |
| ) |
| |
| def rm_file(self, path): |
| pvfs_identifier = self._extract_pvfs_identifier(path) |
| if isinstance(pvfs_identifier, PVFSTableIdentifier): |
| rest_api = self.__rest_api(pvfs_identifier) |
| table_path = self._get_table_store(rest_api, pvfs_identifier).path |
| if pvfs_identifier.sub_path is not None: |
| storage_type = self._get_storage_type(table_path) |
| storage_location = table_path |
| actual_path = pvfs_identifier.get_actual_path(storage_location) |
| fs = self._get_filesystem(pvfs_identifier, storage_type, storage_location) |
| return fs.rm_file( |
| self._strip_storage_protocol(storage_type, actual_path), |
| ) |
| raise Exception("Rm is not supported for path: {}.".format(path)) |
| |
| def rmdir(self, path): |
| files = self.ls(path) |
| if len(files) == 0: |
| pvfs_identifier = self._extract_pvfs_identifier(path) |
| rest_api = self.__rest_api(pvfs_identifier) |
| if isinstance(pvfs_identifier, PVFSDatabaseIdentifier): |
| database_name = pvfs_identifier.database |
| rest_api.drop_database(database_name) |
| return True |
| elif isinstance(pvfs_identifier, PVFSTableIdentifier): |
| table_identifier = pvfs_identifier.get_identifier() |
| table_path = self._get_table_store(rest_api, pvfs_identifier).path |
| if pvfs_identifier.sub_path is None: |
| rest_api.drop_table(table_identifier) |
| self._cache.pop(pvfs_identifier) |
| return True |
| storage_type = self._get_storage_type(table_path) |
| storage_location = table_path |
| actual_path = pvfs_identifier.get_actual_path(storage_location) |
| fs = self._get_filesystem(pvfs_identifier, storage_type, storage_location) |
| return fs.rmdir( |
| self._strip_storage_protocol(storage_type, actual_path) |
| ) |
| raise Exception("Rm dir is not supported for path: {}.".format(path)) |
| else: |
| raise Exception("Rm dir is not supported for path: {} as it is not empty.".format(path)) |
| |
| def open( |
| self, |
| path, |
| mode="rb", |
| block_size=None, |
| cache_options=None, |
| compression=None, |
| **kwargs |
| ): |
| pvfs_identifier = self._extract_pvfs_identifier(path) |
| if isinstance(pvfs_identifier, PVFSCatalogIdentifier): |
| raise Exception("open is not supported for path: {}".format(path)) |
| elif isinstance(pvfs_identifier, PVFSDatabaseIdentifier): |
| raise Exception( |
| "open is not supported for path: {}".format(path) |
| ) |
| elif isinstance(pvfs_identifier, PVFSTableIdentifier): |
| rest_api = self.__rest_api(pvfs_identifier) |
| table_path = self._get_table_store(rest_api, pvfs_identifier).path |
| if pvfs_identifier.sub_path is None: |
| raise Exception("open is not supported for path: {}".format(path)) |
| else: |
| storage_type = self._get_storage_type(table_path) |
| storage_location = table_path |
| actual_path = pvfs_identifier.get_actual_path(storage_location) |
| fs = self._get_filesystem(pvfs_identifier, storage_type, storage_location) |
| return fs.open( |
| self._strip_storage_protocol(storage_type, actual_path), |
| mode, |
| block_size, |
| cache_options, |
| compression, |
| **kwargs |
| ) |
| |
| def mkdir(self, path, create_parents=True, **kwargs): |
| pvfs_identifier = self._extract_pvfs_identifier(path) |
| rest_api = self.__rest_api(pvfs_identifier) |
| if isinstance(pvfs_identifier, PVFSCatalogIdentifier): |
| raise Exception("mkdir is not supported for path: {}".format(path)) |
| elif isinstance(pvfs_identifier, PVFSDatabaseIdentifier): |
| rest_api.create_database(pvfs_identifier.database, {}) |
| elif isinstance(pvfs_identifier, PVFSTableIdentifier): |
| table_path: str |
| if pvfs_identifier.sub_path is None: |
| if create_parents: |
| try: |
| rest_api.create_database(pvfs_identifier.database, {}) |
| except AlreadyExistsException: |
| pass |
| self._create_object_table(pvfs_identifier) |
| else: |
| if create_parents: |
| try: |
| rest_api.create_database(pvfs_identifier.database, {}) |
| except AlreadyExistsException: |
| pass |
| try: |
| table_path = self._get_table_store(rest_api, pvfs_identifier).path |
| except NoSuchResourceException: |
| try: |
| self._create_object_table(pvfs_identifier) |
| except AlreadyExistsException: |
| pass |
| finally: |
| table_path = self._get_table_store(rest_api, pvfs_identifier).path |
| storage_type = self._get_storage_type(table_path) |
| storage_location = table_path |
| actual_path = pvfs_identifier.get_actual_path(storage_location) |
| fs = self._get_filesystem(pvfs_identifier, storage_type, storage_location) |
| return fs.mkdir( |
| self._strip_storage_protocol(storage_type, actual_path), |
| create_parents, |
| **kwargs |
| ) |
| |
| def makedirs(self, path, exist_ok=True): |
| pvfs_identifier = self._extract_pvfs_identifier(path) |
| rest_api = self.__rest_api(pvfs_identifier) |
| if isinstance(pvfs_identifier, PVFSCatalogIdentifier): |
| raise Exception("makedirs is not supported for path: {}".format(path)) |
| elif isinstance(pvfs_identifier, PVFSDatabaseIdentifier): |
| try: |
| rest_api.create_database(pvfs_identifier.catalog, {}) |
| except AlreadyExistsException as e: |
| if exist_ok: |
| pass |
| raise e |
| elif isinstance(pvfs_identifier, PVFSTableIdentifier): |
| if pvfs_identifier.sub_path is None: |
| try: |
| self._create_object_table(pvfs_identifier) |
| except AlreadyExistsException as e: |
| if exist_ok: |
| pass |
| raise e |
| else: |
| try: |
| self._create_object_table(pvfs_identifier) |
| except AlreadyExistsException as e: |
| if exist_ok: |
| pass |
| else: |
| raise e |
| table_path = self._get_table_store(rest_api, pvfs_identifier).path |
| storage_type = self._get_storage_type(table_path) |
| storage_location = table_path |
| actual_path = pvfs_identifier.get_actual_path(storage_location) |
| fs = self._get_filesystem(pvfs_identifier, storage_type, storage_location) |
| return fs.makedirs( |
| self._strip_storage_protocol(storage_type, actual_path), |
| exist_ok |
| ) |
| |
| def created(self, path): |
| pvfs_identifier = self._extract_pvfs_identifier(path) |
| rest_api = self.__rest_api(pvfs_identifier) |
| if isinstance(pvfs_identifier, PVFSCatalogIdentifier): |
| raise Exception("created is not supported for path: {}".format(path)) |
| elif isinstance(pvfs_identifier, PVFSDatabaseIdentifier): |
| return self.__converse_ts_to_datatime(rest_api.get_database(pvfs_identifier.database).created_at) |
| elif isinstance(pvfs_identifier, PVFSTableIdentifier): |
| if pvfs_identifier.sub_path is None: |
| return self._get_table_store(rest_api, pvfs_identifier).created |
| else: |
| table = self._get_table_store(rest_api, pvfs_identifier) |
| storage_type = self._get_storage_type(table.path) |
| storage_location = table.path |
| actual_path = pvfs_identifier.get_actual_path(storage_location) |
| fs = self._get_filesystem(pvfs_identifier, storage_type, storage_location) |
| return fs.created( |
| self._strip_storage_protocol(storage_type, actual_path) |
| ) |
| |
| def modified(self, path): |
| pvfs_identifier = self._extract_pvfs_identifier(path) |
| rest_api = self.__rest_api(pvfs_identifier) |
| if isinstance(pvfs_identifier, PVFSCatalogIdentifier): |
| raise Exception("modified is not supported for path: {}".format(path)) |
| elif isinstance(pvfs_identifier, PVFSDatabaseIdentifier): |
| return self.__converse_ts_to_datatime(rest_api.get_database(pvfs_identifier.database).updated_at) |
| elif isinstance(pvfs_identifier, PVFSTableIdentifier): |
| table_identifier = pvfs_identifier.get_identifier() |
| if pvfs_identifier.sub_path is None: |
| return self._get_table_store(rest_api, pvfs_identifier).modified |
| else: |
| table = rest_api.get_table(table_identifier) |
| storage_type = self._get_storage_type(table.path) |
| storage_location = table.path |
| actual_path = pvfs_identifier.get_actual_path(storage_location) |
| fs = self._get_filesystem(pvfs_identifier, storage_type, storage_location) |
| return fs.modified( |
| self._strip_storage_protocol(storage_type, actual_path) |
| ) |
| |
| def cat_file(self, path, start=None, end=None, **kwargs): |
| pvfs_identifier = self._extract_pvfs_identifier(path) |
| if isinstance(pvfs_identifier, PVFSCatalogIdentifier): |
| raise Exception( |
| "cat file is not supported for path: {}".format(path) |
| ) |
| elif isinstance(pvfs_identifier, PVFSDatabaseIdentifier): |
| raise Exception( |
| "cat file is not supported for path: {}".format(path) |
| ) |
| elif isinstance(pvfs_identifier, PVFSTableIdentifier): |
| if pvfs_identifier.sub_path is None: |
| raise Exception( |
| "cat file is not supported for path: {}".format(path) |
| ) |
| else: |
| rest_api = self.__rest_api(pvfs_identifier) |
| table = self._get_table_store(rest_api, pvfs_identifier) |
| storage_type = self._get_storage_type(table.path) |
| storage_location = table.path |
| actual_path = pvfs_identifier.get_actual_path(storage_location) |
| fs = self._get_filesystem(pvfs_identifier, storage_type, storage_location) |
| return fs.cat_file( |
| self._strip_storage_protocol(storage_type, actual_path), |
| start, |
| end, |
| **kwargs, |
| ) |
| |
| def get_file(self, rpath, lpath, callback=None, outfile=None, **kwargs): |
| pvfs_identifier = self._extract_pvfs_identifier(rpath) |
| if isinstance(pvfs_identifier, PVFSCatalogIdentifier): |
| raise Exception( |
| "get file is not supported for path: {}".format(rpath) |
| ) |
| elif isinstance(pvfs_identifier, PVFSDatabaseIdentifier): |
| raise Exception( |
| "get file is not supported for path: {}".format(rpath) |
| ) |
| elif isinstance(pvfs_identifier, PVFSTableIdentifier): |
| rest_api = self.__rest_api(pvfs_identifier) |
| if pvfs_identifier.sub_path is None: |
| raise Exception( |
| "get file is not supported for path: {}".format(rpath) |
| ) |
| else: |
| table = self._get_table_store(rest_api, pvfs_identifier) |
| storage_type = self._get_storage_type(table.path) |
| storage_location = table.path |
| actual_path = pvfs_identifier.get_actual_path(storage_location) |
| fs = self._get_filesystem(pvfs_identifier, storage_type, storage_location) |
| return fs.get_file( |
| self._strip_storage_protocol(storage_type, actual_path), |
| lpath, |
| **kwargs |
| ) |
| |
| def _rm(self, path): |
| raise Exception( |
| "_rm is not implemented for Paimon Virtual FileSystem." |
| ) |
| |
| def _create_object_table(self, pvfs_identifier: PVFSTableIdentifier): |
| rest_api = self.__rest_api(pvfs_identifier) |
| schema = Schema(options={'type': 'object-table'}) |
| table_identifier = pvfs_identifier.get_identifier() |
| rest_api.create_table(table_identifier, schema) |
| |
| @staticmethod |
| def __converse_ts_to_datatime(ts: int): |
| return datetime.datetime.fromtimestamp(ts / 1000, tz=datetime.timezone.utc) |
| |
| def _strip_storage_protocol(self, storage_type: StorageType, path: str): |
| if storage_type == StorageType.LOCAL: |
| return path[len("{}:".format(StorageType.LOCAL.value)):] |
| |
| # The two OSS backends want opposite path forms. The legacy ossfs |
| # backend mishandles an oss://-prefixed path (older versions return an |
| # empty array from ls), so the scheme is stripped for it. The jindo |
| # backend (pyjindo) requires the oss:// scheme and crashes without it, |
| # so the path is passed through unchanged. |
| if storage_type == StorageType.OSS: |
| if path.startswith("{}://".format(StorageType.OSS.value)): |
| if self._use_jindo_oss_backend(): |
| return path |
| return path[len("{}://".format(StorageType.OSS.value)):] |
| return path |
| |
| raise Exception( |
| "Storage type:{} doesn't support now.".format(storage_type) |
| ) |
| |
| @staticmethod |
| def _convert_actual_info( |
| entry: Dict, |
| storage_type: StorageType, |
| storage_location: str, |
| virtual_location: str, |
| ): |
| path = PaimonVirtualFileSystem._convert_actual_path(storage_type, entry["name"], storage_location, |
| virtual_location) |
| |
| if "mtime" in entry: |
| # HDFS and GCS |
| return PaimonVirtualFileSystem._create_file_detail(path, entry["size"], entry["type"], entry["mtime"]) |
| elif "LastModified" in entry: |
| # S3 and OSS |
| return PaimonVirtualFileSystem._create_file_detail(path, entry["size"], entry["type"], |
| entry["LastModified"]) |
| # Unknown |
| return PaimonVirtualFileSystem._create_file_detail(path, entry["size"], entry["type"]) |
| |
| @staticmethod |
| def _create_dir_detail(path: str) -> Dict[str, Any]: |
| return PaimonVirtualFileSystem._create_file_detail(path, 0, 'directory', None) |
| |
| @staticmethod |
| def _create_file_detail(name: str, size: int, type: str, mtime: int = None) -> Dict[str, Any]: |
| return { |
| "name": name, |
| "size": size, |
| "type": type, |
| "mtime": mtime, |
| } |
| |
| @staticmethod |
| def _convert_database_virtual_path( |
| catalog_name: str, |
| database_name: str |
| ): |
| return '{}://{}/{}'.format(PROTOCOL_NAME, catalog_name, database_name) |
| |
| @staticmethod |
| def _convert_table_virtual_path( |
| catalog_name: str, |
| database_name: str, |
| table_name: str |
| ): |
| return '{}://{}/{}/{}'.format(PROTOCOL_NAME, catalog_name, database_name, table_name) |
| |
| @staticmethod |
| def _convert_actual_path( |
| storage_type: StorageType, |
| actual_path: str, |
| storage_location: str, |
| virtual_location: str, |
| ): |
| actual_path = PaimonVirtualFileSystem._get_path_without_schema(storage_type, actual_path) |
| storage_location = PaimonVirtualFileSystem._get_path_without_schema(storage_type, storage_location) |
| normalized_pvfs = virtual_location.rstrip('/') |
| sub_location = actual_path[len(storage_location):].lstrip("/") |
| if len(sub_location) == 0: |
| return normalized_pvfs |
| else: |
| return '{}/{}'.format(normalized_pvfs, sub_location) |
| |
| @staticmethod |
| def _get_path_without_schema(storage_type: StorageType, path: str) -> str: |
| if storage_type == StorageType.LOCAL and path.startswith(StorageType.LOCAL.value): |
| return path[len("{}://".format(StorageType.LOCAL.value)):] |
| elif storage_type == StorageType.OSS and path.startswith(StorageType.OSS.value): |
| return path[len("{}://".format(StorageType.OSS.value)):] |
| return path |
| |
| def _extract_pvfs_identifier(self, path: str) -> Optional['PVFSIdentifier']: |
| if not isinstance(path, str): |
| raise Exception("path is not a string") |
| path_without_protocol = path |
| if path.startswith('{}://'.format(PROTOCOL_NAME)): |
| path_without_protocol = path[7:] |
| |
| if not path_without_protocol: |
| return None |
| |
| components = [component for component in path_without_protocol.rstrip('/').split('/') if component] |
| for component in components: |
| if component == '..' or '\x00' in component: |
| raise ValueError( |
| "Invalid path: path traversal components are not allowed" |
| ) |
| catalog: str = None |
| endpoint: str = self.options.get(CatalogOptions.URI) |
| if len(components) > 0: |
| if '.' in components[0]: |
| (catalog, endpoint) = components[0].split('.', 1) |
| else: |
| catalog = components[0] |
| if len(components) == 0: |
| return None |
| elif len(components) == 1: |
| return (PVFSCatalogIdentifier(endpoint=endpoint, catalog=catalog)) |
| elif len(components) == 2: |
| return PVFSDatabaseIdentifier(endpoint=endpoint, catalog=catalog, database=components[1]) |
| elif len(components) == 3: |
| return PVFSTableIdentifier(endpoint=endpoint, catalog=catalog, database=components[1], table=components[2]) |
| elif len(components) > 3: |
| sub_path = '/'.join(components[3:]) |
| return PVFSTableIdentifier( |
| endpoint=endpoint, |
| catalog=catalog, database=components[1], |
| table=components[2], sub_path=sub_path |
| ) |
| return None |
| |
| def _get_table_store(self, rest_api: RESTApi, pvfs_identifier: PVFSTableIdentifier) -> Optional['TableStore']: |
| if not self._cache_enable: |
| table = rest_api.get_table(Identifier.create(pvfs_identifier.database, pvfs_identifier.table)) |
| return self._create_table_store(table) |
| read_lock = self._table_cache_lock.gen_rlock() |
| try: |
| read_lock.acquire() |
| cache_value: Tuple[str, TableStore] = self._table_cache.get( |
| pvfs_identifier.name() |
| ) |
| if cache_value is not None: |
| return cache_value |
| finally: |
| read_lock.release() |
| write_lock = self._table_cache_lock.gen_wlock() |
| try: |
| write_lock.acquire() |
| table = rest_api.get_table(Identifier.create(pvfs_identifier.database, pvfs_identifier.table)) |
| if table is not None: |
| table_store = self._create_table_store(table) |
| self._table_cache[pvfs_identifier.name()] = table_store |
| return table_store |
| else: |
| return None |
| finally: |
| write_lock.release() |
| |
| def _create_table_store(self, table: GetTableResponse): |
| created = self.__converse_ts_to_datatime(table.created_at) |
| modified = self.__converse_ts_to_datatime(table.updated_at) |
| return TableStore(path=table.path, created=created, modified=modified) |
| |
| def __rest_api(self, pvfs_identifier: PVFSIdentifier): |
| read_lock = self._rest_api_cache_lock.gen_rlock() |
| catalog = pvfs_identifier.catalog |
| if pvfs_identifier.endpoint is None or catalog is None: |
| raise ValueError("Endpoint or catalog is not set.") |
| key = pvfs_identifier.get_cache_key() |
| try: |
| read_lock.acquire() |
| rest_api = self._rest_api_cache.get(key) |
| if rest_api is not None: |
| return rest_api |
| finally: |
| read_lock.release() |
| |
| write_lock = self._rest_api_cache_lock.gen_wlock() |
| try: |
| write_lock.acquire() |
| rest_api = self._rest_api_cache.get(key) |
| if rest_api is None: |
| options = self.options.copy() |
| options.data.update( |
| {CatalogOptions.WAREHOUSE.key(): catalog, CatalogOptions.URI.key(): pvfs_identifier.endpoint}) |
| rest_api = RESTApi(options) |
| self._rest_api_cache[catalog] = rest_api |
| return rest_api |
| finally: |
| write_lock.release() |
| |
| def _get_filesystem(self, pvfs_table_identifier: PVFSTableIdentifier, storage_type: StorageType, |
| storage_location: str) -> 'FileSystem': |
| read_lock = self._fs_cache_lock.gen_rlock() |
| try: |
| read_lock.acquire() |
| cache_value: Tuple[StorageType, AbstractFileSystem] = self._fs_cache.get( |
| storage_type |
| ) |
| if cache_value is not None: |
| return cache_value |
| finally: |
| read_lock.release() |
| |
| # The previous filesystem (if any) is closed after the write lock is |
| # released. For the jindo backend, fs.close() tears down a native |
| # JindoSDK connection, which can block on flushing buffers and |
| # terminating worker threads; holding the write lock through that |
| # would stall every other OSS filesystem rebuild. |
| stale_fs = None |
| write_lock = self._table_cache_lock.gen_wlock() |
| try: |
| write_lock.acquire() |
| cache_value: PaimonRealStorage = self._fs_cache.get(pvfs_table_identifier) |
| if cache_value is not None and cache_value.need_refresh() is False: |
| return cache_value.file_system |
| if storage_type == StorageType.LOCAL: |
| fs = LocalFileSystem() |
| elif storage_type == StorageType.OSS: |
| rest_api = self.__rest_api(pvfs_table_identifier) |
| table_identifier = Identifier.create( |
| pvfs_table_identifier.database, pvfs_table_identifier.table) |
| load_token_response: GetTableTokenResponse = rest_api.load_table_token(table_identifier) |
| merged_token = self._merge_token_with_catalog_options(load_token_response.token) |
| fs = self._get_oss_filesystem(Options(merged_token), storage_location) |
| paimon_real_storage = PaimonRealStorage( |
| token=load_token_response.token, |
| expires_at_millis=load_token_response.expires_at_millis, |
| file_system=fs |
| ) |
| self._fs_cache[pvfs_table_identifier] = paimon_real_storage |
| if cache_value is not None: |
| stale_fs = cache_value.file_system |
| else: |
| raise Exception( |
| "Storage type: `{}` doesn't support now.".format(storage_type) |
| ) |
| return fs |
| finally: |
| write_lock.release() |
| if stale_fs is not None: |
| PaimonVirtualFileSystem._close_filesystem_quietly(stale_fs) |
| |
| def _merge_token_with_catalog_options(self, token: dict) -> dict: |
| """Merge token with catalog options, DLF OSS endpoint should override the standard OSS endpoint.""" |
| merged_token = dict(token) |
| dlf_oss_endpoint = self.options.get(CatalogOptions.DLF_OSS_ENDPOINT) |
| if dlf_oss_endpoint and dlf_oss_endpoint.strip(): |
| merged_token[OssOptions.OSS_ENDPOINT.key()] = dlf_oss_endpoint |
| return merged_token |
| |
| @staticmethod |
| def _get_storage_type(path: str): |
| if path.startswith("{}:/".format(StorageType.LOCAL.value)): |
| return StorageType.LOCAL |
| elif path.startswith("{}://".format(StorageType.OSS.value)): |
| return StorageType.OSS |
| raise Exception( |
| "Storage type doesn't support now. Path:{}".format(path) |
| ) |
| |
| def _get_oss_filesystem(self, options: Options, storage_location: str) -> AbstractFileSystem: |
| """Build the fsspec filesystem backing OSS reads/writes. |
| |
| Honors ``fs.oss.impl`` (the same option PyArrowFileIO uses): ``jindo`` |
| backs OSS with the native JindoSDK (PutObject / multipart upload), |
| ``legacy`` backs it with ``ossfs``. ``jindo`` is the default; when |
| pyjindosdk is not installed it falls back to ``ossfs`` -- consistent |
| with PyArrowFileIO. |
| |
| ``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) |
| if oss_impl not in ("jindo", "legacy"): |
| raise Exception( |
| "Unsupported fs.oss.impl value: '{}'. " |
| "Supported values are 'jindo' and 'legacy'.".format(oss_impl) |
| ) |
| if self._use_jindo_oss_backend(): |
| bucket = PaimonVirtualFileSystem._extract_oss_bucket(storage_location) |
| 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. " |
| "Falling back to the ossfs (OSS AppendObject) implementation. " |
| "Install pyjindosdk for native multipart upload: pip install pyjindosdk" |
| ) |
| return PaimonVirtualFileSystem._get_ossfs_filesystem(options) |
| |
| def _use_jindo_oss_backend(self) -> bool: |
| """Whether OSS access uses the native jindo backend rather than ossfs. |
| |
| Decided per filesystem instance from ``fs.oss.impl`` plus the |
| availability of the two pyjindosdk surfaces this path requires |
| (``pyjindo.fs``/``pyjindo.util`` for credential config and |
| ``pyjindo.ossfs`` for the fsspec backend). This is the single |
| condition both ``_get_oss_filesystem`` and ``_strip_storage_protocol`` |
| rely on, so the backend choice and the path form handed to it cannot |
| drift apart. |
| """ |
| return ( |
| self.options.get(OssOptions.OSS_IMPL) == "jindo" |
| and JINDO_AVAILABLE |
| and JINDO_OSSFS_AVAILABLE |
| ) |
| |
| @staticmethod |
| def _close_filesystem_quietly(fs) -> None: |
| """Best-effort release of a filesystem that is being evicted from cache.""" |
| close = getattr(fs, "close", None) |
| if not callable(close): |
| return |
| try: |
| close() |
| except Exception: |
| logger.warning( |
| "ignoring error while closing stale OSS filesystem", |
| exc_info=True, |
| ) |
| |
| @staticmethod |
| def _extract_oss_bucket(oss_path: str) -> str: |
| scheme = "{}://".format(StorageType.OSS.value) |
| if not oss_path.startswith(scheme): |
| raise Exception("Invalid OSS path: {}".format(oss_path)) |
| bucket = oss_path[len(scheme):].split("/", 1)[0] |
| if not bucket: |
| raise Exception("Invalid OSS path without bucket: {}".format(oss_path)) |
| return bucket |
| |
| @staticmethod |
| def _get_ossfs_filesystem(options: Options) -> AbstractFileSystem: |
| access_key_id = options.get(OssOptions.OSS_ACCESS_KEY_ID) |
| if access_key_id is None: |
| raise Exception( |
| "OSS access key id is not found in the options." |
| ) |
| |
| access_key_secret = options.get( |
| OssOptions.OSS_ACCESS_KEY_SECRET |
| ) |
| if access_key_secret is None: |
| raise Exception( |
| "OSS access key secret is not found in the options." |
| ) |
| oss_endpoint_url = options.get(OssOptions.OSS_ENDPOINT) |
| if oss_endpoint_url is None: |
| raise Exception( |
| "OSS endpoint url is not found in the options." |
| ) |
| token = options.get(OssOptions.OSS_SECURITY_TOKEN) |
| return importlib.import_module("ossfs").OSSFileSystem( |
| key=access_key_id, |
| secret=access_key_secret, |
| token=token, |
| endpoint=oss_endpoint_url |
| ) |
