This is an automated email from the ASF dual-hosted git repository.
yihua pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/hudi-rs.git
The following commit(s) were added to refs/heads/main by this push:
new ac404bdb fix(python): run the DataFusion FFI provider on the binding's
tokio runtime (#829)
ac404bdb is described below
commit ac404bdb58bc84f7fd57af3bddcad16b12f364a3
Author: Y Ethan Guo <[email protected]>
AuthorDate: Wed Oct 28 12:25:19 2026 -0700
fix(python): run the DataFusion FFI provider on the binding's tokio runtime
(#829)
datafusion-python polls the provider's stream on its own compiled copy of
tokio, so Lance's I/O scheduler found no current runtime in the binding's
copy and aborted the interpreter. Pass the binding's runtime handle to
FFI_TableProvider, which datafusion-ffi enters around execution and every
poll.
---
python/src/datafusion_internal.rs | 7 ++++++-
python/tests/test_datafusion_ffi.py | 25 +++++++++++++++++++++++++
2 files changed, 31 insertions(+), 1 deletion(-)
diff --git a/python/src/datafusion_internal.rs
b/python/src/datafusion_internal.rs
index 6c1d7006..40cd80b2 100644
--- a/python/src/datafusion_internal.rs
+++ b/python/src/datafusion_internal.rs
@@ -90,7 +90,12 @@ impl HudiDataFusionDataSource {
) -> PyResult<Bound<'py, PyCapsule>> {
let provider = Arc::new(self.table.clone());
let codec = extract_codec(session)?;
- let ffi_provider = FFI_TableProvider::new_with_ffi_codec(provider,
false, None, codec);
+ // datafusion-python polls the provider's stream on its own tokio
runtime,
+ // which this extension's copy of tokio cannot see. Hand over the
binding's
+ // runtime so datafusion-ffi enters it around execution and every poll;
+ // Lance's I/O scheduler spawns onto the current runtime and aborts
without one.
+ let runtime = Some(rt().handle().clone());
+ let ffi_provider = FFI_TableProvider::new_with_ffi_codec(provider,
false, runtime, codec);
PyCapsule::new_with_value(py, ffi_provider,
c"datafusion_table_provider")
}
}
diff --git a/python/tests/test_datafusion_ffi.py
b/python/tests/test_datafusion_ffi.py
index 51628fba..e69fdda4 100644
--- a/python/tests/test_datafusion_ffi.py
+++ b/python/tests/test_datafusion_ffi.py
@@ -27,6 +27,7 @@ import pytest
datafusion = pytest.importorskip("datafusion")
+from hudi import HudiTable # noqa: E402
from hudi._internal import HudiDataFusionDataSource, get_test_table_path #
noqa: E402
@@ -75,3 +76,27 @@ def
test_datafusion_ffi_plan_goes_through_the_ffi_boundary(v6_table):
plans.append(columns["plan"][i])
assert any("FFI_ExecutionPlan" in plan for plan in plans), plans
+
+
+def test_datafusion_ffi_reads_a_lance_table():
+ # Lance's I/O scheduler spawns onto the current tokio runtime. The
provider's
+ # stream is polled from datafusion-python's runtime, which the binding's
copy
+ # of tokio cannot see, so without a handle on the FFI provider this read
+ # aborted the interpreter. A local Parquet table never asks for a runtime,
+ # which is why the tests above did not catch it.
+ lance_table = get_test_table_path("v9_lance_nonpartitioned", "cow")
+ expected = sorted(
+ (row["id"], row["name"])
+ for batch in HudiTable(lance_table).read()
+ for row in batch.to_pylist()
+ )
+ assert expected, "fixture should hold rows"
+
+ ctx = datafusion.SessionContext()
+ ctx.register_table("lance_t", HudiDataFusionDataSource(lance_table))
+ got = sorted(
+ (row["id"], row["name"])
+ for batch in ctx.sql("SELECT id, name FROM lance_t").collect()
+ for row in batch.to_pylist()
+ )
+ assert got == expected
