chore(cubestore): New cubestore rebased to master

Author: waralexrom

rust/cubestore/cubestore/src/lib.rs

@@ -1,4 +1,3 @@
-#![feature(async_closure)]
 #![feature(box_patterns)]
 #![feature(hash_set_entry)]
 

rust/cubestore/cubestore/src/metastore/mod.rs

@@ -4972,7 +4972,7 @@ mod tests {
 
     #[test]
     fn test_structures_size() {
-        assert_eq!(std::mem::size_of::<MetaStoreEvent>(), 672);
+        assert_eq!(std::mem::size_of::<MetaStoreEvent>(), 640);
     }
 
     #[tokio::test]

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

@@ -105,7 +105,7 @@ impl InfoSchemaTableDef for SystemChunksTableDef {
                         row.get_row()
                             .created_at()
                             .as_ref()
-                            .map(|t| t.timestamp_nanos())
+                            .map(timestamp_nanos_or_panic)
                     },
                 )))
             }),

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

@@ -181,26 +181,23 @@ impl InfoSchemaTableDef for SystemTablesTableDef {
                 )))
             }),
             Box::new(|tables| {
-                Arc::new(StringArray::from_iter_values(
-                    tables
-                        .iter()
-                        .map(|row| {
-                            row.table
-                                .get_row()
-                                .seal_at()
-                                .as_ref()
-                                .map(timestamp_nanos_or_panic)
-                        })
-                        .collect::<Vec<_>>(),
-                ))
+                Arc::new(TimestampNanosecondArray::from_iter(tables.iter().map(
+                    |row| {
+                        row.table
+                            .get_row()
+                            .seal_at()
+                            .as_ref()
+                            .map(timestamp_nanos_or_panic)
+                    },
+                )))
             }),
             Box::new(|tables| {
-                Arc::new(BooleanArray::from_iter_values(
-                    tables.iter().map(|row| row.table.get_row().sealed()),
+                Arc::new(BooleanArray::from_iter(
+                    tables.iter().map(|row| Some(row.table.get_row().sealed())),
                 ))
             }),
             Box::new(|tables| {
-                Arc::new(StringArray::from_iter_values(tables.iter().map(|row| {
+                Arc::new(StringArray::from_iter(tables.iter().map(|row| {
                     row.table
                         .get_row()
                         .select_statement()

rust/cubestore/cubestore/src/queryplanner/topk/plan.rs

@@ -321,7 +321,7 @@ struct ColumnProjection<'a> {
 
 fn extract_projections_and_havings(
     p: &Arc<LogicalPlan>,
-) -> Result<Option<ColumnProjection>, DataFusionError> {
+) -> Result<Option<ColumnProjection<'_>>, DataFusionError> {
     // Goal:  Deal with arbitrary series of Projection and Filter, where the Projections are column
     // projections (or cardinality(column)), on top of an underlying node.
     //

rust/cubestore/cubestore/src/queryplanner/udfs.rs

@@ -1,5 +1,3 @@
-//TODO rebase use super::udf_xirr::XirrAccumulator;
-use crate::queryplanner::coalesce::SUPPORTED_COALESCE_TYPES;
 use crate::queryplanner::hll::{Hll, HllUnion};
 use crate::queryplanner::info_schema::timestamp_nanos_or_panic;
 use crate::queryplanner::udf_xirr::{XirrUDF, XIRR_UDAF_NAME};
@@ -32,6 +30,7 @@ pub fn registerable_scalar_udfs_iter() -> impl Iterator<Item = ScalarUDF> {
         ScalarUDF::new_from_impl(DateAddSub::new_sub()),
         ScalarUDF::new_from_impl(UnixTimestamp::new()),
         ScalarUDF::new_from_impl(ConvertTz::new()),
+        ScalarUDF::new_from_impl(Now::new()),
     ]
     .into_iter()
 }
@@ -72,62 +71,74 @@ pub fn aggregate_kind_by_name(n: &str) -> Option<CubeAggregateUDFKind> {
 // TODO: add custom type and use it instead of `Binary` for HLL columns.
 
 #[derive(Debug)]
-struct UnixTimestamp {
+struct Now {
     signature: Signature,
 }
 
-struct Now {}
 impl Now {
+    fn new() -> Self {
+        Now {
+            signature: Self::signature(),
+        }
+    }
+
     fn signature() -> Signature {
-        Signature::Exact(Vec::new())
+        Signature::exact(Vec::new(), Volatility::Stable)
     }
 }
-/* TODO rebase - reimplement for new interface
-impl CubeScalarUDF for Now {
-    fn kind(&self) -> CubeScalarUDFKind {
-        CubeScalarUDFKind::Now
-    }
 
+impl ScalarUDFImpl for Now {
     fn name(&self) -> &str {
         "NOW"
     }
 
-    fn descriptor(&self) -> ScalarUDF {
-        ScalarUDF {
-            name: self.name().to_string(),
-            signature: Self::signature(),
-            return_type: Arc::new(|inputs| {
-                assert!(inputs.is_empty());
-                Ok(Arc::new(DataType::Timestamp(TimeUnit::Nanosecond, None)))
-            }),
-            fun: Arc::new(|_| {
-                let t = match SystemTime::now().duration_since(SystemTime::UNIX_EPOCH) {
-                    Ok(t) => t,
-                    Err(e) => {
-                        return Err(DataFusionError::Internal(format!(
-                            "Failed to get current timestamp: {}",
-                            e
-                        )))
-                    }
-                };
-
-                let nanos = match i64::try_from(t.as_nanos()) {
-                    Ok(t) => t,
-                    Err(e) => {
-                        return Err(DataFusionError::Internal(format!(
-                            "Failed to convert timestamp to i64: {}",
-                            e
-                        )))
-                    }
-                };
+    fn as_any(&self) -> &dyn Any {
+        self
+    }
 
-                Ok(ColumnarValue::Scalar(ScalarValue::TimestampNanosecond(
-                    Some(nanos),
+    fn signature(&self) -> &Signature {
+        &self.signature
+    }
+
+    fn return_type(&self, _arg_types: &[DataType]) -> datafusion::common::Result<DataType> {
+        Ok(DataType::Timestamp(TimeUnit::Nanosecond, None))
+    }
+
+    fn invoke_with_args(
+        &self,
+        _args: datafusion::logical_expr::ScalarFunctionArgs,
+    ) -> datafusion::error::Result<ColumnarValue> {
+        let t = match SystemTime::now().duration_since(SystemTime::UNIX_EPOCH) {
+            Ok(t) => t,
+            Err(e) => {
+                return Err(DataFusionError::Internal(format!(
+                    "Failed to get current timestamp: {}",
+                    e
                 )))
-            }),
-        }
+            }
+        };
+
+        let nanos = match i64::try_from(t.as_nanos()) {
+            Ok(t) => t,
+            Err(e) => {
+                return Err(DataFusionError::Internal(format!(
+                    "Failed to convert timestamp to i64: {}",
+                    e
+                )))
+            }
+        };
+
+        Ok(ColumnarValue::Scalar(ScalarValue::TimestampNanosecond(
+            Some(nanos),
+            None,
+        )))
     }
-} */
+}
+
+#[derive(Debug)]
+struct UnixTimestamp {
+    signature: Signature,
+}
 
 impl UnixTimestamp {
     pub fn new() -> Self {
@@ -145,40 +156,6 @@ impl ScalarUDFImpl for UnixTimestamp {
         "unix_timestamp"
     }
 
-    /* TODO rebase - reimplement for new interface
-     * fn descriptor(&self) -> ScalarUDF {
-        ScalarUDF {
-            name: self.name().to_string(),
-            signature: Self::signature(),
-            return_type: Arc::new(|inputs| {
-                assert!(inputs.is_empty());
-                Ok(Arc::new(DataType::Int64))
-            }),
-            fun: Arc::new(|_| {
-                let t = match SystemTime::now().duration_since(SystemTime::UNIX_EPOCH) {
-                    Ok(t) => t,
-                    Err(e) => {
-                        return Err(DataFusionError::Internal(format!(
-                            "Failed to get current timestamp: {}",
-                            e
-                        )))
-                    }
-                };
-
-                let seconds = match i64::try_from(t.as_secs()) {
-                    Ok(t) => t,
-                    Err(e) => {
-                        return Err(DataFusionError::Internal(format!(
-                            "Failed to convert timestamp to i64: {}",
-                            e
-                        )))
-                    }
-                };
-
-                Ok(ColumnarValue::Scalar(ScalarValue::Int64(Some(seconds))))
-            }),
-        }
-    }*/
     fn as_any(&self) -> &dyn Any {
         self
     }
@@ -191,16 +168,31 @@ impl ScalarUDFImpl for UnixTimestamp {
         Ok(DataType::Int64)
     }
 
-    fn invoke(&self, _args: &[ColumnarValue]) -> datafusion::common::Result<ColumnarValue> {
-        Err(DataFusionError::Internal(
-            "UNIX_TIMESTAMP() was not optimized away".to_string(),
-        ))
-    }
+    fn invoke_with_args(
+        &self,
+        _args: datafusion::logical_expr::ScalarFunctionArgs,
+    ) -> datafusion::error::Result<ColumnarValue> {
+        let t = match SystemTime::now().duration_since(SystemTime::UNIX_EPOCH) {
+            Ok(t) => t,
+            Err(e) => {
+                return Err(DataFusionError::Internal(format!(
+                    "Failed to get current timestamp: {}",
+                    e
+                )))
+            }
+        };
 
-    fn invoke_no_args(&self, _number_rows: usize) -> datafusion::common::Result<ColumnarValue> {
-        Err(DataFusionError::Internal(
-            "UNIX_TIMESTAMP() was not optimized away".to_string(),
-        ))
+        let seconds = match i64::try_from(t.as_secs()) {
+            Ok(t) => t,
+            Err(e) => {
+                return Err(DataFusionError::Internal(format!(
+                    "Failed to convert timestamp to i64: {}",
+                    e
+                )))
+            }
+        };
+
+        Ok(ColumnarValue::Scalar(ScalarValue::Int64(Some(seconds))))
     }
 
     fn simplify(

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

@@ -808,12 +808,13 @@ impl SqlService for SqlServiceImpl {
 
                 let disable_quoting = with_options
                     .iter()
-                    .find(|&opt| opt.name.value == "disable_quoting")
-                    .map_or(Ok(false), |option| match &option.value {
-                        Value::Boolean(value) => Ok(*value),
+                    .filter_map(filter_sql_option_key_value)
+                    .find(|&(name, _)| name.value == "disable_quoting")
+                    .map_or(Ok(false), |(_, value)| match value {
+                        Expr::Value(Value::Boolean(value)) => Ok(*value),
                         _ => Err(CubeError::user(format!(
                             "Bad disable_quoting flag (expected boolean) {}",
-                            option.value
+                            value
                         ))),
                     })?;
 
@@ -4349,9 +4350,9 @@ mod tests {
             };
             assert_eq!(
                 pp_plan,
-                "Projection, [information_schema.tables.table_name]\
+                "Projection, [information_schema.tables.table_name:table_name]\
                 \n  Filter\
-                \n    Scan information_schema.tables, source: InfoSchemaTableProvider, fields: [table_schema, table_name]"
+                \n    Scan information_schema.tables, source: InfoSchemaTableProvider(table: Tables), fields: [table_schema, table_name]"
             );
         }).await;
     }