Reject reads on table with unsupported type changes (delta-io#2787)

## Description

Adds a guardrail for the type widening table feature to reject reads
when an unsupported change was applied to the table.
This should never happen unless an implementation doesn't respect the
type widening feature specification, which explicitly lists type changes
that are allowed.

## How was this patch tested?
- Added a test manually committing an invalid type change.

Author: johanl-db

spark/src/main/resources/error/delta-error-classes.json

@@ -2655,6 +2655,12 @@
     ],
     "sqlState" : "0AKDC"
   },
+  "DELTA_UNSUPPORTED_TYPE_CHANGE_IN_SCHEMA" : {
+    "message" : [
+      "Unable to operate on this table because an unsupported type change was applied. Field <fieldName> was changed from <fromType> to <toType>."
+    ],
+    "sqlState" : "0AKDC"
+  },
   "DELTA_UNSUPPORTED_VACUUM_SPECIFIC_PARTITION" : {
     "message" : [
       "Please provide the base path (<baseDeltaPath>) when Vacuuming Delta tables. Vacuuming specific partitions is currently not supported."

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

@@ -659,6 +659,17 @@ trait DeltaErrorsBase
     )
   }
 
+  def unsupportedTypeChangeInSchema(
+      fieldPath: Seq[String],
+      fromType: DataType,
+      toType: DataType)
+    : Throwable = {
+    new DeltaIllegalStateException(
+      errorClass = "DELTA_UNSUPPORTED_TYPE_CHANGE_IN_SCHEMA",
+      messageParameters = Array(SchemaUtils.prettyFieldName(fieldPath), fromType.sql, toType.sql)
+    )
+  }
+
   def cannotWriteIntoView(table: TableIdentifier): Throwable = {
     new DeltaAnalysisException(
       errorClass = "DELTA_CANNOT_WRITE_INTO_VIEW",

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

@@ -68,6 +68,8 @@ case class DeltaParquetFileFormat(
       "Wrong arguments for Delta table scan with deletion vectors")
   }
 
+  TypeWidening.assertTableReadable(protocol, metadata)
+
   val columnMappingMode: DeltaColumnMappingMode = metadata.columnMappingMode
   val referenceSchema: StructType = metadata.schema
 

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

@@ -89,6 +89,29 @@ object TypeWidening {
       case _ => false
     }
 
+  /**
+   * Asserts that the given table doesn't contain any unsupported type changes. This should never
+   * happen unless a non-compliant writer applied a type change that is not part of the feature
+   * specification.
+   */
+  def assertTableReadable(protocol: Protocol, metadata: Metadata): Unit = {
+    if (!isSupported(protocol) ||
+      !TypeWideningMetadata.containsTypeWideningMetadata(metadata.schema)) {
+      return
+    }
+
+    TypeWideningMetadata.getAllTypeChanges(metadata.schema).foreach {
+      case (_, TypeChange(_, from: AtomicType, to: AtomicType, _))
+        if isTypeChangeSupported(from, to) =>
+      case (fieldPath, invalidChange) =>
+        throw DeltaErrors.unsupportedTypeChangeInSchema(
+          fieldPath ++ invalidChange.fieldPath,
+          invalidChange.fromType,
+          invalidChange.toType
+        )
+    }
+  }
+
   /**
    * Filter the given list of files to only keep files that were written before the latest type
    * change, if any. These older files contain a column or field with a type that is different than

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

@@ -18,6 +18,7 @@ package org.apache.spark.sql.delta
 
 import scala.collection.mutable
 
+import org.apache.spark.sql.delta.actions.Protocol
 import org.apache.spark.sql.delta.schema.{SchemaMergingUtils, SchemaUtils}
 import org.apache.spark.sql.util.ScalaExtensions._
 
@@ -225,6 +226,27 @@ private[delta] object TypeWideningMetadata {
       case _ => false
     }
 
+  /**
+   * Return all type changes recorded in the table schema.
+   * @return A list of tuples (field path, type change).
+   */
+  def getAllTypeChanges(schema: StructType): Seq[(Seq[String], TypeChange)] = {
+    if (!containsTypeWideningMetadata(schema)) return Seq.empty
+
+    val allStructFields = SchemaUtils.filterRecursively(schema, checkComplexTypes = true) {
+      _ => true
+    }
+
+    def getTypeChanges(field: StructField): Seq[TypeChange] =
+      fromField(field)
+        .map(_.typeChanges)
+        .getOrElse(Seq.empty)
+
+    allStructFields.flatMap { case (fieldPath, field) =>
+      getTypeChanges(field).map((fieldPath :+ field.name, _))
+    }
+  }
+
   /** Return the version of the latest type change recorded in the schema metadata */
   def getLatestTypeChangeVersion(schema: StructType): Option[Long] = {
     val allStructFields = SchemaUtils.filterRecursively(schema, checkComplexTypes = true) {

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

@@ -1068,6 +1068,15 @@ trait DeltaErrorsSuiteBase
           |""".stripMargin
       ))
     }
+    {
+      val e = intercept[DeltaIllegalStateException] {
+        throw DeltaErrors.unsupportedTypeChangeInSchema(Seq("s", "a"), IntegerType, StringType)
+      }
+      checkErrorMessage(e, Some("DELTA_UNSUPPORTED_TYPE_CHANGE_IN_SCHEMA"), Some("0AKDC"),
+        Some("Unable to operate on this table because an unsupported type change was applied. " +
+          "Field s.a was changed from INT to STRING."
+      ))
+    }
     {
       val e = intercept[DeltaAnalysisException] {
         val classConf = Seq(("classKey", "classVal"))

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

@@ -18,6 +18,7 @@ package org.apache.spark.sql.delta
 
 import java.util.concurrent.TimeUnit
 
+import org.apache.spark.sql.delta.DeltaOperations.ManualUpdate
 import org.apache.spark.sql.delta.actions.AddFile
 import org.apache.spark.sql.delta.catalog.DeltaTableV2
 import org.apache.spark.sql.delta.commands.AlterTableDropFeatureDeltaCommand
@@ -979,4 +980,40 @@ trait DeltaTypeWideningTableFeatureTests extends BeforeAndAfterEach {
       }
     }
   }
+
+  test("unsupported type changes applied to the table") {
+    sql(s"CREATE TABLE delta.`$tempDir` (a array<int>) USING DELTA")
+    val metadata = new MetadataBuilder()
+      .putMetadataArray("delta.typeChanges", Array(
+        new MetadataBuilder()
+          .putString("toType", "string")
+          .putString("fromType", "int")
+          .putLong("tableVersion", 2)
+          .putString("fieldPath", "element")
+          .build()
+      )).build()
+
+    // Add an unsupported type change to the table schema. Only an implementation that isn't
+    // compliant with the feature specification would allow this.
+    deltaLog.withNewTransaction { txn =>
+      txn.commit(
+        Seq(txn.snapshot.metadata.copy(
+          schemaString = new StructType()
+            .add("a", StringType, nullable = true, metadata).json
+        )),
+        ManualUpdate)
+    }
+
+    checkError(
+      exception = intercept[DeltaIllegalStateException] {
+        readDeltaTable(tempPath).collect()
+      },
+      errorClass = "DELTA_UNSUPPORTED_TYPE_CHANGE_IN_SCHEMA",
+      parameters = Map(
+        "fieldName" -> "a.element",
+        "fromType" -> "INT",
+        "toType" -> "STRING"
+      )
+    )
+  }
 }