resolve conflicts

Author: xudong963

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

@@ -57,7 +57,6 @@ use datafusion_common::stats::Precision;
 use datafusion_datasource::compute_all_files_statistics;
 use datafusion_datasource::file_groups::FileGroup;
 use datafusion_physical_expr_common::sort_expr::LexRequirement;
-use datafusion_physical_plan::statistics::add_row_stats;
 use futures::{future, stream, Stream, StreamExt, TryStreamExt};
 use itertools::Itertools;
 use object_store::ObjectStore;

datafusion/datasource/src/source.rs

@@ -25,6 +25,7 @@ use std::sync::Arc;
 use datafusion_physical_plan::execution_plan::{Boundedness, EmissionType};
 use datafusion_physical_plan::metrics::{ExecutionPlanMetricsSet, MetricsSet};
 use datafusion_physical_plan::projection::ProjectionExec;
+use datafusion_physical_plan::statistics::PartitionedStatistics;
 use datafusion_physical_plan::{
     DisplayAs, DisplayFormatType, ExecutionPlan, PlanProperties,
 };
@@ -188,9 +189,7 @@ impl ExecutionPlan for DataSourceExec {
         self.data_source.statistics()
     }
 
-    fn statistics_by_partition(
-        &self,
-    ) -> datafusion_common::Result<PartitionedStatistics> {
+    fn statistics_by_partition(&self) -> Result<PartitionedStatistics> {
         let mut statistics = {
             let mut v =
                 Vec::with_capacity(self.properties().partitioning.partition_count());

datafusion/datasource/src/statistics.rs

@@ -36,9 +36,6 @@ use datafusion_common::stats::Precision;
 use datafusion_common::{plan_datafusion_err, plan_err, DataFusionError, Result};
 use datafusion_physical_expr::{expressions::Column, PhysicalSortExpr};
 use datafusion_physical_expr_common::sort_expr::LexOrdering;
-pub(crate) use datafusion_physical_plan::statistics::{
-    add_row_stats, compute_summary_statistics, set_max_if_greater, set_min_if_lesser,
-};
 use datafusion_physical_plan::{ColumnStatistics, Statistics};
 
 /// A normalized representation of file min/max statistics that allows for efficient sorting & comparison.
@@ -479,7 +476,7 @@ pub fn compute_all_files_statistics(
     // Then summary statistics across all file groups
     let file_groups_statistics = file_groups_with_stats
         .iter()
-        .filter_map(|file_group| file_group.statistics());
+        .filter_map(|file_group| file_group.statistics().as_deref());
 
     let mut statistics =
         Statistics::try_merge_iter(file_groups_statistics, &table_schema)?;

datafusion/physical-optimizer/src/enforce_sorting/replace_with_order_preserving_variants.rs

@@ -34,7 +34,7 @@ use datafusion_physical_plan::execution_plan::EmissionType;
 use datafusion_physical_plan::repartition::RepartitionExec;
 use datafusion_physical_plan::sorts::sort_preserving_merge::SortPreservingMergeExec;
 use datafusion_physical_plan::tree_node::PlanContext;
-use datafusion_physical_plan::{ExecutionPlan, ExecutionPlanProperties};
+use datafusion_physical_plan::ExecutionPlanProperties;
 
 use itertools::izip;
 
@@ -205,9 +205,8 @@ pub fn plan_with_order_breaking_variants(
         // Replace `SortPreservingMergeExec` with a `CoalescePartitionsExec`
         // SPM may have `fetch`, so pass it to the `CoalescePartitionsExec`
         let child = Arc::clone(&sort_input.children[0].plan);
-        let coalesce = CoalescePartitionsExec::new(child)
-            .with_fetch(plan.fetch())
-            .unwrap();
+        let coalesce =
+            Arc::new(CoalescePartitionsExec::new(child).with_fetch(plan.fetch()));
         sort_input.plan = coalesce;
     } else {
         return sort_input.update_plan_from_children();

datafusion/physical-plan/src/coalesce_batches.rs

@@ -24,6 +24,7 @@ use std::task::{Context, Poll};
 
 use super::metrics::{BaselineMetrics, ExecutionPlanMetricsSet, MetricsSet};
 use super::{DisplayAs, ExecutionPlanProperties, PlanProperties, Statistics};
+use crate::statistics::PartitionedStatistics;
 use crate::{
     DisplayFormatType, ExecutionPlan, RecordBatchStream, SendableRecordBatchStream,
 };