[SPARK-57590][SQL] Address review: use CANNOT_MERGE_SCHEMAS for archive merge conflicts and cover mergeSchema inference

Author: akshatshenoi-db

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFileFormat.scala

@@ -39,7 +39,7 @@ import org.apache.parquet.format.converter.ParquetMetadataConverter.SKIP_ROW_GRO
 import org.apache.parquet.hadoop._
 import org.apache.parquet.hadoop.util.HadoopInputFile
 
-import org.apache.spark.TaskContext
+import org.apache.spark.{SparkException, TaskContext}
 import org.apache.spark.internal.Logging
 import org.apache.spark.internal.LogKeys.{PATH, SCHEMA}
 import org.apache.spark.sql.SparkSession
@@ -749,8 +749,8 @@ object ParquetFileFormat extends Logging {
             fileSourceOptions.ignoreMissingFiles).foreach { schema =>
           merged = Some(merged.fold(schema) { acc =>
             try acc.merge(schema, caseSensitive) catch {
-              case e: Throwable =>
-                throw QueryExecutionErrors.failedToMergeIncompatibleSchemasError(acc, schema, e)
+              case cause: SparkException =>
+                throw QueryExecutionErrors.failedMergingSchemaError(acc, schema, cause)
             }
           })
         }

sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/ParquetTarArchiveReadSuite.scala

@@ -17,6 +17,9 @@
 
 package org.apache.spark.sql.execution.datasources
 
+import java.io.File
+import java.nio.file.Files
+
 import org.apache.spark.sql.functions.input_file_name
 import org.apache.spark.sql.internal.SQLConf
 
@@ -75,4 +78,25 @@ class ParquetTarArchiveReadSuite
     assertArchiveMatchesDir(
       Seq(entryName(0) -> encodeFile(withName), entryName(1) -> encodeFile(idOnly)))
   }
+
+  test("archive inference unions differing fields across entries with mergeSchema=true") {
+    // Parquet does not union schemas during default inference, but `mergeSchema=true` folds every
+    // entry's schema; over an archive that folds each unpacked entry one at a time. The unioned
+    // schema must match a directory read of the same files under the same option.
+    val withName = sampleDf((1, "Alice"), (2, "Bob"))
+    val idExtra = Seq((3, 30)).toDF("id", "extra")
+    val entries = Seq(entryName(0) -> encodeFile(withName), entryName(1) -> encodeFile(idExtra))
+    val merge = Map("mergeSchema" -> "true")
+    withArchiveFile() { archive =>
+      writeArchive(archive, entries)
+      val archiveSchema = inferredSchema(Seq(archive.getCanonicalPath), merge)
+      withTempDir { dir =>
+        entries.foreach { case (n, b) => Files.write(new File(dir, n).toPath, b) }
+        assert(archiveSchema.fieldNames.toSet == Set("id", "name", "extra"),
+          s"expected the union of entry fields, got $archiveSchema")
+        assert(archiveSchema == inferredSchema(Seq(dir.getCanonicalPath), merge),
+          s"archive mergeSchema inference diverged from a directory read; got $archiveSchema")
+      }
+    }
+  }
 }