[python] Introduce Python SQL to PyPaimon

Author: JingsongLi

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

@@ -71,6 +71,8 @@ jobs:
             build-essential \
             git \
             curl \
+            pkg-config \
+            libssl-dev \
             && apt-get clean \
             && rm -rf /var/lib/apt/lists/*
 
@@ -139,12 +141,22 @@ jobs:
         if: matrix.python-version != '3.6.15'
         shell: bash
         run: |
-          pip install maturin
+          pip install maturin[patchelf]
           git clone -b support_directory https://github.com/JingsongLi/tantivy-py.git /tmp/tantivy-py
           cd /tmp/tantivy-py
           maturin build --release
           pip install target/wheels/tantivy-*.whl
 
+      - name: Build and install pypaimon-rust from source
+        if: matrix.python-version != '3.6.15'
+        shell: bash
+        run: |
+          git clone https://github.com/apache/paimon-rust.git /tmp/paimon-rust
+          cd /tmp/paimon-rust/bindings/python
+          maturin build --release -o dist
+          pip install dist/pypaimon_rust-*.whl
+          pip install 'datafusion>=52'
+
       - name: Run lint-python.sh
         shell: bash
         run: |

docs/content/pypaimon/cli.md

@@ -621,3 +621,105 @@ default
 mydb
 analytics
 ```
+
+## SQL Command
+
+Execute SQL queries on Paimon tables directly from the command line. This feature is powered by pypaimon-rust and DataFusion.
+
+**Prerequisites:**
+
+```shell
+pip install pypaimon[sql]
+```
+
+### One-Shot Query
+
+Execute a single SQL query and display the result:
+
+```shell
+paimon sql "SELECT * FROM users LIMIT 10"
+```
+
+Output:
+```
+ id    name  age      city
+  1   Alice   25   Beijing
+  2     Bob   30  Shanghai
+  3 Charlie   35 Guangzhou
+```
+
+**Options:**
+
+- `--format, -f`: Output format: `table` (default) or `json`
+
+**Examples:**
+
+```shell
+# Direct table name (uses default catalog and database)
+paimon sql "SELECT * FROM users"
+
+# Two-part: database.table
+paimon sql "SELECT * FROM mydb.users"
+
+# Query with filter and aggregation
+paimon sql "SELECT city, COUNT(*) AS cnt FROM users GROUP BY city ORDER BY cnt DESC"
+
+# Output as JSON
+paimon sql "SELECT * FROM users LIMIT 5" --format json
+```
+
+### Interactive REPL
+
+Start an interactive SQL session by running `paimon sql` without a query argument. The REPL supports arrow keys for line editing, and command history is persisted across sessions in `~/.paimon_history`.
+
+```shell
+paimon sql
+```
+
+Output:
+```
+    ____        _
+   / __ \____ _(_)___ ___  ____  ____
+  / /_/ / __ `/ / __ `__ \/ __ \/ __ \
+ / ____/ /_/ / / / / / / / /_/ / / / /
+/_/    \__,_/_/_/ /_/ /_/\____/_/ /_/
+
+  Powered by pypaimon-rust + DataFusion
+  Type 'help' for usage, 'exit' to quit.
+
+paimon> SHOW DATABASES;
+default
+mydb
+
+paimon> USE mydb;
+Using database 'mydb'.
+
+paimon> SHOW TABLES;
+orders
+users
+
+paimon> SELECT count(*) AS cnt
+     > FROM users
+     > WHERE age > 18;
+ cnt
+  42
+(1 row in 0.05s)
+
+paimon> exit
+Bye!
+```
+
+SQL statements end with `;` and can span multiple lines. The continuation prompt `     >` indicates that more input is expected.
+
+**REPL Commands:**
+
+| Command | Description |
+|---|---|
+| `USE <database>;` | Switch the default database |
+| `SHOW DATABASES;` | List all databases |
+| `SHOW TABLES;` | List tables in the current database |
+| `SELECT ...;` | Execute a SQL query |
+| `help` | Show usage information |
+| `exit` / `quit` | Exit the REPL |
+
+For more details on SQL syntax and the Python API, see [SQL Query]({{< ref "pypaimon/sql" >}}).

docs/content/pypaimon/sql.md

@@ -0,0 +1,156 @@
+---
+title: "SQL Query"
+weight: 8
+type: docs
+aliases:
+  - /pypaimon/sql.html
+---
+
+<!--
+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.
+-->
+
+# SQL Query
+
+PyPaimon supports executing SQL queries on Paimon tables, powered by [pypaimon-rust](https://github.com/apache/paimon-rust/tree/main/bindings/python) and [DataFusion](https://datafusion.apache.org/python/).
+
+## Installation
+
+SQL query support requires additional dependencies. Install them with:
+
+```shell
+pip install pypaimon[sql]
+```
+
+This will install `pypaimon-rust` and `datafusion`.
+
+## Usage
+
+SQL query capability is available directly on the `Catalog` object via the `sql()` and `sql_to_pandas()` methods.
+
+### Basic Query
+
+```python
+from pypaimon import CatalogFactory
+
+catalog = CatalogFactory.create({"warehouse": "/path/to/warehouse"})
+
+# Execute SQL and get a PyArrow Table
+result = catalog.sql("SELECT * FROM my_table")
+print(result)
+
+# Execute SQL and get a Pandas DataFrame
+df = catalog.sql_to_pandas("SELECT * FROM my_table")
+print(df)
+```
+
+### Table Reference Format
+
+The default catalog and default database (`default`) are pre-configured, so you can reference tables in two ways:
+
+```python
+# Direct table name (uses default database)
+catalog.sql("SELECT * FROM my_table")
+
+# Two-part: database.table
+catalog.sql("SELECT * FROM mydb.my_table")
+```
+
+### Filtering
+
+```python
+result = catalog.sql("""
+    SELECT id, name, age 
+    FROM users 
+    WHERE age > 18 AND city = 'Beijing'
+""")
+```
+
+### Aggregation
+
+```python
+result = catalog.sql("""
+    SELECT city, COUNT(*) AS cnt, AVG(age) AS avg_age
+    FROM users
+    GROUP BY city
+    ORDER BY cnt DESC
+""")
+```
+
+### Join
+
+```python
+result = catalog.sql("""
+    SELECT u.name, o.order_id, o.amount
+    FROM users u
+    JOIN orders o ON u.id = o.user_id
+    WHERE o.amount > 100
+""")
+```
+
+### Subquery
+
+```python
+result = catalog.sql("""
+    SELECT * FROM users
+    WHERE id IN (
+        SELECT user_id FROM orders
+        WHERE amount > 1000
+    )
+""")
+```
+
+### Cross-Database Query
+
+```python
+# Query a table in another database using two-part syntax
+result = catalog.sql("""
+    SELECT u.name, o.amount
+    FROM default.users u
+    JOIN analytics.orders o ON u.id = o.user_id
+""")
+```
+
+## REST Catalog
+
+SQL query also works with REST catalogs:
+
+```python
+from pypaimon import CatalogFactory
+
+catalog = CatalogFactory.create({
+    "metastore": "rest",
+    "uri": "http://localhost:8080",
+    "warehouse": "my_warehouse",
+})
+
+result = catalog.sql("SELECT * FROM my_table LIMIT 10")
+```
+
+## Supported SQL Syntax
+
+The SQL engine is powered by Apache DataFusion, which supports a rich set of SQL syntax including:
+
+- `SELECT`, `WHERE`, `GROUP BY`, `HAVING`, `ORDER BY`, `LIMIT`
+- `JOIN` (INNER, LEFT, RIGHT, FULL, CROSS)
+- Subqueries and CTEs (`WITH`)
+- Aggregate functions (`COUNT`, `SUM`, `AVG`, `MIN`, `MAX`, etc.)
+- Window functions (`ROW_NUMBER`, `RANK`, `LAG`, `LEAD`, etc.)
+- `UNION`, `INTERSECT`, `EXCEPT`
+
+For the full SQL reference, see the [DataFusion SQL documentation](https://datafusion.apache.org/user-guide/sql/index.html).

paimon-python/pypaimon/__init__.py

@@ -35,4 +35,12 @@
     "Schema",
     "Tag",
     "TagManager",
+    "SQLContext",
 ]
+
+
+def __getattr__(name):
+    if name == "SQLContext":
+        from pypaimon.sql.sql_context import SQLContext
+        return SQLContext
+    raise AttributeError(f"module 'pypaimon' has no attribute {name}")

paimon-python/pypaimon/catalog/catalog.py

@@ -293,3 +293,37 @@ def list_branches(self, identifier: Identifier) -> List[str]:
         raise NotImplementedError(
             "list_branches is not supported by this catalog."
         )
+
+    @abstractmethod
+    def _get_catalog_options_map(self) -> Dict[str, str]:
+        """Return catalog options as a string-to-string dict for pypaimon-rust."""
+
+    def _get_sql_context(self):
+        if self._sql_context is None:
+            from pypaimon.sql.sql_context import SQLContext
+            self._sql_context = SQLContext(self._get_catalog_options_map())
+        return self._sql_context
+
+    def sql(self, query: str):
+        """Execute a SQL query and return the result as a PyArrow Table.
+
+        Requires pypaimon-rust and datafusion. Install with: pip install pypaimon[sql]
+
+        Tables can be referenced by name directly or with two-part (database.table) syntax.
+
+        Example::
+
+            catalog = CatalogFactory.create({"warehouse": "/path/to/warehouse"})
+            # Direct table name (uses default database)
+            catalog.sql("SELECT * FROM my_table WHERE id > 10")
+            # Two-part: database.table
+            catalog.sql("SELECT * FROM mydb.my_table")
+        """
+        return self._get_sql_context().sql(query)
+
+    def sql_to_pandas(self, query: str):
+        """Execute a SQL query and return the result as a Pandas DataFrame.
+
+        Requires pypaimon-rust and datafusion. Install with: pip install pypaimon[sql]
+        """
+        return self._get_sql_context().sql_to_pandas(query)

paimon-python/pypaimon/catalog/filesystem_catalog.py

@@ -47,6 +47,10 @@ def __init__(self, catalog_options: Options):
         self.warehouse = catalog_options.get(CatalogOptions.WAREHOUSE)
         self.catalog_options = catalog_options
         self.file_io = FileIO.get(self.warehouse, self.catalog_options)
+        self._sql_context = None
+
+    def _get_catalog_options_map(self) -> dict:
+        return {str(k): str(v) for k, v in self.catalog_options.to_map().items()}
 
     def list_databases(self) -> list:
         statuses = self.file_io.list_status(self.warehouse)

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

@@ -68,10 +68,14 @@ def __init__(self, context: CatalogContext, config_required: Optional[bool] = Tr
         # FUSE support (lazy import only when enabled)
         self.fuse_enabled = self.context.options.get(FuseOptions.FUSE_ENABLED, False)
         self._fuse_resolver = None
+        self._sql_context = None
         if self.fuse_enabled:
             from pypaimon.catalog.rest.fuse_support import FusePathResolver
             self._fuse_resolver = FusePathResolver(self.context.options, self.rest_api)
 
+    def _get_catalog_options_map(self) -> dict:
+        return {str(k): str(v) for k, v in self.context.options.to_map().items()}
+
     def catalog_loader(self):
         """
         Create and return a RESTCatalogLoader for this catalog.

paimon-python/pypaimon/cli/cli.py

@@ -121,6 +121,10 @@ def main():
     from pypaimon.cli.cli_catalog import add_catalog_subcommands
     add_catalog_subcommands(catalog_parser)
 
+    # SQL command
+    from pypaimon.cli.cli_sql import add_sql_subcommand
+    add_sql_subcommand(subparsers)
+
     args = parser.parse_args()
 
     if args.command is None: