[SPARK] Support CTAS from a table with a deleted default value

spark/src/main/scala/org/apache/spark/sql/delta/commands/CreateDeltaTableCommand.scala

@@ -29,7 +29,7 @@ import org.apache.spark.sql.delta.coordinatedcommits.CoordinatedCommitsUtils
 import org.apache.spark.sql.delta.hooks.{HudiConverterHook, IcebergConverterHook, UpdateCatalog, UpdateCatalogFactory}
 import org.apache.spark.sql.delta.logging.DeltaLogKeys
 import org.apache.spark.sql.delta.metering.DeltaLogging
-import org.apache.spark.sql.delta.schema.SchemaUtils
+import org.apache.spark.sql.delta.schema.{SchemaMergingUtils, SchemaUtils}
 import org.apache.spark.sql.delta.sources.DeltaSQLConf
 import org.apache.hadoop.conf.Configuration
 import org.apache.hadoop.fs.{FileSystem, Path}
@@ -40,12 +40,13 @@ import org.apache.spark.sql._
 import org.apache.spark.sql.catalyst.catalog.{CatalogTable, CatalogTableType}
 import org.apache.spark.sql.catalyst.expressions.Attribute
 import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.util.ResolveDefaultColumnsUtils
 import org.apache.spark.sql.connector.catalog.Identifier
 import org.apache.spark.sql.execution.command.{LeafRunnableCommand, RunnableCommand}
 import org.apache.spark.sql.execution.metric.SQLMetric
 import org.apache.spark.sql.execution.metric.SQLMetrics.createMetric
 import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.types.{MetadataBuilder, StructField, StructType}
 
 /**
  * Single entry point for all write or declaration operations for Delta tables accessed through
@@ -314,7 +315,29 @@ case class CreateDeltaTableCommand(
             txn.snapshot.domainMetadata, newDomainMetadata)
         taggedCommitData = taggedCommitData.copy(actions = newActions)
       }
-      val op = getOperation(txn.metadata, isManagedTable, Some(options),
+
+      def schemaContainsExistsColumnMetadata(schema: StructType): Boolean = {
+        schema.exists(
+          _.metadata.contains(ResolveDefaultColumnsUtils.EXISTS_DEFAULT_COLUMN_METADATA_KEY))
+      }
+      val metadataForOp =
+        if (schemaContainsExistsColumnMetadata(txn.metadata.schema)) {
+          val newSchema = SchemaMergingUtils.transformColumns(schema) {
+            case (_, field: StructField, _) if field.metadata.contains(
+                ResolveDefaultColumnsUtils.EXISTS_DEFAULT_COLUMN_METADATA_KEY) =>
+              val newMetadata = new MetadataBuilder()
+                .withMetadata(field.metadata)
+                .remove(ResolveDefaultColumnsUtils.EXISTS_DEFAULT_COLUMN_METADATA_KEY)
+                .build()
+              field.copy(metadata = newMetadata)
+            case (_, field: StructField, _) => field
+          }
+          txn.metadata.copy(schemaString = newSchema.json)
+        } else {
+          txn.metadata
+        }
+
+      val op = getOperation(metadataForOp, isManagedTable, Some(options),
         clusterBy = ClusteredTableUtils.getLogicalClusteringColumnNames(
           txn, taggedCommitData.actions)
       )

spark/src/test/scala/org/apache/spark/sql/delta/DeltaTableCreationTests.scala

@@ -2409,6 +2409,31 @@ class DeltaTableCreationSuite
       }
     }
   }
+
+  test("CREATE TABLE selecting from a table with removed column defaults") {
+      // Defaults are only possible for top level columns.
+      sql("""CREATE TABLE test_table(int_col INT DEFAULT 2)
+            |USING delta
+            |TBLPROPERTIES ('delta.feature.allowColumnDefaults' = 'supported')""".stripMargin)
+
+      def defaultsTableFeatureEnabled(tableName: String): Boolean = {
+        val deltaLog = DeltaLog.forTable(spark, TableIdentifier(tableName))
+        val protocol = deltaLog.unsafeVolatileSnapshot.protocol
+        protocol.writerFeatureNames.contains(AllowColumnDefaultsTableFeature.name)
+      }
+
+      assert(defaultsTableFeatureEnabled("test_table"))
+
+      // @TODO: It is currently not possible to CTAS from a table with an active column default
+      //        without explicitly enabling the table feature.
+      assertThrows[DeltaAnalysisException] {
+        sql("CREATE TABLE test_table_2 USING DELTA AS SELECT * FROM test_table")
+      }
+      sql("ALTER TABLE test_table ALTER COLUMN int_col DROP DEFAULT")
+      sql("CREATE TABLE test_table_3 USING DELTA AS SELECT * FROM test_table")
+
+      assert(!defaultsTableFeatureEnabled("test_table_3"))
+  }
 }
 
 trait DeltaTableCreationColumnMappingSuiteBase extends DeltaColumnMappingSelectedTestMixin {