[SPARK][VARIANT] Add minimal support for variant type in delta-spark (#2923)

<!--
Thanks for sending a pull request!  Here are some tips for you:
1. If this is your first time, please read our contributor guidelines:
https://github.com/delta-io/delta/blob/master/CONTRIBUTING.md
2. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP]
Your PR title ...'.
  3. Be sure to keep the PR description updated to reflect all changes.
  4. Please write your PR title to summarize what this PR proposes.
5. If possible, provide a concise example to reproduce the issue for a
faster review.
6. If applicable, include the corresponding issue number in the PR title
and link it in the body.
-->

#### Which Delta project/connector is this regarding?
<!--
Please add the component selected below to the beginning of the pull
request title
For example: [Spark] Title of my pull request
-->

- [x] Spark
- [ ] Standalone
- [ ] Flink
- [ ] Kernel
- [ ] Other (fill in here)

## Description

<!--
- Describe what this PR changes.
- Describe why we need the change.
 
If this PR resolves an issue be sure to include "Resolves #XXX" to
correctly link and close the issue upon merge.
-->

Adds the variant table feature to minimally implement the variant type
as described in the RFC in #2867.

Also disables using variant columns as partition columns.

## How was this patch tested?

Added some UTs. More UTs will be merged in followup PRs

tested against both spark 3.5 and 4.0 snapshot with
```
build/sbt -DsparkVersion=latest spark/'testOnly org.apache.spark.sql.delta.DeltaVariantSuite'
build/sbt -DsparkVersion=master spark/'testOnly org.apache.spark.sql.delta.DeltaVariantSuite'
```


## Does this PR introduce _any_ user-facing changes?

<!--
If yes, please clarify the previous behavior and the change this PR
proposes - provide the console output, description and/or an example to
show the behavior difference if possible.
If possible, please also clarify if this is a user-facing change
compared to the released Delta Lake versions or within the unreleased
branches such as master.
If no, write 'No'.
-->

no

Author: richardc-db

spark/src/main/scala-spark-3.5/shims/VariantShims.scala

@@ -0,0 +1,26 @@
+/*
+ * Copyright (2024) The Delta Lake Project Authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.types
+
+object VariantShims {
+
+  /**
+   * Spark's variant type is implemented for Spark 4.0 and is not implemented in Spark 3.5. Thus,
+   * any Spark 3.5 DataType cannot be a variant type.
+   */
+  def isVariantType(dt: DataType): Boolean = false
+}

spark/src/main/scala-spark-master/shims/VariantShims.scala

@@ -0,0 +1,23 @@
+/*
+ * Copyright (2021) The Delta Lake Project Authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.types
+
+object VariantShims {
+
+  /** Spark's variant type is only implemented in Spark 4.0 and above. */
+  def isVariantType(dt: DataType): Boolean = dt.isInstanceOf[VariantType]
+}

spark/src/main/scala/org/apache/spark/sql/delta/TableFeature.scala

@@ -339,7 +339,8 @@ object TableFeature {
       VacuumProtocolCheckTableFeature,
       V2CheckpointTableFeature,
       RowTrackingFeature,
-      InCommitTimestampTableFeature)
+      InCommitTimestampTableFeature,
+      VariantTypeTableFeature)
     if (DeltaUtils.isTesting) {
       features ++= Set(
         TestLegacyWriterFeature,
@@ -502,6 +503,14 @@ object TimestampNTZTableFeature extends ReaderWriterFeature(name = "timestampNtz
   }
 }
 
+object VariantTypeTableFeature extends ReaderWriterFeature(name = "variantType-dev")
+    with FeatureAutomaticallyEnabledByMetadata {
+  override def metadataRequiresFeatureToBeEnabled(
+      metadata: Metadata, spark: SparkSession): Boolean = {
+    SchemaUtils.checkForVariantTypeColumnsRecursively(metadata.schema)
+  }
+}
+
 object DeletionVectorsTableFeature
   extends ReaderWriterFeature(name = "deletionVectors")
   with FeatureAutomaticallyEnabledByMetadata {

spark/src/main/scala/org/apache/spark/sql/delta/schema/SchemaUtils.scala

@@ -1271,6 +1271,14 @@ def normalizeColumnNamesInDataType(
     SchemaUtils.typeExistsRecursively(schema)(_.isInstanceOf[TimestampNTZType])
   }
 
+
+  /**
+   * Returns 'true' if any VariantType exists in the table schema.
+   */
+  def checkForVariantTypeColumnsRecursively(schema: StructType): Boolean = {
+    SchemaUtils.typeExistsRecursively(schema)(VariantShims.isVariantType(_))
+  }
+
   /**
    * Find the unsupported data types in a `DataType` recursively. Add the unsupported data types to
    * the provided `unsupportedDataTypes` buffer.
@@ -1303,6 +1311,7 @@ def normalizeColumnNamesInDataType(
     case DateType =>
     case TimestampType =>
     case TimestampNTZType =>
+    case dt if VariantShims.isVariantType(dt) =>
     case BinaryType =>
     case _: DecimalType =>
     case a: ArrayType =>

spark/src/main/scala/org/apache/spark/sql/delta/util/PartitionUtils.scala

@@ -605,7 +605,8 @@ private[delta] object PartitionUtils {
 
     partitionColumnsSchema(schema, partitionColumns, caseSensitive).foreach {
       field => field.dataType match {
-        case _: AtomicType => // OK
+        // Variant types are not orderable and thus cannot be partition columns.
+        case a: AtomicType if !VariantShims.isVariantType(a) => // OK
         case _ => throw DeltaErrors.cannotUseDataTypeForPartitionColumnError(field)
       }
     }

spark/src/test/scala-spark-master/org/apache/spark/sql/delta/DeltaVariantSuite.scala

@@ -0,0 +1,129 @@
+/*
+ * Copyright (2021) The Delta Lake Project Authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.delta
+
+import org.apache.spark.sql.delta.actions.Protocol
+import org.apache.spark.sql.delta.actions.TableFeatureProtocolUtils
+import org.apache.spark.sql.delta.test.DeltaSQLCommandTest
+
+import org.apache.spark.SparkThrowable
+import org.apache.spark.sql.{AnalysisException, QueryTest, Row}
+import org.apache.spark.sql.catalyst.TableIdentifier
+import org.apache.spark.sql.test.SharedSparkSession
+import org.apache.spark.sql.types.StructType
+
+class DeltaVariantSuite
+    extends QueryTest
+    with SharedSparkSession
+    with DeltaSQLCommandTest {
+
+  private def getProtocolForTable(table: String): Protocol = {
+    val deltaLog = DeltaLog.forTable(spark, TableIdentifier(table))
+    deltaLog.unsafeVolatileSnapshot.protocol
+  }
+
+  test("create a new table with Variant, higher protocol and feature should be picked.") {
+    withTable("tbl") {
+      sql("CREATE TABLE tbl(s STRING, v VARIANT) USING DELTA")
+      sql("INSERT INTO tbl (SELECT 'foo', parse_json(cast(id + 99 as string)) FROM range(1))")
+      assert(spark.table("tbl").selectExpr("v::int").head == Row(99))
+      assert(
+        getProtocolForTable("tbl") ==
+        VariantTypeTableFeature.minProtocolVersion.withFeature(VariantTypeTableFeature)
+      )
+    }
+  }
+
+  test("creating a table without Variant should use the usual minimum protocol") {
+    withTable("tbl") {
+      sql("CREATE TABLE tbl(s STRING, i INTEGER) USING DELTA")
+      assert(getProtocolForTable("tbl") == Protocol(1, 2))
+
+      val deltaLog = DeltaLog.forTable(spark, TableIdentifier("tbl"))
+      assert(
+        !deltaLog.unsafeVolatileSnapshot.protocol.isFeatureSupported(VariantTypeTableFeature),
+        s"Table tbl contains VariantTypeFeature descriptor when its not supposed to"
+      )
+    }
+  }
+
+  test("add a new Variant column should upgrade to the correct protocol versions") {
+    withTable("tbl") {
+      sql("CREATE TABLE tbl(s STRING) USING delta")
+      assert(getProtocolForTable("tbl") == Protocol(1, 2))
+
+      // Should throw error
+      val e = intercept[SparkThrowable] {
+        sql("ALTER TABLE tbl ADD COLUMN v VARIANT")
+      }
+      // capture the existing protocol here.
+      // we will check the error message later in this test as we need to compare the
+      // expected schema and protocol
+      val deltaLog = DeltaLog.forTable(spark, TableIdentifier("tbl"))
+      val currentProtocol = deltaLog.unsafeVolatileSnapshot.protocol
+      val currentFeatures = currentProtocol.implicitlyAndExplicitlySupportedFeatures
+        .map(_.name)
+        .toSeq
+        .sorted
+        .mkString(", ")
+
+      // add table feature
+      sql(
+        s"ALTER TABLE tbl " +
+        s"SET TBLPROPERTIES('delta.feature.variantType-dev' = 'supported')"
+      )
+
+      sql("ALTER TABLE tbl ADD COLUMN v VARIANT")
+
+      // check previously thrown error message
+      checkError(
+        e,
+        errorClass = "DELTA_FEATURES_REQUIRE_MANUAL_ENABLEMENT",
+        parameters = Map(
+          "unsupportedFeatures" -> VariantTypeTableFeature.name,
+          "supportedFeatures" -> currentFeatures
+        )
+      )
+
+      sql("INSERT INTO tbl (SELECT 'foo', parse_json(cast(id + 99 as string)) FROM range(1))")
+      assert(spark.table("tbl").selectExpr("v::int").head == Row(99))
+
+      assert(
+        getProtocolForTable("tbl") ==
+        VariantTypeTableFeature.minProtocolVersion
+          .withFeature(VariantTypeTableFeature)
+          .withFeature(InvariantsTableFeature)
+          .withFeature(AppendOnlyTableFeature)
+      )
+    }
+  }
+
+  test("VariantType may not be used as a partition column") {
+    withTable("delta_test") {
+      checkError(
+        exception = intercept[AnalysisException] {
+          sql(
+            """CREATE TABLE delta_test(s STRING, v VARIANT)
+              |USING delta
+              |PARTITIONED BY (v)""".stripMargin)
+        },
+        errorClass = "INVALID_PARTITION_COLUMN_DATA_TYPE",
+        parameters = Map("type" -> "\"VARIANT\"")
+      )
+    }
+  }
+}