fix

Author: zhuqi-lucas

datafusion/datasource-parquet/src/opener.rs

@@ -495,11 +495,11 @@ impl FileOpener for ParquetOpener {
 
             // If reverse scanning is enabled, reverse the prepared plan
             if reverse_row_groups {
-                info!("reversing parquet file scan for file {}", file_name);
+                info!("reversing parquet file scan for file {file_name}");
                 prepared_plan = prepared_plan.reverse(file_metadata.as_ref())?;
             }
 
-            info!("parquet file scan for file {}", file_name);
+            info!("parquet file scan for file {file_name}");
 
             // Apply the prepared plan to the builder
             builder = prepared_plan.apply_to_builder(builder);

datafusion/datasource-parquet/src/source.rs

@@ -55,9 +55,8 @@ use datafusion_physical_plan::SortOrderPushdownResult;
 
 #[cfg(feature = "parquet_encryption")]
 use datafusion_execution::parquet_encryption::EncryptionFactory;
-use datafusion_physical_expr_common::sort_expr::{LexOrdering, PhysicalSortExpr};
+use datafusion_physical_expr_common::sort_expr::PhysicalSortExpr;
 use itertools::Itertools;
-use log::info;
 use object_store::ObjectStore;
 #[cfg(feature = "parquet_encryption")]
 use parquet::encryption::decrypt::FileDecryptionProperties;
@@ -381,7 +380,6 @@ impl ParquetSource {
         self
     }
 
-
     /// Return the value described in [`Self::with_pushdown_filters`]
     pub(crate) fn pushdown_filters(&self) -> bool {
         self.table_parquet_options.global.pushdown_filters

datafusion/datasource/src/file_scan_config.rs

@@ -66,7 +66,7 @@ use datafusion_physical_plan::execution_plan::SchedulingType;
 use datafusion_physical_plan::projection::{
     all_alias_free_columns, new_projections_for_columns,
 };
-use log::{debug, info, warn};
+use log::{debug, warn};
 use std::borrow::Cow;
 use std::collections::HashMap;
 use std::marker::PhantomData;

datafusion/datasource/src/source.rs

@@ -22,28 +22,28 @@ use std::fmt;
 use std::fmt::{Debug, Formatter};
 use std::sync::Arc;
 
+use crate::file_scan_config::FileScanConfig;
+use datafusion_common::config::ConfigOptions;
+use datafusion_common::{Constraints, Result, Statistics};
+use datafusion_execution::{SendableRecordBatchStream, TaskContext};
+use datafusion_physical_expr::{EquivalenceProperties, Partitioning, PhysicalExpr};
+use datafusion_physical_expr_common::sort_expr::{LexOrdering, PhysicalSortExpr};
 use datafusion_physical_plan::execution_plan::{
     Boundedness, EmissionType, SchedulingType,
 };
+use datafusion_physical_plan::filter_pushdown::{
+    ChildPushdownResult, FilterPushdownPhase, FilterPushdownPropagation, PushedDown,
+};
 use datafusion_physical_plan::metrics::SplitMetrics;
 use datafusion_physical_plan::metrics::{ExecutionPlanMetricsSet, MetricsSet};
 use datafusion_physical_plan::projection::{ProjectionExec, ProjectionExpr};
 use datafusion_physical_plan::stream::BatchSplitStream;
+use datafusion_physical_plan::SortOrderPushdownResult;
 use datafusion_physical_plan::{
     DisplayAs, DisplayFormatType, ExecutionPlan, PlanProperties,
 };
 use itertools::Itertools;
 use log::info;
-use crate::file_scan_config::FileScanConfig;
-use datafusion_common::config::ConfigOptions;
-use datafusion_common::{Constraints, Result, Statistics};
-use datafusion_execution::{SendableRecordBatchStream, TaskContext};
-use datafusion_physical_expr::{EquivalenceProperties, Partitioning, PhysicalExpr};
-use datafusion_physical_expr_common::sort_expr::{LexOrdering, PhysicalSortExpr};
-use datafusion_physical_plan::filter_pushdown::{
-    ChildPushdownResult, FilterPushdownPhase, FilterPushdownPropagation, PushedDown,
-};
-use datafusion_physical_plan::SortOrderPushdownResult;
 
 /// A source of data, typically a list of files or memory
 ///
@@ -396,7 +396,7 @@ impl ExecutionPlan for DataSourceExec {
         &self,
         order: &[PhysicalSortExpr],
     ) -> Result<SortOrderPushdownResult<Arc<dyn ExecutionPlan>>> {
-        info!("trying to pushdown sort in DataSource: {:?}", order);
+        info!("trying to pushdown sort in DataSource: {order:?}");
         // Delegate to the data source and wrap result with DataSourceExec
         self.data_source
             .try_pushdown_sort(order)?

datafusion/physical-optimizer/src/pushdown_sort.rs

@@ -57,7 +57,6 @@ use datafusion_physical_plan::sorts::sort::SortExec;
 use datafusion_physical_plan::ExecutionPlan;
 use datafusion_physical_plan::SortOrderPushdownResult;
 use std::sync::Arc;
-use log::info;
 
 /// A PhysicalOptimizerRule that attempts to push down sort requirements to data sources.
 ///
@@ -92,8 +91,6 @@ impl PhysicalOptimizerRule for PushdownSort {
             let sort_input = Arc::clone(sort_exec.input());
             let required_ordering = sort_exec.expr();
 
-            info!("trying to pushdown sort: {:?}", required_ordering);
-
             // Try to push the sort requirement down through the plan tree
             // Each node type defines its own pushdown behavior via try_pushdown_sort()
             match sort_input.try_pushdown_sort(required_ordering)? {
@@ -102,7 +99,6 @@ impl PhysicalOptimizerRule for PushdownSort {
                     Ok(Transformed::yes(inner))
                 }
                 SortOrderPushdownResult::Inexact { inner } => {
-                    info!("inexact pushdown sort : {:?}", required_ordering);
                     // Data source is optimized for the ordering but not perfectly sorted
                     // Keep the Sort operator but use the optimized input
                     // Benefits: TopK queries can terminate early, better cache locality
@@ -115,7 +111,6 @@ impl PhysicalOptimizerRule for PushdownSort {
                     )))
                 }
                 SortOrderPushdownResult::Unsupported => {
-                    info!("unsupported pushdown sort : {:?}", required_ordering);
                     // Cannot optimize for this ordering - no change
                     Ok(Transformed::no(plan))
                 }

datafusion/physical-plan/src/coop.rs

@@ -79,17 +79,20 @@ use crate::filter_pushdown::{
     ChildPushdownResult, FilterDescription, FilterPushdownPhase,
     FilterPushdownPropagation,
 };
-use crate::{DisplayAs, DisplayFormatType, ExecutionPlan, PlanProperties, RecordBatchStream, SendableRecordBatchStream, SortOrderPushdownResult};
+use crate::{
+    DisplayAs, DisplayFormatType, ExecutionPlan, PlanProperties, RecordBatchStream,
+    SendableRecordBatchStream, SortOrderPushdownResult,
+};
 use arrow::record_batch::RecordBatch;
 use arrow_schema::Schema;
 use datafusion_common::{internal_err, Result, Statistics};
 use datafusion_execution::TaskContext;
 
 use crate::execution_plan::SchedulingType;
 use crate::stream::RecordBatchStreamAdapter;
+use datafusion_physical_expr_common::sort_expr::PhysicalSortExpr;
 use futures::{Stream, StreamExt};
 use log::info;
-use datafusion_physical_expr_common::sort_expr::PhysicalSortExpr;
 
 /// A stream that passes record batches through unchanged while cooperating with the Tokio runtime.
 /// It consumes cooperative scheduling budget for each returned [`RecordBatch`],
@@ -313,13 +316,14 @@ impl ExecutionPlan for CooperativeExec {
         Ok(FilterPushdownPropagation::if_all(child_pushdown_result))
     }
 
-    fn try_pushdown_sort(&self, order: &[PhysicalSortExpr]) -> Result<SortOrderPushdownResult<Arc<dyn ExecutionPlan>>> {
-        info!("CooperativeExec trying to pushdown sort: {:?}", order);
-        self.input
-            .try_pushdown_sort(order)?
-            .try_map(|inner| {
-                Ok(Arc::new(CooperativeExec::new(inner)) as Arc<dyn ExecutionPlan>)
-            })
+    fn try_pushdown_sort(
+        &self,
+        order: &[PhysicalSortExpr],
+    ) -> Result<SortOrderPushdownResult<Arc<dyn ExecutionPlan>>> {
+        info!("CooperativeExec trying to pushdown sort: {order:?}");
+        self.input.try_pushdown_sort(order)?.try_map(|inner| {
+            Ok(Arc::new(CooperativeExec::new(inner)) as Arc<dyn ExecutionPlan>)
+        })
     }
 }
 

datafusion/physical-plan/src/filter.rs

@@ -24,7 +24,7 @@ use itertools::Itertools;
 
 use super::{
     ColumnStatistics, DisplayAs, ExecutionPlanProperties, PlanProperties,
-    RecordBatchStream, SendableRecordBatchStream, Statistics,
+    RecordBatchStream, SendableRecordBatchStream, SortOrderPushdownResult, Statistics,
 };
 use crate::common::can_project;
 use crate::execution_plan::CardinalityEffect;
@@ -63,6 +63,7 @@ use datafusion_physical_expr::{
 };
 
 use datafusion_physical_expr_common::physical_expr::fmt_sql;
+use datafusion_physical_expr_common::sort_expr::PhysicalSortExpr;
 use futures::stream::{Stream, StreamExt};
 use log::trace;
 
@@ -570,6 +571,25 @@ impl ExecutionPlan for FilterExec {
             updated_node,
         })
     }
+    fn try_pushdown_sort(
+        &self,
+        order: &[PhysicalSortExpr],
+    ) -> Result<SortOrderPushdownResult<Arc<dyn ExecutionPlan>>> {
+        let child = self.input();
+        match child.try_pushdown_sort(order)? {
+            SortOrderPushdownResult::Exact { inner } => {
+                let new_exec = Arc::new(self.clone()).with_new_children(vec![inner])?;
+                Ok(SortOrderPushdownResult::Exact { inner: new_exec })
+            }
+            SortOrderPushdownResult::Inexact { inner } => {
+                let new_exec = Arc::new(self.clone()).with_new_children(vec![inner])?;
+                Ok(SortOrderPushdownResult::Inexact { inner: new_exec })
+            }
+            SortOrderPushdownResult::Unsupported => {
+                Ok(SortOrderPushdownResult::Unsupported)
+            }
+        }
+    }
 }
 
 impl EmbeddedProjection for FilterExec {

datafusion/physical-plan/src/projection.rs

@@ -22,7 +22,10 @@
 
 use super::expressions::{Column, Literal};
 use super::metrics::{BaselineMetrics, ExecutionPlanMetricsSet, MetricsSet};
-use super::{DisplayAs, ExecutionPlanProperties, PlanProperties, RecordBatchStream, SendableRecordBatchStream, SortOrderPushdownResult, Statistics};
+use super::{
+    DisplayAs, ExecutionPlanProperties, PlanProperties, RecordBatchStream,
+    SendableRecordBatchStream, SortOrderPushdownResult, Statistics,
+};
 use crate::execution_plan::CardinalityEffect;
 use crate::filter_pushdown::{
     ChildPushdownResult, FilterDescription, FilterPushdownPhase,
@@ -47,7 +50,9 @@ use datafusion_execution::TaskContext;
 use datafusion_physical_expr::equivalence::ProjectionMapping;
 use datafusion_physical_expr::utils::collect_columns;
 use datafusion_physical_expr_common::physical_expr::{fmt_sql, PhysicalExprRef};
-use datafusion_physical_expr_common::sort_expr::{LexOrdering, LexRequirement, PhysicalSortExpr};
+use datafusion_physical_expr_common::sort_expr::{
+    LexOrdering, LexRequirement, PhysicalSortExpr,
+};
 // Re-exported from datafusion-physical-expr for backwards compatibility
 // We recommend updating your imports to use datafusion-physical-expr directly
 pub use datafusion_physical_expr::projection::{
@@ -344,19 +349,18 @@ impl ExecutionPlan for ProjectionExec {
         Ok(FilterPushdownPropagation::if_all(child_pushdown_result))
     }
 
-    fn try_pushdown_sort(&self, order: &[PhysicalSortExpr]) -> Result<SortOrderPushdownResult<Arc<dyn ExecutionPlan>>> {
+    fn try_pushdown_sort(
+        &self,
+        order: &[PhysicalSortExpr],
+    ) -> Result<SortOrderPushdownResult<Arc<dyn ExecutionPlan>>> {
         let child = self.input();
-
-        println!("try pushdown_sort in ProjectionExec: {:?}", order);
         match child.try_pushdown_sort(order)? {
             SortOrderPushdownResult::Exact { inner } => {
-                let new_exec = Arc::new(self.clone())
-                    .with_new_children(vec![inner])?;
+                let new_exec = Arc::new(self.clone()).with_new_children(vec![inner])?;
                 Ok(SortOrderPushdownResult::Exact { inner: new_exec })
             }
             SortOrderPushdownResult::Inexact { inner } => {
-                let new_exec = Arc::new(self.clone())
-                    .with_new_children(vec![inner])?;
+                let new_exec = Arc::new(self.clone()).with_new_children(vec![inner])?;
                 Ok(SortOrderPushdownResult::Inexact { inner: new_exec })
             }
             SortOrderPushdownResult::Unsupported => {