This is an automated email from the ASF dual-hosted git repository. JingsongLi pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/paimon-rust.git
The following commit(s) were added to refs/heads/main by this push:
new 7fd9bab9 fix(python): rebuild the fallback runtime after fork (#832)
7fd9bab9 is described below
commit 7fd9bab99f63a3860a72cd93f31b047f9f770f89
Author: Jingsong Lee <[email protected]>
AuthorDate: Tue Sep 15 10:18:05 2026 +0800
fix(python): rebuild the fallback runtime after fork (#832)
---
Cargo.lock | 16 ++--
DEPENDENCIES.rust.tsv | 8 +-
benchmarks/tpcds/DEPENDENCIES.rust.tsv | 8 +-
bindings/c/DEPENDENCIES.rust.tsv | 8 +-
bindings/go/DEPENDENCIES.rust.tsv | 8 +-
bindings/python/DEPENDENCIES.rust.tsv | 8 +-
bindings/python/tests/test_runtime_fork.py | 79 ++++++++++++++++++
crates/integration_tests/DEPENDENCIES.rust.tsv | 8 +-
.../integrations/datafusion/DEPENDENCIES.rust.tsv | 8 +-
crates/integrations/datafusion/src/runtime.rs | 97 ++++++++++++++++++++--
crates/paimon-rest-server/DEPENDENCIES.rust.tsv | 8 +-
crates/paimon/DEPENDENCIES.rust.tsv | 8 +-
12 files changed, 213 insertions(+), 51 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index e13b4b93..6b3bc679 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -658,9 +658,9 @@ checksum =
"f2032f911046de80f0a198e0901378627c33f59ea0ac00e363d481118bd70a53"
[[package]]
name = "aws-lc-rs"
-version = "1.17.3"
+version = "1.18.1"
source = "registry+https://github.com/rust-lang/crates.io-index";
-checksum = "00bdb5da18dac48ca2cc7cd4a98e533e8635a58e2361d13a1a4ee3888e0d72f1"
+checksum = "b281d307588d634de920874890732659e2e7672f72b5e10e81badc1a8a83621e"
dependencies = [
"aws-lc-sys",
"zeroize",
@@ -668,9 +668,9 @@ dependencies = [
[[package]]
name = "aws-lc-sys"
-version = "0.43.0"
+version = "0.45.0"
source = "registry+https://github.com/rust-lang/crates.io-index";
-checksum = "43103168cc76fe62678a375e722fc9cb3a0146159ac5828bc4f0dfd755c2224c"
+checksum = "9bff6c3b54fad79a2e60b8102caf565819711497c1f5f092f49508e2f5c31b27"
dependencies = [
"cc",
"cmake",
@@ -6031,9 +6031,9 @@ dependencies = [
[[package]]
name = "rustls"
-version = "0.23.42"
+version = "0.23.45"
source = "registry+https://github.com/rust-lang/crates.io-index";
-checksum = "3c54fcab019b409d04215d3a17cb438fd7fbf192ee61461f20f4fe18704bc138"
+checksum = "0d41d731c7d2f962d1ccc364cec258de3c0e93b38c2fb3ba97ac74513048d634"
dependencies = [
"aws-lc-rs",
"once_cell",
@@ -6094,9 +6094,9 @@ checksum =
"f87165f0995f63a9fbeea62b64d10b4d9d8e78ec6d7d51fb2125fda7bb36788f"
[[package]]
name = "rustls-webpki"
-version = "0.103.13"
+version = "0.103.15"
source = "registry+https://github.com/rust-lang/crates.io-index";
-checksum = "61c429a8649f110dddef65e2a5ad240f747e85f7758a6bccc7e5777bd33f756e"
+checksum = "f3c3cf1d8b1e7d4927e2d154c3fcb02979afb9939629c62cd9048d4f07b60ac2"
dependencies = [
"aws-lc-rs",
"ring",
diff --git a/DEPENDENCIES.rust.tsv b/DEPENDENCIES.rust.tsv
index 5bb7096b..0e823a2b 100644
--- a/DEPENDENCIES.rust.tsv
+++ b/DEPENDENCIES.rust.tsv
@@ -55,8 +55,8 @@ [email protected] X
[email protected]
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X X
X X X
[email protected] X
X
[email protected] X X
X X X
[email protected]
X
[email protected]
X
[email protected]
X
@@ -504,12 +504,12 @@ [email protected]
X X
[email protected] X
X
[email protected] X
X
[email protected] X X
X
[email protected] X
X X
[email protected] X
X X
[email protected] X
X X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected]
X
[email protected]
X
[email protected] X
X
[email protected] X X
[email protected] X
X X
diff --git a/benchmarks/tpcds/DEPENDENCIES.rust.tsv
b/benchmarks/tpcds/DEPENDENCIES.rust.tsv
index a58f9739..15d9276a 100644
--- a/benchmarks/tpcds/DEPENDENCIES.rust.tsv
+++ b/benchmarks/tpcds/DEPENDENCIES.rust.tsv
@@ -39,8 +39,8 @@ [email protected] X
[email protected]
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X X
X X X
[email protected] X
X
[email protected] X X
X X X
[email protected] X
[email protected] X
X
[email protected] X
X
@@ -327,12 +327,12 @@ [email protected] X
X
[email protected]
X
[email protected] X
X
[email protected] X X
X
[email protected] X
X X
[email protected] X
X X
[email protected] X
X X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected]
X
[email protected]
X
[email protected] X
X
[email protected] X X
[email protected] X
X X
diff --git a/bindings/c/DEPENDENCIES.rust.tsv b/bindings/c/DEPENDENCIES.rust.tsv
index 47cb3e4b..ba261e8e 100644
--- a/bindings/c/DEPENDENCIES.rust.tsv
+++ b/bindings/c/DEPENDENCIES.rust.tsv
@@ -31,8 +31,8 @@ [email protected] X
[email protected]
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X X
X X X
[email protected] X
X
[email protected] X X
X X X
[email protected] X
[email protected] X
X
[email protected] X
X
@@ -281,12 +281,12 @@ [email protected] X
X
[email protected]
X
[email protected] X
X
[email protected] X X
X
[email protected] X
X X
[email protected] X
X X
[email protected] X
X X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected]
X
[email protected]
X
[email protected] X
X
[email protected] X X
[email protected] X
X X
diff --git a/bindings/go/DEPENDENCIES.rust.tsv
b/bindings/go/DEPENDENCIES.rust.tsv
index 47cb3e4b..ba261e8e 100644
--- a/bindings/go/DEPENDENCIES.rust.tsv
+++ b/bindings/go/DEPENDENCIES.rust.tsv
@@ -31,8 +31,8 @@ [email protected] X
[email protected]
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X X
X X X
[email protected] X
X
[email protected] X X
X X X
[email protected] X
[email protected] X
X
[email protected] X
X
@@ -281,12 +281,12 @@ [email protected] X
X
[email protected]
X
[email protected] X
X
[email protected] X X
X
[email protected] X
X X
[email protected] X
X X
[email protected] X
X X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected]
X
[email protected]
X
[email protected] X
X
[email protected] X X
[email protected] X
X X
diff --git a/bindings/python/DEPENDENCIES.rust.tsv
b/bindings/python/DEPENDENCIES.rust.tsv
index 1957e800..341eda3a 100644
--- a/bindings/python/DEPENDENCIES.rust.tsv
+++ b/bindings/python/DEPENDENCIES.rust.tsv
@@ -40,8 +40,8 @@ [email protected] X
[email protected]
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X X
X X X
[email protected] X
X
[email protected] X X
X X X
[email protected] X
[email protected] X
X
[email protected] X
X
@@ -428,12 +428,12 @@ [email protected]
X X
[email protected] X
X
[email protected] X
X
[email protected] X X
X
[email protected] X
X X
[email protected] X
X X
[email protected] X
X X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected]
X
[email protected]
X
[email protected] X
X
[email protected] X X
[email protected] X
X X
diff --git a/bindings/python/tests/test_runtime_fork.py
b/bindings/python/tests/test_runtime_fork.py
new file mode 100644
index 00000000..740e42fe
--- /dev/null
+++ b/bindings/python/tests/test_runtime_fork.py
@@ -0,0 +1,79 @@
+# 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 multiprocessing
+
+import pytest
+
+from pypaimon_rust.datafusion import PaimonCatalog, SQLContext
+
+
+def _plan_rows(catalog):
+ assert "fork_db" in catalog.list_databases()
+ builder = catalog.get_table("fork_db.t").new_read_builder()
+ plan = builder.new_scan().plan()
+ assert plan.snapshot_id() == 1
+ assert sum(split.row_count() for split in plan.splits()) == 3
+ return [split.serialize() for split in plan.splits()]
+
+
+def _plan_in_child(warehouse, inherited_catalog, result):
+ catalog = inherited_catalog or PaimonCatalog({"warehouse": warehouse})
+ # The second call also exercises reuse of the child's runtime.
+ expected = _plan_rows(catalog)
+ assert _plan_rows(catalog) == expected
+ result.send(expected)
+ result.close()
+
+
[email protected]("fork" not in multiprocessing.get_all_start_methods(),
reason="requires fork")
[email protected]("reuse_catalog", [False, True])
+def test_fork_after_parent_planning(tmp_path, reuse_catalog):
+ warehouse = str(tmp_path)
+ ctx = SQLContext()
+ ctx.register_catalog("paimon", {"warehouse": warehouse})
+ ctx.sql("CREATE SCHEMA paimon.fork_db")
+ ctx.sql("CREATE TABLE paimon.fork_db.t (id INT)")
+ ctx.sql("INSERT INTO paimon.fork_db.t VALUES (1), (2), (3)")
+ catalog = PaimonCatalog({"warehouse": warehouse})
+ expected = _plan_rows(catalog)
+
+ context = multiprocessing.get_context("fork")
+ received, sent = context.Pipe(duplex=False)
+ child = context.Process(
+ target=_plan_in_child,
+ args=(warehouse, catalog if reuse_catalog else None, sent),
+ )
+ child.start()
+ sent.close()
+ try:
+ assert received.poll(20), "child planning blocked after inheriting the
parent runtime"
+ assert received.recv() == expected
+ child.join(5)
+ assert child.exitcode == 0
+ finally:
+ received.close()
+ if child.is_alive():
+ child.terminate()
+ child.join(5)
+ if child.is_alive():
+ child.kill()
+ child.join(5)
+ child.close()
+
+ # Replacing the child's runtime must not disrupt its parent.
+ assert _plan_rows(catalog) == expected
diff --git a/crates/integration_tests/DEPENDENCIES.rust.tsv
b/crates/integration_tests/DEPENDENCIES.rust.tsv
index b33de207..81971be1 100644
--- a/crates/integration_tests/DEPENDENCIES.rust.tsv
+++ b/crates/integration_tests/DEPENDENCIES.rust.tsv
@@ -30,8 +30,8 @@ [email protected] X
[email protected]
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X X
X X X
[email protected] X
X
[email protected] X X
X X X
[email protected] X
[email protected] X
X
[email protected] X
X
@@ -251,12 +251,12 @@ [email protected] X
X
[email protected]
X
[email protected] X
X
[email protected] X X
X
[email protected] X
X X
[email protected] X
X X
[email protected] X
X X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected]
X
[email protected]
X
[email protected] X
X
[email protected] X X
[email protected] X
X X
diff --git a/crates/integrations/datafusion/DEPENDENCIES.rust.tsv
b/crates/integrations/datafusion/DEPENDENCIES.rust.tsv
index 6e690bf9..4d4a003d 100644
--- a/crates/integrations/datafusion/DEPENDENCIES.rust.tsv
+++ b/crates/integrations/datafusion/DEPENDENCIES.rust.tsv
@@ -46,8 +46,8 @@ [email protected] X
[email protected]
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X X
X X X
[email protected] X
X
[email protected] X X
X X X
[email protected] X
[email protected] X
X
[email protected] X
X
@@ -423,12 +423,12 @@ [email protected]
X X
[email protected] X
X
[email protected] X
X
[email protected] X X
X
[email protected] X
X X
[email protected] X
X X
[email protected] X
X X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected]
X
[email protected]
X
[email protected] X
X
[email protected] X X
[email protected] X
X X
diff --git a/crates/integrations/datafusion/src/runtime.rs
b/crates/integrations/datafusion/src/runtime.rs
index 4da502ac..aed51093 100644
--- a/crates/integrations/datafusion/src/runtime.rs
+++ b/crates/integrations/datafusion/src/runtime.rs
@@ -16,23 +16,63 @@
// under the License.
use std::future::Future;
+use std::sync::atomic::{AtomicPtr, Ordering};
use std::sync::OnceLock;
use tokio::runtime::{Handle, Runtime};
-static RUNTIME: OnceLock<Runtime> = OnceLock::new();
+struct ProcessRuntime {
+ pid: u32,
+ runtime: OnceLock<Runtime>,
+}
+
+static RUNTIME: AtomicPtr<ProcessRuntime> =
AtomicPtr::new(std::ptr::null_mut());
fn global_runtime() -> &'static Runtime {
- RUNTIME.get_or_init(|| {
- Runtime::new()
- .expect("failed to build global tokio runtime for paimon
datafusion integration")
- })
+ let pid = std::process::id();
+ let mut current = RUNTIME.load(Ordering::Acquire);
+ loop {
+ if !current.is_null() {
+ // SAFETY: Published states are never freed. Acquire pairs with the
+ // successful publication below, so both fields are initialized.
+ let state = unsafe { &*current };
+ if state.pid == pid {
+ return state.runtime.get_or_init(|| {
+ Runtime::new().expect(
+ "failed to build global tokio runtime for paimon
datafusion integration",
+ )
+ });
+ }
+ }
+
+ // A fork inherits the runtime but none of its worker threads. Publish
+ // a fresh, uninitialized state before constructing a runtime so a fork
+ // during initialization also bypasses the parent's locked OnceLock.
+ let next = Box::into_raw(Box::new(ProcessRuntime {
+ pid,
+ runtime: OnceLock::new(),
+ }));
+ match RUNTIME.compare_exchange(current, next, Ordering::AcqRel,
Ordering::Acquire) {
+ Ok(_) => {
+ // Do not drop the inherited state: runtime shutdown can wait
+ // for threads which no longer exist in the child process.
+ current = next;
+ }
+ Err(actual) => {
+ // SAFETY: This state was never published or initialized; no
+ // other thread can hold a reference to it.
+ unsafe { drop(Box::from_raw(next)) };
+ current = actual;
+ }
+ }
+ }
}
/// Returns a [`Handle`] to the global Tokio runtime.
///
/// If a Tokio runtime is already entered on the current thread, its handle is
-/// returned directly. Otherwise a lazily-initialised global runtime is used.
+/// returned directly. Otherwise a lazily-initialised runtime for the current
+/// process is used, replacing any state inherited from a parent after `fork`.
pub fn runtime() -> Handle {
match Handle::try_current() {
Ok(h) => h,
@@ -42,7 +82,7 @@ pub fn runtime() -> Handle {
// These helpers work around DataFusion FFI callbacks that may run without an
// entered Tokio runtime. See
https://github.com/apache/datafusion/issues/16312.
-// A global OnceLock<Runtime> avoids creating a new runtime on every call;
+// A runtime shared within each process avoids creating one on every call;
// if DataFusion fixes runtime propagation end-to-end, we should be able to
// remove these manual fallback runtimes.
pub(crate) async fn await_with_runtime<F>(future: F) -> F::Output
@@ -72,3 +112,46 @@ where
global_runtime().block_on(future)
}
}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn concurrent_callers_share_the_process_runtime() {
+ let barrier = std::sync::Barrier::new(8);
+ std::thread::scope(|scope| {
+ let threads: Vec<_> = (0..8)
+ .map(|_| {
+ scope.spawn(|| {
+ barrier.wait();
+ let runtime = global_runtime();
+ assert_eq!(
+ runtime.block_on(async { tokio::spawn(async { 42
}).await.unwrap() }),
+ 42
+ );
+ runtime as *const Runtime as usize
+ })
+ })
+ .collect();
+ let runtimes: Vec<_> = threads
+ .into_iter()
+ .map(|thread| thread.join().unwrap())
+ .collect();
+ assert!(runtimes.iter().all(|runtime| *runtime == runtimes[0]));
+ });
+ }
+
+ #[test]
+ fn entered_runtime_is_preserved() {
+ let local = tokio::runtime::Builder::new_current_thread()
+ .enable_all()
+ .build()
+ .unwrap();
+ let _entered = local.enter();
+ assert_eq!(
+ runtime().runtime_flavor(),
+ tokio::runtime::RuntimeFlavor::CurrentThread
+ );
+ }
+}
diff --git a/crates/paimon-rest-server/DEPENDENCIES.rust.tsv
b/crates/paimon-rest-server/DEPENDENCIES.rust.tsv
index 374c700d..8c60ff42 100644
--- a/crates/paimon-rest-server/DEPENDENCIES.rust.tsv
+++ b/crates/paimon-rest-server/DEPENDENCIES.rust.tsv
@@ -30,8 +30,8 @@ [email protected] X
[email protected]
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X X
X X X
[email protected] X
X
[email protected] X X
X X X
[email protected]
X
[email protected]
X
[email protected]
X
@@ -255,12 +255,12 @@ [email protected] X
X
[email protected]
X
[email protected] X
X
[email protected] X X
X
[email protected] X
X X
[email protected] X
X X
[email protected] X
X X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected]
X
[email protected]
X
[email protected] X
X
[email protected] X X
[email protected] X
X X
diff --git a/crates/paimon/DEPENDENCIES.rust.tsv
b/crates/paimon/DEPENDENCIES.rust.tsv
index 9273ed30..b4befc39 100644
--- a/crates/paimon/DEPENDENCIES.rust.tsv
+++ b/crates/paimon/DEPENDENCIES.rust.tsv
@@ -44,8 +44,8 @@ [email protected] X
[email protected]
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X X
X X X
[email protected] X
X
[email protected] X X
X X X
[email protected] X
[email protected] X
X
[email protected] X
X
@@ -415,12 +415,12 @@ [email protected]
X X
[email protected] X
X
[email protected] X
X
[email protected] X X
X
[email protected] X
X X
[email protected] X
X X
[email protected] X
X X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected]
X
[email protected]
X
[email protected] X
X
[email protected] X X
[email protected] X
X X
