feat: impl partitions and region_peers information schema

src/catalog/src/error.rs

@@ -164,6 +164,15 @@ pub enum Error {
         location: Location,
     },
 
+    #[snafu(display("Failed to find table partitions: #{table}"))]
+    FindPartitions {
+        source: partition::error::Error,
+        table: String,
+    },
+
+    #[snafu(display("Failed to find region routes"))]
+    FindRegionRoutes { source: partition::error::Error },
+
     #[snafu(display("Failed to read system catalog table records"))]
     ReadSystemCatalog {
         location: Location,
@@ -254,11 +263,14 @@ impl ErrorExt for Error {
         match self {
             Error::InvalidKey { .. }
             | Error::SchemaNotFound { .. }
-            | Error::TableNotFound { .. }
             | Error::CatalogNotFound { .. }
+            | Error::FindPartitions { .. }
+            | Error::FindRegionRoutes { .. }
             | Error::InvalidEntryType { .. }
             | Error::ParallelOpenTable { .. } => StatusCode::Unexpected,
 
+            Error::TableNotFound { .. } => StatusCode::TableNotFound,
+
             Error::SystemCatalog { .. }
             | Error::EmptyValue { .. }
             | Error::ValueDeserialize { .. } => StatusCode::StorageUnavailable,

src/catalog/src/information_schema.rs

@@ -15,7 +15,9 @@
 mod columns;
 mod key_column_usage;
 mod memory_table;
+mod partitions;
 mod predicate;
+mod region_peers;
 mod runtime_metrics;
 mod schemata;
 mod table_names;
@@ -47,6 +49,8 @@ use self::columns::InformationSchemaColumns;
 use crate::error::Result;
 use crate::information_schema::key_column_usage::InformationSchemaKeyColumnUsage;
 use crate::information_schema::memory_table::{get_schema_columns, MemoryTable};
+use crate::information_schema::partitions::InformationSchemaPartitions;
+use crate::information_schema::region_peers::InformationSchemaRegionPeers;
 use crate::information_schema::runtime_metrics::InformationSchemaMetrics;
 use crate::information_schema::schemata::InformationSchemaSchemata;
 use crate::information_schema::tables::InformationSchemaTables;
@@ -74,6 +78,7 @@ lazy_static! {
         TRIGGERS,
         GLOBAL_STATUS,
         SESSION_STATUS,
+        PARTITIONS,
     ];
 }
 
@@ -156,6 +161,10 @@ impl InformationSchemaProvider {
                 BUILD_INFO.to_string(),
                 self.build_table(BUILD_INFO).unwrap(),
             );
+            tables.insert(
+                REGION_PEERS.to_string(),
+                self.build_table(REGION_PEERS).unwrap(),
+            );
         }
 
         tables.insert(TABLES.to_string(), self.build_table(TABLES).unwrap());
@@ -226,6 +235,14 @@ impl InformationSchemaProvider {
                 self.catalog_manager.clone(),
             )) as _),
             RUNTIME_METRICS => Some(Arc::new(InformationSchemaMetrics::new())),
+            PARTITIONS => Some(Arc::new(InformationSchemaPartitions::new(
+                self.catalog_name.clone(),
+                self.catalog_manager.clone(),
+            )) as _),
+            REGION_PEERS => Some(Arc::new(InformationSchemaRegionPeers::new(
+                self.catalog_name.clone(),
+                self.catalog_manager.clone(),
+            )) as _),
             _ => None,
         }
     }

src/catalog/src/information_schema/columns.rs

@@ -58,6 +58,7 @@ const COLUMN_DEFAULT: &str = "column_default";
 const IS_NULLABLE: &str = "is_nullable";
 const COLUMN_TYPE: &str = "column_type";
 const COLUMN_COMMENT: &str = "column_comment";
+const INIT_CAPACITY: usize = 42;
 
 impl InformationSchemaColumns {
     pub(super) fn new(catalog_name: String, catalog_manager: Weak<dyn CatalogManager>) -> Self {
@@ -154,16 +155,16 @@ impl InformationSchemaColumnsBuilder {
             schema,
             catalog_name,
             catalog_manager,
-            catalog_names: StringVectorBuilder::with_capacity(42),
-            schema_names: StringVectorBuilder::with_capacity(42),
-            table_names: StringVectorBuilder::with_capacity(42),
-            column_names: StringVectorBuilder::with_capacity(42),
-            data_types: StringVectorBuilder::with_capacity(42),
-            semantic_types: StringVectorBuilder::with_capacity(42),
-            column_defaults: StringVectorBuilder::with_capacity(42),
-            is_nullables: StringVectorBuilder::with_capacity(42),
-            column_types: StringVectorBuilder::with_capacity(42),
-            column_comments: StringVectorBuilder::with_capacity(42),
+            catalog_names: StringVectorBuilder::with_capacity(INIT_CAPACITY),
+            schema_names: StringVectorBuilder::with_capacity(INIT_CAPACITY),
+            table_names: StringVectorBuilder::with_capacity(INIT_CAPACITY),
+            column_names: StringVectorBuilder::with_capacity(INIT_CAPACITY),
+            data_types: StringVectorBuilder::with_capacity(INIT_CAPACITY),
+            semantic_types: StringVectorBuilder::with_capacity(INIT_CAPACITY),
+            column_defaults: StringVectorBuilder::with_capacity(INIT_CAPACITY),
+            is_nullables: StringVectorBuilder::with_capacity(INIT_CAPACITY),
+            column_types: StringVectorBuilder::with_capacity(INIT_CAPACITY),
+            column_comments: StringVectorBuilder::with_capacity(INIT_CAPACITY),
         }
     }
 
@@ -177,13 +178,6 @@ impl InformationSchemaColumnsBuilder {
         let predicates = Predicates::from_scan_request(&request);
 
         for schema_name in catalog_manager.schema_names(&catalog_name).await? {
-            if !catalog_manager
-                .schema_exists(&catalog_name, &schema_name)
-                .await?
-            {
-                continue;
-            }
-
             let mut stream = catalog_manager.tables(&catalog_name, &schema_name).await;
 
             while let Some(table) = stream.try_next().await? {

src/catalog/src/information_schema/key_column_usage.rs

@@ -23,10 +23,10 @@ use common_recordbatch::{RecordBatch, SendableRecordBatchStream};
 use datafusion::physical_plan::stream::RecordBatchStreamAdapter as DfRecordBatchStreamAdapter;
 use datafusion::physical_plan::streaming::PartitionStream as DfPartitionStream;
 use datafusion::physical_plan::SendableRecordBatchStream as DfSendableRecordBatchStream;
-use datatypes::prelude::{ConcreteDataType, ScalarVectorBuilder, VectorRef};
+use datatypes::prelude::{ConcreteDataType, MutableVector, ScalarVectorBuilder, VectorRef};
 use datatypes::schema::{ColumnSchema, Schema, SchemaRef};
 use datatypes::value::Value;
-use datatypes::vectors::{StringVectorBuilder, UInt32VectorBuilder};
+use datatypes::vectors::{ConstantVector, StringVector, StringVectorBuilder, UInt32VectorBuilder};
 use snafu::{OptionExt, ResultExt};
 use store_api::storage::{ScanRequest, TableId};
 
@@ -44,6 +44,7 @@ const TABLE_SCHEMA: &str = "table_schema";
 const TABLE_NAME: &str = "table_name";
 const COLUMN_NAME: &str = "column_name";
 const ORDINAL_POSITION: &str = "ordinal_position";
+const INIT_CAPACITY: usize = 42;
 
 /// The virtual table implementation for `information_schema.KEY_COLUMN_USAGE`.
 pub(super) struct InformationSchemaKeyColumnUsage {
@@ -162,9 +163,6 @@ struct InformationSchemaKeyColumnUsageBuilder {
     column_name: StringVectorBuilder,
     ordinal_position: UInt32VectorBuilder,
     position_in_unique_constraint: UInt32VectorBuilder,
-    referenced_table_schema: StringVectorBuilder,
-    referenced_table_name: StringVectorBuilder,
-    referenced_column_name: StringVectorBuilder,
 }
 
 impl InformationSchemaKeyColumnUsageBuilder {
@@ -177,18 +175,15 @@ impl InformationSchemaKeyColumnUsageBuilder {
             schema,
             catalog_name,
             catalog_manager,
-            constraint_catalog: StringVectorBuilder::with_capacity(42),
-            constraint_schema: StringVectorBuilder::with_capacity(42),
-            constraint_name: StringVectorBuilder::with_capacity(42),
-            table_catalog: StringVectorBuilder::with_capacity(42),
-            table_schema: StringVectorBuilder::with_capacity(42),
-            table_name: StringVectorBuilder::with_capacity(42),
-            column_name: StringVectorBuilder::with_capacity(42),
-            ordinal_position: UInt32VectorBuilder::with_capacity(42),
-            position_in_unique_constraint: UInt32VectorBuilder::with_capacity(42),
-            referenced_table_schema: StringVectorBuilder::with_capacity(42),
-            referenced_table_name: StringVectorBuilder::with_capacity(42),
-            referenced_column_name: StringVectorBuilder::with_capacity(42),
+            constraint_catalog: StringVectorBuilder::with_capacity(INIT_CAPACITY),
+            constraint_schema: StringVectorBuilder::with_capacity(INIT_CAPACITY),
+            constraint_name: StringVectorBuilder::with_capacity(INIT_CAPACITY),
+            table_catalog: StringVectorBuilder::with_capacity(INIT_CAPACITY),
+            table_schema: StringVectorBuilder::with_capacity(INIT_CAPACITY),
+            table_name: StringVectorBuilder::with_capacity(INIT_CAPACITY),
+            column_name: StringVectorBuilder::with_capacity(INIT_CAPACITY),
+            ordinal_position: UInt32VectorBuilder::with_capacity(INIT_CAPACITY),
+            position_in_unique_constraint: UInt32VectorBuilder::with_capacity(INIT_CAPACITY),
         }
     }
 
@@ -301,12 +296,15 @@ impl InformationSchemaKeyColumnUsageBuilder {
         self.column_name.push(Some(column_name));
         self.ordinal_position.push(Some(ordinal_position));
         self.position_in_unique_constraint.push(None);
-        self.referenced_table_schema.push(None);
-        self.referenced_table_name.push(None);
-        self.referenced_column_name.push(None);
     }
 
     fn finish(&mut self) -> Result<RecordBatch> {
+        let rows_num = self.table_catalog.len();
+
+        let null_string_vector = Arc::new(ConstantVector::new(
+            Arc::new(StringVector::from(vec![None as Option<&str>])),
+            rows_num,
+        ));
         let columns: Vec<VectorRef> = vec![
             Arc::new(self.constraint_catalog.finish()),
             Arc::new(self.constraint_schema.finish()),
@@ -317,9 +315,9 @@ impl InformationSchemaKeyColumnUsageBuilder {
             Arc::new(self.column_name.finish()),
             Arc::new(self.ordinal_position.finish()),
             Arc::new(self.position_in_unique_constraint.finish()),
-            Arc::new(self.referenced_table_schema.finish()),
-            Arc::new(self.referenced_table_name.finish()),
-            Arc::new(self.referenced_column_name.finish()),
+            null_string_vector.clone(),
+            null_string_vector.clone(),
+            null_string_vector,
         ];
         RecordBatch::new(self.schema.clone(), columns).context(CreateRecordBatchSnafu)
     }