fix(python): guard DataFusion FFI export on datafusion major version (delta-io#4142)

# Description

- Add a runtime version check in __datafusion_table_provider__ to
prevent FFI ABI mismatch segfaults
- Block capsule export when installed datafusion major != 52
- Provide actionable error text with QueryBuilder workaround

Changes:

- lib.rs: add REQUIRED_DATAFUSION_PY_MAJOR, datafusion_python_version(),
guard at method start
- test_datafusion.py: add incompatible version and not installed tests

Note: This guard is a temporary safety net to prevent segfaults until
DataFusion 52 Python wheels are available on PyPI. Once wheels land,
users can install datafusion==52.* and use SessionContext registration
normally.

# Related Issue(s)
- delta-io#4135 
<!---
For example:

- closes delta-io#106
--->

# Documentation

<!---
Share links to useful documentation
--->

Signed-off-by: Ethan Urbanski <ethan@urbanskitech.com>

Author: ethan-tyler

python/src/lib.rs

@@ -52,7 +52,7 @@ use deltalake::{DeltaResult, DeltaTable, DeltaTableBuilder, init_client_version}
 use futures::TryStreamExt;
 use pyo3::exceptions::{PyRuntimeError, PyValueError};
 use pyo3::pybacked::PyBackedStr;
-use pyo3::types::{PyCapsule, PyDict, PyFrozenSet};
+use pyo3::types::{PyCapsule, PyDict, PyFrozenSet, PyModule};
 use pyo3::{IntoPyObjectExt, prelude::*};
 use pyo3_arrow::export::{Arro3RecordBatch, Arro3RecordBatchReader};
 use pyo3_arrow::{PyRecordBatchReader, PySchema as PyArrowSchema};
@@ -115,6 +115,17 @@ struct RawDeltaTableMetaData {
 
 type StringVec = Vec<String>;
 
+const REQUIRED_DATAFUSION_PY_MAJOR: u32 = 52;
+
+fn datafusion_python_version(py: Python<'_>) -> Option<String> {
+    let importlib_metadata = PyModule::import(py, "importlib.metadata").ok()?;
+    importlib_metadata
+        .call_method1("version", ("datafusion",))
+        .ok()?
+        .extract()
+        .ok()
+}
+
 /// Segmented impl for RawDeltaTable to avoid these methods being exposed via the pymethods macro.
 ///
 /// In essence all these functions should be considered internal to the Rust code and not exposed
@@ -1838,6 +1849,24 @@ impl RawDeltaTable {
         &self,
         py: Python<'py>,
     ) -> PyResult<Bound<'py, PyCapsule>> {
+        let found_version = datafusion_python_version(py);
+        let found_major = found_version
+            .as_deref()
+            .and_then(|version| version.split('.').next()?.parse().ok());
+        if found_major != Some(REQUIRED_DATAFUSION_PY_MAJOR) {
+            let found = found_version.unwrap_or_else(|| "not installed".to_string());
+            return Err(PyRuntimeError::new_err(format!(
+                "DataFusion Python integration requires datafusion=={required}.x (found: {found}).\n\n\
+This deltalake build exports a DataFusion {required}.x FFI TableProvider; mismatched majors can segfault.\n\n\
+Workaround (no DataFusion-Python required): use deltalake.QueryBuilder:\n\n\
+    from deltalake import DeltaTable, QueryBuilder\n\n\
+    dt = DeltaTable('path/to/table')\n\
+    data = QueryBuilder().register('tbl', dt).execute('SELECT * FROM tbl')\n\n\
+Install datafusion=={required}.* (matching major) to use DataFusion SessionContext registration, or keep using QueryBuilder.",
+                required = REQUIRED_DATAFUSION_PY_MAJOR,
+            )));
+        }
+
         let handle = rt().handle();
         let name = CString::new("datafusion_table_provider").unwrap();
         let table = self.with_table(|t| Ok(t.clone()))?;

python/tests/test_datafusion.py

@@ -14,6 +14,58 @@ def _datafusion_major_version() -> int | None:
         return None
 
 
+def test_datafusion_table_provider_incompatible_version_errors(tmp_path, monkeypatch):
+    # Force the runtime check to behave like "datafusion==51.x" even if datafusion isn't installed.
+    call_count = {"count": 0}
+
+    def fake_version(pkg: str) -> str:
+        assert pkg == "datafusion"
+        call_count["count"] += 1
+        if call_count["count"] == 1:
+            return "51.0.0"
+        return "52.0.0"
+
+    monkeypatch.setattr("importlib.metadata.version", fake_version)
+
+    table = Table(
+        {"id": Array([1, 2, 3], Field("id", type=DataType.int64(), nullable=True))}
+    )
+    write_deltalake(tmp_path, table)
+    dt = DeltaTable(tmp_path)
+
+    # Call the FFI export hook directly; DO NOT call SessionContext.register_* (that is what segfaults).
+    with pytest.raises(RuntimeError) as exc_info:
+        dt.__datafusion_table_provider__()  # type: ignore[attr-defined]
+
+    assert call_count["count"] == 1
+
+    msg = str(exc_info.value)
+    assert "datafusion" in msg
+    assert "datafusion==52" in msg
+    assert "QueryBuilder" in msg
+
+
+def test_datafusion_table_provider_not_installed_errors(tmp_path, monkeypatch):
+    def fake_version(pkg: str) -> str:
+        raise PackageNotFoundError(pkg)
+
+    monkeypatch.setattr("importlib.metadata.version", fake_version)
+
+    table = Table(
+        {"id": Array([1, 2, 3], Field("id", type=DataType.int64(), nullable=True))}
+    )
+    write_deltalake(tmp_path, table)
+    dt = DeltaTable(tmp_path)
+
+    with pytest.raises(RuntimeError) as exc_info:
+        dt.__datafusion_table_provider__()  # type: ignore[attr-defined]
+
+    msg = str(exc_info.value)
+    assert "datafusion" in msg
+    assert "not installed" in msg.lower()
+    assert "QueryBuilder" in msg
+
+
 @pytest.mark.datafusion
 def test_datafusion_table_provider(tmp_path):
     if os.environ.get("DELTALAKE_RUN_DATAFUSION_TESTS") != "1":