#49 Implement Python native write with PyArrow

Author: chenghuichen

pypaimon/__init__.py

@@ -16,36 +16,38 @@
 # limitations under the License.
 #################################################################################
 
-from .api import Schema
-from .py4j import Catalog
-from .py4j import CommitMessage
-from .py4j import Predicate
-from .py4j import PredicateBuilder
-from .py4j import ReadBuilder
-from .py4j import RowType
-from .py4j import Split
-from .py4j import Table
-from .py4j import BatchTableCommit
-from .py4j import TableRead
-from .py4j import TableScan
-from .py4j import Plan
-from .py4j import BatchTableWrite
-from .py4j import BatchWriteBuilder
-
-__all__ = [
-    'Schema',
-    'Catalog',
-    'CommitMessage',
-    'Predicate',
-    'PredicateBuilder',
-    'ReadBuilder',
-    'RowType',
-    'Split',
-    'Table',
-    'BatchTableCommit',
-    'TableRead',
-    'TableScan',
-    'Plan',
-    'BatchTableWrite',
-    'BatchWriteBuilder'
-]
+# from .api import Schema
+# from .api import Database
+# from .py4j import Catalog
+# from .py4j import CommitMessage
+# from .py4j import Predicate
+# from .py4j import PredicateBuilder
+# from .py4j import ReadBuilder
+# from .py4j import RowType
+# from .py4j import Split
+# from .py4j import Table
+# from .py4j import BatchTableCommit
+# from .py4j import TableRead
+# from .py4j import TableScan
+# from .py4j import Plan
+# from .py4j import BatchTableWrite
+# from .py4j import BatchWriteBuilder
+#
+# __all__ = [
+#     'Schema',
+#     'Database',
+#     'Catalog',
+#     'CommitMessage',
+#     'Predicate',
+#     'PredicateBuilder',
+#     'ReadBuilder',
+#     'RowType',
+#     'Split',
+#     'Table',
+#     'BatchTableCommit',
+#     'TableRead',
+#     'TableScan',
+#     'Plan',
+#     'BatchTableWrite',
+#     'BatchWriteBuilder'
+# ]

pypaimon/api/__init__.py

@@ -25,7 +25,9 @@
 from .table_commit import BatchTableCommit
 from .table_write import BatchTableWrite
 from .write_builder import BatchWriteBuilder
-from .table import Table, Schema
+from .schema import Schema
+from .table import Table
+from .database import Database
 from .catalog import Catalog
 
 __all__ = [

pypaimon/api/catalog.py

@@ -18,7 +18,7 @@
 
 from abc import ABC, abstractmethod
 from typing import Optional
-from pypaimon.api import Table, Schema
+from pypaimon.api import Table, Schema, Database
 
 
 class Catalog(ABC):
@@ -27,10 +27,9 @@ class Catalog(ABC):
     metadata such as database/table from a paimon catalog.
     """
 
-    @staticmethod
     @abstractmethod
-    def create(catalog_options: dict) -> 'Catalog':
-        """Create catalog from configuration."""
+    def get_database(self, name: str) -> 'Database':
+        """Get paimon database identified by the given name."""
 
     @abstractmethod
     def get_table(self, identifier: str) -> Table:

pypaimon/api/catalog_factory.py

@@ -0,0 +1,18 @@
+from pypaimon.api.catalog import Catalog
+
+
+class CatalogFactory:
+
+    @staticmethod
+    def create(catalog_options: dict) -> Catalog:
+        from pypaimon.pynative.catalog.catalog_option import CatalogOptions
+        from pypaimon.pynative.catalog.abstract_catalog import AbstractCatalog
+        from pypaimon.pynative.catalog.filesystem_catalog import FileSystemCatalog  # noqa: F401
+        from pypaimon.pynative.catalog.hive_catalog import HiveCatalog  # noqa: F401
+
+        identifier = catalog_options.get(CatalogOptions.METASTORE, "filesystem")
+        subclasses = AbstractCatalog.__subclasses__()
+        for subclass in subclasses:
+            if subclass.identifier() == identifier:
+                return subclass(catalog_options)
+        raise ValueError(f"Unknown catalog identifier: {identifier}")

pypaimon/api/commit_message.py

@@ -16,8 +16,12 @@
 # limitations under the License.
 #################################################################################
 
-from abc import ABC
+from abc import ABC, abstractmethod
 
 
 class CommitMessage(ABC):
     """Commit message collected from writer."""
+
+    @abstractmethod
+    def is_empty(self):
+        """"""

pypaimon/api/database.py

@@ -0,0 +1,28 @@
+################################################################################
+#  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.
+#################################################################################
+
+from typing import Optional
+
+
+class Database:
+    """Structure of a Database."""
+
+    def __init__(self, name: str, properties: dict, comment: Optional[str] = None):
+        self.name = name
+        self.properties = properties
+        self.comment = comment

pypaimon/api/schema.py

@@ -0,0 +1,37 @@
+################################################################################
+#  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 pyarrow as pa
+
+from typing import Optional, List
+
+
+class Schema:
+    """Schema of a table."""
+
+    def __init__(self,
+                 pa_schema: pa.Schema,
+                 partition_keys: Optional[List[str]] = None,
+                 primary_keys: Optional[List[str]] = None,
+                 options: Optional[dict] = None,
+                 comment: Optional[str] = None):
+        self.pa_schema = pa_schema
+        self.partition_keys = partition_keys
+        self.primary_keys = primary_keys
+        self.options = options
+        self.comment = comment

pypaimon/api/table.py

@@ -16,11 +16,8 @@
 # limitations under the License.
 #################################################################################
 
-import pyarrow as pa
-
 from abc import ABC, abstractmethod
 from pypaimon.api import ReadBuilder, BatchWriteBuilder
-from typing import Optional, List
 
 
 class Table(ABC):
@@ -33,19 +30,3 @@ def new_read_builder(self) -> ReadBuilder:
     @abstractmethod
     def new_batch_write_builder(self) -> BatchWriteBuilder:
         """Returns a builder for building batch table write and table commit."""
-
-
-class Schema:
-    """Schema of a table."""
-
-    def __init__(self,
-                 pa_schema: pa.Schema,
-                 partition_keys: Optional[List[str]] = None,
-                 primary_keys: Optional[List[str]] = None,
-                 options: Optional[dict] = None,
-                 comment: Optional[str] = None):
-        self.pa_schema = pa_schema
-        self.partition_keys = partition_keys
-        self.primary_keys = primary_keys
-        self.options = options
-        self.comment = comment

pypaimon/py4j/java_implementation.py

@@ -33,7 +33,7 @@
 
 from pypaimon.pynative.common.exception import PyNativeNotImplementedError
 from pypaimon.pynative.common.predicate import PyNativePredicate
-from pypaimon.pynative.common.row.internal_row import InternalRow
+from pypaimon.pynative.reader.row.internal_row import InternalRow
 from pypaimon.pynative.util.reader_converter import ReaderConverter
 
 if TYPE_CHECKING:
@@ -43,6 +43,10 @@
 
 class Catalog(catalog.Catalog):
 
+    @staticmethod
+    def identifier() -> str:
+        pass # TODO
+
     def __init__(self, j_catalog, catalog_options: dict):
         self._j_catalog = j_catalog
         self._catalog_options = catalog_options