[GLUTEN-8050][VL] Add viewfs support in scan validation (#8049)

apache · Dec 24, 2024 · 0570203 · 0570203
1 parent 1d12c4a
commit 0570203
Show file tree

Hide file tree

Showing 5 changed files with 55 additions and 15 deletions.
diff --git a/backends-clickhouse/src/main/scala/org/apache/gluten/backendsapi/clickhouse/CHBackend.scala b/backends-clickhouse/src/main/scala/org/apache/gluten/backendsapi/clickhouse/CHBackend.scala
@@ -43,6 +43,7 @@ import org.apache.spark.sql.execution.datasources.orc.OrcFileFormat
 import org.apache.spark.sql.execution.datasources.parquet.ParquetFileFormat
 import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.types._
+import org.apache.spark.util.SerializableConfiguration
 
 import java.util.Locale
 
@@ -166,7 +167,8 @@ object CHBackendSettings extends BackendSettingsApi with Logging {
       format: ReadFileFormat,
       fields: Array[StructField],
       rootPaths: Seq[String],
-      properties: Map[String, String]): ValidationResult = {
+      properties: Map[String, String],
+      serializableHadoopConf: Option[SerializableConfiguration] = None): ValidationResult = {
 
     // Validate if all types are supported.
     def hasComplexType: Boolean = {

diff --git a/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxBackend.scala b/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxBackend.scala
@@ -47,8 +47,9 @@ import org.apache.spark.sql.execution.datasources.parquet.{ParquetFileFormat, Pa
 import org.apache.spark.sql.hive.execution.HiveFileFormat
 import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.types._
+import org.apache.spark.util.SerializableConfiguration
 
-import org.apache.hadoop.fs.Path
+import org.apache.hadoop.fs.{FileSystem, Path}
 
 import scala.util.control.Breaks.breakable
 
@@ -99,15 +100,36 @@ object VeloxBackendSettings extends BackendSettingsApi {
       format: ReadFileFormat,
       fields: Array[StructField],
       rootPaths: Seq[String],
-      properties: Map[String, String]): ValidationResult = {
+      properties: Map[String, String],
+      serializableHadoopConf: Option[SerializableConfiguration] = None): ValidationResult = {
 
     def validateScheme(): Option[String] = {
       val filteredRootPaths = distinctRootPaths(rootPaths)
-      if (
-        filteredRootPaths.nonEmpty && !VeloxFileSystemValidationJniWrapper
-          .allSupportedByRegisteredFileSystems(filteredRootPaths.toArray)
-      ) {
-        Some(s"Scheme of [$filteredRootPaths] is not supported by registered file systems.")
+      if (filteredRootPaths.nonEmpty) {
+        val resolvedPaths =
+          if (
+            GlutenConfig.getConf.enableHdfsViewfs && filteredRootPaths.head.startsWith("viewfs")
+          ) {
+            // Convert the viewfs path to hdfs path.
+            filteredRootPaths.map {
+              viewfsPath =>
+                val viewPath = new Path(viewfsPath)
+                val viewFileSystem =
+                  FileSystem.get(viewPath.toUri, serializableHadoopConf.get.value)
+                viewFileSystem.resolvePath(viewPath).toString
+            }
+          } else {
+            filteredRootPaths
+          }
+
+        if (
+          !VeloxFileSystemValidationJniWrapper.allSupportedByRegisteredFileSystems(
+            resolvedPaths.toArray)
+        ) {
+          Some(s"Scheme of [$filteredRootPaths] is not supported by registered file systems.")
+        } else {
+          None
+        }
       } else {
         None
       }

diff --git a/gluten-substrait/src/main/scala/org/apache/gluten/backendsapi/BackendSettingsApi.scala b/gluten-substrait/src/main/scala/org/apache/gluten/backendsapi/BackendSettingsApi.scala
@@ -29,6 +29,7 @@ import org.apache.spark.sql.connector.read.Scan
 import org.apache.spark.sql.execution.command.CreateDataSourceTableAsSelectCommand
 import org.apache.spark.sql.execution.datasources.{FileFormat, InsertIntoHadoopFsRelationCommand}
 import org.apache.spark.sql.types.StructField
+import org.apache.spark.util.SerializableConfiguration
 
 trait BackendSettingsApi {
 
@@ -39,7 +40,9 @@ trait BackendSettingsApi {
       format: ReadFileFormat,
       fields: Array[StructField],
       rootPaths: Seq[String],
-      properties: Map[String, String]): ValidationResult = ValidationResult.succeeded
+      properties: Map[String, String],
+      serializableHadoopConf: Option[SerializableConfiguration] = None): ValidationResult =
+    ValidationResult.succeeded
 
   def getSubstraitReadFileFormatV1(fileFormat: FileFormat): LocalFilesNode.ReadFileFormat
 

diff --git a/gluten-substrait/src/main/scala/org/apache/gluten/execution/BasicScanExecTransformer.scala b/gluten-substrait/src/main/scala/org/apache/gluten/execution/BasicScanExecTransformer.scala
@@ -31,6 +31,7 @@ import org.apache.spark.sql.catalyst.expressions._
 import org.apache.spark.sql.connector.read.InputPartition
 import org.apache.spark.sql.hive.HiveTableScanExecTransformer
 import org.apache.spark.sql.types.{StringType, StructField, StructType}
+import org.apache.spark.util.SerializableConfiguration
 
 import com.google.protobuf.StringValue
 import io.substrait.proto.NamedStruct
@@ -77,6 +78,9 @@ trait BasicScanExecTransformer extends LeafTransformSupport with BaseDataSource
           getProperties))
   }
 
+  val serializableHadoopConf: SerializableConfiguration = new SerializableConfiguration(
+    sparkContext.hadoopConfiguration)
+
   override protected def doValidateInternal(): ValidationResult = {
     var fields = schema.fields
 
@@ -91,7 +95,12 @@ trait BasicScanExecTransformer extends LeafTransformSupport with BaseDataSource
     }
 
     val validationResult = BackendsApiManager.getSettings
-      .validateScanExec(fileFormat, fields, getRootFilePaths, getProperties)
+      .validateScanExec(
+        fileFormat,
+        fields,
+        getRootFilePaths,
+        getProperties,
+        Some(serializableHadoopConf))
     if (!validationResult.ok()) {
       return validationResult
     }

diff --git a/gluten-substrait/src/main/scala/org/apache/gluten/execution/WholeStageTransformer.scala b/gluten-substrait/src/main/scala/org/apache/gluten/execution/WholeStageTransformer.scala
@@ -382,14 +382,18 @@ case class WholeStageTransformer(child: SparkPlan, materializeInput: Boolean = f
             splitInfos.foreach {
               case splitInfo: LocalFilesNode =>
                 val paths = splitInfo.getPaths.asScala
-                if (paths.nonEmpty && paths.head.startsWith("viewfs")) {
+                if (paths.nonEmpty && paths.exists(_.startsWith("viewfs"))) {
                   // Convert the viewfs path into hdfs
                   val newPaths = paths.map {
                     viewfsPath =>
-                      val viewPath = new Path(viewfsPath)
-                      val viewFileSystem =
-                        FileSystem.get(viewPath.toUri, serializableHadoopConf.value)
-                      viewFileSystem.resolvePath(viewPath).toString
+                      var finalPath = viewfsPath
+                      while (finalPath.startsWith("viewfs")) {
+                        val viewPath = new Path(finalPath)
+                        val viewFileSystem =
+                          FileSystem.get(viewPath.toUri, serializableHadoopConf.value)
+                        finalPath = viewFileSystem.resolvePath(viewPath).toString
+                      }
+                      finalPath
                   }
                   splitInfo.setPaths(newPaths.asJava)
                 }