fix

Author: adriangb

datafusion/catalog/src/default_table_source.rs

@@ -73,7 +73,6 @@ impl TableSource for DefaultTableSource {
         &self,
         filter: &[&Expr],
     ) -> datafusion_common::Result<Vec<TableProviderFilterPushDown>> {
-        #[allow(deprecated)]
         self.table_provider.supports_filters_pushdown(filter)
     }
 

datafusion/catalog/src/memory/table.rs

@@ -138,6 +138,7 @@ impl MemTable {
     ) -> Result<Self> {
         let schema = t.schema();
         let constraints = t.constraints();
+        #[expect(deprecated)]
         let exec = t.scan(state, None, &[], None).await?;
         let partition_count = exec.output_partitioning().partition_count();
 

datafusion/catalog/src/table.rs

@@ -185,11 +185,22 @@ pub trait TableProvider: Debug + Sync + Send {
             limit,
         } = args;
         let filters = filters.unwrap_or_default();
-        #[allow(deprecated)]
+        let unsupported_filters = self
+            .supports_filters_pushdown(&filters.iter().collect_vec())?
+            .into_iter()
+            .zip(&filters)
+            .filter_map(|(support, expr)| match support {
+                TableProviderFilterPushDown::Inexact | TableProviderFilterPushDown::Unsupported => {
+                    Some(expr.clone())
+                }
+                TableProviderFilterPushDown::Exact => None,
+            })
+            .collect_vec();
+        #[expect(deprecated)]
         let plan = self
             .scan(state, projection.as_ref(), &filters, limit)
             .await?;
-        Ok(ScanResult::new(plan, vec![]))
+        Ok(ScanResult::new(plan, unsupported_filters))
     }
 
     /// Specify if DataFusion should provide filter expressions to the
@@ -272,7 +283,6 @@ pub trait TableProvider: Debug + Sync + Send {
     ///     }
     /// }
     /// ```
-    #[deprecated(since = "50.0.0", note = "Use `scan_with_args` instead")]
     fn supports_filters_pushdown(
         &self,
         filters: &[&Expr],

datafusion/core/src/datasource/listing/table.rs

@@ -1178,9 +1178,9 @@ impl TableProvider for ListingTable {
         state: &dyn Session,
         args: ScanArgs,
     ) -> Result<ScanResult> {
-        let projection = options.projection();
-        let filters = options.filters().map(|f| f.to_vec()).unwrap_or_default();
-        let limit = options.limit();
+        let projection = args.projection();
+        let filters = args.filters().map(|f| f.to_vec()).unwrap_or_default();
+        let limit = args.limit();
 
         // extract types of partition columns
         let table_partition_cols = self
@@ -1219,7 +1219,7 @@ impl TableProvider for ListingTable {
         }
 
         let known_file_ordering = self.try_create_output_ordering()?;
-        let desired_file_ordering = match options.preferred_ordering() {
+        let desired_file_ordering = match args.preferred_ordering() {
             Some(ordering) if !ordering.is_empty() => {
                 // Prefer the ordering requested by the query to any inherint file ordering
                 create_ordering(&self.table_schema, &[ordering.to_vec()])?
@@ -1696,6 +1696,7 @@ mod tests {
 
         let table = load_table(&ctx, "alltypes_plain.parquet").await?;
         let projection = None;
+        #[expect(deprecated)]
         let exec = table
             .scan(&ctx.state(), projection, &[], None)
             .await
@@ -1833,6 +1834,7 @@ mod tests {
         // this will filter out the only file in the store
         let filter = Expr::not_eq(col("p1"), lit("v1"));
 
+        #[expect(deprecated)]
         let scan = table
             .scan(&ctx.state(), None, &[filter], None)
             .await
@@ -2733,6 +2735,7 @@ mod tests {
             .with_schema(schema_default);
         let table_default = ListingTable::try_new(config_default)?;
 
+        #[expect(deprecated)]
         let exec_default = table_default.scan(&state, None, &[], None).await?;
         assert_eq!(
             exec_default.partition_statistics(None)?.num_rows,
@@ -2754,6 +2757,7 @@ mod tests {
             .with_schema(schema_disabled);
         let table_disabled = ListingTable::try_new(config_disabled)?;
 
+        #[expect(deprecated)]
         let exec_disabled = table_disabled.scan(&state, None, &[], None).await?;
         assert_eq!(
             exec_disabled.partition_statistics(None)?.num_rows,
@@ -2773,6 +2777,7 @@ mod tests {
             .with_schema(schema_enabled);
         let table_enabled = ListingTable::try_new(config_enabled)?;
 
+        #[expect(deprecated)]
         let exec_enabled = table_enabled.scan(&state, None, &[], None).await?;
         assert_eq!(
             exec_enabled.partition_statistics(None)?.num_rows,
@@ -2879,6 +2884,7 @@ mod tests {
         assert!(table.schema_adapter_factory().is_none());
 
         // The scan should work correctly with the default schema adapter
+        #[expect(deprecated)]
         let scan_result = table.scan(&ctx.state(), None, &[], None).await;
         assert!(
             scan_result.is_ok(),
@@ -2915,6 +2921,7 @@ mod tests {
         )?;
 
         // The error should bubble up from the scan operation when schema mapping fails
+        #[expect(deprecated)]
         let scan_result = table.scan(&ctx.state(), None, &[], None).await;
 
         assert!(scan_result.is_err());

datafusion/core/src/datasource/memory_test.rs

@@ -59,6 +59,7 @@ mod tests {
         let provider = MemTable::try_new(schema, vec![vec![batch]])?;
 
         // scan with projection
+        #[expect(deprecated)]
         let exec = provider
             .scan(&session_ctx.state(), Some(&vec![2, 1]), &[], None)
             .await?;
@@ -94,6 +95,7 @@ mod tests {
 
         let provider = MemTable::try_new(schema, vec![vec![batch]])?;
 
+        #[expect(deprecated)]
         let exec = provider.scan(&session_ctx.state(), None, &[], None).await?;
         let mut it = exec.execute(0, task_ctx)?;
         let batch1 = it.next().await.unwrap()?;
@@ -126,6 +128,7 @@ mod tests {
 
         let projection: Vec<usize> = vec![0, 4];
 
+        #[expect(deprecated)]
         match provider
             .scan(&session_ctx.state(), Some(&projection), &[], None)
             .await
@@ -254,6 +257,7 @@ mod tests {
         let provider =
             MemTable::try_new(Arc::new(merged_schema), vec![vec![batch1, batch2]])?;
 
+        #[expect(deprecated)]
         let exec = provider.scan(&session_ctx.state(), None, &[], None).await?;
         let mut it = exec.execute(0, task_ctx)?;
         let batch1 = it.next().await.unwrap()?;

datafusion/core/src/physical_planner.rs

@@ -86,7 +86,7 @@ use datafusion_expr::{
 use datafusion_physical_expr::aggregate::{AggregateExprBuilder, AggregateFunctionExpr};
 use datafusion_physical_expr::expressions::{Column, Literal};
 use datafusion_physical_expr::{
-    conjunction, create_physical_sort_exprs, LexOrdering, PhysicalSortExpr,
+    create_physical_sort_exprs, LexOrdering, PhysicalSortExpr,
 };
 use datafusion_physical_optimizer::PhysicalOptimizerRule;
 use datafusion_physical_plan::empty::EmptyExec;

datafusion/core/tests/parquet/file_statistics.rs

@@ -60,6 +60,7 @@ async fn check_stats_precision_with_filter_pushdown() {
     options.execution.parquet.pushdown_filters = true;
 
     // Scan without filter, stats are exact
+    #[expect(deprecated)]
     let exec = table.scan(&state, None, &[], None).await.unwrap();
     assert_eq!(
         exec.partition_statistics(None).unwrap().num_rows,
@@ -71,6 +72,7 @@ async fn check_stats_precision_with_filter_pushdown() {
     // (it is not a partition filter). Therefore; it will be pushed down to the
     // source operator after the appropriate optimizer pass.
     let filter_expr = Expr::gt(col("id"), lit(1));
+    #[expect(deprecated)]
     let exec_with_filter = table
         .scan(&state, None, std::slice::from_ref(&filter_expr), None)
         .await
@@ -119,6 +121,7 @@ async fn load_table_stats_with_session_level_cache() {
 
     //Session 1 first time list files
     assert_eq!(get_static_cache_size(&state1), 0);
+    #[expect(deprecated)]
     let exec1 = table1.scan(&state1, None, &[], None).await.unwrap();
 
     assert_eq!(
@@ -135,6 +138,7 @@ async fn load_table_stats_with_session_level_cache() {
     //Session 2 first time list files
     //check session 1 cache result not show in session 2
     assert_eq!(get_static_cache_size(&state2), 0);
+    #[expect(deprecated)]
     let exec2 = table2.scan(&state2, None, &[], None).await.unwrap();
     assert_eq!(
         exec2.partition_statistics(None).unwrap().num_rows,
@@ -150,6 +154,7 @@ async fn load_table_stats_with_session_level_cache() {
     //Session 1 second time list files
     //check session 1 cache result not show in session 2
     assert_eq!(get_static_cache_size(&state1), 1);
+    #[expect(deprecated)]
     let exec3 = table1.scan(&state1, None, &[], None).await.unwrap();
     assert_eq!(
         exec3.partition_statistics(None).unwrap().num_rows,
@@ -195,6 +200,7 @@ async fn list_files_with_session_level_cache() {
 
     //Session 1 first time list files
     assert_eq!(get_list_file_cache_size(&state1), 0);
+    #[expect(deprecated)]
     let exec1 = table1.scan(&state1, None, &[], None).await.unwrap();
     let data_source_exec = exec1.as_any().downcast_ref::<DataSourceExec>().unwrap();
     let data_source = data_source_exec.data_source();
@@ -211,6 +217,7 @@ async fn list_files_with_session_level_cache() {
     //Session 2 first time list files
     //check session 1 cache result not show in session 2
     assert_eq!(get_list_file_cache_size(&state2), 0);
+    #[expect(deprecated)]
     let exec2 = table2.scan(&state2, None, &[], None).await.unwrap();
     let data_source_exec = exec2.as_any().downcast_ref::<DataSourceExec>().unwrap();
     let data_source = data_source_exec.data_source();
@@ -227,6 +234,7 @@ async fn list_files_with_session_level_cache() {
     //Session 1 second time list files
     //check session 1 cache result not show in session 2
     assert_eq!(get_list_file_cache_size(&state1), 1);
+    #[expect(deprecated)]
     let exec3 = table1.scan(&state1, None, &[], None).await.unwrap();
     let data_source_exec = exec3.as_any().downcast_ref::<DataSourceExec>().unwrap();
     let data_source = data_source_exec.data_source();

datafusion/core/tests/physical_optimizer/partition_statistics.rs

@@ -102,6 +102,7 @@ mod test {
             .downcast_ref::<ListingTable>()
             .unwrap()
             .clone();
+        #[expect(deprecated)]
         listing_table
             .scan(&ctx.state(), None, &[], None)
             .await

datafusion/ffi/src/table_provider.rs

@@ -260,6 +260,7 @@ unsafe extern "C" fn scan_fn_wrapper(
 
         let projections: Vec<_> = projections.into_iter().collect();
 
+        #[expect(deprecated)]
         let plan = rresult_return!(
             internal_provider
                 .scan(&ctx.state(), Some(&projections), &filters, limit.into())