[CARMEL-5867] ShowCreateTableCommand and DescribeTableCommand miss nullable information (#880)

Author: fenzhu

sql/catalyst/src/main/scala/org/apache/spark/sql/types/StructField.scala

@@ -93,6 +93,8 @@ case class StructField(
       .map(escapeSingleQuotedString)
       .map(" COMMENT '" + _ + "'")
 
-    s"${quoteIdentifier(name)} ${dataType.sql}${comment.getOrElse("")}"
+    val notNull = if (nullable) "" else " NOT NULL"
+
+    s"${quoteIdentifier(name)} ${dataType.sql}$notNull${comment.getOrElse("")}"
   }
 }

sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala

@@ -972,12 +972,15 @@ abstract class DescribeCommandBase extends RunnableCommand {
   protected def describeSchema(
       schema: StructType,
       buffer: ArrayBuffer[Row],
-      header: Boolean): Unit = {
+      header: Boolean,
+      ignoreNullable: Boolean = true): Unit = {
     if (header) {
       append(buffer, s"# ${output.head.name}", output(1).name, output(2).name)
     }
     schema.foreach { column =>
-      append(buffer, column.name, column.dataType.simpleString, column.getComment().orNull)
+      val columnNullableInfo = if (ignoreNullable || column.nullable) "" else " not null"
+      val columnTypeInfo = column.dataType.simpleString + columnNullableInfo
+      append(buffer, column.name, columnTypeInfo, column.getComment().orNull)
     }
   }
 
@@ -1001,24 +1004,28 @@ case class DescribeTableCommand(
   override def run(sparkSession: SparkSession): Seq[Row] = {
     val result = new ArrayBuffer[Row]
     val catalog = sparkSession.sessionState.catalog
+    val enforceNotNull = sparkSession.sessionState.conf.enforceSchemaNotNull
 
     if (catalog.isTemporaryView(table)) {
       if (partitionSpec.nonEmpty) {
         throw new AnalysisException(
           s"DESC PARTITION is not allowed on a temporary view: ${table.identifier}")
       }
-      describeSchema(catalog.lookupRelation(table).schema, result, header = false)
+      describeSchema(catalog.lookupRelation(table).schema, result,
+        header = false, ignoreNullable = !enforceNotNull)
     } else {
       val metadata = catalog.getTableMetadata(table)
       if (metadata.schema.isEmpty) {
         // In older version(prior to 2.1) of Spark, the table schema can be empty and should be
         // inferred at runtime. We should still support it.
-        describeSchema(sparkSession.table(metadata.identifier).schema, result, header = false)
+        describeSchema(sparkSession.table(metadata.identifier).schema, result,
+          header = false, ignoreNullable = !enforceNotNull)
       } else {
-        describeSchema(metadata.schema, result, header = false)
+        describeSchema(metadata.schema, result,
+          header = false, ignoreNullable = !enforceNotNull)
       }
 
-      describePartitionInfo(metadata, result)
+      describePartitionInfo(metadata, result, !enforceNotNull)
 
       if (partitionSpec.nonEmpty) {
         // Outputs the partition-specific info for the DDL command:
@@ -1032,10 +1039,12 @@ case class DescribeTableCommand(
     result
   }
 
-  private def describePartitionInfo(table: CatalogTable, buffer: ArrayBuffer[Row]): Unit = {
+  private def describePartitionInfo(table: CatalogTable,
+      buffer: ArrayBuffer[Row], ignoreNullable: Boolean): Unit = {
     if (table.partitionColumnNames.nonEmpty) {
       append(buffer, "# Partition Information", "", "")
-      describeSchema(table.partitionSchema, buffer, header = true)
+      describeSchema(table.partitionSchema, buffer,
+        header = true, ignoreNullable = ignoreNullable)
     }
   }
 
@@ -1111,7 +1120,9 @@ case class DescribeQueryCommand(queryText: String, plan: LogicalPlan)
   override def run(sparkSession: SparkSession): Seq[Row] = {
     val result = new ArrayBuffer[Row]
     val queryExecution = sparkSession.sessionState.executePlan(plan)
-    describeSchema(queryExecution.analyzed.schema, result, header = false)
+    val enforceNotNull = sparkSession.sessionState.conf.enforceSchemaNotNull
+    describeSchema(queryExecution.analyzed.schema, result,
+      header = false, ignoreNullable = !enforceNotNull)
     result
   }
 }

sql/core/src/test/scala/org/apache/spark/sql/ShowCreateTableSuite.scala

@@ -19,6 +19,7 @@ package org.apache.spark.sql
 
 import org.apache.spark.sql.catalyst.TableIdentifier
 import org.apache.spark.sql.catalyst.catalog.CatalogTable
+import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.test.{SharedSparkSession, SQLTestUtils}
 import org.apache.spark.util.Utils
 
@@ -179,6 +180,26 @@ abstract class ShowCreateTableSuite extends QueryTest with SQLTestUtils {
     }
   }
 
+  test("CARMEL-5867: ShowCreateTableCommand and DescribeTableCommand miss nullable information") {
+    withSQLConf(SQLConf.ENFORCE_SCHEMA_NOT_NULL.key -> "true") {
+      withTable("tab") {
+        sql("create table tab(id int not null comment 'test'," +
+          " name string, score double not null) using parquet")
+        checkAnswer(
+          sql("show create table tab"),
+          Row("CREATE TABLE `default`.`tab` (" +
+            "\n  `id` INT NOT NULL COMMENT 'test'," +
+            "\n  `name` STRING," +
+            "\n  `score` DOUBLE NOT NULL)" +
+            "\nUSING parquet" +
+            "\n"))
+        val descResult = sql("desc extended tab").collect()
+        val res = descResult.filter(_.getString(1).contains(" not null"))
+        assert(res.length == 2)
+      }
+    }
+  }
+
   protected def getShowDDL(showCreateTableSql: String): String = {
     val result = sql(showCreateTableSql)
       .head()