Fix logical vs physical schema mismatch for aliased now() (#12951)

* test: reproducer of error with aliased now field

* fix: now() UDF is not nullable

* fix: when extracting metadata from expr, handle CastExpr

* test: update tests now that the fixes are working

datafusion/functions/src/datetime/now.rs

@@ -21,7 +21,7 @@ use arrow::datatypes::DataType;
 use arrow::datatypes::DataType::Timestamp;
 use arrow::datatypes::TimeUnit::Nanosecond;
 
-use datafusion_common::{internal_err, Result, ScalarValue};
+use datafusion_common::{internal_err, ExprSchema, Result, ScalarValue};
 use datafusion_expr::simplify::{ExprSimplifyResult, SimplifyInfo};
 use datafusion_expr::{ColumnarValue, Expr, ScalarUDFImpl, Signature, Volatility};
 
@@ -84,4 +84,8 @@ impl ScalarUDFImpl for NowFunc {
             ScalarValue::TimestampNanosecond(now_ts, Some("+00:00".into())),
         )))
     }
+
+    fn is_nullable(&self, _args: &[Expr], _schema: &dyn ExprSchema) -> bool {
+        false
+    }
 }

datafusion/physical-plan/src/projection.rs

@@ -40,7 +40,7 @@ use datafusion_common::stats::Precision;
 use datafusion_common::Result;
 use datafusion_execution::TaskContext;
 use datafusion_physical_expr::equivalence::ProjectionMapping;
-use datafusion_physical_expr::expressions::Literal;
+use datafusion_physical_expr::expressions::{CastExpr, Literal};
 
 use crate::execution_plan::CardinalityEffect;
 use futures::stream::{Stream, StreamExt};
@@ -246,6 +246,10 @@ pub(crate) fn get_field_metadata(
     e: &Arc<dyn PhysicalExpr>,
     input_schema: &Schema,
 ) -> Option<HashMap<String, String>> {
+    if let Some(cast) = e.as_any().downcast_ref::<CastExpr>() {
+        return get_field_metadata(cast.expr(), input_schema);
+    }
+
     // Look up field by index in schema (not NAME as there can be more than one
     // column with the same name)
     e.as_any()

datafusion/sqllogictest/src/test_context.rs

@@ -319,18 +319,29 @@ pub async fn register_metadata_tables(ctx: &SessionContext) {
             String::from("metadata_key"),
             String::from("the l_name field"),
         )]));
+    let ts = Field::new("ts", DataType::Timestamp(TimeUnit::Nanosecond, None), false)
+        .with_metadata(HashMap::from([(
+            String::from("metadata_key"),
+            String::from("ts non-nullable field"),
+        )]));
 
-    let schema = Schema::new(vec![id, name, l_name]).with_metadata(HashMap::from([(
-        String::from("metadata_key"),
-        String::from("the entire schema"),
-    )]));
+    let schema =
+        Schema::new(vec![id, name, l_name, ts]).with_metadata(HashMap::from([(
+            String::from("metadata_key"),
+            String::from("the entire schema"),
+        )]));
 
     let batch = RecordBatch::try_new(
         Arc::new(schema),
         vec![
             Arc::new(Int32Array::from(vec![Some(1), None, Some(3)])) as _,
             Arc::new(StringArray::from(vec![None, Some("bar"), Some("baz")])) as _,
             Arc::new(StringArray::from(vec![None, Some("l_bar"), Some("l_baz")])) as _,
+            Arc::new(TimestampNanosecondArray::from(vec![
+                1599572549190855123,
+                1599572549190855123,
+                1599572549190855123,
+            ])) as _,
         ],
     )
     .unwrap();

datafusion/sqllogictest/test_files/metadata.slt

@@ -125,5 +125,21 @@ NULL NULL l_bar
 
 
 
+query P rowsort
+SELECT ts
+FROM ((
+    SELECT now() AS ts
+    FROM table_with_metadata
+) UNION ALL (
+        SELECT ts
+    FROM table_with_metadata
+))
+GROUP BY ts
+ORDER BY ts
+LIMIT 1;
+----
+2020-09-08T13:42:29.190855123Z
+
+
 statement ok
 drop table table_with_metadata;