initial commit

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

@@ -20,6 +20,7 @@ package org.apache.spark.sql.delta.commands
 import java.util.concurrent.TimeUnit.NANOSECONDS
 
 import scala.util.control.NonFatal
+import scala.collection.mutable.{Map => MutableMap, Set => MutableSet}
 
 import org.apache.spark.sql.delta.{DeltaAnalysisException, DeltaErrors, DeltaLog, DeltaOptions, DeltaTableIdentifier, DeltaTableUtils, OptimisticTransaction, ResolvedPathBasedNonDeltaTable}
 import org.apache.spark.sql.delta.actions._
@@ -100,11 +101,30 @@ trait DeltaCommand extends DeltaLogging {
   def generateCandidateFileMap(
       basePath: Path,
       candidateFiles: Seq[AddFile]): Map[String, AddFile] = {
-    val nameToAddFileMap = candidateFiles.map(add =>
-      DeltaFileOperations.absolutePath(basePath.toString, add.path).toString -> add).toMap
-    assert(nameToAddFileMap.size == candidateFiles.length,
-      s"File name collisions found among:\n${candidateFiles.map(_.path).mkString("\n")}")
-    nameToAddFileMap
+    val nameToAddFileMap = MutableMap[String, AddFile]()
+    val duplicatePaths = MutableMap[String, MutableSet[AddFile]]()
+
+    candidateFiles.foreach { add: AddFile =>
+      val filePath = DeltaFileOperations.absolutePath(basePath.toString, add.path).toString
+      if (nameToAddFileMap.contains(filePath)) {
+        // Add the file to the set of conflicts for this path
+        duplicatePaths.getOrElseUpdate(filePath, MutableSet(nameToAddFileMap(filePath))).add(add)
+      } else {
+        nameToAddFileMap += (filePath -> add)
+      }
+    }
+
+    // Check if there were any duplicates and throw an exception if found
+    if (duplicatePaths.nonEmpty) {
+      throw new IllegalArgumentException(
+        s"File name collisions detected for the following paths:\n" +
+            duplicatePaths.map { case (filePath, files) =>
+              s"Path: $filePath\nConflicting files: ${files.map(_.path).mkString(", ")}"
+            }.mkString("\n\n")
+      )
+    }
+
+    nameToAddFileMap.toMap
   }
 
   /**