feat: expose newest table provider to python (delta-io#4057)

# Description
The description of the main changes of your pull request

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

- closes #106
--->

# Documentation

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

Signed-off-by: Ion Koutsouris <15728914+ion-elgreco@users.noreply.github.com>

Author: ion-elgreco

crates/core/src/delta_datafusion/mod.rs

@@ -75,6 +75,7 @@ pub use self::session::{
     create_session,
 };
 pub use self::table_provider::next::DeltaScan as DeltaScanNext;
+pub use self::table_provider::next::SnapshotWrapper;
 pub(crate) use find_files::*;
 
 pub(crate) const PATH_COLUMN: &str = "__delta_rs_path";

crates/core/src/delta_datafusion/table_provider/next/mod.rs

@@ -54,7 +54,7 @@ mod scan;
 const FILE_ID_COLUMN_DEFAULT: &str = "__delta_rs_file_id__";
 
 #[derive(Clone, Debug, Serialize, Deserialize)]
-pub(crate) enum SnapshotWrapper {
+pub enum SnapshotWrapper {
     Snapshot(Arc<Snapshot>),
     EagerSnapshot(Arc<EagerSnapshot>),
 }
@@ -85,7 +85,8 @@ pub struct DeltaScan {
 }
 
 impl DeltaScan {
-    pub(super) fn new(snapshot: SnapshotWrapper, config: DeltaScanConfig) -> Result<Self> {
+    // create new delta scan
+    pub fn new(snapshot: SnapshotWrapper, config: DeltaScanConfig) -> Result<Self> {
         let scan_schema = config.table_schema(snapshot.table_configuration())?;
         let full_schema = if config.retain_file_id() {
             let mut fields = scan_schema.fields().to_vec();

python/src/lib.rs

@@ -23,7 +23,10 @@ use deltalake::datafusion::catalog::TableProvider;
 use deltalake::datafusion::datasource::provider_as_source;
 use deltalake::datafusion::logical_expr::LogicalPlanBuilder;
 use deltalake::datafusion::prelude::SessionContext;
-use deltalake::delta_datafusion::DeltaCdfTableProvider;
+use deltalake::delta_datafusion::{
+    DeltaCdfTableProvider, DeltaScanConfig, DeltaScanNext, SnapshotWrapper,
+};
+
 use deltalake::errors::DeltaTableError;
 use deltalake::kernel::scalars::ScalarExt;
 use deltalake::kernel::transaction::{CommitBuilder, CommitProperties, TableReference};
@@ -1839,8 +1842,20 @@ impl RawDeltaTable {
         let handle = None;
         let name = CString::new("datafusion_table_provider").unwrap();
 
-        let table = self.with_table(|t| Ok(Arc::new(t.clone())))?;
-        let provider = FFI_TableProvider::new(table, false, handle);
+        let table = self.with_table(|t| Ok(t.clone()))?;
+
+        let config = DeltaScanConfig::new();
+        let snapshot_wrapped = SnapshotWrapper::EagerSnapshot(Arc::new(
+            table
+                .snapshot()
+                .map_err(PythonError::from)?
+                .snapshot()
+                .clone(),
+        ));
+        let scan =
+            Arc::new(DeltaScanNext::new(snapshot_wrapped, config).map_err(PythonError::from)?)
+                as Arc<dyn TableProvider>;
+        let provider = FFI_TableProvider::new(scan, false, handle);
 
         PyCapsule::new(py, provider, Some(name.clone()))
     }