Implement crude stats for mem

Author: ThomasAWalker

datafusion/core/src/datasource/memory.rs

@@ -37,9 +37,11 @@ use crate::physical_planner::create_physical_sort_exprs;
 use arrow::datatypes::SchemaRef;
 use arrow::record_batch::RecordBatch;
 use datafusion_catalog::Session;
-use datafusion_common::{not_impl_err, plan_err, Constraints, DFSchema, SchemaExt};
+use datafusion_common::stats::Precision;
+use datafusion_common::{not_impl_err, plan_err, Constraints, DFSchema, SchemaExt, Statistics};
 pub use datafusion_datasource::memory::MemorySourceConfig;
 pub use datafusion_datasource::source::DataSourceExec;
+
 use datafusion_execution::TaskContext;
 use datafusion_expr::dml::InsertOp;
 use datafusion_expr::SortExpr;
@@ -67,11 +69,13 @@ pub struct MemTable {
     /// Optional pre-known sort order(s). Must be `SortExpr`s.
     /// inserting data into this table removes the order
     pub sort_order: Arc<Mutex<Vec<Vec<SortExpr>>>>,
+    num_rows: usize,
 }
 
 impl MemTable {
     /// Create a new in-memory table from the provided schema and record batches
     pub fn try_new(schema: SchemaRef, partitions: Vec<Vec<RecordBatch>>) -> Result<Self> {
+        let mut num_rows = 0_usize;
         for batches in partitions.iter().flatten() {
             let batches_schema = batches.schema();
             if !schema.contains(&batches_schema) {
@@ -81,6 +85,7 @@ impl MemTable {
                 );
                 return plan_err!("Mismatch between schema and batches");
             }
+            num_rows += batches.num_rows();
         }
 
         Ok(Self {
@@ -92,6 +97,7 @@ impl MemTable {
             constraints: Constraints::empty(),
             column_defaults: HashMap::new(),
             sort_order: Arc::new(Mutex::new(vec![])),
+            num_rows
         })
     }
 
@@ -215,6 +221,12 @@ impl TableProvider for MemTable {
         TableType::Base
     }
 
+    fn statistics(&self) -> Option<Statistics> {
+        let mut stats = Statistics::new_unknown(&self.schema);
+        stats.num_rows = Precision::Inexact(self.num_rows);
+        Some(stats)
+    }
+
     async fn scan(
         &self,
         state: &dyn Session,