fix: Use Unparser for expr to sql

src/duckdb/sql_table.rs

@@ -153,6 +153,7 @@ impl<T, P> DuckSqlExec<T, P> {
 
     fn sql(&self) -> SqlResult<String> {
         let sql = self.base_exec.sql()?;
+
         Ok(format!(
             "{cte_expr} {sql}",
             cte_expr = get_cte(&self.table_functions)

src/sql/sql_provider_datafusion/expr.rs

@@ -1,7 +1,12 @@
+use std::sync::Arc;
+
 use bigdecimal::{num_bigint::BigInt, BigDecimal};
 use datafusion::{
     logical_expr::{Cast, Expr},
     scalar::ScalarValue,
+    sql::unparser::dialect::{
+        DefaultDialect, Dialect, MySqlDialect, PostgreSqlDialect, SqliteDialect,
+    },
 };
 
 pub const SECONDS_IN_DAY: i32 = 86_400;
@@ -27,6 +32,18 @@ pub enum Engine {
     MySQL,
 }
 
+impl Engine {
+    /// Get the corresponding `Dialect` to use for unparsing
+    pub fn dialect(&self) -> Arc<dyn Dialect + Send + Sync> {
+        match self {
+            Engine::SQLite => Arc::new(SqliteDialect {}),
+            Engine::Postgres => Arc::new(PostgreSqlDialect {}),
+            Engine::MySQL => Arc::new(MySqlDialect {}),
+            Engine::Spark | Engine::DuckDB | Engine::ODBC => Arc::new(DefaultDialect {}),
+        }
+    }
+}
+
 #[allow(clippy::too_many_lines)]
 #[allow(clippy::cast_precision_loss)]
 pub fn to_sql_with_engine(expr: &Expr, engine: Option<Engine>) -> Result<String> {
@@ -236,6 +253,16 @@ fn handle_cast(cast: &Cast, engine: Option<Engine>, expr: &Expr) -> Result<Strin
                 to_sql_with_engine(&cast.expr, engine)?,
             )),
         },
+        arrow::datatypes::DataType::Int64 => match engine {
+            Some(Engine::DuckDB | Engine::SQLite | Engine::Postgres) => Ok(format!(
+                "CAST({} AS BIGINT)",
+                to_sql_with_engine(&cast.expr, engine)?,
+            )),
+            _ => Ok(format!(
+                "CAST({} AS INT64)",
+                to_sql_with_engine(&cast.expr, engine)?,
+            )),
+        },
         _ => Err(Error::UnsupportedFilterExpr {
             expr: format!("{expr}"),
         }),

src/sql/sql_provider_datafusion/mod.rs

@@ -10,7 +10,13 @@ use crate::sql::db_connection_pool::{
     DbConnectionPool,
 };
 use async_trait::async_trait;
-use datafusion::{catalog::Session, sql::unparser::dialect::Dialect};
+use datafusion::{
+    catalog::Session,
+    sql::unparser::{
+        dialect::{DefaultDialect, Dialect},
+        Unparser,
+    },
+};
 use expr::Engine;
 use futures::TryStreamExt;
 use snafu::prelude::*;
@@ -46,6 +52,9 @@ pub enum Error {
 
     #[snafu(display("Unable to generate SQL: {source}"))]
     UnableToGenerateSQL { source: expr::Error },
+
+    #[snafu(display("Unable to generate SQL: {source}"))]
+    UnableToGenerateSQLDataFusion { source: DataFusionError },
 }
 
 pub type Result<T, E = Error> = std::result::Result<T, E>;
@@ -174,9 +183,15 @@ impl<T, P> TableProvider for SqlTable<T, P> {
         &self,
         filters: &[&Expr],
     ) -> DataFusionResult<Vec<TableProviderFilterPushDown>> {
+        let dialect = self.engine.map_or(
+            Arc::new(DefaultDialect {}) as Arc<dyn Dialect + Send + Sync>,
+            |e| e.dialect(),
+        );
+        let unparser = Unparser::new(dialect.as_ref());
+
         let filter_push_down: Vec<TableProviderFilterPushDown> = filters
             .iter()
-            .map(|f| match expr::to_sql_with_engine(f, self.engine) {
+            .map(|f| match unparser.expr_to_sql(f) {
                 Ok(_) => TableProviderFilterPushDown::Exact,
                 Err(_) => TableProviderFilterPushDown::Unsupported,
             })
@@ -284,19 +299,18 @@ impl<T, P> SqlExec<T, P> {
         let where_expr = if self.filters.is_empty() {
             String::new()
         } else {
+            let dialect = self.engine.map_or(
+                Arc::new(DefaultDialect {}) as Arc<dyn Dialect + Send + Sync>,
+                |e| e.dialect(),
+            );
+            let unparser = Unparser::new(dialect.as_ref());
+
             let filter_expr = self
                 .filters
                 .iter()
-                .map(|f| match f {
-                    // DataFusion optimization uses aliases to preserve original expression names during optimizations and type coercions:
-                    // https://github.com/apache/datafusion/issues/3794
-                    // If there is a filter with additional alias information, we must use the expression only
-                    // as original expression name is unnecessary and the alias can't be part of the WHERE clause
-                    Expr::Alias(alias) => expr::to_sql_with_engine(&alias.expr, self.engine),
-                    _ => expr::to_sql_with_engine(f, self.engine),
-                })
-                .collect::<expr::Result<Vec<_>>>()
-                .context(UnableToGenerateSQLSnafu)?;
+                .map(|f| unparser.expr_to_sql(f).map(|s| s.to_string()))
+                .collect::<Result<Vec<String>, DataFusionError>>()
+                .context(UnableToGenerateSQLDataFusionSnafu)?;
             format!("WHERE {}", filter_expr.join(" AND "))
         };
 

src/util/mod.rs

@@ -1,7 +1,10 @@
+use datafusion::error::Result as DataFusionResult;
 use datafusion::logical_expr::Expr;
+use datafusion::sql::unparser::dialect::DefaultDialect;
+use datafusion::sql::unparser::Unparser;
 use snafu::prelude::*;
-use std::collections::HashMap;
 use std::hash::Hash;
+use std::{collections::HashMap, sync::Arc};
 
 use crate::{
     sql::sql_provider_datafusion::expr::{self, Engine},
@@ -23,14 +26,23 @@ pub mod test;
 #[derive(Debug, Snafu)]
 pub enum Error {
     #[snafu(display("Unable to generate SQL: {source}"))]
-    UnableToGenerateSQL { source: expr::Error },
+    UnableToGenerateSQL {
+        source: datafusion::error::DataFusionError,
+    },
 }
 
 pub fn filters_to_sql(filters: &[Expr], engine: Option<Engine>) -> Result<String, Error> {
+    let dialect = engine
+        .map(|e| e.dialect())
+        .unwrap_or(Arc::new(DefaultDialect {}));
     Ok(filters
         .iter()
-        .map(|expr| expr::to_sql_with_engine(expr, engine))
-        .collect::<expr::Result<Vec<_>>>()
+        .map(|f| {
+            Unparser::new(dialect.as_ref())
+                .expr_to_sql(f)
+                .map(|e| e.to_string())
+        })
+        .collect::<DataFusionResult<Vec<String>>>()
         .context(UnableToGenerateSQLSnafu)?
         .join(" AND "))
 }