Add base implementation framework (#3)

Author: yuzelin

.github/workflows/paimon-python-checks.yml

@@ -23,7 +23,7 @@ on:
   pull_request:
     paths-ignore:
       - 'dev/**'
-      - 'java-based-implementation/paimon-python-java-bridge/**'
+      - 'java_based_implementation/paimon-python-java-bridge/**'
       - '**/*.md'
 
 concurrency:

README.md

@@ -31,7 +31,7 @@ We can use `py4j` to leverage Java code to read Paimon data. This section descri
 ### Build paimon-python-java-bridge
 
 ```bash
-cd java-based-implementation/paimon-python-java-bridge/
+cd java_based_implementation/paimon-python-java-bridge/
 mvn clean install -DskipTests
 ```
 The built target is java-based-implementation/paimon-python-java-bridge/target/paimon-python-java-bridge-<version>.jar

dev/dev-requirements.txt

@@ -19,4 +19,6 @@
 pip>=20.3
 setuptools>=18.0
 wheel
+py4j==0.10.9.7
+pyarrow>=5.0.0
 pytest~=7.0

java_based_implementation/__init__.py

@@ -0,0 +1,17 @@
+################################################################################
+#  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.
+################################################################################

java_based_implementation/api_impl.py

@@ -0,0 +1,112 @@
+################################################################################
+#  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 java_based_implementation.java_gateway import get_gateway
+from java_based_implementation.util.java_utils import to_j_catalog_context
+from paimon_python_api import catalog, read_builder, table_scan, split, table_read
+from paimon_python_api import table
+from pyarrow import RecordBatchReader
+from typing import List
+from typing_extensions import Self
+
+
+class Catalog(catalog.Catalog):
+
+    def __init__(self, j_catalog):
+        self._j_catalog = j_catalog
+
+    @staticmethod
+    def create(catalog_context: dict) -> 'Catalog':
+        j_catalog_context = to_j_catalog_context(catalog_context)
+        gateway = get_gateway()
+        j_catalog = gateway.jvm.CatalogFactory.createCatalog(j_catalog_context)
+        return Catalog(j_catalog)
+
+    def get_table(self, identifier: tuple) -> 'Table':
+        gateway = get_gateway()
+        j_identifier = gateway.jvm.Identifier.fromString(identifier)
+        j_table = self._j_catalog.getTable(j_identifier)
+        return Table(j_table)
+
+
+class Table(table.Table):
+
+    def __init__(self, j_table):
+        self._j_table = j_table
+
+    def new_read_builder(self) -> 'ReadBuilder':
+        j_read_builder = self._j_table.newReadBuilder()
+        return ReadBuilder(j_read_builder)
+
+
+class ReadBuilder(read_builder.ReadBuilder):
+
+    def __init__(self, j_read_builder):
+        self._j_read_builder = j_read_builder
+
+    def with_projection(self, projection: List[List[int]]) -> Self:
+        self._j_read_builder.withProjection(projection)
+        return self
+
+    def with_limit(self, limit: int) -> Self:
+        self._j_read_builder.withLimit(limit)
+        return self
+
+    def new_scan(self) -> 'TableScan':
+        j_table_scan = self._j_read_builder.newScan()
+        return TableScan(j_table_scan)
+
+    def new_read(self) -> 'TableRead':
+        # TODO
+        pass
+
+
+class TableScan(table_scan.TableScan):
+
+    def __init__(self, j_table_scan):
+        self._j_table_scan = j_table_scan
+
+    def plan(self) -> 'Plan':
+        j_plan = self._j_table_scan.plan()
+        j_splits = j_plan.splits()
+        return Plan(j_splits)
+
+
+class Plan(table_scan.Plan):
+
+    def __init__(self, j_splits):
+        self._j_splits = j_splits
+
+    def splits(self) -> List['Split']:
+        return list(map(lambda s: Split(s), self._j_splits))
+
+
+class Split(split.Split):
+
+    def __init__(self, j_split):
+        self._j_split = j_split
+
+    def to_j_split(self):
+        return self._j_split
+
+
+class TableRead(table_read.TableRead):
+
+    def create_reader(self, split: Split) -> RecordBatchReader:
+        # TODO
+        pass

java_based_implementation/gateway_server.py

@@ -0,0 +1,73 @@
+################################################################################
+#  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 os
+import platform
+import signal
+
+from subprocess import Popen, PIPE
+from java_based_implementation.util.constants import (PYPAIMON_JVM_ARGS, PYPAIMON_JAVA_CLASSPATH,
+                                                      PYPAIMON_MAIN_ARGS, PYPAIMON_MAIN_CLASS)
+
+
+def on_windows():
+    return platform.system() == "Windows"
+
+
+def find_java_executable():
+    java_executable = "java.exe" if on_windows() else "java"
+    java_home = None
+
+    if java_home is None and "JAVA_HOME" in os.environ:
+        java_home = os.environ["JAVA_HOME"]
+
+    if java_home is not None:
+        java_executable = os.path.join(java_home, "bin", java_executable)
+
+    return java_executable
+
+
+def launch_gateway_server_process(env):
+    java_executable = find_java_executable()
+    # TODO construct Java module log settings
+    log_settings = []
+    jvm_args = env.get(PYPAIMON_JVM_ARGS, '').split()
+    classpath = env.get(PYPAIMON_JAVA_CLASSPATH)
+    main_args = env.get(PYPAIMON_MAIN_ARGS, '').split()
+    command = [
+        java_executable,
+        *jvm_args,
+        # default jvm args
+        "-XX:+IgnoreUnrecognizedVMOptions",
+        "--add-opens=jdk.proxy2/jdk.proxy2=ALL-UNNAMED",
+        *log_settings,
+        "-cp",
+        classpath,
+        "-c",
+        PYPAIMON_MAIN_CLASS,
+        *main_args
+    ]
+
+    if not on_windows():
+        def preexec_func():
+            # ignore ctrl-c / SIGINT
+            signal.signal(signal.SIGINT, signal.SIG_IGN)
+
+        preexec_fn = preexec_func
+    return Popen(list(filter(lambda c: len(c) != 0, command)),
+                 stdin=PIPE, stderr=PIPE, preexec_fn=preexec_fn, env=env)

java_based_implementation/java_gateway.py

@@ -0,0 +1,119 @@
+################################################################################
+#  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 os
+import shutil
+import struct
+import tempfile
+import time
+from logging import WARN
+from py4j.java_gateway import (java_import, logger, JavaGateway, GatewayParameters,
+                               CallbackServerParameters)
+from threading import RLock
+
+from java_based_implementation.gateway_server import launch_gateway_server_process
+from java_based_implementation.util.constants import PYPAIMON_CONN_INFO_PATH
+from java_based_implementation.util.exceptions import install_py4j_hooks
+
+_gateway = None
+_lock = RLock()
+
+
+def get_gateway():
+    # type: () -> JavaGateway
+    global _gateway
+    global _lock
+    with _lock:
+        if _gateway is None:
+            # Set the level to WARN to mute the noisy INFO level logs
+            logger.level = WARN
+            _gateway = launch_gateway()
+
+            callback_server = _gateway.get_callback_server()
+            callback_server_listening_address = callback_server.get_listening_address()
+            callback_server_listening_port = callback_server.get_listening_port()
+            _gateway.jvm.org.apache.paimon.python.PythonEnvUtils.resetCallbackClient(
+                _gateway.java_gateway_server,
+                callback_server_listening_address,
+                callback_server_listening_port)
+            # import the paimon view
+            import_paimon_view(_gateway)
+            # TODO add exception handler for better exception stacktrace
+            install_py4j_hooks()
+            _gateway.entry_point.put("Watchdog", Watchdog())
+    return _gateway
+
+
+def launch_gateway():
+    # type: () -> JavaGateway
+    """
+    launch jvm gateway
+    """
+
+    # Create a temporary directory where the gateway server should write the connection information.
+    conn_info_dir = tempfile.mkdtemp()
+    try:
+        fd, conn_info_file = tempfile.mkstemp(dir=conn_info_dir)
+        os.close(fd)
+        os.unlink(conn_info_file)
+
+        env = dict(os.environ)
+        env[PYPAIMON_CONN_INFO_PATH] = conn_info_file
+
+        p = launch_gateway_server_process(env)
+
+        while not p.poll() and not os.path.isfile(conn_info_file):
+            time.sleep(0.1)
+
+        if not os.path.isfile(conn_info_file):
+            stderr_info = p.stderr.read().decode('utf-8')
+            raise RuntimeError(
+                "Java gateway process exited before sending its port number.\nStderr:\n"
+                + stderr_info
+            )
+
+        with open(conn_info_file, "rb") as info:
+            gateway_port = struct.unpack("!I", info.read(4))[0]
+    finally:
+        shutil.rmtree(conn_info_dir)
+
+    # Connect to the gateway
+    gateway = JavaGateway(
+        gateway_parameters=GatewayParameters(port=gateway_port, auto_convert=True),
+        callback_server_parameters=CallbackServerParameters(
+            port=0, daemonize=True, daemonize_connections=True))
+
+    return gateway
+
+
+# TODO: import more
+def import_paimon_view(gateway):
+    java_import(gateway.jvm, "org.apache.paimon.table.*")
+
+
+class Watchdog(object):
+    """
+    Used to provide to Java side to check whether its parent process is alive.
+    """
+
+    def ping(self):
+        time.sleep(10)
+        return True
+
+    class Java:
+        implements = ["org.apache.paimon.python.PythonGatewayServer$Watchdog"]