feat: table metrics (#1469)

* feat: Statistic

* add todo

* fmt: cargo fmt

* feat: some simple impl for MemTable

* chore: a try on adding statistics

* Update src/table/src/stats.rs

Co-authored-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com>

* docs: fix typo

* newlines unnecessary

---------

Co-authored-by: Ruihang Xia <waynestxia@gmail.com>
Co-authored-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com>

src/catalog/src/system.rs

@@ -80,6 +80,10 @@ impl Table for SystemCatalogTable {
     async fn delete(&self, request: DeleteRequest) -> table::Result<usize> {
         self.0.delete(request).await
     }
+
+    fn statistics(&self) -> Option<table::stats::TableStatistics> {
+        self.0.statistics()
+    }
 }
 
 impl SystemCatalogTable {

src/mito/src/table.rs

@@ -38,7 +38,6 @@ use store_api::storage::{
     AddColumn, AlterOperation, AlterRequest, ChunkReader, FlushContext, ReadContext, Region,
     RegionMeta, RegionNumber, ScanRequest, SchemaRef, Snapshot, WriteContext, WriteRequest,
 };
-use table::error as table_error;
 use table::error::{
     InvalidTableSnafu, RegionSchemaMismatchSnafu, Result as TableResult, TableOperationSnafu,
 };
@@ -49,7 +48,8 @@ use table::requests::{
     AddColumnRequest, AlterKind, AlterTableRequest, DeleteRequest, InsertRequest,
 };
 use table::table::scan::SimpleTableScan;
-use table::table::{AlterContext, RegionStat, Table};
+use table::table::{AlterContext, Table};
+use table::{error as table_error, RegionStat};
 use tokio::sync::Mutex;
 
 use crate::error;

src/store-api/src/storage.rs

@@ -34,7 +34,7 @@ pub use self::chunk::{Chunk, ChunkReader};
 pub use self::descriptors::*;
 pub use self::engine::{CreateOptions, EngineContext, OpenOptions, StorageEngine};
 pub use self::metadata::RegionMeta;
-pub use self::region::{FlushContext, Region, WriteContext};
+pub use self::region::{FlushContext, Region, RegionStat, WriteContext};
 pub use self::requests::{
     AddColumn, AlterOperation, AlterRequest, GetRequest, ScanRequest, WriteRequest,
 };

src/store-api/src/storage/region.rs

@@ -77,10 +77,23 @@ pub trait Region: Send + Sync + Clone + std::fmt::Debug + 'static {
 
     fn disk_usage_bytes(&self) -> u64;
 
+    fn region_stat(&self) -> RegionStat {
+        RegionStat {
+            region_id: self.id(),
+            disk_usage_bytes: self.disk_usage_bytes(),
+        }
+    }
+
     /// Flush memtable of the region to disk.
     async fn flush(&self, ctx: &FlushContext) -> Result<(), Self::Error>;
 }
 
+#[derive(Default, Debug)]
+pub struct RegionStat {
+    pub region_id: u64,
+    pub disk_usage_bytes: u64,
+}
+
 /// Context for write operations.
 #[derive(Debug, Clone, Default)]
 pub struct WriteContext {}

src/table/src/lib.rs

@@ -18,8 +18,12 @@ pub mod error;
 pub mod metadata;
 pub mod predicate;
 pub mod requests;
+pub mod stats;
 pub mod table;
 pub mod test_util;
 
+pub use store_api::storage::RegionStat;
+
 pub use crate::error::{Error, Result};
+pub use crate::stats::{ColumnStatistics, TableStatistics};
 pub use crate::table::{Table, TableRef};

src/table/src/stats.rs

@@ -0,0 +1,47 @@
+// Copyright 2023 Greptime Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use datatypes::value::Value;
+
+/// Statistics for a relation
+/// Fields are optional and can be inexact because the sources
+/// sometimes provide approximate estimates for performance reasons
+/// and the transformations output are not always predictable.
+#[derive(Debug, Clone, Default, PartialEq, Eq)]
+pub struct TableStatistics {
+    /// The number of table rows
+    pub num_rows: Option<usize>,
+    /// total bytes of the table rows
+    pub total_byte_size: Option<usize>,
+    /// Statistics on a column level
+    pub column_statistics: Option<Vec<ColumnStatistics>>,
+    /// If true, any field that is `Some(..)` is the actual value in the data provided by the operator (it is not
+    /// an estimation). Any or all other fields might still be None, in which case no information is known.
+    /// if false, any field that is `Some(..)` may contain an inexact estimate and may not be the actual value.
+    pub is_exact: bool,
+}
+
+/// Statistics for a column within a relation
+#[derive(Clone, Debug, Default, PartialEq, Eq)]
+pub struct ColumnStatistics {
+    /// Number of null values on column
+    pub null_count: Option<usize>,
+    /// Maximum value of column
+    pub max_value: Option<Value>,
+    /// Minimum value of column
+    pub min_value: Option<Value>,
+    /// Number of distinct values
+    pub distinct_count: Option<usize>,
+    // TODO(discord9): histogram of values
+}

src/table/src/table.rs

@@ -28,6 +28,8 @@ use store_api::storage::RegionNumber;
 use crate::error::{Result, UnsupportedSnafu};
 use crate::metadata::{FilterPushDownType, TableId, TableInfoRef, TableType};
 use crate::requests::{AlterTableRequest, DeleteRequest, InsertRequest};
+use crate::stats::TableStatistics;
+use crate::RegionStat;
 
 pub type AlterContext = anymap::Map<dyn Any + Send + Sync>;
 
@@ -117,6 +119,11 @@ pub trait Table: Send + Sync {
         }
         .fail()?
     }
+
+    /// Get statistics for this table, if available
+    fn statistics(&self) -> Option<TableStatistics> {
+        None
+    }
 }
 
 pub type TableRef = Arc<dyn Table>;
@@ -127,9 +134,3 @@ pub trait TableIdProvider {
 }
 
 pub type TableIdProviderRef = Arc<dyn TableIdProvider + Send + Sync>;
-
-#[derive(Default, Debug)]
-pub struct RegionStat {
-    pub region_id: u64,
-    pub disk_usage_bytes: u64,
-}

src/table/src/test_util/memtable.rs

@@ -35,7 +35,7 @@ use crate::metadata::{
     TableId, TableInfoBuilder, TableInfoRef, TableMetaBuilder, TableType, TableVersion,
 };
 use crate::table::scan::SimpleTableScan;
-use crate::Table;
+use crate::{ColumnStatistics, Table, TableStatistics};
 
 #[derive(Debug, Clone)]
 pub struct MemTable {
@@ -172,6 +172,30 @@ impl Table for MemTable {
             recordbatch: Some(recordbatch),
         }))))
     }
+
+    fn statistics(&self) -> Option<TableStatistics> {
+        let df_recordbatch = self.recordbatch.df_record_batch();
+        let num_rows = df_recordbatch.num_rows();
+        let total_byte_size = df_recordbatch.get_array_memory_size();
+        let column_statistics: Vec<_> = df_recordbatch
+            .columns()
+            .iter()
+            .map(|col| {
+                let null_count = col.null_count();
+                ColumnStatistics {
+                    null_count: Some(null_count),
+                    // TODO(discord9): implement more statistics
+                    ..Default::default()
+                }
+            })
+            .collect();
+        Some(TableStatistics {
+            num_rows: Some(num_rows),
+            total_byte_size: Some(total_byte_size),
+            column_statistics: Some(column_statistics),
+            is_exact: true,
+        })
+    }
 }
 
 impl RecordBatchStream for MemtableStream {