Revert "[Delta] Support absolute path in Clone" (delta-io#4740)

Author: vkorukanti

iceberg/src/main/scala/org/apache/spark/sql/delta/IcebergFileManifest.scala

@@ -79,8 +79,6 @@ class IcebergFileManifest(
     _sizeInBytes.get
   }
 
-  override def supportAbsolutePath: Boolean = true
-
   def allFiles: Dataset[ConvertTargetFile] = {
     if (fileSparkResults.isEmpty) fileSparkResults = Some(getFileSparkResults())
     fileSparkResults.get

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

@@ -239,14 +239,11 @@ abstract class CloneConvertedSource(spark: SparkSession) extends CloneSource {
     val partitionSchema = convertTargetTable.partitionSchema
 
     {
-      val fileManifest = convertTargetTable.fileManifest
-      val supportAbsolutePath = fileManifest.supportAbsolutePath
-      fileManifest.allFiles.mapPartitions { targetFile =>
+      convertTargetTable.fileManifest.allFiles.mapPartitions { targetFile =>
         val basePath = new Path(baseDir)
         val fs = basePath.getFileSystem(conf.value.value)
         targetFile.map(ConvertUtils.createAddFile(
-          _, basePath, fs, SQLConf.get, Some(partitionSchema),
-          supportAbsolutePath))
+          _, basePath, fs, SQLConf.get, Some(partitionSchema)))
       }
     }
   }

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

@@ -305,13 +305,10 @@ abstract class ConvertToDeltaCommandBase(
     val shouldCollectStats = collectStats && statsEnabled
     val statsBatchSize = conf.getConf(DeltaSQLConf.DELTA_IMPORT_BATCH_SIZE_STATS_COLLECTION)
     var numFiles = 0L
-    val useAbsolutePath = manifest.supportAbsolutePath
     manifest.getFiles.grouped(statsBatchSize).flatMap { batch =>
       val adds = batch.map(
         ConvertUtils.createAddFile(
-          _, txn.deltaLog.dataPath, fs, conf, Some(partitionSchema),
-          useAbsolutePath || deltaPath.isDefined
-        ))
+          _, txn.deltaLog.dataPath, fs, conf, Some(partitionSchema), deltaPath.isDefined))
       if (shouldCollectStats) {
         logInfo(
           log"Collecting stats for a batch of " +

spark/src/main/scala/org/apache/spark/sql/delta/commands/convert/ConvertUtils.scala

@@ -221,7 +221,7 @@ trait ConvertUtilsBase extends DeltaLogging {
         s"Fail to relativize path $path against base path $basePath.")
       relativePath.toUri.toString
     } else {
-      fs.makeQualified(path).toUri.toString
+      path.toUri.toString
     }
 
     AddFile(pathStrForAddFile, partition, file.length, file.modificationTime, dataChange = true,

spark/src/main/scala/org/apache/spark/sql/delta/commands/convert/interfaces.scala

@@ -67,8 +67,6 @@ trait ConvertTargetFileManifest extends Closeable {
   /** Return the active files for a table in sequence */
   def getFiles: Iterator[ConvertTargetFile] = allFiles.toLocalIterator().asScala
 
-  /** All data file paths are required to be relative to table path when false */
-  def supportAbsolutePath: Boolean = false
   /** Return the number of files for the table */
   def numFiles: Long = allFiles.count()