fix streaming sql

Author: zhuqi-lucas

datafusion/datasource/src/file_scan_config.rs

@@ -595,8 +595,11 @@ impl DataSource for FileScanConfig {
                         &mut eq_properties,
                         &schema,
                     ) {
-                        Ok(()) => {}
+                        Ok(()) => {
+                            println!("Added filter equivalence info: {eq_properties:?} ");
+                        }
                         Err(e) => {
+                            println!("Fail to add filter equivalence info: {e}");
                             warn!("Failed to add filter equivalence info: {e}");
                             #[cfg(debug_assertions)]
                             panic!("Failed to add filter equivalence info: {e}");
@@ -764,10 +767,12 @@ impl FileScanConfig {
         eq_properties: &mut EquivalenceProperties,
         schema: &Schema,
     ) -> Result<()> {
+        println!("Adding filter equivalence info from filter: {filter:?} ");
         macro_rules! ignore_dangling_col {
             ($col:expr) => {
                 if let Some(col) = $col.as_any().downcast_ref::<Column>() {
                     if schema.index_of(col.name()).is_err() {
+                        print!("error ignoring dangling column {}", col.name());
                         continue;
                     }
                 }

datafusion/datasource/src/source.rs

@@ -357,6 +357,7 @@ impl ExecutionPlan for DataSourceExec {
                 let mut new_node = self.clone();
                 new_node.data_source = data_source;
                 // Re-compute properties since we have new filters which will impact equivalence info
+                println!("Recomputing properties for DataSourceExec after filter pushdown");
                 new_node.cache =
                     Self::compute_properties(Arc::clone(&new_node.data_source));