Merge remote-tracking branch 'origin/branch-48-stream' into branch_49_fix

Author: zhuqi-lucas

.github/workflows/audit.yml

@@ -44,4 +44,4 @@ jobs:
       - name: Run audit check
         # Ignored until https://github.com/apache/datafusion/issues/15571
         # ignored py03 warning until arrow 55 upgrade
-        run: cargo audit --ignore RUSTSEC-2024-0370 --ignore RUSTSEC-2025-0020
+        run: cargo audit --ignore RUSTSEC-2024-0370 --ignore RUSTSEC-2025-0020 --ignore RUSTSEC-2025-0047

Cargo.lock

@@ -80,35 +80,9 @@ async fn to_physical_plan_in_one_api_demo(
         displayable(physical_plan.as_ref()).indent(false)
     );
 
-    let traversal = extract_node_ids_from_execution_plan_tree(physical_plan.as_ref());
-    let expected_traversal = vec![
-        Some(0),
-        Some(1),
-        Some(2),
-        Some(3),
-        Some(4),
-        Some(5),
-        Some(6),
-        Some(7),
-        Some(8),
-        Some(9),
-    ];
-    assert_eq!(expected_traversal, traversal);
     Ok(())
 }
 
-fn extract_node_ids_from_execution_plan_tree(
-    physical_plan: &dyn ExecutionPlan,
-) -> Vec<Option<usize>> {
-    let mut traversed_nodes: Vec<Option<usize>> = vec![];
-    for child in physical_plan.children() {
-        let node_ids = extract_node_ids_from_execution_plan_tree(child.as_ref());
-        traversed_nodes.extend(node_ids);
-    }
-    traversed_nodes.push(physical_plan.properties().node_id());
-    traversed_nodes
-}
-
 /// Converts a logical plan into a physical plan by utilizing the analyzer,
 /// optimizer, and query planner APIs separately. This flavor gives more
 /// control over the planning process.

datafusion-examples/examples/planner_api.rs

@@ -224,7 +224,7 @@ impl SortTest {
     /// Sort the input using SortExec and ensure the results are
     /// correct according to `Vec::sort` both with and without spilling
     async fn run(&self) -> (Vec<Vec<RecordBatch>>, Vec<RecordBatch>) {
-        let input = Arc::clone(self.input());
+        let input = self.input.clone();
         let first_batch = input
             .iter()
             .flat_map(|p| p.iter())

datafusion/core/tests/fuzz_cases/sort_fuzz.rs

@@ -165,7 +165,9 @@ async fn page_index_filter_one_col() {
 
     // 5.create filter date_string_col == "01/01/09"`;
     // Note this test doesn't apply type coercion so the literal must match the actual view type
-    let filter = col("date_string_col").eq(lit(ScalarValue::new_utf8view("01/01/09")));
+    // xudong: use new_utf8, because schema_force_view_types was changed to false now.
+    // qi: when schema_force_view_types setting to true, we should change back to utf8view
+    let filter = col("date_string_col").eq(lit(ScalarValue::new_utf8("01/01/09")));
     let batches = get_filter_results(&state, filter.clone(), false).await;
     assert_eq!(batches[0].num_rows(), 14);
 

datafusion/core/tests/parquet/page_pruning.rs

@@ -3612,18 +3612,19 @@ fn test_replace_order_preserving_variants_with_fetch() -> Result<()> {
     );
 
     // Apply the function
-    let result = replace_order_preserving_variants(dist_context)?;
+    let result = replace_order_preserving_variants(dist_context, false)?;
 
     // Verify the plan was transformed to CoalescePartitionsExec
     result
+        .0
         .plan
         .as_any()
         .downcast_ref::<CoalescePartitionsExec>()
         .expect("Expected CoalescePartitionsExec");
 
     // Verify fetch was preserved
     assert_eq!(
-        result.plan.fetch(),
+        result.0.plan.fetch(),
         Some(5),
         "Fetch value was not preserved after transformation"
     );

datafusion/core/tests/physical_optimizer/enforce_distribution.rs

@@ -252,14 +252,14 @@ impl ExecutionPlan for DataSinkExec {
 
     fn with_node_id(
         self: Arc<Self>,
-        _node_id: usize,
+        node_id: usize,
     ) -> Result<Option<Arc<dyn ExecutionPlan>>> {
         let mut new_plan = DataSinkExec::new(
             Arc::clone(self.input()),
             Arc::clone(&self.sink),
             self.sort_order.clone(),
         );
-        let new_props = new_plan.cache.clone().with_node_id(_node_id);
+        let new_props = new_plan.cache.clone().with_node_id(node_id);
         new_plan.cache = new_props;
         Ok(Some(Arc::new(new_plan)))
     }

datafusion/datasource/src/sink.rs

@@ -362,10 +362,10 @@ impl ExecutionPlan for DataSourceExec {
 
     fn with_node_id(
         self: Arc<Self>,
-        _node_id: usize,
+        node_id: usize,
     ) -> Result<Option<Arc<dyn ExecutionPlan>>> {
-        let mut new_plan = DataSourceExec::new(self.data_source.clone());
-        let new_props = new_plan.cache.clone().with_node_id(_node_id);
+        let mut new_plan = DataSourceExec::new(Arc::clone(&self.data_source));
+        let new_props = new_plan.cache.clone().with_node_id(node_id);
         new_plan.cache = new_props;
         Ok(Some(Arc::new(new_plan)))
     }

datafusion/datasource/src/source.rs

@@ -61,7 +61,6 @@ pub mod push_down_limit;
 pub mod replace_distinct_aggregate;
 pub mod scalar_subquery_to_join;
 pub mod simplify_expressions;
-mod simplify_predicates;
 pub mod single_distinct_to_groupby;
 pub mod utils;
 

datafusion/optimizer/src/lib.rs

@@ -41,7 +41,7 @@ use datafusion_expr::{
 
 use crate::optimizer::ApplyOrder;
 use crate::utils::{has_all_column_refs, is_restrict_null_predicate};
-use crate::{simplify_predicates::simplify_predicates, OptimizerConfig, OptimizerRule};
+use crate::{simplify_expressions::simplify_predicates, OptimizerConfig, OptimizerRule};
 
 /// Optimizer rule for pushing (moving) filter expressions down in a plan so
 /// they are applied as early as possible.
@@ -2311,7 +2311,7 @@ mod tests {
             plan,
             @r"
         Projection: test.a, test1.d
-          Cross Join:
+          Cross Join: 
             Projection: test.a, test.b, test.c
               TableScan: test, full_filters=[test.a = Int32(1)]
             Projection: test1.d, test1.e, test1.f
@@ -2341,7 +2341,7 @@ mod tests {
             plan,
             @r"
         Projection: test.a, test1.a
-          Cross Join:
+          Cross Join: 
             Projection: test.a, test.b, test.c
               TableScan: test, full_filters=[test.a = Int32(1)]
             Projection: test1.a, test1.b, test1.c