[logical-types] use Scalar in Expr::Logical (#12793)

* [logical-types] use Scalar in Expr::Logical

* Fix rust lint

* Fix CI

* Fix docs

Author: notfilippo

datafusion-cli/src/functions.rs

@@ -27,7 +27,7 @@ use datafusion::common::{plan_err, Column};
 use datafusion::datasource::function::TableFunctionImpl;
 use datafusion::datasource::TableProvider;
 use datafusion::error::Result;
-use datafusion::logical_expr::Expr;
+use datafusion::logical_expr::{Expr, Scalar};
 use datafusion::physical_plan::memory::MemoryExec;
 use datafusion::physical_plan::ExecutionPlan;
 use datafusion::scalar::ScalarValue;
@@ -321,7 +321,10 @@ pub struct ParquetMetadataFunc {}
 impl TableFunctionImpl for ParquetMetadataFunc {
     fn call(&self, exprs: &[Expr]) -> Result<Arc<dyn TableProvider>> {
         let filename = match exprs.first() {
-            Some(Expr::Literal(ScalarValue::Utf8(Some(s)))) => s, // single quote: parquet_metadata('x.parquet')
+            Some(Expr::Literal(Scalar {
+                value: ScalarValue::Utf8(Some(s)),
+                ..
+            })) => s, // single quote: parquet_metadata('x.parquet')
             Some(Expr::Column(Column { name, .. })) => name, // double quote: parquet_metadata("x.parquet")
             _ => {
                 return plan_err!(

datafusion-examples/examples/expr_api.rs

@@ -61,7 +61,7 @@ async fn main() -> Result<()> {
     let expr2 = Expr::BinaryExpr(BinaryExpr::new(
         Box::new(col("a")),
         Operator::Plus,
-        Box::new(Expr::Literal(ScalarValue::Int32(Some(5)))),
+        Box::new(Expr::from(ScalarValue::Int32(Some(5)))),
     ));
     assert_eq!(expr, expr2);
 

datafusion-examples/examples/simple_udtf.rs

@@ -30,7 +30,7 @@ use datafusion::physical_plan::ExecutionPlan;
 use datafusion::prelude::SessionContext;
 use datafusion_common::{plan_err, ScalarValue};
 use datafusion_expr::simplify::SimplifyContext;
-use datafusion_expr::{Expr, TableType};
+use datafusion_expr::{Expr, Scalar, TableType};
 use datafusion_optimizer::simplify_expressions::ExprSimplifier;
 use std::fs::File;
 use std::io::Seek;
@@ -133,7 +133,11 @@ struct LocalCsvTableFunc {}
 
 impl TableFunctionImpl for LocalCsvTableFunc {
     fn call(&self, exprs: &[Expr]) -> Result<Arc<dyn TableProvider>> {
-        let Some(Expr::Literal(ScalarValue::Utf8(Some(ref path)))) = exprs.first() else {
+        let Some(Expr::Literal(Scalar {
+            value: ScalarValue::Utf8(Some(ref path)),
+            ..
+        })) = exprs.first()
+        else {
             return plan_err!("read_csv requires at least one string argument");
         };
 
@@ -145,7 +149,11 @@ impl TableFunctionImpl for LocalCsvTableFunc {
                 let info = SimplifyContext::new(&execution_props);
                 let expr = ExprSimplifier::new(info).simplify(expr.clone())?;
 
-                if let Expr::Literal(ScalarValue::Int64(Some(limit))) = expr {
+                if let Expr::Literal(Scalar {
+                    value: ScalarValue::Int64(Some(limit)),
+                    ..
+                }) = expr
+                {
                     Ok(limit as usize)
                 } else {
                     plan_err!("Limit must be an integer")

datafusion/core/benches/map_query_sql.rs

@@ -71,8 +71,8 @@ fn criterion_benchmark(c: &mut Criterion) {
     let mut value_buffer = Vec::new();
 
     for i in 0..1000 {
-        key_buffer.push(Expr::Literal(ScalarValue::Utf8(Some(keys[i].clone()))));
-        value_buffer.push(Expr::Literal(ScalarValue::Int32(Some(values[i]))));
+        key_buffer.push(Expr::from(ScalarValue::Utf8(Some(keys[i].clone()))));
+        value_buffer.push(Expr::from(ScalarValue::Int32(Some(values[i]))));
     }
     c.bench_function("map_1000_1", |b| {
         b.iter(|| {

datafusion/core/src/dataframe/mod.rs

@@ -1188,7 +1188,7 @@ impl DataFrame {
     /// ```
     pub async fn count(self) -> Result<usize> {
         let rows = self
-            .aggregate(vec![], vec![count(Expr::Literal(COUNT_STAR_EXPANSION))])?
+            .aggregate(vec![], vec![count(Expr::from(COUNT_STAR_EXPANSION))])?
             .collect()
             .await?;
         let len = *rows
@@ -2985,7 +2985,7 @@ mod tests {
         let join = left.clone().join_on(
             right.clone(),
             JoinType::Inner,
-            Some(Expr::Literal(ScalarValue::Null)),
+            Some(Expr::from(ScalarValue::Null)),
         )?;
         let expected_plan = "CrossJoin:\
         \n  TableScan: a projection=[c1], full_filters=[Boolean(NULL)]\

datafusion/core/src/datasource/listing/helpers.rs

@@ -868,7 +868,7 @@ mod tests {
         assert_eq!(
             evaluate_partition_prefix(
                 partitions,
-                &[col("a").eq(Expr::Literal(ScalarValue::Date32(Some(3))))],
+                &[col("a").eq(Expr::from(ScalarValue::Date32(Some(3))))],
             ),
             Some(Path::from("a=1970-01-04")),
         );
@@ -877,7 +877,7 @@ mod tests {
         assert_eq!(
             evaluate_partition_prefix(
                 partitions,
-                &[col("a").eq(Expr::Literal(ScalarValue::Date64(Some(
+                &[col("a").eq(Expr::from(ScalarValue::Date64(Some(
                     4 * 24 * 60 * 60 * 1000
                 )))),],
             ),

datafusion/core/src/datasource/listing/table.rs

@@ -1941,7 +1941,7 @@ mod tests {
         let filter_predicate = Expr::BinaryExpr(BinaryExpr::new(
             Box::new(Expr::Column("column1".into())),
             Operator::GtEq,
-            Box::new(Expr::Literal(ScalarValue::Int32(Some(0)))),
+            Box::new(Expr::from(ScalarValue::Int32(Some(0)))),
         ));
 
         // Create a new batch of data to insert into the table

datafusion/core/src/datasource/physical_plan/parquet/row_filter.rs

@@ -373,7 +373,7 @@ impl<'schema> TreeNodeRewriter for PushdownChecker<'schema> {
                         //
                         // See comments on `FilterCandidateBuilder` for more information
                         let null_value = ScalarValue::try_from(field.data_type())?;
-                        Ok(Transformed::yes(Arc::new(Literal::new(null_value)) as _))
+                        Ok(Transformed::yes(Arc::new(Literal::from(null_value)) as _))
                     })
                     // If the column is not in the table schema, should throw the error
                     .map_err(|e| arrow_datafusion_err!(e));
@@ -699,9 +699,10 @@ mod test {
             .expect("expected error free record batch");
 
         // Test all should fail
-        let expr = col("timestamp_col").lt(Expr::Literal(
-            ScalarValue::TimestampNanosecond(Some(1), Some(Arc::from("UTC"))),
-        ));
+        let expr = col("timestamp_col").lt(Expr::from(ScalarValue::TimestampNanosecond(
+            Some(1),
+            Some(Arc::from("UTC")),
+        )));
         let expr = logical2physical(&expr, &table_schema);
         let candidate = FilterCandidateBuilder::new(expr, &file_schema, &table_schema)
             .build(&metadata)
@@ -723,9 +724,10 @@ mod test {
         assert!(matches!(filtered, Ok(a) if a == BooleanArray::from(vec![false; 8])));
 
         // Test all should pass
-        let expr = col("timestamp_col").gt(Expr::Literal(
-            ScalarValue::TimestampNanosecond(Some(0), Some(Arc::from("UTC"))),
-        ));
+        let expr = col("timestamp_col").gt(Expr::from(ScalarValue::TimestampNanosecond(
+            Some(0),
+            Some(Arc::from("UTC")),
+        )));
         let expr = logical2physical(&expr, &table_schema);
         let candidate = FilterCandidateBuilder::new(expr, &file_schema, &table_schema)
             .build(&metadata)
@@ -826,7 +828,7 @@ mod test {
 
         let expr = col("str_col")
             .is_not_null()
-            .or(col("int_col").gt(Expr::Literal(ScalarValue::UInt64(Some(5)))));
+            .or(col("int_col").gt(Expr::from(ScalarValue::UInt64(Some(5)))));
 
         assert!(can_expr_be_pushed_down_with_schemas(
             &expr,

datafusion/core/src/datasource/physical_plan/parquet/row_group_filter.rs

@@ -1237,10 +1237,10 @@ mod tests {
             .run(
                 lit("1").eq(lit("1")).and(
                     col(r#""String""#)
-                        .eq(Expr::Literal(ScalarValue::Utf8View(Some(String::from(
+                        .eq(Expr::from(ScalarValue::Utf8View(Some(String::from(
                             "Hello_Not_Exists",
                         )))))
-                        .or(col(r#""String""#).eq(Expr::Literal(ScalarValue::Utf8View(
+                        .or(col(r#""String""#).eq(Expr::from(ScalarValue::Utf8View(
                             Some(String::from("Hello_Not_Exists2")),
                         )))),
                 ),
@@ -1322,15 +1322,15 @@ mod tests {
             // generate pruning predicate `(String = "Hello") OR (String = "the quick") OR (String = "are you")`
             .run(
                 col(r#""String""#)
-                    .eq(Expr::Literal(ScalarValue::Utf8View(Some(String::from(
+                    .eq(Expr::from(ScalarValue::Utf8View(Some(String::from(
                         "Hello",
                     )))))
-                    .or(col(r#""String""#).eq(Expr::Literal(ScalarValue::Utf8View(
-                        Some(String::from("the quick")),
-                    ))))
-                    .or(col(r#""String""#).eq(Expr::Literal(ScalarValue::Utf8View(
-                        Some(String::from("are you")),
-                    )))),
+                    .or(col(r#""String""#).eq(Expr::from(ScalarValue::Utf8View(Some(
+                        String::from("the quick"),
+                    )))))
+                    .or(col(r#""String""#).eq(Expr::from(ScalarValue::Utf8View(Some(
+                        String::from("are you"),
+                    ))))),
             )
             .await
     }

datafusion/core/src/physical_optimizer/enforce_distribution.rs

@@ -1722,7 +1722,7 @@ pub(crate) mod tests {
         let predicate = Arc::new(BinaryExpr::new(
             col("c", &schema()).unwrap(),
             Operator::Eq,
-            Arc::new(Literal::new(ScalarValue::Int64(Some(0)))),
+            Arc::new(Literal::from(ScalarValue::Int64(Some(0)))),
         ));
         Arc::new(FilterExec::try_new(predicate, input).unwrap())
     }