diff --git a/datafusion/functions/src/datetime/now.rs b/datafusion/functions/src/datetime/now.rs
index b2221215b94b..74eb5aea4255 100644
--- a/datafusion/functions/src/datetime/now.rs
+++ b/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
+    }
 }
diff --git a/datafusion/physical-plan/src/projection.rs b/datafusion/physical-plan/src/projection.rs
index 49bf05964226..a28328fb5d43 100644
--- a/datafusion/physical-plan/src/projection.rs
+++ b/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()
diff --git a/datafusion/sqllogictest/src/test_context.rs b/datafusion/sqllogictest/src/test_context.rs
index 9a0db1c41c71..2143b3089ee5 100644
--- a/datafusion/sqllogictest/src/test_context.rs
+++ b/datafusion/sqllogictest/src/test_context.rs
@@ -319,11 +319,17 @@ 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),
@@ -331,6 +337,11 @@ pub async fn register_metadata_tables(ctx: &SessionContext) {
             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();
diff --git a/datafusion/sqllogictest/test_files/metadata.slt b/datafusion/sqllogictest/test_files/metadata.slt
index d0853b9e4983..588a36e3d515 100644
--- a/datafusion/sqllogictest/test_files/metadata.slt
+++ b/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;