diff --git a/datafusion/src/logical_plan/builder.rs b/datafusion/src/logical_plan/builder.rs
index 09c3a14513e50..d1360c332e836 100644
--- a/datafusion/src/logical_plan/builder.rs
+++ b/datafusion/src/logical_plan/builder.rs
@@ -149,7 +149,7 @@ impl LogicalPlanBuilder {
                 .iter()
                 .enumerate()
                 .map(|(j, expr)| {
-                    if let Expr::Literal(ScalarValue::Utf8(None)) = expr {
+                    if let Expr::Literal(ScalarValue::Null) = expr {
                         nulls.push((i, j));
                         Ok(field_types[j].clone())
                     } else {
diff --git a/datafusion/src/physical_plan/planner.rs b/datafusion/src/physical_plan/planner.rs
index fd0421b5e0bad..d1a9dbdb77726 100644
--- a/datafusion/src/physical_plan/planner.rs
+++ b/datafusion/src/physical_plan/planner.rs
@@ -1087,7 +1087,7 @@ impl DefaultPhysicalPlanner {
                 list,
                 negated,
             } => match expr.as_ref() {
-                Expr::Literal(ScalarValue::Utf8(None)) => {
+                Expr::Literal(ScalarValue::Null) => {
                     Ok(expressions::lit(ScalarValue::Boolean(None)))
                 }
                 _ => {
@@ -1102,7 +1102,7 @@ impl DefaultPhysicalPlanner {
                     let list_exprs = list
                         .iter()
                         .map(|expr| match expr {
-                            Expr::Literal(ScalarValue::Utf8(None)) => self
+                            Expr::Literal(ScalarValue::Null) => self
                                 .create_physical_expr(
                                     expr,
                                     input_dfschema,
diff --git a/datafusion/src/scalar.rs b/datafusion/src/scalar.rs
index 00586bf5549ef..95326ca05e707 100644
--- a/datafusion/src/scalar.rs
+++ b/datafusion/src/scalar.rs
@@ -37,6 +37,8 @@ use std::{convert::TryFrom, fmt, iter::repeat, sync::Arc};
 /// This is the single-valued counter-part of arrow’s `Array`.
 #[derive(Clone)]
 pub enum ScalarValue {
+    /// untyped null
+    Null,
     /// true or false value
     Boolean(Option<bool>),
     /// 32bit float
@@ -100,6 +102,7 @@ impl PartialEq for ScalarValue {
         // any newly added enum variant will require editing this list
         // or else face a compile error
         match (self, other) {
+            (Null, _) | (_, Null) => false,
             (Boolean(v1), Boolean(v2)) => v1.eq(v2),
             (Boolean(_), _) => false,
             (Float32(v1), Float32(v2)) => {
@@ -171,6 +174,7 @@ impl PartialOrd for ScalarValue {
         // any newly added enum variant will require editing this list
         // or else face a compile error
         match (self, other) {
+            (Null, _) | (_, Null) => None,
             (Boolean(v1), Boolean(v2)) => v1.partial_cmp(v2),
             (Boolean(_), _) => None,
             (Float32(v1), Float32(v2)) => {
@@ -253,6 +257,7 @@ impl std::hash::Hash for ScalarValue {
     fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
         use ScalarValue::*;
         match self {
+            Null => {}
             Boolean(v) => v.hash(state),
             Float32(v) => {
                 let v = v.map(OrderedFloat);
@@ -456,6 +461,7 @@ impl ScalarValue {
     /// Getter for the `DataType` of the value
     pub fn get_datatype(&self) -> DataType {
         match self {
+            ScalarValue::Null => DataType::Null,
             ScalarValue::Boolean(_) => DataType::Boolean,
             ScalarValue::UInt8(_) => DataType::UInt8,
             ScalarValue::UInt16(_) => DataType::UInt16,
@@ -521,7 +527,8 @@ impl ScalarValue {
     pub fn is_null(&self) -> bool {
         matches!(
             *self,
-            ScalarValue::Boolean(None)
+            ScalarValue::Null
+                | ScalarValue::Boolean(None)
                 | ScalarValue::UInt8(None)
                 | ScalarValue::UInt16(None)
                 | ScalarValue::UInt32(None)
@@ -908,6 +915,7 @@ impl ScalarValue {
     /// Converts a scalar value into an array of `size` rows.
     pub fn to_array_of_size(&self, size: usize) -> ArrayRef {
         match self {
+            ScalarValue::Null => new_null_array(&DataType::Null, size),
             ScalarValue::Boolean(e) => {
                 Arc::new(BooleanArray::from(vec![*e; size])) as ArrayRef
             }
@@ -1082,6 +1090,7 @@ impl ScalarValue {
         }
 
         Ok(match array.data_type() {
+            DataType::Null => ScalarValue::Null,
             DataType::Boolean => typed_cast!(array, index, BooleanArray, Boolean),
             DataType::Float64 => typed_cast!(array, index, Float64Array, Float64),
             DataType::Float32 => typed_cast!(array, index, Float32Array, Float32),
@@ -1218,6 +1227,7 @@ impl ScalarValue {
         }
 
         match self {
+            ScalarValue::Null => false,
             ScalarValue::Boolean(val) => {
                 eq_array_primitive!(array, index, BooleanArray, val)
             }
@@ -1443,6 +1453,7 @@ impl TryFrom<&DataType> for ScalarValue {
     /// Create a Null instance of ScalarValue for this datatype
     fn try_from(datatype: &DataType) -> Result<Self> {
         Ok(match datatype {
+            DataType::Null => ScalarValue::Null,
             DataType::Boolean => ScalarValue::Boolean(None),
             DataType::Float64 => ScalarValue::Float64(None),
             DataType::Float32 => ScalarValue::Float32(None),
@@ -1501,6 +1512,7 @@ macro_rules! format_option {
 impl fmt::Display for ScalarValue {
     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
         match self {
+            ScalarValue::Null => write!(f, "NULL")?,
             ScalarValue::Boolean(e) => format_option!(f, e)?,
             ScalarValue::Float32(e) => format_option!(f, e)?,
             ScalarValue::Float64(e) => format_option!(f, e)?,
@@ -1575,6 +1587,7 @@ impl fmt::Display for ScalarValue {
 impl fmt::Debug for ScalarValue {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         match self {
+            ScalarValue::Null => write!(f, "UntypedNull"),
             ScalarValue::Boolean(_) => write!(f, "Boolean({})", self),
             ScalarValue::Float32(_) => write!(f, "Float32({})", self),
             ScalarValue::Float64(_) => write!(f, "Float64({})", self),
diff --git a/datafusion/src/sql/planner.rs b/datafusion/src/sql/planner.rs
index 60d2da8be2c71..96f9bdbcca410 100644
--- a/datafusion/src/sql/planner.rs
+++ b/datafusion/src/sql/planner.rs
@@ -1162,7 +1162,7 @@ impl<'a, S: ContextProvider> SqlToRel<'a, S> {
                             Ok(lit(s.clone()))
                         }
                         SQLExpr::Value(Value::Null) => {
-                            Ok(Expr::Literal(ScalarValue::Utf8(None)))
+                            Ok(Expr::Literal(ScalarValue::Null))
                         }
                         SQLExpr::Value(Value::Boolean(n)) => Ok(lit(*n)),
                         SQLExpr::UnaryOp { ref op, ref expr } => {
@@ -1189,7 +1189,7 @@ impl<'a, S: ContextProvider> SqlToRel<'a, S> {
             SQLExpr::Value(Value::Number(n, _)) => parse_sql_number(n),
             SQLExpr::Value(Value::SingleQuotedString(ref s)) => Ok(lit(s.clone())),
             SQLExpr::Value(Value::Boolean(n)) => Ok(lit(*n)),
-            SQLExpr::Value(Value::Null) => Ok(Expr::Literal(ScalarValue::Utf8(None))),
+            SQLExpr::Value(Value::Null) => Ok(Expr::Literal(ScalarValue::Null)),
             SQLExpr::Extract { field, expr } => Ok(Expr::ScalarFunction {
                 fun: functions::BuiltinScalarFunction::DatePart,
                 args: vec![