[GLUTEN-8050][VL] Add viewfs support in scan validation

backends-clickhouse/src/main/scala/org/apache/gluten/backendsapi/clickhouse/CHBackend.scala

@@ -41,6 +41,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
 
@@ -164,7 +165,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 = {

backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxBackend.scala

@@ -45,8 +45,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
 
@@ -96,15 +97,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
       }

gluten-substrait/src/main/scala/org/apache/gluten/backendsapi/BackendSettingsApi.scala

@@ -27,6 +27,7 @@ import org.apache.spark.sql.catalyst.plans._
 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 {
 
@@ -37,7 +38,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 supportWriteFilesExec(
       format: FileFormat,

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.{BooleanType, 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
     }