Skip to content

Commit 16597db

Browse files
committed
Add table provider capsule to PyTableProvider
1 parent d23d72b commit 16597db

File tree

1 file changed

+18
-1
lines changed

1 file changed

+18
-1
lines changed

src/dataframe.rs

Lines changed: 18 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -34,6 +34,7 @@ use datafusion::error::DataFusionError;
3434
use datafusion::execution::SendableRecordBatchStream;
3535
use datafusion::parquet::basic::{BrotliLevel, Compression, GzipLevel, ZstdLevel};
3636
use datafusion::prelude::*;
37+
use datafusion_ffi::table_provider::FFI_TableProvider;
3738
use futures::{StreamExt, TryStreamExt};
3839
use pyo3::exceptions::PyValueError;
3940
use pyo3::prelude::*;
@@ -58,7 +59,7 @@ use crate::{
5859
// this is an interim implementation
5960
#[pyclass(name = "TableProvider", module = "datafusion")]
6061
pub struct PyTableProvider {
61-
provider: Arc<dyn TableProvider>,
62+
provider: Arc<dyn TableProvider + Send>,
6263
}
6364

6465
impl PyTableProvider {
@@ -71,6 +72,22 @@ impl PyTableProvider {
7172
PyTable::new(table_provider)
7273
}
7374
}
75+
76+
#[pymethods]
77+
impl PyTableProvider {
78+
fn __datafusion_table_provider__<'py>(
79+
&self,
80+
py: Python<'py>,
81+
) -> PyResult<Bound<'py, PyCapsule>> {
82+
let name = CString::new("datafusion_table_provider").unwrap();
83+
84+
let runtime = get_tokio_runtime().0.handle().clone();
85+
let provider = FFI_TableProvider::new(Arc::clone(&self.provider), false, Some(runtime));
86+
87+
PyCapsule::new(py, provider, Some(name.clone()))
88+
}
89+
}
90+
7491
const MAX_TABLE_BYTES_TO_DISPLAY: usize = 2 * 1024 * 1024; // 2 MB
7592
const MIN_TABLE_ROWS_TO_DISPLAY: usize = 20;
7693

0 commit comments

Comments
 (0)