[BLAZE-805] Support statistics of ExecutionPlan for SortExec

native-engine/datafusion-ext-plans/src/sort_exec.rs

@@ -185,7 +185,7 @@ impl ExecutionPlan for SortExec {
     }
 
     fn statistics(&self) -> Result<Statistics> {
-        todo!()
+        Statistics::with_fetch(self.input.statistics()?, self.schema(), self.fetch, 0, 1)
     }
 }
 
@@ -1393,9 +1393,9 @@ mod fuzztest {
         record_batch::RecordBatch,
     };
     use datafusion::{
-        common::Result,
+        common::{stats::Precision, Result},
         physical_expr::{expressions::Column, PhysicalSortExpr},
-        physical_plan::memory::MemoryExec,
+        physical_plan::{memory::MemoryExec, ExecutionPlan},
         prelude::{SessionConfig, SessionContext},
     };
 
@@ -1460,8 +1460,9 @@ mod fuzztest {
             None,
         )?);
         let sort = Arc::new(SortExec::new(input, sort_exprs.clone(), None));
-        let output = datafusion::physical_plan::collect(sort, task_ctx.clone()).await?;
+        let output = datafusion::physical_plan::collect(sort.clone(), task_ctx.clone()).await?;
         let a = concat_batches(&schema, &output)?;
+        let a_row_count = sort.clone().statistics()?.num_rows;
 
         let input = Arc::new(MemoryExec::try_new(
             &[batches.clone()],
@@ -1472,10 +1473,13 @@ mod fuzztest {
             sort_exprs.clone(),
             input,
         ));
-        let output = datafusion::physical_plan::collect(sort, task_ctx.clone()).await?;
+        let output = datafusion::physical_plan::collect(sort.clone(), task_ctx.clone()).await?;
         let b = concat_batches(&schema, &output)?;
+        let b_row_count = sort.clone().statistics()?.num_rows;
 
         assert_eq!(a.num_rows(), b.num_rows());
+        assert_eq!(a_row_count, b_row_count);
+        assert_eq!(a_row_count, Precision::Exact(n));
         assert!(a == b);
         Ok(())
     }