From d30ac53d93bb436f2b851d31e0770e5e113d69d1 Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Fri, 14 Aug 2026 23:28:22 -0700 Subject: [PATCH] fix(python): release GIL during catalog I/O --- bindings/python/src/context.rs | 38 ++++++---- bindings/python/tests/test_catalog_gil.py | 86 +++++++++++++++++++++++ 2 files changed, 110 insertions(+), 14 deletions(-) create mode 100644 bindings/python/tests/test_catalog_gil.py diff --git a/bindings/python/src/context.rs b/bindings/python/src/context.rs index 6ab04ad7..b5dbc6a0 100644 --- a/bindings/python/src/context.rs +++ b/bindings/python/src/context.rs @@ -119,8 +119,8 @@ pub struct PaimonCatalog { impl PaimonCatalog { /// Create a Paimon catalog that can be registered into a DataFusion session. #[new] - fn new(catalog_options: HashMap) -> PyResult { - let catalog = build_paimon_catalog(catalog_options)?; + fn new(py: Python<'_>, catalog_options: HashMap) -> PyResult { + let catalog = py.detach(|| build_paimon_catalog(catalog_options))?; let provider = Arc::new( PaimonCatalogProvider::new( None, @@ -148,21 +148,28 @@ impl PaimonCatalog { } /// List all databases in this catalog. - fn list_databases(&self) -> PyResult> { - runtime() - .block_on(self.catalog.list_databases()) - .map_err(to_py_err) + fn list_databases(&self, py: Python<'_>) -> PyResult> { + let catalog = Arc::clone(&self.catalog); + py.detach(|| { + runtime() + .block_on(catalog.list_databases()) + .map_err(to_py_err) + }) } /// List all tables in the given database. - fn list_tables(&self, database_name: &str) -> PyResult> { - runtime() - .block_on(self.catalog.list_tables(database_name)) - .map_err(to_py_err) + fn list_tables(&self, py: Python<'_>, database_name: &str) -> PyResult> { + let catalog = Arc::clone(&self.catalog); + let database_name = database_name.to_string(); + py.detach(|| { + runtime() + .block_on(catalog.list_tables(&database_name)) + .map_err(to_py_err) + }) } /// Get a table handle by `"db.table"` identifier. - fn get_table(&self, identifier: &str) -> PyResult { + fn get_table(&self, py: Python<'_>, identifier: &str) -> PyResult { let parts: Vec<&str> = identifier.splitn(2, '.').collect(); if parts.len() != 2 || parts[0].is_empty() || parts[1].is_empty() { return Err(PyValueError::new_err(format!( @@ -170,9 +177,12 @@ impl PaimonCatalog { ))); } let id = Identifier::new(parts[0], parts[1]); - let table = runtime() - .block_on(self.catalog.get_table(&id)) - .map_err(to_py_err)?; + let catalog = Arc::clone(&self.catalog); + let table = py.detach(|| { + runtime() + .block_on(catalog.get_table(&id)) + .map_err(to_py_err) + })?; Ok(PyTable::new(Arc::new(table))) } } diff --git a/bindings/python/tests/test_catalog_gil.py b/bindings/python/tests/test_catalog_gil.py new file mode 100644 index 00000000..4cb38b4c --- /dev/null +++ b/bindings/python/tests/test_catalog_gil.py @@ -0,0 +1,86 @@ +# 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 json +import threading +from http.server import BaseHTTPRequestHandler, HTTPServer + +import pytest + +from pypaimon_rust.datafusion import PaimonCatalog + + +class _RESTHandler(BaseHTTPRequestHandler): + def do_GET(self): + if self.path.startswith("/v1/config?"): + self._respond({"defaults": {"prefix": "test"}}) + elif self.path == "/v1/test/databases": + self._respond({"databases": ["db"], "nextPageToken": None}) + elif self.path == "/v1/test/databases/db/tables": + self._respond({"tables": ["table"], "nextPageToken": None}) + else: + self._respond( + { + "resourceType": "table", + "resourceName": "missing", + "message": "Not Found", + "code": 404, + }, + 404, + ) + + def log_message(self, format, *args): + pass + + def _respond(self, payload, status=200): + body = json.dumps(payload).encode() + self.send_response(status) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + +@pytest.fixture +def rest_server(): + server = HTTPServer(("localhost", 0), _RESTHandler) + thread = threading.Thread(target=server.serve_forever) + thread.start() + try: + yield "http://localhost:%d" % server.server_port + finally: + server.shutdown() + server.server_close() + thread.join() + + +def test_rest_catalog_calls_release_gil(rest_server, monkeypatch): + monkeypatch.setenv("NO_PROXY", "localhost,127.0.0.1") + monkeypatch.setenv("no_proxy", "localhost,127.0.0.1") + + catalog = PaimonCatalog( + { + "metastore": "rest", + "uri": rest_server, + "warehouse": "warehouse", + "token.provider": "bear", + "token": "test-token", + } + ) + assert catalog.list_databases() == ["db"] + assert catalog.list_tables("db") == ["table"] + with pytest.raises(ValueError, match="does not exist"): + catalog.get_table("db.missing")