feat(cubestore): Introduce information_schema.columns (#6152)

Author: ovr

rust/cubestore/cubestore/src/queryplanner/info_schema/info_schema_columns.rs

@@ -0,0 +1,88 @@
+use crate::metastore::table::TablePath;
+use crate::metastore::Column;
+use crate::queryplanner::{InfoSchemaTableDef, InfoSchemaTableDefContext};
+use crate::CubeError;
+use arrow::array::{ArrayRef, StringArray};
+use arrow::datatypes::{DataType, Field};
+use async_trait::async_trait;
+use std::sync::Arc;
+
+pub struct ColumnsInfoSchemaTableDef;
+
+#[async_trait]
+impl InfoSchemaTableDef for ColumnsInfoSchemaTableDef {
+    type T = (Column, TablePath);
+
+    async fn rows(
+        &self,
+        ctx: InfoSchemaTableDefContext,
+    ) -> Result<Arc<Vec<(Column, TablePath)>>, CubeError> {
+        let rows = ctx.meta_store.get_tables_with_path(false).await?;
+        let mut res = Vec::new();
+
+        for row in rows.iter() {
+            let columns = row.table.get_row().get_columns();
+            for column in columns {
+                res.push((column.clone(), row.clone()));
+            }
+        }
+
+        Ok(Arc::new(res))
+    }
+
+    fn columns(
+        &self,
+    ) -> Vec<(
+        Field,
+        Box<dyn Fn(Arc<Vec<(Column, TablePath)>>) -> ArrayRef>,
+    )> {
+        vec![
+            (
+                Field::new("table_schema", DataType::Utf8, false),
+                Box::new(|tables| {
+                    Arc::new(StringArray::from(
+                        tables
+                            .iter()
+                            .map(|(_, row)| row.schema.get_row().get_name().as_str())
+                            .collect::<Vec<_>>(),
+                    ))
+                }),
+            ),
+            (
+                Field::new("table_name", DataType::Utf8, false),
+                Box::new(|tables| {
+                    Arc::new(StringArray::from(
+                        tables
+                            .iter()
+                            .map(|(_, row)| row.table.get_row().get_table_name().as_str())
+                            .collect::<Vec<_>>(),
+                    ))
+                }),
+            ),
+            (
+                Field::new("column_name", DataType::Utf8, false),
+                Box::new(|tables| {
+                    Arc::new(StringArray::from(
+                        tables
+                            .iter()
+                            .map(|(column, _)| column.get_name().as_str())
+                            .collect::<Vec<_>>(),
+                    ))
+                }),
+            ),
+            (
+                Field::new("data_type", DataType::Utf8, false),
+                Box::new(|tables| {
+                    Arc::new(StringArray::from(
+                        tables
+                            .iter()
+                            .map(|(column, _)| column.get_column_type().to_string())
+                            .collect::<Vec<_>>(),
+                    ))
+                }),
+            ),
+        ]
+    }
+}
+
+crate::base_info_schema_table_def!(ColumnsInfoSchemaTableDef);

rust/cubestore/cubestore/src/queryplanner/info_schema/mod.rs

@@ -1,3 +1,4 @@
+mod info_schema_columns;
 mod info_schema_schemata;
 mod info_schema_tables;
 mod system_cache;
@@ -10,6 +11,7 @@ mod system_replay_handles;
 mod system_snapshots;
 mod system_tables;
 
+pub use info_schema_columns::*;
 pub use info_schema_schemata::*;
 pub use info_schema_tables::*;
 pub use system_cache::*;

rust/cubestore/cubestore/src/queryplanner/mod.rs

@@ -25,9 +25,10 @@ use crate::metastore::table::{Table, TablePath};
 use crate::metastore::{IdRow, MetaStore};
 use crate::queryplanner::flatten_union::FlattenUnion;
 use crate::queryplanner::info_schema::{
-    SchemataInfoSchemaTableDef, SystemCacheTableDef, SystemChunksTableDef, SystemIndexesTableDef,
-    SystemJobsTableDef, SystemPartitionsTableDef, SystemQueueTableDef, SystemReplayHandlesTableDef,
-    SystemSnapshotsTableDef, SystemTablesTableDef, TablesInfoSchemaTableDef,
+    ColumnsInfoSchemaTableDef, SchemataInfoSchemaTableDef, SystemCacheTableDef,
+    SystemChunksTableDef, SystemIndexesTableDef, SystemJobsTableDef, SystemPartitionsTableDef,
+    SystemQueueTableDef, SystemReplayHandlesTableDef, SystemSnapshotsTableDef,
+    SystemTablesTableDef, TablesInfoSchemaTableDef,
 };
 use crate::queryplanner::now::MaterializeNow;
 use crate::queryplanner::planning::{choose_index_ext, ClusterSendNode};
@@ -297,6 +298,11 @@ impl ContextProvider for MetaStoreSchemaProvider {
                 })
             });
         res.or_else(|| match (schema, table) {
+            ("information_schema", "columns") => Some(Arc::new(InfoSchemaTableProvider::new(
+                self.meta_store.clone(),
+                self.cache_store.clone(),
+                InfoSchemaTable::Columns,
+            ))),
             ("information_schema", "tables") => Some(Arc::new(InfoSchemaTableProvider::new(
                 self.meta_store.clone(),
                 self.cache_store.clone(),
@@ -382,6 +388,7 @@ impl ContextProvider for MetaStoreSchemaProvider {
 
 #[derive(Clone, Debug)]
 pub enum InfoSchemaTable {
+    Columns,
     Tables,
     Schemata,
     SystemJobs,
@@ -450,6 +457,7 @@ macro_rules! base_info_schema_table_def {
 impl InfoSchemaTable {
     fn table_def(&self) -> Box<dyn BaseInfoSchemaTableDef + Send + Sync> {
         match self {
+            InfoSchemaTable::Columns => Box::new(ColumnsInfoSchemaTableDef),
             InfoSchemaTable::Tables => Box::new(TablesInfoSchemaTableDef),
             InfoSchemaTable::Schemata => Box::new(SchemataInfoSchemaTableDef),
             InfoSchemaTable::SystemTables => Box::new(SystemTablesTableDef),