Mark scalar subquery results nullable

datafusion/expr/src/expr_schema.rs

@@ -374,9 +374,8 @@ impl ExprSchemable for Expr {
 
                 Ok(expr_nullable | subquery_nullable)
             }
-            Expr::ScalarSubquery(subquery) => {
-                Ok(subquery.subquery.schema().field(0).is_nullable())
-            }
+            // A scalar subquery returns NULL when it produces no rows.
+            Expr::ScalarSubquery(_) => Ok(true),
             Expr::BinaryExpr(BinaryExpr { left, right, .. }) => {
                 Ok(left.nullable(input_schema)? || right.nullable(input_schema)?)
             }
@@ -522,9 +521,12 @@ impl ExprSchemable for Expr {
             | Expr::Exists { .. } => {
                 Ok(Arc::new(Field::new(&schema_name, DataType::Boolean, false)))
             }
-            Expr::ScalarSubquery(subquery) => {
-                Ok(Arc::clone(&subquery.subquery.schema().fields()[0]))
-            }
+            Expr::ScalarSubquery(subquery) => Ok(Arc::new(
+                subquery.subquery.schema().fields()[0]
+                    .as_ref()
+                    .clone()
+                    .with_nullable(true),
+            )),
             Expr::BinaryExpr(BinaryExpr { left, right, op }) => {
                 let (left_field, right_field) =
                     (left.to_field(schema)?.1, right.to_field(schema)?.1);
@@ -1035,6 +1037,22 @@ mod tests {
         assert!(expr.nullable(&get_schema(false)).unwrap());
     }
 
+    #[test]
+    fn scalar_subquery_is_nullable_even_when_its_field_is_not() -> Result<()> {
+        let subquery = LogicalPlanBuilder::empty(false)
+            .project(vec![lit(1)])?
+            .build()?;
+        assert!(!subquery.schema().field(0).is_nullable());
+
+        let expr = crate::scalar_subquery(Arc::new(subquery));
+        assert!(expr.nullable(&MockExprSchema::new())?);
+
+        let field = expr.to_field(&MockExprSchema::new())?.1;
+        assert_eq!(field.data_type(), &DataType::Int32);
+        assert!(field.is_nullable());
+        Ok(())
+    }
+
     #[test]
     fn expr_schema_data_type() {
         let expr = col("foo");