diff --git a/integration_tests/run_pyspark_from_build.sh b/integration_tests/run_pyspark_from_build.sh
index 9bd72b2ada0..b3a474e7565 100755
--- a/integration_tests/run_pyspark_from_build.sh
+++ b/integration_tests/run_pyspark_from_build.sh
@@ -364,6 +364,21 @@ EOF
     fi
     export PYSP_TEST_spark_rapids_memory_gpu_allocSize=${PYSP_TEST_spark_rapids_memory_gpu_allocSize:-'1536m'}
 
+    if [[ "$VELOX_TEST" -eq 1 ]]; then
+      if [ -z "${VELOX_JARS}" ]; then
+        echo "Error: Environment VELOX_JARS is not set."
+        exit 1
+      fi
+      export PYSP_TEST_spark_jars="${PYSP_TEST_spark_jars},${VELOX_JARS//:/,}"
+      export PYSP_TEST_spark_memory_offHeap_enabled=true
+      export PYSP_TEST_spark_memory_offHeap_size=512M
+      export PYSP_TEST_spark_gluten_loadLibFromJar=true
+      export PYSP_TEST_spark_rapids_sql_loadVelox=true
+      if [[ "$VELOX_HDFS_TEST" -eq 1 ]]; then
+        export PYSP_TEST_spark_rapids_sql_velox_useVeloxHDFS=true
+      fi
+    fi
+
     SPARK_SHELL_SMOKE_TEST="${SPARK_SHELL_SMOKE_TEST:-0}"
     if [[ "${SPARK_SHELL_SMOKE_TEST}" != "0" ]]; then
         echo "Running spark-shell smoke test..."
diff --git a/pom.xml b/pom.xml
index 893af998540..677ec189752 100644
--- a/pom.xml
+++ b/pom.xml
@@ -825,6 +825,7 @@
         <maven.scalastyle.skip>false</maven.scalastyle.skip>
         <dist.jar.compress>true</dist.jar.compress>
         <spark330.iceberg.version>0.14.1</spark330.iceberg.version>
+        <gluten.version>1.2.0</gluten.version>
         <!--
             If true, disables verification that all Shims be built as of one and the same git
             commit hash. Do not use for CI!
@@ -1067,6 +1068,26 @@
                 <type>test-jar</type>
                 <scope>test</scope>
             </dependency>
+            <dependency>
+                <groupId>org.apache.gluten</groupId>
+                <artifactId>spark-sql-columnar-shims-spark32</artifactId>
+                <version>${gluten.version}</version>
+            </dependency>
+            <dependency>
+                <groupId>org.apache.gluten</groupId>
+                <artifactId>backends-velox</artifactId>
+                <version>${gluten.version}</version>
+                <exclusions>
+                    <exclusion>
+                        <groupId>org.apache.gluten</groupId>
+                        <artifactId>spark-sql-columnar-shims-spark34</artifactId>
+                    </exclusion>
+                    <exclusion>
+                        <groupId>com.fasterxml.jackson.core</groupId>
+                        <artifactId>jackson-core</artifactId>
+                    </exclusion>
+                </exclusions>
+            </dependency>
         </dependencies>
     </dependencyManagement>
 
diff --git a/sql-plugin/pom.xml b/sql-plugin/pom.xml
index c9cfb8ce99f..c50adec664f 100644
--- a/sql-plugin/pom.xml
+++ b/sql-plugin/pom.xml
@@ -102,6 +102,34 @@
             <artifactId>mockito-core</artifactId>
             <scope>test</scope>
         </dependency>
+
+        <dependency>
+            <groupId>org.apache.gluten</groupId>
+            <artifactId>spark-sql-columnar-shims-spark32</artifactId>
+            <version>${gluten.version}</version>
+            <scope>provided</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.gluten</groupId>
+            <artifactId>backends-velox</artifactId>
+            <version>${gluten.version}</version>
+            <scope>provided</scope>
+            <exclusions>
+                <exclusion>
+                    <groupId>org.apache.gluten</groupId>
+                    <artifactId>spark-sql-columnar-shims-spark34</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>com.fasterxml.jackson.core</groupId>
+                    <artifactId>jackson-core</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+        <dependency>
+            <groupId>com.nvidia</groupId>
+            <artifactId>rapids-4-spark-velox_${scala.binary.version}</artifactId>
+            <version>${project.version}</version>
+        </dependency>
     </dependencies>
     <!-- #if scala-2.13 --><!--
     <profiles>
diff --git a/sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuParquetScan.scala b/sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuParquetScan.scala
index f9c0d88ba6c..a73a7c5a7ae 100644
--- a/sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuParquetScan.scala
+++ b/sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuParquetScan.scala
@@ -21,7 +21,6 @@ import java.net.URI
 import java.nio.ByteBuffer
 import java.nio.channels.SeekableByteChannel
 import java.nio.charset.StandardCharsets
-import java.util
 import java.util.{Collections, Locale}
 import java.util.concurrent._
 
@@ -58,6 +57,7 @@ import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName
 import org.apache.spark.TaskContext
 import org.apache.spark.broadcast.Broadcast
 import org.apache.spark.internal.Logging
+import org.apache.spark.rapids.velox.{FileCopyRange, VeloxHDFS}
 import org.apache.spark.sql.SparkSession
 import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.catalyst.expressions.Expression
@@ -464,7 +464,6 @@ private case class GpuParquetFileFilterHandler(
 
   private val PARQUET_ENCRYPTION_CONFS = Seq("parquet.encryption.kms.client.class",
     "parquet.encryption.kms.client.class", "parquet.crypto.factory.class")
-  private val PARQUET_MAGIC_ENCRYPTED = "PARE".getBytes(StandardCharsets.US_ASCII)
 
   private def isParquetTimeInInt96(parquetType: Type): Boolean = {
     parquetType match {
@@ -536,8 +535,12 @@ private case class GpuParquetFileFilterHandler(
   private def readFooterBuffer(
       filePath: Path,
       conf: Configuration): HostMemoryBuffer = {
-    PerfIO.readParquetFooterBuffer(filePath, conf, verifyParquetMagic)
-      .getOrElse(readFooterBufUsingHadoop(filePath, conf))
+    PerfIO.readParquetFooterBuffer(
+        filePath, conf, GpuParquetUtils.verifyParquetMagic)
+      .getOrElse(
+        VeloxHDFS.readParquetFooterBuffer(filePath, conf)
+          .getOrElse(readFooterBufUsingHadoop(filePath, conf))
+      )
   }
 
   private def readFooterBufUsingHadoop(filePath: Path, conf: Configuration): HostMemoryBuffer = {
@@ -558,7 +561,7 @@ private case class GpuParquetFileFilterHandler(
         val magic = new Array[Byte](MAGIC.length)
         inputStream.readFully(magic)
         val footerIndex = footerLengthIndex - footerLength
-        verifyParquetMagic(filePath, magic)
+        GpuParquetUtils.verifyParquetMagic(filePath, magic)
         if (footerIndex < MAGIC.length || footerIndex >= footerLengthIndex) {
           throw new RuntimeException(s"corrupted file: the footer index is not within " +
             s"the file: $footerIndex")
@@ -583,21 +586,6 @@ private case class GpuParquetFileFilterHandler(
     }
   }
 
-
-  private def verifyParquetMagic(filePath: Path, magic: Array[Byte]): Unit = {
-    if (!util.Arrays.equals(MAGIC, magic)) {
-      if (util.Arrays.equals(PARQUET_MAGIC_ENCRYPTED, magic)) {
-        throw new RuntimeException("The GPU does not support reading encrypted Parquet " +
-          "files. To read encrypted or columnar encrypted files, disable the GPU Parquet " +
-          s"reader via ${RapidsConf.ENABLE_PARQUET_READ.key}.")
-      } else {
-        throw new RuntimeException(s"$filePath is not a Parquet file. " +
-          s"Expected magic number at tail ${util.Arrays.toString(MAGIC)} " +
-          s"but found ${util.Arrays.toString(magic)}")
-      }
-    }
-  }
-
   private def readAndFilterFooter(
       file: PartitionedFile,
       conf : Configuration,
@@ -659,8 +647,15 @@ private case class GpuParquetFileFilterHandler(
             }
           }
         }.getOrElse {
-          ParquetFileReader.readFooter(conf, filePath,
-            ParquetMetadataConverter.range(file.start, file.start + file.length))
+          VeloxHDFS.readParquetFooterBuffer(filePath, conf).map { hmb =>
+            withResource(hmb) { _ =>
+              ParquetFileReader.readFooter(new HMBInputFile(hmb),
+                ParquetMetadataConverter.range(file.start, file.start + file.length))
+            }
+          }.getOrElse {
+            ParquetFileReader.readFooter(conf, filePath,
+              ParquetMetadataConverter.range(file.start, file.start + file.length))
+          }
         }
       }
     }
@@ -1604,6 +1599,28 @@ trait ParquetPartitionReaderBase extends Logging with ScanWithMetrics
         conf, out.buffer, filePath.toUri,
         coalescedRanges.map(r => IntRangeWithOffset(r.offset, r.length, r.outputOffset))
       ).getOrElse {
+      // try to read data through VeloxHDFS if necessary
+      val streamHandle = VeloxHDFS.createInputFileStream(filePathString)
+      if (streamHandle > 0) {
+        // Builds ParquetCopyRange while computing total read size
+        val ranges = ArrayBuffer.empty[FileCopyRange]
+        val bufferAddr = out.buffer.getAddress
+        val readSize = coalescedRanges.foldLeft(0L) { (acc, blockCopy) =>
+          ranges += FileCopyRange(
+            blockCopy.offset,
+            blockCopy.length,
+            bufferAddr + blockCopy.outputOffset
+          )
+          out.seek(blockCopy.outputOffset + blockCopy.length)
+          acc + blockCopy.length
+        }
+        // leverage velox::HdfsReadFile to buffer Hdfs files (which based on libhdfs3)
+        VeloxHDFS.copyRangesFromFile(
+          filePathString, streamHandle, ranges,
+          closeAfterFinished = true
+        )
+        readSize
+      } else {
         withResource(filePath.getFileSystem(conf).open(filePath)) { in =>
           val copyBuffer: Array[Byte] = new Array[Byte](copyBufferSize)
           coalescedRanges.foldLeft(0L) { (acc, blockCopy) =>
@@ -1611,6 +1628,7 @@ trait ParquetPartitionReaderBase extends Logging with ScanWithMetrics
           }
         }
       }
+    }
     // try to cache the remote ranges that were copied
     remoteCopies.foreach { range =>
       metrics.getOrElse(GpuMetric.FILECACHE_DATA_RANGE_MISSES, NoopMetric) += 1
diff --git a/sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuParquetUtils.scala b/sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuParquetUtils.scala
index 9ca568335e6..aaf9d3a4d6c 100644
--- a/sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuParquetUtils.scala
+++ b/sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuParquetUtils.scala
@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2022, NVIDIA CORPORATION.
+ * Copyright (c) 2022-2024, NVIDIA CORPORATION.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -16,10 +16,14 @@
 
 package com.nvidia.spark.rapids
 
+import java.nio.charset.StandardCharsets
+import java.util
 import java.util.Locale
 
 import scala.collection.JavaConverters._
 
+import org.apache.hadoop.fs.Path
+import org.apache.parquet.hadoop.ParquetFileWriter.MAGIC
 import org.apache.parquet.hadoop.metadata.{BlockMetaData, ColumnChunkMetaData, ColumnPath}
 import org.apache.parquet.schema.MessageType
 
@@ -83,4 +87,27 @@ object GpuParquetUtils extends Logging {
 
     block
   }
+
+  /**
+   * Verify the Magic code stored in the Parquet Footer
+   *
+   * @param filePath the path of Parquet file
+   * @param magic the Magic code extracted from the file
+   */
+  def verifyParquetMagic(filePath: Path, magic: Array[Byte]): Unit = {
+    if (!util.Arrays.equals(MAGIC, magic)) {
+      if (util.Arrays.equals(PARQUET_MAGIC_ENCRYPTED, magic)) {
+        throw new RuntimeException("The GPU does not support reading encrypted Parquet " +
+          "files. To read encrypted or columnar encrypted files, disable the GPU Parquet " +
+          s"reader via ${RapidsConf.ENABLE_PARQUET_READ.key}.")
+      } else {
+        throw new RuntimeException(s"$filePath is not a Parquet file. " +
+          s"Expected magic number at tail ${util.Arrays.toString(MAGIC)} " +
+          s"but found ${util.Arrays.toString(magic)}")
+      }
+    }
+  }
+
+  private val PARQUET_MAGIC_ENCRYPTED = "PARE".getBytes(StandardCharsets.US_ASCII)
+
 }
diff --git a/sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuRowToColumnarExec.scala b/sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuRowToColumnarExec.scala
index 51b6645d7b7..8588812cdbe 100644
--- a/sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuRowToColumnarExec.scala
+++ b/sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuRowToColumnarExec.scala
@@ -34,7 +34,7 @@ import org.apache.spark.sql.rapids.execution.TrampolineUtil
 import org.apache.spark.sql.types._
 import org.apache.spark.sql.vectorized.ColumnarBatch
 
-private class GpuRowToColumnConverter(schema: StructType) extends Serializable {
+class GpuRowToColumnConverter(schema: StructType) extends Serializable {
   private val converters = schema.fields.map {
     f => GpuRowToColumnConverter.getConverterForType(f.dataType, f.nullable)
   }
@@ -594,7 +594,8 @@ class RowToColumnarIterator(
     numOutputRows: GpuMetric = NoopMetric,
     numOutputBatches: GpuMetric = NoopMetric,
     streamTime: GpuMetric = NoopMetric,
-    opTime: GpuMetric = NoopMetric) extends Iterator[ColumnarBatch] {
+    opTime: GpuMetric = NoopMetric,
+    acquireGpuTime: GpuMetric = NoopMetric) extends Iterator[ColumnarBatch] {
 
   private val targetSizeBytes = localGoal.targetSizeBytes
   private var targetRows = 0
@@ -650,7 +651,11 @@ class RowToColumnarIterator(
         // note that TaskContext.get() can return null during unit testing so we wrap it in an
         // option here
         Option(TaskContext.get())
-            .foreach(ctx => GpuSemaphore.acquireIfNecessary(ctx))
+          .foreach { ctx =>
+            val acquireGpuStart = System.nanoTime()
+            GpuSemaphore.acquireIfNecessary(ctx)
+            acquireGpuTime += System.nanoTime() - acquireGpuStart
+          }
 
         val ret = withResource(new NvtxWithMetrics("RowToColumnar", NvtxColor.GREEN,
             opTime)) { _ =>
diff --git a/sql-plugin/src/main/scala/com/nvidia/spark/rapids/RapidsConf.scala b/sql-plugin/src/main/scala/com/nvidia/spark/rapids/RapidsConf.scala
index a83ad716d34..409f588de27 100644
--- a/sql-plugin/src/main/scala/com/nvidia/spark/rapids/RapidsConf.scala
+++ b/sql-plugin/src/main/scala/com/nvidia/spark/rapids/RapidsConf.scala
@@ -1684,6 +1684,53 @@ val GPU_COREDUMP_PIPE_PATTERN = conf("spark.rapids.gpu.coreDump.pipePattern")
     .booleanConf
     .createWithDefault(false)
 
+  val PARQUET_VELOX_READER = conf("spark.rapids.sql.parquet.useVelox")
+    .doc("Use velox to do ParquetScan (on CPUs)")
+    .internal()
+    .booleanConf
+    .createWithDefault(true)
+
+  object VeloxFilterPushdownType extends Enumeration {
+    val ALL_SUPPORTED, NONE, UNCHANGED = Value
+  }
+
+  val PUSH_DOWN_FILTERS_TO_VELOX = conf("spark.rapids.sql.parquet.pushDownFiltersToVelox")
+    .doc("Push down all supported filters to Velox if set to ALL_SUPPORTED. " +
+      "If set to NONE, no filters will be pushed down so all filters are on the GPU. " +
+      "If set to UNCHANGED, filters will be both pushed down and keeped on the GPU. " +
+      "UNCHANGED is to make the behavior same as before.")
+    .internal()
+    .stringConf
+    .transform(_.toUpperCase(java.util.Locale.ROOT))
+    .checkValues(VeloxFilterPushdownType.values.map(_.toString))
+    .createWithDefault(VeloxFilterPushdownType.ALL_SUPPORTED.toString)
+
+  val ENABLE_NATIVE_VELOX_CONVERTER = conf("spark.rapids.sql.enableNativeVeloxConverter")
+    .doc("Re-formatting VeloxColumn to align with the memory layout of GpuColumn directly")
+    .internal()
+    .booleanConf
+    .createWithDefault(true)
+
+  val PARQUET_VELOX_PRELOAD_CAP = conf("spark.rapids.sql.parquet.veloxPreloadedBatches")
+    .doc("Preloading capacity of VeloxParquetScan. If > 0, will enable preloading" +
+      " VeloxScanIterator asynchronously in a separate thread")
+    .internal()
+    .integerConf
+    .createWithDefault(0)
+
+  val ENABLE_VELOX_HDFS = conf("spark.rapids.sql.velox.useVeloxHDFS")
+    .doc("Use HDFS reader of velox to do buffering instead of Hadoop Java API")
+    .internal()
+    .startupOnly()
+    .booleanConf
+    .createWithDefault(false)
+
+  val LOAD_VELOX = conf("spark.rapids.sql.loadVelox")
+    .doc("Load Velox (through Gluten) as a spark driver plugin")
+    .startupOnly()
+    .booleanConf
+    .createWithDefault(true)
+
   val HASH_AGG_REPLACE_MODE = conf("spark.rapids.sql.hashAgg.replaceMode")
     .doc("Only when hash aggregate exec has these modes (\"all\" by default): " +
       "\"all\" (try to replace all aggregates, default), " +
@@ -2816,6 +2863,18 @@ class RapidsConf(conf: Map[String, String]) extends Logging {
 
   lazy val avroDebugDumpAlways: Boolean = get(AVRO_DEBUG_DUMP_ALWAYS)
 
+  lazy val parquetVeloxReader: Boolean = get(PARQUET_VELOX_READER)
+
+  lazy val pushDownFiltersToVelox: String = get(PUSH_DOWN_FILTERS_TO_VELOX)
+
+  lazy val enableNativeVeloxConverter: Boolean = get(ENABLE_NATIVE_VELOX_CONVERTER)
+
+  lazy val parquetVeloxPreloadCapacity: Int = get(PARQUET_VELOX_PRELOAD_CAP)
+
+  lazy val enableVeloxHDFS: Boolean = get(ENABLE_VELOX_HDFS)
+
+  lazy val loadVelox: Boolean = get(LOAD_VELOX)
+
   lazy val hashAggReplaceMode: String = get(HASH_AGG_REPLACE_MODE)
 
   lazy val partialMergeDistinctEnabled: Boolean = get(PARTIAL_MERGE_DISTINCT_ENABLED)
diff --git a/sql-plugin/src/main/scala/org/apache/spark/rapids/velox/VeloxColumnarBatchConverter.scala b/sql-plugin/src/main/scala/org/apache/spark/rapids/velox/VeloxColumnarBatchConverter.scala
new file mode 100644
index 00000000000..b574d3f9f52
--- /dev/null
+++ b/sql-plugin/src/main/scala/org/apache/spark/rapids/velox/VeloxColumnarBatchConverter.scala
@@ -0,0 +1,242 @@
+/*
+ * Copyright (c) 2024, NVIDIA CORPORATION.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.rapids.velox
+
+import scala.collection.mutable
+
+import ai.rapids.cudf.NvtxColor
+import com.nvidia.spark.rapids.{AcquireFailed, CoalesceSizeGoal, CudfRowTransitions, GeneratedInternalRowToCudfRowIterator, GpuColumnVector, GpuMetric, GpuRowToColumnConverter, GpuSemaphore, NvtxWithMetrics, RowToColumnarIterator}
+import com.nvidia.spark.rapids.Arm._
+import com.nvidia.spark.rapids.RapidsPluginImplicits._
+import com.nvidia.spark.rapids.velox.{RapidsHostColumn, VeloxBatchConverter}
+import org.apache.gluten.execution.VeloxColumnarToRowExec
+import org.apache.gluten.rapids.GlutenJniWrapper
+
+import org.apache.spark.TaskContext
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.Attribute
+import org.apache.spark.sql.types.{StructField, StructType}
+import org.apache.spark.sql.vectorized.{ColumnarBatch, ColumnVector}
+
+class CoalesceNativeConverter(veloxIter: Iterator[ColumnarBatch],
+                              targetBatchSizeInBytes: Int,
+                              schema: StructType,
+                              metrics: Map[String, GpuMetric])
+  extends Iterator[Array[RapidsHostColumn]] with Logging {
+
+  private var converterImpl: Option[VeloxBatchConverter] = None
+
+  private var srcExhausted = false
+
+  private val c2cMetrics = Map(
+    "C2COutputSize" -> GpuMetric.unwrap(metrics("C2COutputSize")))
+
+  @transient private var runtime: GlutenJniWrapper = _
+
+  def setRuntime(): Unit = {
+    runtime = GlutenJniWrapper.create()
+  }
+
+  override def hasNext(): Boolean = {
+    // either converter holds data or upstreaming iterator holds data
+    val ret = withResource(new NvtxWithMetrics("VeloxC2CHasNext", NvtxColor.WHITE,
+      metrics("C2CStreamTime"))) { _ =>
+      converterImpl.exists(c => c.isDeckFilled || c.hasProceedingBuilders) ||
+        (!srcExhausted && veloxIter.hasNext)
+    }
+    if (!ret) {
+      if (!srcExhausted) {
+        srcExhausted = true
+      }
+      converterImpl.foreach { c =>
+        // VeloxBatchConverter collects the eclipsedTime of C2C_Conversion by itself.
+        // Here we fetch the final value before closing it.
+        metrics("C2CTime") += c.eclipsedNanoSecond
+        // release the native instance when upstreaming iterator has been exhausted
+        val detailedMetrics = c.close()
+        val tID = TaskContext.get().taskAttemptId()
+        logError(s"task[$tID] CoalesceNativeConverter finished:\n$detailedMetrics")
+        converterImpl = None
+      }
+    }
+    ret
+  }
+
+  override def next(): Array[RapidsHostColumn] = {
+    val ntvx = new NvtxWithMetrics("VeloxC2CNext", NvtxColor.YELLOW, metrics("C2CStreamTime"))
+    withResource(ntvx) { _ =>
+      while (true) {
+        converterImpl.foreach { impl =>
+          val needFlush = if (veloxIter.hasNext) {
+            // The only condition leading to a nonEmpty deck is targetBuffers are unset after
+            // the previous flushing
+            if (impl.isDeckFilled) {
+              impl.resetTargetBuffers()
+            }
+            // tryAppendBatch, if failed, the batch will be placed on the deck
+            metrics("VeloxOutputBatches") += 1
+            !impl.tryAppendBatch(veloxIter.next())
+          } else {
+            srcExhausted = true
+            true
+          }
+          if (needFlush) {
+            metrics("C2COutputBatches") += 1
+            val rapidsHostBatch = impl.flushAndConvert()
+            // It is essential to check and tidy up the deck right after flushing. Because if
+            // the next call of veloxIter.hasNext will release the batch which the deck holds
+            // its reference.
+            if (impl.isDeckFilled) {
+              impl.resetTargetBuffers()
+            }
+            return rapidsHostBatch
+          }
+        }
+        if (converterImpl.isEmpty) {
+          require(runtime != null, "Please setRuntime before fetching the iterator")
+          val converter = VeloxBatchConverter(
+            runtime,
+            veloxIter.next(),
+            targetBatchSizeInBytes, schema, c2cMetrics
+          )
+          converterImpl = Some(converter)
+        }
+      }
+
+      throw new RuntimeException("should NOT reach this line")
+    }
+  }
+
+}
+
+trait RapidsHostBatchProducer {
+  def hasNext: Boolean
+
+  def waitForNext(): Unit
+
+  def takeNext: Array[RapidsHostColumn]
+}
+
+case class SyncHostBatchProducer(base: Iterator[Array[RapidsHostColumn]])
+  extends RapidsHostBatchProducer {
+
+  @transient
+  private lazy val deck = mutable.Queue.empty[Array[RapidsHostColumn]]
+
+  override def hasNext: Boolean = base.hasNext
+
+  override def waitForNext(): Unit = {
+    if (deck.isEmpty) {
+      deck.enqueue(base.next())
+    }
+  }
+
+  override def takeNext: Array[RapidsHostColumn] = {
+    require(deck.nonEmpty, "The deck is NOT filled")
+    deck.dequeue()
+  }
+}
+
+object VeloxColumnarBatchConverter extends Logging {
+
+  def hostToDevice(hostProducer: RapidsHostBatchProducer,
+                   outputAttr: Seq[Attribute],
+                   metrics: Map[String, GpuMetric]): Iterator[ColumnarBatch] = {
+
+    new Iterator[ColumnarBatch] {
+
+      private val dataTypes = outputAttr.map(_.dataType).toArray
+
+      override def hasNext: Boolean = hostProducer.hasNext
+
+      override def next(): ColumnarBatch = {
+        // 1. Preparing Stage
+        // Firstly, waits for the next host batch being ready
+        hostProducer.waitForNext()
+        // Then, acquires GpuSemaphore
+        Option(TaskContext.get()).foreach { ctx =>
+          GpuSemaphore.tryAcquire(ctx) match {
+            case AcquireFailed(_) =>
+              withResource(new NvtxWithMetrics("gpuAcquireC2C", NvtxColor.GREEN,
+                metrics("GpuAcquireTime"))) { _ =>
+                GpuSemaphore.acquireIfNecessary(ctx)
+              }
+            case _ =>
+          }
+        }
+        // Finally, take the ownership of the next host batch, which might trigger the potential
+        // asynchronous producer through creating an empty slot on the buffer.
+        val hostColumns = hostProducer.takeNext
+
+        // 2. Transferring Stage
+        val deviceVectors: Array[ColumnVector] = hostColumns.zip(dataTypes).safeMap {
+          case (RapidsHostColumn(hcv, isPinned, totalBytes), dt) =>
+            val nvtxMetric = if (isPinned) {
+              metrics("PinnedH2DSize") += totalBytes
+              new NvtxWithMetrics("pinnedH2D", NvtxColor.DARK_GREEN, metrics("PinnedH2DTime"))
+            } else {
+              metrics("PageableH2DSize") += totalBytes
+              new NvtxWithMetrics("PageableH2D", NvtxColor.GREEN, metrics("PageableH2DTime"))
+            }
+            withResource(hcv) { _ =>
+              withResource(nvtxMetric) { _ =>
+                GpuColumnVector.from(hcv.copyToDevice(), dt)
+              }
+            }
+        }
+        new ColumnarBatch(deviceVectors, hostColumns.head.vector.getRowCount.toInt)
+      }
+    }
+  }
+
+  def roundTripConvert(iter: Iterator[ColumnarBatch],
+                       outputAttr: Seq[Attribute],
+                       coalesceGoal: CoalesceSizeGoal,
+                       metrics: Map[String, GpuMetric]): Iterator[ColumnarBatch] = {
+    val rowIter: Iterator[InternalRow] = VeloxColumnarToRowExec.toRowIterator(
+      iter,
+      outputAttr,
+      GpuMetric.unwrap(metrics("C2ROutputRows")),
+      GpuMetric.unwrap(metrics("C2ROutputBatches")),
+      GpuMetric.unwrap(metrics("VeloxC2RTime"))
+    )
+    val useCudfRowTransition: Boolean = {
+      outputAttr.nonEmpty && outputAttr.length < 100000000 &&
+        CudfRowTransitions.areAllSupported(outputAttr)
+    }
+
+    if (useCudfRowTransition) {
+      GeneratedInternalRowToCudfRowIterator(
+        rowIter,
+        outputAttr.toArray, coalesceGoal,
+        metrics("R2CStreamTime"), metrics("R2CTime"),
+        metrics("R2CInputRows"), metrics("R2COutputRows"), metrics("R2COutputBatches"))
+    } else {
+      val fullOutputSchema = StructType(outputAttr.map { ar =>
+        StructField(ar.name, ar.dataType, ar.nullable)
+      })
+      val converters = new GpuRowToColumnConverter(fullOutputSchema)
+      new RowToColumnarIterator(
+        rowIter,
+        fullOutputSchema, coalesceGoal, converters,
+        metrics("R2CInputRows"), metrics("R2COutputRows"), metrics("R2COutputBatches"),
+        metrics("R2CStreamTime"), metrics("R2CTime"), metrics("GpuAcquireTime"))
+    }
+  }
+
+}
diff --git a/sql-plugin/src/main/scala/org/apache/spark/rapids/velox/VeloxFileSourceScanExec.scala b/sql-plugin/src/main/scala/org/apache/spark/rapids/velox/VeloxFileSourceScanExec.scala
new file mode 100644
index 00000000000..ca4f59e2b10
--- /dev/null
+++ b/sql-plugin/src/main/scala/org/apache/spark/rapids/velox/VeloxFileSourceScanExec.scala
@@ -0,0 +1,233 @@
+/*
+ * Copyright (c) 2024, NVIDIA CORPORATION.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.rapids.velox
+
+import java.util.concurrent.TimeUnit.NANOSECONDS
+
+import scala.collection.mutable
+
+import com.nvidia.spark.rapids.{GpuExec, GpuMetric, RapidsConf, TargetSize}
+import org.apache.gluten.execution.{FileSourceScanExecTransformer, WholeStageTransformer}
+
+import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.catalyst.{InternalRow, TableIdentifier}
+import org.apache.spark.sql.catalyst.expressions.{Attribute, DynamicPruningExpression, Expression, Literal}
+import org.apache.spark.sql.catalyst.plans.QueryPlan
+import org.apache.spark.sql.execution.datasources.HadoopFsRelation
+import org.apache.spark.sql.rapids.GpuDataSourceScanExec
+import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.vectorized.ColumnarBatch
+import org.apache.spark.util.collection.BitSet
+
+case class VeloxFileSourceScanExec(
+    @transient relation: HadoopFsRelation,
+    originalOutput: Seq[Attribute],
+    requiredSchema: StructType,
+    partitionFilters: Seq[Expression],
+    optionalBucketSet: Option[BitSet],
+    optionalNumCoalescedBuckets: Option[Int],
+    dataFilters: Seq[Expression],
+    tableIdentifier: Option[TableIdentifier],
+    disableBucketedScan: Boolean = false,
+    queryUsesInputFile: Boolean = false,
+    alluxioPathsMap: Option[Map[String, String]],
+    filteredOutput: Option[Seq[Attribute]] = None)(@transient val rapidsConf: RapidsConf)
+  extends GpuDataSourceScanExec with GpuExec {
+  import GpuMetric._
+
+  // All expressions are on the CPU.
+  override def gpuExpressions: Seq[Expression] = Nil
+
+  private val glutenScan: FileSourceScanExecTransformer = {
+    FileSourceScanExecTransformer(
+      relation,
+      output,
+      requiredSchema,
+      partitionFilters,
+      optionalBucketSet,
+      optionalNumCoalescedBuckets,
+      dataFilters,
+      tableIdentifier,
+      disableBucketedScan)
+  }
+
+  private val coalesceSizeGoal = rapidsConf.gpuTargetBatchSizeBytes
+
+  override def output: Seq[Attribute] = filteredOutput.getOrElse(originalOutput)
+
+  override lazy val metadata: Map[String, String] = {
+    def seqToString(seq: Seq[Any]) = seq.mkString("[", ", ", "]")
+
+    val location = relation.location
+    val locationDesc =
+      location.getClass.getSimpleName +
+        GpuDataSourceScanExec.buildLocationMetadata(location.rootPaths, maxMetadataValueLength)
+    Map(
+      "Format" -> relation.fileFormat.toString,
+      "ReadSchema" -> requiredSchema.catalogString,
+      "Batched" -> supportsColumnar.toString,
+      "PartitionFilters" -> seqToString(partitionFilters),
+      "PushedFilters" -> seqToString(glutenScan.filterExprs()),
+      "DataFilters" -> seqToString(dataFilters),
+      "Location" -> locationDesc)
+  }
+
+  lazy val inputRDD: RDD[InternalRow] = {
+    // invoke a whole stage transformer
+    val glutenPipeline = WholeStageTransformer(glutenScan, materializeInput = false)(1)
+    logInfo(s"Using Velox to read parquet: ${metadata.toString()}")
+    logInfo(s"SubstraitPlan: ${glutenPipeline.substraitPlanJson}")
+    logInfo(s"NativePlan for VeloxParquetScan: ${glutenPipeline.nativePlanString(true)}")
+
+    val glutenScanRDD = glutenPipeline.doExecuteColumnar()
+
+    // Narrow down all used metrics on the Executor side, in case transferring unnecessary metrics
+    val embeddedMetrics = {
+      val mapBuilder = mutable.Map.empty[String, GpuMetric]
+      if (rapidsConf.enableNativeVeloxConverter) {
+        commonMetrics.keys.foreach(key => mapBuilder += key -> allMetrics(key))
+        nativeMetrics.keys.foreach(key => mapBuilder += key -> allMetrics(key))
+        if (rapidsConf.parquetVeloxPreloadCapacity > 0) {
+          preloadingMetrics.keys.foreach(key => mapBuilder += key -> allMetrics(key))
+        }
+      } else {
+        commonMetrics.keys.foreach(key => mapBuilder += key -> allMetrics(key))
+        roundTripMetrics.keys.foreach(key => mapBuilder += key -> allMetrics(key))
+      }
+      mapBuilder.toMap
+    }
+
+    new VeloxParquetScanRDD(glutenScanRDD,
+      output,
+      requiredSchema,
+      TargetSize(coalesceSizeGoal),
+      embeddedMetrics,
+      useNativeConverter = rapidsConf.enableNativeVeloxConverter,
+      preloadedCapacity = rapidsConf.parquetVeloxPreloadCapacity)
+  }
+
+  override def inputRDDs(): Seq[RDD[InternalRow]] = {
+    inputRDD :: Nil
+  }
+
+  private val commonMetrics = Map[String, () => GpuMetric](
+    "VeloxScanTime" -> (() => createNanoTimingMetric(MODERATE_LEVEL, "VeloxScanTime")),
+    "GpuAcquireTime" -> (() => createNanoTimingMetric(MODERATE_LEVEL, "GpuAcquireTime")),
+  )
+  private val nativeMetrics = Map[String, () => GpuMetric](
+    "C2COutputBatches" -> (() => createMetric(MODERATE_LEVEL, "C2COutputBatches")),
+    "VeloxOutputBatches" -> (() => createMetric(MODERATE_LEVEL, "VeloxOutputBatches")),
+    "C2COutputSize" -> (() => createSizeMetric(MODERATE_LEVEL, "C2COutputSize")),
+    "C2CTime" -> (() => createNanoTimingMetric(MODERATE_LEVEL, "C2CTime")),
+    "C2CStreamTime" -> (() => createNanoTimingMetric(MODERATE_LEVEL, "C2CStreamTime")),
+    "PageableH2DTime" -> (() => createNanoTimingMetric(MODERATE_LEVEL, "PageableH2DTime")),
+    "PinnedH2DTime" -> (() => createNanoTimingMetric(MODERATE_LEVEL, "PinnedH2DTime")),
+    "PageableH2DSize" -> (() => createSizeMetric(MODERATE_LEVEL, "PageableH2DSize")),
+    "PinnedH2DSize" -> (() => createSizeMetric(MODERATE_LEVEL, "PinnedH2DSize")),
+  )
+  private val preloadingMetrics = Map[String, () => GpuMetric](
+    "preloadWaitTime" -> (() => createNanoTimingMetric(MODERATE_LEVEL, "preloadWaitTime"))
+  )
+  private val roundTripMetrics = Map[String, () => GpuMetric](
+    "C2ROutputRows" -> (() => createMetric(MODERATE_LEVEL, "C2ROutputRows")),
+    "C2ROutputBatches" -> (() => createMetric(MODERATE_LEVEL, "C2ROutputBatches")),
+    "R2CInputRows" -> (() => createMetric(MODERATE_LEVEL, "R2CInputRows")),
+    "R2COutputRows" -> (() => createMetric(MODERATE_LEVEL, "R2COutputRows")),
+    "R2COutputBatches" -> (() => createMetric(MODERATE_LEVEL, "R2COutputBatches")),
+    "VeloxC2RTime" -> (() => createTimingMetric(MODERATE_LEVEL, "VeloxC2RTime")),
+    "R2CTime" -> (() => createNanoTimingMetric(MODERATE_LEVEL, "R2CTime")),
+    "R2CStreamTime" -> (() => createNanoTimingMetric(MODERATE_LEVEL, "R2CStreamTime")),
+  )
+
+  override lazy val allMetrics: Map[String, GpuMetric] = {
+    val mapBuilder = Map.newBuilder[String, GpuMetric]
+    mapBuilder += "scanTime" -> createTimingMetric(ESSENTIAL_LEVEL, "TotalTime")
+    // Add common embedded metrics
+    commonMetrics.foreach { case (key, generator) =>
+      mapBuilder += key -> generator()
+    }
+    // NativeConverter and RoundTripConverter uses different metrics
+    if (rapidsConf.enableNativeVeloxConverter) {
+      nativeMetrics.foreach { case (key, generator) =>
+        mapBuilder += key -> generator()
+      }
+      if (rapidsConf.parquetVeloxPreloadCapacity > 0) {
+        preloadingMetrics.foreach { case (key, generator) =>
+          mapBuilder += key -> generator()
+        }
+      }
+    } else {
+      roundTripMetrics.foreach { case (key, generator) =>
+        mapBuilder += key -> generator()
+      }
+    }
+    // Expose all metrics of the underlying GlutenScanExec
+    glutenScan.metrics.foreach { case (key, metric) =>
+      mapBuilder += s"GLUTEN_$key" -> GpuMetric.wrap(metric)
+    }
+    mapBuilder.result()
+  }
+
+  override protected def doExecute(): RDD[InternalRow] =
+    throw new IllegalStateException(s"Row-based execution should not occur for $this")
+
+  override protected def internalDoExecuteColumnar(): RDD[ColumnarBatch] = {
+    val scanTime = gpuLongMetric("scanTime")
+    inputRDD.asInstanceOf[RDD[ColumnarBatch]].mapPartitionsInternal { batches =>
+      new Iterator[ColumnarBatch] {
+
+        override def hasNext: Boolean = {
+          val startNs = System.nanoTime()
+          val hasNext = batches.hasNext
+          scanTime += NANOSECONDS.toMillis(System.nanoTime() - startNs)
+          hasNext
+        }
+
+        override def next(): ColumnarBatch = {
+          val startNs = System.nanoTime()
+          val batch = batches.next()
+          scanTime += NANOSECONDS.toMillis(System.nanoTime() - startNs)
+          batch
+        }
+      }
+    }
+  }
+
+  override def doCanonicalize(): VeloxFileSourceScanExec = {
+    VeloxFileSourceScanExec(
+      relation,
+      originalOutput.map(QueryPlan.normalizeExpressions(_, originalOutput)),
+      requiredSchema,
+      QueryPlan.normalizePredicates(
+        filterUnusedDynamicPruningExpressions(partitionFilters), originalOutput),
+      optionalBucketSet,
+      optionalNumCoalescedBuckets,
+      QueryPlan.normalizePredicates(dataFilters, originalOutput),
+      None,
+      queryUsesInputFile,
+      alluxioPathsMap = alluxioPathsMap)(rapidsConf)
+  }
+
+  // Filters unused DynamicPruningExpression expressions - one which has been replaced
+  // with DynamicPruningExpression(Literal.TrueLiteral) during Physical Planning
+  private def filterUnusedDynamicPruningExpressions(predicates: Seq[Expression]) = {
+    predicates.filterNot(_ == DynamicPruningExpression(Literal.TrueLiteral))
+  }
+
+  override def otherCopyArgs: Seq[AnyRef] = Seq(rapidsConf)
+
+}
diff --git a/sql-plugin/src/main/scala/org/apache/spark/rapids/velox/VeloxHDFS.scala b/sql-plugin/src/main/scala/org/apache/spark/rapids/velox/VeloxHDFS.scala
new file mode 100644
index 00000000000..060008f4944
--- /dev/null
+++ b/sql-plugin/src/main/scala/org/apache/spark/rapids/velox/VeloxHDFS.scala
@@ -0,0 +1,133 @@
+/*
+ * Copyright (c) 2024, NVIDIA CORPORATION.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.rapids.velox
+
+import ai.rapids.cudf.{HostMemoryBuffer, NvtxColor, NvtxRange}
+import com.nvidia.spark.rapids.{GpuParquetUtils, HostMemoryOutputStream}
+import com.nvidia.spark.rapids.Arm.{closeOnExcept, withResource}
+import com.nvidia.spark.rapids.velox.VeloxBackendApis
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.fs.Path
+import org.apache.parquet.hadoop.ParquetFileWriter.MAGIC
+
+import org.apache.spark.internal.Logging
+
+case class FileCopyRange(offset: Long, length: Long, targetMemAddr: Long)
+
+object VeloxHDFS extends Logging {
+
+  private val PARQUET_FOOTER_LENGTH_SIZE = 4
+
+  /**
+   * Returns object handle of VeloxHadoopInputFileStream. Returns 0 if fails.
+   */
+  def createInputFileStream(filePath: String): Long = {
+    if (!isVeloxHDFSEnabled) {
+      return 0L
+    }
+    // If openHadoopFile returns 0, it means the specific filePath cannot be parsed by veloxHDFS
+    val ptr = runtime.hdfsOpenFile(filePath)
+    if (ptr == 0L) {
+      logError(s"failed to open file($filePath) via VeloxHDFS")
+    }
+    ptr
+  }
+
+  /**
+   * Reads range data contiguously from HDFS files via VeloxHDFS.
+   */
+  def copyRangesFromFile(fileName: String,
+                         fileStream: Long,
+                         ranges: Seq[FileCopyRange],
+                         closeAfterFinished: Boolean = false): Unit = {
+    try {
+      ranges.foreach {
+        case FileCopyRange(offset, len, tgtAddr) =>
+          runtime.hdfsCopyFromFile(fileStream, offset, len, tgtAddr, false)
+      }
+      logError(s"successfully read file($fileName) through VeloxHDFS")
+    } finally {
+      if (closeAfterFinished) {
+        runtime.hdfsCloseFile(fileStream)
+      }
+    }
+  }
+
+  /**
+   * The alternative implementation of GpuParquetFileFilterHandler.readFooterBufUsingHadoop
+   * via VeloxHDFS (backed by libhdfs3) instead of parquet-mr
+   */
+  def readParquetFooterBuffer(filePath: Path, conf: Configuration): Option[HostMemoryBuffer] = {
+    if (!isVeloxHDFSEnabled) {
+      return None
+    }
+    // If openHadoopFile returns 0, it means the specific filePath cannot be parsed by veloxHDFS
+    val handle = runtime.hdfsOpenFile(filePath.toString)
+    if (handle == 0L) {
+      logError(s"failed to open file($filePath) via VeloxHDFS")
+      return None
+    }
+
+    val fileLen = runtime.hdfsGetFileSize(handle)
+    if (fileLen < MAGIC.length + PARQUET_FOOTER_LENGTH_SIZE + MAGIC.length) {
+      throw new RuntimeException(s"$filePath is not a Parquet file (too small length: $fileLen )")
+    }
+    val footerLengthIndex = fileLen - PARQUET_FOOTER_LENGTH_SIZE - MAGIC.length
+    try {
+      withResource(new NvtxRange("ReadFooterBytes", NvtxColor.YELLOW)) { _ =>
+        runtime.hdfsReaderSeekTo(handle, footerLengthIndex)
+        val footerLength = runtime.hdfsReadIntLittleEndian(handle)
+        val magic = new Array[Byte](MAGIC.length)
+        withResource(HostMemoryBuffer.allocate(MAGIC.length, false)) { buf =>
+          runtime.hdfsReadFully(handle, buf.getAddress, buf.getLength)
+          buf.getBytes(magic, 0, 0, buf.getLength)
+        }
+        val footerIndex = footerLengthIndex - footerLength
+        GpuParquetUtils.verifyParquetMagic(filePath, magic)
+        if (footerIndex < MAGIC.length || footerIndex >= footerLengthIndex) {
+          throw new RuntimeException(s"corrupted file: the footer index is not within " +
+            s"the file: $footerIndex")
+        }
+        val hmbLength = (fileLen - footerIndex).toInt
+        closeOnExcept(HostMemoryBuffer.allocate(hmbLength + MAGIC.length, false)) { buf =>
+          val out = new HostMemoryOutputStream(buf)
+          out.write(MAGIC)
+          // close HadoopFileInputStream after read
+          runtime.hdfsCopyFromFile(handle, footerIndex, hmbLength,
+            buf.getAddress + out.getPos,
+            true
+          )
+          logError(s"successfully read ParquetFooter($filePath) through VeloxHDFS")
+          Some(buf)
+        }
+      }
+    } catch {
+      case ex: Throwable =>
+        runtime.hdfsCloseFile(handle)
+        throw ex
+    }
+  }
+
+  // It's safe to assert VeloxRuntime is defined when useVeloxHdfs returns true, because
+  // useVeloxHdfs shall certainly return false if VeloxRuntime has NOT been setup.
+  @transient
+  private lazy val isVeloxHDFSEnabled = VeloxBackendApis.useVeloxHdfs
+
+  @transient
+  private lazy val runtime = VeloxBackendApis.getRuntime.get
+
+}
diff --git a/sql-plugin/src/main/scala/org/apache/spark/rapids/velox/VeloxParquetScanRDD.scala b/sql-plugin/src/main/scala/org/apache/spark/rapids/velox/VeloxParquetScanRDD.scala
new file mode 100644
index 00000000000..1ce8e411d1e
--- /dev/null
+++ b/sql-plugin/src/main/scala/org/apache/spark/rapids/velox/VeloxParquetScanRDD.scala
@@ -0,0 +1,253 @@
+/*
+ * Copyright (c) 2024, NVIDIA CORPORATION.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.rapids.velox
+
+import java.util.concurrent.locks.ReentrantLock
+
+import ai.rapids.cudf.NvtxColor
+import com.nvidia.spark.rapids.{CoalesceSizeGoal, GpuMetric, NvtxWithMetrics}
+import com.nvidia.spark.rapids.Arm.withResource
+import com.nvidia.spark.rapids.velox.RapidsHostColumn
+
+import org.apache.spark.{InterruptibleIterator, Partition, TaskContext}
+import org.apache.spark.internal.Logging
+import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.Attribute
+import org.apache.spark.sql.execution.metric.SQLMetric
+import org.apache.spark.sql.rapids.execution.TrampolineUtil
+import org.apache.spark.sql.types.{StructField, StructType}
+import org.apache.spark.sql.vectorized.ColumnarBatch
+
+class VeloxParquetScanRDD(scanRDD: RDD[ColumnarBatch],
+                          outputAttr: Seq[Attribute],
+                          outputSchema: StructType,
+                          coalesceGoal: CoalesceSizeGoal,
+                          metrics: Map[String, GpuMetric],
+                          useNativeConverter: Boolean,
+                          preloadedCapacity: Int
+                         ) extends RDD[InternalRow](scanRDD.sparkContext, Nil) {
+
+  private val veloxScanTime = GpuMetric.unwrap(metrics("VeloxScanTime"))
+
+  override protected def getPartitions: Array[Partition] = scanRDD.partitions
+
+  override def compute(split: Partition, context: TaskContext): Iterator[InternalRow] = {
+    // the wrapping Iterator for the underlying VeloxScan task
+    val veloxIter = new VeloxScanMetricsIter(scanRDD.compute(split, context), veloxScanTime)
+
+    val resIter = if (!useNativeConverter) {
+      VeloxColumnarBatchConverter.roundTripConvert(
+        veloxIter, outputAttr, coalesceGoal, metrics)
+    } else {
+      // Preloading only works for NativeConverter because using roundTripConverter we
+      // can NOT split the building process of HostColumnVector and the host2device process,
+      // since they are completed in one by GpuRowToColumnConverter.
+
+      val schema = StructType(outputAttr.map { ar =>
+        StructField(ar.name, ar.dataType, ar.nullable)
+      })
+      require(coalesceGoal.targetSizeBytes <= Int.MaxValue,
+        s"targetSizeBytes should be smaller than 2GB, but got ${coalesceGoal.targetSizeBytes}"
+      )
+      val coalesceConverter = new CoalesceNativeConverter(
+        veloxIter, coalesceGoal.targetSizeBytes.toInt, schema, metrics
+      )
+
+      val hostIter: RapidsHostBatchProducer = if (preloadedCapacity > 0) {
+        val producerInitFn = () => {
+          coalesceConverter.setRuntime()
+        }
+        PrefetchHostBatchProducer(context.taskAttemptId(),
+          coalesceConverter,
+          producerInitFn,
+          preloadedCapacity,
+          metrics("preloadWaitTime")
+        )
+      } else {
+        coalesceConverter.setRuntime()
+        SyncHostBatchProducer(coalesceConverter)
+      }
+
+      VeloxColumnarBatchConverter.hostToDevice(hostIter, outputAttr, metrics)
+    }
+
+    // TODO: SPARK-25083 remove the type erasure hack in data source scan
+    new InterruptibleIterator(context, resIter.asInstanceOf[Iterator[InternalRow]])
+  }
+
+}
+
+private class VeloxScanMetricsIter(iter: Iterator[ColumnarBatch],
+                                   scanTime: SQLMetric
+                                  ) extends Iterator[ColumnarBatch] {
+  override def hasNext: Boolean = {
+    val start = System.nanoTime()
+    try {
+      iter.hasNext
+    } finally {
+      scanTime += System.nanoTime() - start
+    }
+  }
+
+  override def next(): ColumnarBatch = {
+    val start = System.nanoTime()
+    try {
+      iter.next()
+    } finally {
+      scanTime += System.nanoTime() - start
+    }
+  }
+}
+
+private case class PrefetchHostBatchProducer(taskAttId: Long,
+                                             iterImpl: Iterator[Array[RapidsHostColumn]],
+                                             producerInitFn: () => Unit,
+                                             capacity: Int,
+                                             waitTimeMetric: GpuMetric
+                                            ) extends RapidsHostBatchProducer with Logging {
+
+  @transient
+  @volatile private var isInit: Boolean = false
+  @transient
+  @volatile private var isProducing: Boolean = false
+  @transient @volatile private var readIndex: Int = 0
+  @transient @volatile private var writeIndex: Int = 0
+  // This lock guarantees anytime if ProducerStatus == running there must be a working batch
+  // being produced or waiting to be put into the queue.
+  @transient private lazy val hasNextLock = new ReentrantLock()
+
+  @transient private lazy val emptyLock = new ReentrantLock()
+  @transient private lazy val fullLock = new ReentrantLock()
+
+  private var producer: Thread = _
+
+  @transient private lazy val buffer: Array[Either[Throwable, Array[RapidsHostColumn]]] = {
+    Array.ofDim[Either[Throwable, Array[RapidsHostColumn]]](capacity)
+  }
+
+  @transient private lazy val produceFn: Runnable = new Runnable {
+
+    // This context will be got in the main Thread during the initialization of `produceFn`
+    private val taskContext: TaskContext = TaskContext.get()
+
+    override def run(): Unit = {
+      TrampolineUtil.setTaskContext(taskContext)
+      hasNextLock.lock()
+      try {
+        do {
+          isProducing = true
+          hasNextLock.unlock()
+
+          do {
+            fullLock.synchronized {
+              if (writeIndex - readIndex == capacity) {
+                fullLock.wait()
+              }
+            }
+          } while (writeIndex - readIndex == capacity)
+
+          buffer(writeIndex % capacity) = Right(iterImpl.next())
+          emptyLock.synchronized {
+            writeIndex += 1
+            emptyLock.notify()
+          }
+
+          hasNextLock.lock()
+          isProducing = false
+          logInfo(s"[$taskAttId] PreloadedIterator produced $writeIndex batches, " +
+            s"currently preloaded batchNum: ${writeIndex - readIndex}"
+          )
+        }
+        while (iterImpl.hasNext)
+        hasNextLock.unlock()
+      } catch {
+        case ex: Throwable =>
+          // transfer the exception info to the main thread as an interrupted signal
+          buffer(writeIndex % capacity) = Left(ex)
+          writeIndex += 1
+          isProducing = false
+          if (hasNextLock.isHeldByCurrentThread) {
+            hasNextLock.unlock()
+          }
+          throw new RuntimeException(ex)
+      } finally {
+        TrampolineUtil.unsetTaskContext()
+      }
+    }
+  }
+
+  override def hasNext: Boolean = {
+    if (!isInit) {
+      withResource(new NvtxWithMetrics("waitForCPU", NvtxColor.RED, waitTimeMetric)) { _ =>
+        if (!iterImpl.hasNext) {
+          return false
+        }
+        isInit = true
+        isProducing = true
+        producerInitFn()
+        producer = new Thread(produceFn)
+        producer.start()
+        return true
+      }
+    }
+
+    writeIndex > readIndex || {
+      hasNextLock.lock()
+      val ret = writeIndex > readIndex || isProducing
+      hasNextLock.unlock()
+      ret
+    }
+  }
+
+  override def waitForNext(): Unit = {
+    // Return if buffer is not empty
+    if (writeIndex > readIndex) {
+      return
+    }
+    // Waiting for "emptyLock"
+    withResource(new NvtxWithMetrics("waitForCPU", NvtxColor.RED, waitTimeMetric)) { _ =>
+      do {
+        emptyLock.synchronized {
+          if (writeIndex == readIndex) {
+            emptyLock.wait()
+          }
+        }
+      } while (writeIndex == readIndex)
+    }
+  }
+
+  override def takeNext: Array[RapidsHostColumn] = {
+    require(writeIndex > readIndex, "The RingBuffer is EMPTY")
+
+    buffer(readIndex % capacity) match {
+      case Left(ex: Throwable) =>
+        logError(s"[$taskAttId] PreloadedIterator: AsyncProducer failed with exceptions")
+        throw new RuntimeException(s"[$taskAttId] PreloadedIterator", ex)
+      case Right(ret: Array[RapidsHostColumn]) =>
+        logInfo(s"[$taskAttId] PreloadedIterator consumed $readIndex batches, " +
+          s"currently preloaded batchNum: ${writeIndex - readIndex}"
+        )
+        // Update the readIndex and activate "fullLock"
+        fullLock.synchronized {
+          readIndex += 1
+          fullLock.notify()
+        }
+        ret
+    }
+  }
+}
diff --git a/sql-plugin/src/main/spark320/scala/com/nvidia/spark/rapids/shims/spark320/VeloxFileSourceScanExecMeta.scala b/sql-plugin/src/main/spark320/scala/com/nvidia/spark/rapids/shims/spark320/VeloxFileSourceScanExecMeta.scala
new file mode 100644
index 00000000000..bb25958c07d
--- /dev/null
+++ b/sql-plugin/src/main/spark320/scala/com/nvidia/spark/rapids/shims/spark320/VeloxFileSourceScanExecMeta.scala
@@ -0,0 +1,136 @@
+/*
+ * Copyright (c) 2024, NVIDIA CORPORATION.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+/*** spark-rapids-shim-json-lines
+{"spark": "320"}
+{"spark": "321"}
+{"spark": "321cdh"}
+{"spark": "322"}
+{"spark": "323"}
+{"spark": "324"}
+spark-rapids-shim-json-lines ***/
+package com.nvidia.spark.rapids.shims
+
+import com.nvidia.spark.rapids._
+import org.apache.spark.rapids.velox.VeloxFileSourceScanExec
+
+import org.apache.spark.sql.catalyst.expressions.{Attribute, DynamicPruningExpression}
+import org.apache.spark.sql.execution._
+import org.apache.spark.sql.execution.datasources.{HadoopFsRelation, InMemoryFileIndex}
+
+class VeloxFileSourceScanExecMeta(plan: FileSourceScanExec,
+                                  conf: RapidsConf,
+                                  parent: Option[RapidsMeta[_, _, _]],
+                                  rule: DataFromReplacementRule,
+                                  pushedFilterSchema: Option[Seq[Attribute]] = None)
+  extends SparkPlanMeta[FileSourceScanExec](plan, conf, parent, rule) {
+
+  // Replaces SubqueryBroadcastExec inside dynamic pruning filters with GPU counterpart
+  // if possible. Instead regarding filters as childExprs of current Meta, we create
+  // a new meta for SubqueryBroadcastExec. The reason is that the GPU replacement of
+  // FileSourceScan is independent from the replacement of the partitionFilters. It is
+  // possible that the FileSourceScan is on the CPU, while the dynamic partitionFilters
+  // are on the GPU. And vice versa.
+  private lazy val partitionFilters = {
+    val convertBroadcast = (bc: SubqueryBroadcastExec) => {
+      val meta = GpuOverrides.wrapAndTagPlan(bc, conf)
+      meta.tagForExplain()
+      meta.convertIfNeeded().asInstanceOf[BaseSubqueryExec]
+    }
+    wrapped.partitionFilters.map { filter =>
+      filter.transformDown {
+        case dpe@DynamicPruningExpression(inSub: InSubqueryExec) =>
+          inSub.plan match {
+            case bc: SubqueryBroadcastExec =>
+              dpe.copy(inSub.copy(plan = convertBroadcast(bc)))
+            case reuse@ReusedSubqueryExec(bc: SubqueryBroadcastExec) =>
+              dpe.copy(inSub.copy(plan = reuse.copy(convertBroadcast(bc))))
+            case _ =>
+              dpe
+          }
+      }
+    }
+  }
+
+  // partition filters and data filters are not run on the GPU
+  override val childExprs: Seq[ExprMeta[_]] = Seq.empty
+
+  override def tagPlanForGpu(): Unit = ScanExecShims.tagGpuFileSourceScanExecSupport(this)
+
+  override def convertToGpu(): GpuExec = {
+    val sparkSession = wrapped.relation.sparkSession
+    val options = wrapped.relation.options
+    val (location, _) =
+      if (AlluxioCfgUtils.enabledAlluxioReplacementAlgoConvertTime(conf)) {
+        val shouldReadFromS3 = wrapped.relation.location match {
+          // Only handle InMemoryFileIndex
+          //
+          // skip handle `MetadataLogFileIndex`, from the description of this class:
+          // it's about the files generated by the `FileStreamSink`.
+          // The streaming data source is not in our scope.
+          //
+          // For CatalogFileIndex and FileIndex of `delta` data source,
+          // need more investigation.
+          case inMemory: InMemoryFileIndex =>
+            // List all the partitions to reduce overhead, pass in 2 empty filters.
+            // Subsequent process will do the right partition pruning.
+            // This operation is fast, because it lists files from the caches and the caches
+            // already exist in the `InMemoryFileIndex`.
+            val pds = inMemory.listFiles(Seq.empty, Seq.empty)
+            AlluxioUtils.shouldReadDirectlyFromS3(conf, pds)
+          case _ =>
+            false
+        }
+
+        if (!shouldReadFromS3) {
+          // it's convert time algorithm and some paths are not large tables
+          AlluxioUtils.replacePathIfNeeded(
+            conf,
+            wrapped.relation,
+            partitionFilters,
+            wrapped.dataFilters)
+        } else {
+          // convert time algorithm and read large files
+          (wrapped.relation.location, None)
+        }
+      } else {
+        // it's not convert time algorithm or read large files, do not replace
+        (wrapped.relation.location, None)
+      }
+
+    val newRelation = HadoopFsRelation(
+      location,
+      wrapped.relation.partitionSchema,
+      wrapped.relation.dataSchema,
+      wrapped.relation.bucketSpec,
+      wrapped.relation.fileFormat,
+      options)(sparkSession)
+
+    VeloxFileSourceScanExec(
+      newRelation,
+      wrapped.output,
+      wrapped.requiredSchema,
+      partitionFilters,
+      wrapped.optionalBucketSet,
+      wrapped.optionalNumCoalescedBuckets,
+      wrapped.dataFilters,
+      wrapped.tableIdentifier,
+      wrapped.disableBucketedScan,
+      queryUsesInputFile = false,
+      None,
+      filteredOutput = pushedFilterSchema)(conf)
+  }
+}