feat(cubesql): Support Domo data queries (#7509)

* feat(cubesql): Support `Mon` format in `to_char`

* feat(cubesql): Support casting `Utf8` to `Interval`

* feat(cubesql): Support `TimestampMillisecond` output

* feat(cubesql): Support `DATE` function pushdown

* feat(cubesql): Support Domo date filtering

* fix(cubesql): Fix `LIKE` behavior with `startsWith`, `endsWith`

* fix(cubesql): Fix `ORDER BY` remapping with uppercase alias

* chore(cubesql): Support parsing date with time as `Date32`

packages/cubejs-schema-compiler/src/adapter/BaseQuery.js

@@ -2493,6 +2493,7 @@ export class BaseQuery {
         TO_CHAR: 'TO_CHAR({{ args_concat }})',
         // DATEADD is being rewritten to DATE_ADD
         // DATEADD: 'DATEADD({{ date_part }}, {{ interval }}, {{ args[2] }})',
+        DATE: 'DATE({{ args_concat }})',
       },
       statements: {
         select: 'SELECT {{ select_concat | map(attribute=\'aliased\') | join(\', \') }} \n' +

rust/cubesql/cubesql/Cargo.toml

@@ -10,12 +10,12 @@ homepage = "https://cube.dev"
 
 [dependencies]
 arc-swap = "1"
-datafusion = { git = 'https://github.com/cube-js/arrow-datafusion.git', rev = "5833202f5a9b69fded3cd45ffc9041ca5c404d33", default-features = false, features = ["regex_expressions", "unicode_expressions"] }
+datafusion = { git = 'https://github.com/cube-js/arrow-datafusion.git', rev = "dc15ce71abc4b4a2a41a1c149afc8b293027c49f", default-features = false, features = ["regex_expressions", "unicode_expressions"] }
 anyhow = "1.0"
 thiserror = "1.0.50"
 cubeclient = { path = "../cubeclient" }
 pg-srv = { path = "../pg-srv" }
-sqlparser = { git = 'https://github.com/cube-js/sqlparser-rs.git', rev = "037562f7975fcc92b725efbfbaaf050272668593" }
+sqlparser = { git = 'https://github.com/cube-js/sqlparser-rs.git', rev = "2c1becf622f2283514ecbb7e3f01bd858edeaa71" }
 lazy_static = "1.4.0"
 base64 = "0.13.0"
 tokio = { version = "^1.35", features = ["full", "rt", "tracing"] }

rust/cubesql/cubesql/src/compile/engine/df/scan.rs

@@ -40,7 +40,10 @@ use crate::{
 };
 use chrono::{Datelike, NaiveDate, NaiveDateTime};
 use datafusion::{
-    arrow::{array::TimestampNanosecondBuilder, datatypes::TimeUnit},
+    arrow::{
+        array::{TimestampMillisecondBuilder, TimestampNanosecondBuilder},
+        datatypes::TimeUnit,
+    },
     execution::context::TaskContext,
     logical_plan::JoinType,
     scalar::ScalarValue,
@@ -1061,6 +1064,36 @@ pub fn transform_response<V: ValueObject>(
                     }
                 )
             }
+            DataType::Timestamp(TimeUnit::Millisecond, None) => {
+                build_column!(
+                    DataType::Timestamp(TimeUnit::Millisecond, None),
+                    TimestampMillisecondBuilder,
+                    response,
+                    field_name,
+                    {
+                        (FieldValue::String(s), builder) => {
+                            let timestamp = NaiveDateTime::parse_from_str(s.as_str(), "%Y-%m-%dT%H:%M:%S.%f")
+                                .or_else(|_| NaiveDateTime::parse_from_str(s.as_str(), "%Y-%m-%d %H:%M:%S.%f"))
+                                .or_else(|_| NaiveDateTime::parse_from_str(s.as_str(), "%Y-%m-%dT%H:%M:%S"))
+                                .map_err(|e| {
+                                    DataFusionError::Execution(format!(
+                                        "Can't parse timestamp: '{}': {}",
+                                        s, e
+                                    ))
+                                })?;
+                            // TODO switch parsing to microseconds
+                            if timestamp.timestamp_millis() > (((1 as i64) << 62) / 1_000_000) {
+                                builder.append_null()?;
+                            } else {
+                                builder.append_value(timestamp.timestamp_millis())?;
+                            }
+                        },
+                    },
+                    {
+                        (ScalarValue::TimestampMillisecond(v, None), builder) => builder.append_option(v.clone())?,
+                    }
+                )
+            }
             DataType::Date32 => {
                 build_column!(
                     DataType::Date32,
@@ -1070,6 +1103,9 @@ pub fn transform_response<V: ValueObject>(
                     {
                         (FieldValue::String(s), builder) => {
                             let date = NaiveDate::parse_from_str(s.as_str(), "%Y-%m-%d")
+                                // FIXME: temporary solution for cases when expected type is Date32
+                                // but underlying data is a Timestamp
+                                .or_else(|_| NaiveDate::parse_from_str(s.as_str(), "%Y-%m-%dT00:00:00.000"))
                                 .map_err(|e| {
                                     DataFusionError::Execution(format!(
                                         "Can't parse date: '{}': {}",

rust/cubesql/cubesql/src/compile/engine/df/wrapper.rs

@@ -746,6 +746,16 @@ impl CubeScanWrapperNode {
             sql = new_sql_query;
 
             let original_alias = expr_name(&original_expr, &schema)?;
+            let original_alias_key = Column::from_name(&original_alias);
+            if let Some(alias_column) = next_remapping.get(&original_alias_key) {
+                let alias = alias_column.name.clone();
+                aliased_columns.push(AliasedColumn {
+                    expr: expr_sql,
+                    alias,
+                });
+                continue;
+            }
+
             let alias = if can_rename_columns {
                 let alias = expr_name(&expr, &schema)?;
                 let mut truncated_alias = non_id_regex.replace_all(&alias, "_").to_lowercase();
@@ -766,10 +776,7 @@ impl CubeScanWrapperNode {
             };
             if original_alias != alias {
                 if !next_remapping.contains_key(&Column::from_name(&alias)) {
-                    next_remapping.insert(
-                        Column::from_name(&original_alias),
-                        Column::from_name(&alias),
-                    );
+                    next_remapping.insert(original_alias_key, Column::from_name(&alias));
                     next_remapping.insert(
                         Column {
                             name: original_alias.clone(),

rust/cubesql/cubesql/src/compile/engine/udf.rs

@@ -1467,6 +1467,7 @@ fn postgres_datetime_format_to_iso(format: String) -> String {
         .replace("yyyy", "%Y")
         // NOTE: "%q" is not a part of chrono
         .replace("Q", "%q")
+        .replace("Mon", "%b")
         .replace("DD", "%d")
         .replace("dd", "%d")
         .replace("HH24", "%H")