chore(cubestore): Upgrade DF: fix info schema table providers

Author: paveltiunov

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

@@ -27,12 +27,12 @@ impl InfoSchemaTableDef for TablesInfoSchemaTableDef {
             Field::new(
                 "build_range_end",
                 DataType::Timestamp(TimeUnit::Nanosecond, None),
-                false,
+                true,
             ),
             Field::new(
                 "seal_at",
                 DataType::Timestamp(TimeUnit::Nanosecond, None),
-                false,
+                true,
             ),
         ]
     }

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

@@ -45,15 +45,15 @@ impl InfoSchemaTableDef for SystemTablesTableDef {
             Field::new(
                 "build_range_end",
                 DataType::Timestamp(TimeUnit::Nanosecond, None),
-                false,
+                true,
             ),
             Field::new(
                 "seal_at",
                 DataType::Timestamp(TimeUnit::Nanosecond, None),
-                false,
+                true,
             ),
             Field::new("sealed", DataType::Boolean, false),
-            Field::new("select_statement", DataType::Utf8, false),
+            Field::new("select_statement", DataType::Utf8, true),
             Field::new("extension", DataType::Utf8, true),
         ]
     }

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

@@ -318,7 +318,7 @@ impl ContextProvider for MetaStoreSchemaProvider {
                 let table = self
                     .inline_tables
                     .iter()
-                    .find(|inline_table| inline_table.name == table.as_ref())
+                    .find(|inline_table| inline_table.name.to_lowercase() == table.as_ref())
                     .ok_or_else(|| {
                         DataFusionError::Plan(format!("Inline table {} was not found", name))
                     })?;
@@ -789,11 +789,16 @@ impl ExecutionPlan for InfoSchemaTableExec {
         };
         let table = self.table.clone();
         let limit = self.limit.clone();
+        let projection = self.projection.clone();
         let batch = async move {
-            table
+            let mut batch = table
                 .scan(table_def, limit)
                 .await
-                .map_err(|e| DataFusionError::Execution(e.to_string()))
+                .map_err(|e| DataFusionError::Execution(e.to_string()))?;
+            if let Some(projection) = projection {
+                batch = batch.project(projection.as_slice())?;
+            }
+            Ok(batch)
         };
 
         let stream = futures::stream::once(batch);

rust/cubestore/cubestore/src/queryplanner/pretty_printers.rs

@@ -28,7 +28,7 @@ use crate::queryplanner::tail_limit::TailLimitExec;
 use crate::queryplanner::topk::ClusterAggregateTopK;
 use crate::queryplanner::topk::SortColumn;
 use crate::queryplanner::trace_data_loaded::TraceDataLoadedExec;
-use crate::queryplanner::CubeTableLogical;
+use crate::queryplanner::{CubeTableLogical, InfoSchemaTableProvider};
 use datafusion::physical_plan::empty::EmptyExec;
 use datafusion::physical_plan::expressions::Column;
 use datafusion::physical_plan::joins::HashJoinExec;
@@ -303,6 +303,8 @@ fn pp_source(t: Arc<dyn TableProvider>) -> String {
         format!("CubeTable(index: {})", pp_index(t.index_snapshot()))
     } else if let Some(t) = t.as_any().downcast_ref::<InlineTableProvider>() {
         format!("InlineTableProvider(data: {} rows)", t.get_data().len())
+    } else if let Some(t) = t.as_any().downcast_ref::<InfoSchemaTableProvider>() {
+        format!("InfoSchemaTableProvider(table: {:?})", t.table)
     } else {
         panic!("unknown table provider");
     }

rust/cubestore/cubestore/src/queryplanner/query_executor.rs

@@ -1587,7 +1587,7 @@ impl TableProvider for InlineTableProvider {
                     .collect::<Vec<Field>>(),
             ))
         } else {
-            schema
+            schema.clone()
         };
 
         if !self.inline_table_ids.iter().any(|id| id == &self.id) {
@@ -1599,7 +1599,7 @@ impl TableProvider for InlineTableProvider {
         let projection = projection.cloned();
         Ok(Arc::new(MemoryExec::try_new(
             &vec![batches],
-            projected_schema,
+            schema.clone(),
             projection,
         )?))
     }

rust/cubestore/cubestore/src/sql/mod.rs

@@ -128,7 +128,7 @@ pub type InlineTables = Vec<InlineTable>;
 
 impl InlineTable {
     pub fn new(id: u64, name: String, data: Arc<DataFrame>) -> Self {
-        Self { id, name, data }
+        Self { id, name: name.to_lowercase(), data: Arc::new(data.lowercase()) }
     }
 }
 

rust/cubestore/cubestore/src/store/mod.rs

@@ -59,12 +59,19 @@ pub const ROW_GROUP_SIZE: usize = 16384; // TODO config
 #[derive(Serialize, Deserialize, Hash, Eq, PartialEq, Debug, DeepSizeOf)]
 pub struct DataFrame {
     columns: Vec<Column>,
-    data: Vec<Row>,
+    data: Arc<Vec<Row>>,
 }
 
 impl DataFrame {
     pub fn new(columns: Vec<Column>, data: Vec<Row>) -> DataFrame {
-        DataFrame { columns, data }
+        DataFrame { columns, data: Arc::new(data) }
+    }
+
+    pub fn lowercase(&self) -> Self {
+        Self {
+            columns: self.columns.iter().map(|c| Column::new(c.get_name().to_lowercase(), c.get_column_type().clone(), c.get_index().clone())).collect(),
+            data: self.data.clone(),
+        }
     }
 
     pub fn len(&self) -> usize {
@@ -88,14 +95,6 @@ impl DataFrame {
         &self.data
     }
 
-    pub fn mut_rows(&mut self) -> &mut Vec<Row> {
-        &mut self.data
-    }
-
-    pub fn into_rows(self) -> Vec<Row> {
-        self.data
-    }
-
     pub fn to_execution_plan(
         &self,
         columns: &Vec<Column>,
@@ -166,10 +165,6 @@ impl ChunkData {
     pub fn len(&self) -> usize {
         self.data_frame.len()
     }
-
-    pub fn mut_rows(&mut self) -> &mut Vec<Row> {
-        &mut self.data_frame.data
-    }
 }
 
 pub struct WALStore {