(hudi-rs) branch main updated: fix(python): run the DataFusion FFI provider on the binding's tokio runtime (#829)

[yi...@apache.org]

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 <et...@gmail.com>
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