[SPARK-59147][SQL] Preserve struct nullness when SELECT * EXCEPT drops a nested field

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/unresolved.scala

@@ -760,8 +760,18 @@ case class UnresolvedStarExceptOrReplace(
                   col.toAttribute -> nestedExcept.tail
               }.get
           }
-          Alias(CreateStruct(
-            filterColumns(extractedFields.toImmutableArraySeq, newExcepts)), col.name)()
+          val newStruct = CreateStruct(
+            filterColumns(extractedFields.toImmutableArraySeq, newExcepts))
+          // Dropping a nested field must not change the nullness of the enclosing struct.
+          // CreateStruct is never nullable, so a NULL struct would otherwise be rebuilt as a
+          // non-NULL struct whose remaining fields are NULL. Guard it the same way UpdateFields
+          // does for DropField.
+          val newCol = if (col.nullable) {
+            If(IsNull(col), Literal(null, newStruct.dataType), newStruct)
+          } else {
+            newStruct
+          }
+          Alias(newCol, col.name)()
         // if there are multiple nestedExcepts but one is empty we must have overlapping except
         // columns. throw an error.
         case (_, Some(nestedExcepts)) if nestedExcepts.size > 1 =>

sql/core/src/test/resources/sql-tests/analyzer-results/pipe-operators.sql.out

@@ -1103,7 +1103,7 @@ Project [x#x, y#x]
 table st
 |> drop col.i1
 -- !query analysis
-Project [x#x, named_struct(i2, col#x.i2) AS col#x]
+Project [x#x, if (isnull(col#x)) null else named_struct(i2, col#x.i2) AS col#x]
 +- SubqueryAlias spark_catalog.default.st
    +- Relation spark_catalog.default.st[x#x,col#x] parquet
 
@@ -1112,7 +1112,7 @@ Project [x#x, named_struct(i2, col#x.i2) AS col#x]
 table st
 |> drop col.i1, col.i2
 -- !query analysis
-Project [x#x, named_struct() AS col#x]
+Project [x#x, if (isnull(col#x)) null else named_struct() AS col#x]
 +- SubqueryAlias spark_catalog.default.st
    +- Relation spark_catalog.default.st[x#x,col#x] parquet
 

sql/core/src/test/resources/sql-tests/analyzer-results/selectExcept.sql.out

@@ -530,3 +530,78 @@ Project [x#x]
          +- Project [cast(c1#x as int) AS c1#x, cast(c2#x as int) AS c2#x, cast(c3#x as void) AS c3#x, cast(c4#x as int) AS c4#x, cast(c5#x as int) AS c5#x]
             +- SubqueryAlias T
                +- LocalRelation [c1#x, c2#x, c3#x, c4#x, c5#x]
+
+
+-- !query
+CREATE TEMPORARY VIEW nullable_struct AS SELECT * FROM VALUES
+  (0, CAST(NULL AS STRUCT<a: INT, b: INT, s2: STRUCT<c: INT, d: INT>>)),
+  (1, named_struct("a", 1, "b", 11, "s2", CAST(NULL AS STRUCT<c: INT, d: INT>))),
+  (2, named_struct("a", 2, "b", 22, "s2", named_struct("c", 222, "d", 2222)))
+AS nullable_struct(id, data)
+-- !query analysis
+CreateViewCommand `nullable_struct`, SELECT * FROM VALUES
+  (0, CAST(NULL AS STRUCT<a: INT, b: INT, s2: STRUCT<c: INT, d: INT>>)),
+  (1, named_struct("a", 1, "b", 11, "s2", CAST(NULL AS STRUCT<c: INT, d: INT>))),
+  (2, named_struct("a", 2, "b", 22, "s2", named_struct("c", 222, "d", 2222)))
+AS nullable_struct(id, data), false, false, LocalTempView, UNSUPPORTED, true
+   +- Project [id#x, data#x]
+      +- SubqueryAlias nullable_struct
+         +- LocalRelation [id#x, data#x]
+
+
+-- !query
+SELECT * EXCEPT (data.a) FROM nullable_struct
+-- !query analysis
+Project [id#x, if (isnull(data#x)) null else named_struct(b, data#x.b, s2, data#x.s2) AS data#x]
++- SubqueryAlias nullable_struct
+   +- View (`nullable_struct`, [id#x, data#x])
+      +- Project [cast(id#x as int) AS id#x, cast(data#x as struct<a:int,b:int,s2:struct<c:int,d:int>>) AS data#x]
+         +- Project [id#x, data#x]
+            +- SubqueryAlias nullable_struct
+               +- LocalRelation [id#x, data#x]
+
+
+-- !query
+SELECT id, data IS NULL FROM (SELECT * EXCEPT (data.a) FROM nullable_struct)
+-- !query analysis
+Project [id#x, isnull(data#x) AS (data IS NULL)#x]
++- SubqueryAlias __auto_generated_subquery_name
+   +- Project [id#x, if (isnull(data#x)) null else named_struct(b, data#x.b, s2, data#x.s2) AS data#x]
+      +- SubqueryAlias nullable_struct
+         +- View (`nullable_struct`, [id#x, data#x])
+            +- Project [cast(id#x as int) AS id#x, cast(data#x as struct<a:int,b:int,s2:struct<c:int,d:int>>) AS data#x]
+               +- Project [id#x, data#x]
+                  +- SubqueryAlias nullable_struct
+                     +- LocalRelation [id#x, data#x]
+
+
+-- !query
+SELECT id, data IS NULL, data.s2 IS NULL FROM (SELECT * EXCEPT (data.s2.c) FROM nullable_struct)
+-- !query analysis
+Project [id#x, isnull(data#x) AS (data IS NULL)#x, isnull(data#x.s2) AS (data.s2 IS NULL)#x]
++- SubqueryAlias __auto_generated_subquery_name
+   +- Project [id#x, if (isnull(data#x)) null else named_struct(a, data#x.a, b, data#x.b, s2, if (isnull(data#x.s2)) null else named_struct(d, data#x.s2.d)) AS data#x]
+      +- SubqueryAlias nullable_struct
+         +- View (`nullable_struct`, [id#x, data#x])
+            +- Project [cast(id#x as int) AS id#x, cast(data#x as struct<a:int,b:int,s2:struct<c:int,d:int>>) AS data#x]
+               +- Project [id#x, data#x]
+                  +- SubqueryAlias nullable_struct
+                     +- LocalRelation [id#x, data#x]
+
+
+-- !query
+SELECT data.* EXCEPT (s2.c) FROM nullable_struct
+-- !query analysis
+Project [data#x.a AS a#x, data#x.b AS b#x, if (isnull(data#x.s2)) null else named_struct(d, data#x.s2.d) AS s2#x]
++- SubqueryAlias nullable_struct
+   +- View (`nullable_struct`, [id#x, data#x])
+      +- Project [cast(id#x as int) AS id#x, cast(data#x as struct<a:int,b:int,s2:struct<c:int,d:int>>) AS data#x]
+         +- Project [id#x, data#x]
+            +- SubqueryAlias nullable_struct
+               +- LocalRelation [id#x, data#x]
+
+
+-- !query
+DROP VIEW nullable_struct
+-- !query analysis
+DropTempViewCommand nullable_struct, false

sql/core/src/test/resources/sql-tests/inputs/selectExcept.sql

@@ -76,3 +76,16 @@ SELECT 1 FROM v1 WHERE 4 IN (*);
 SELECT T.* FROM v1, LATERAL (SELECT  v1.*) AS T(c1, c2, c3, c4, c5);
 SELECT T.* FROM v1, LATERAL (SELECT  COALESCE(v1.*)) AS T(x);
 
+-- EXCEPT a nested field preserves the nullness of the enclosing struct
+CREATE TEMPORARY VIEW nullable_struct AS SELECT * FROM VALUES
+  (0, CAST(NULL AS STRUCT<a: INT, b: INT, s2: STRUCT<c: INT, d: INT>>)),
+  (1, named_struct("a", 1, "b", 11, "s2", CAST(NULL AS STRUCT<c: INT, d: INT>))),
+  (2, named_struct("a", 2, "b", 22, "s2", named_struct("c", 222, "d", 2222)))
+AS nullable_struct(id, data);
+
+SELECT * EXCEPT (data.a) FROM nullable_struct;
+SELECT id, data IS NULL FROM (SELECT * EXCEPT (data.a) FROM nullable_struct);
+SELECT id, data IS NULL, data.s2 IS NULL FROM (SELECT * EXCEPT (data.s2.c) FROM nullable_struct);
+SELECT data.* EXCEPT (s2.c) FROM nullable_struct;
+
+DROP VIEW nullable_struct;

sql/core/src/test/resources/sql-tests/results/selectExcept.sql.out

@@ -494,3 +494,63 @@ SELECT T.* FROM v1, LATERAL (SELECT  COALESCE(v1.*)) AS T(x)
 struct<x:int>
 -- !query output
 1
+
+
+-- !query
+CREATE TEMPORARY VIEW nullable_struct AS SELECT * FROM VALUES
+  (0, CAST(NULL AS STRUCT<a: INT, b: INT, s2: STRUCT<c: INT, d: INT>>)),
+  (1, named_struct("a", 1, "b", 11, "s2", CAST(NULL AS STRUCT<c: INT, d: INT>))),
+  (2, named_struct("a", 2, "b", 22, "s2", named_struct("c", 222, "d", 2222)))
+AS nullable_struct(id, data)
+-- !query schema
+struct<>
+-- !query output
+
+
+
+-- !query
+SELECT * EXCEPT (data.a) FROM nullable_struct
+-- !query schema
+struct<id:int,data:struct<b:int,s2:struct<c:int,d:int>>>
+-- !query output
+0	NULL
+1	{"b":11,"s2":null}
+2	{"b":22,"s2":{"c":222,"d":2222}}
+
+
+-- !query
+SELECT id, data IS NULL FROM (SELECT * EXCEPT (data.a) FROM nullable_struct)
+-- !query schema
+struct<id:int,(data IS NULL):boolean>
+-- !query output
+0	true
+1	false
+2	false
+
+
+-- !query
+SELECT id, data IS NULL, data.s2 IS NULL FROM (SELECT * EXCEPT (data.s2.c) FROM nullable_struct)
+-- !query schema
+struct<id:int,(data IS NULL):boolean,(data.s2 IS NULL):boolean>
+-- !query output
+0	true	true
+1	false	true
+2	false	false
+
+
+-- !query
+SELECT data.* EXCEPT (s2.c) FROM nullable_struct
+-- !query schema
+struct<a:int,b:int,s2:struct<d:int>>
+-- !query output
+1	11	NULL
+2	22	{"d":2222}
+NULL	NULL	NULL
+
+
+-- !query
+DROP VIEW nullable_struct
+-- !query schema
+struct<>
+-- !query output
+