Introduce hybrid (CPU) scan for Parquet read #11720

res-life · 2024-11-13T13:50:43Z

Merge C2C code to main

Move code from internal gitlab nvspark/spark-rapids

Signed-off-by: Chong Gao [email protected]

res-life · 2024-11-13T13:52:00Z

It's draft, may missed some code change, will double check later.
This can not pass building, because Gluten backends-velox 1.2.0 jar is not deployed to public maven repo by Gluten community.
The building will pass if the Gluten jars are installed locally by maven install

jlowe

Please elaborate in the headline and description what this PR is doing. C2C is not a well-known acronym in the project and is not very descriptive.

revans2

Just a quick look at the code. Nothing too in depth.

revans2 · 2024-11-20T15:10:52Z

sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuRowToColumnarExec.scala

@@ -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 =>


nit: can this be in a separate PR? It does not appear to have much to do with the hybrid scan or the C2C?

revans2 · 2024-11-20T15:11:55Z

sql-plugin/src/main/scala/com/nvidia/spark/rapids/RapidsConf.scala

+  // spark.rapids.sql.hybrid.loadBackend defined at HybridPluginWrapper of spark-rapids-private
+  val LOAD_HYBRID_BACKEND = conf(HybridBackend.LOAD_BACKEND_KEY)
+    .doc("Load hybrid backend as an extra plugin of spark-rapids during launch time")
+    .startupOnly()


nit: can we mark this as internal until we have the details worked out and we feel the feature is stable enough for a user to stumble on and try it on their own?

revans2 · 2024-11-20T15:13:36Z

sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuParquetScan.scala

@@ -2736,6 +2736,12 @@ case class ParquetTableReader(
  }

  override def close(): Unit = {
+    debugDumpPrefix.foreach { prefix =>


nit: this feels like it is a bug fix that is unrelated to the C2C code change. Could we do it in a separate PR?

winningsix · 2024-11-25T10:14:14Z

integration_tests/src/main/python/parquet_test.py

+2. TimestampType can NOT be the KeyType of MapType
+3. NestedMap is disabled because it may produce incorrect result (usually occurring when table is very small)
+"""
+velox_gens = [


could we align the name to hybrid here?

winningsix · 2024-11-25T10:16:35Z

sql-plugin/src/main/scala/com/nvidia/spark/rapids/RapidsConf.scala

@@ -1684,6 +1685,19 @@ val GPU_COREDUMP_PIPE_PATTERN = conf("spark.rapids.gpu.coreDump.pipePattern")
    .booleanConf
    .createWithDefault(false)

+  val HYBRID_PARQUET_READER = conf("spark.rapids.sql.parquet.useHybridReader")
+    .doc("Use HybridScan to read Parquet data via CPUs")


Better call out Gluten and Velox. "Use HybridScan to read Parquet data using CPUs. The underlying implementation leverages both Gluten and Velox."

Signed-off-by: Chong Gao <[email protected]>

res-life · 2024-11-25T10:30:24Z

Passed IT. Tested conventional Spark-Rapids jar and regular Spark-Rapids jar.
Passed NDS test.
Will fix comments later.
Will push commits related to make a uber jar for all spark versions.

revans2

I need to do some manual testing on my own to try and understand what is happening here and how this is all working. It may take a while.

revans2 · 2024-11-25T14:51:44Z

sql-plugin/src/main/scala/com/nvidia/spark/rapids/Plugin.scala

@@ -31,6 +31,7 @@ import com.nvidia.spark.DFUDFPlugin
 import com.nvidia.spark.rapids.RapidsConf.AllowMultipleJars
 import com.nvidia.spark.rapids.RapidsPluginUtils.buildInfoEvent
 import com.nvidia.spark.rapids.filecache.{FileCache, FileCacheLocalityManager, FileCacheLocalityMsg}
+import com.nvidia.spark.rapids.hybrid.HybridPluginWrapper


Where is this coming from? I don't see the class defined anywhere in this patch. I think previously the plan was to have the jar this is a part of to always be on the classpath, but the requirements changed and this is now coming from a separate jar that is not going to be on the classpath in all cases. So this has to be loaded through reflection and handle it not being there. We cannot just load it all of the time like this code is doing.

revans2 · 2024-11-25T14:53:45Z

sql-plugin/src/main/scala/org/apache/spark/rapids/hybrid/CoalesceConvertIterator.scala

+        // 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")


What is this an error? We are logging details. This should be an info at the most. Probably a debug.

revans2 · 2024-11-25T14:54:25Z

sql-plugin/src/main/scala/org/apache/spark/rapids/hybrid/CoalesceConvertIterator.scala

+
+  override def next(): Array[RapidsHostColumn] = {
+    val ntvx = new NvtxWithMetrics("VeloxC2CNext", NvtxColor.YELLOW, metrics("C2CStreamTime"))
+    withResource(ntvx) { _ =>


Why is the stream time valuable?

revans2 · 2024-11-25T14:56:09Z

sql-plugin/src/main/scala/org/apache/spark/rapids/hybrid/CoalesceConvertIterator.scala

+
+    hostIter.map { hostVectors =>
+      Option(TaskContext.get()).foreach { ctx =>
+        GpuSemaphore.tryAcquire(ctx) match {


Why are you doing a two stage acquire? Is it just so you can get another metric for the acquire time?

revans2 · 2024-11-25T14:57:16Z

sql-plugin/src/main/scala/org/apache/spark/rapids/hybrid/HybridBackend.scala

+import com.nvidia.spark.rapids.hybrid.HybridPluginWrapper
+
+object HybridBackend {
+  val LOAD_BACKEND_KEY: String = HybridPluginWrapper.LOAD_BACKEND_KEY


Same comment here as above. This cannot go in as is. We need to be able to run without the velox jars on the classpath.

revans2 · 2024-11-25T14:59:07Z

sql-plugin/pom.xml

+
+        <dependency>
+            <groupId>com.nvidia</groupId>
+            <artifactId>rapids-4-spark-hybrid_${scala.binary.version}</artifactId>


Like I said previously I don't think that the plan is to have this jar on the classpath all of the time. We need to have a way to make sure that we can run without velox on the classpath. Does this jar pull in gluten/velox? Does it do it dynamically?

revans2 · 2024-11-25T15:05:22Z

sql-plugin/src/main/scala/org/apache/spark/rapids/hybrid/HybridParquetScanRDD.scala

+    val schema = StructType(outputAttr.map { ar =>
+      StructField(ar.name, ar.dataType, ar.nullable)
+    })
+    require(coalesceGoal.targetSizeBytes <= Int.MaxValue,


I am a little concerned about this. We just merged in a code change that allows the batch size to go above 2 GiB. When I grep through the code I don't see anywhere else that has a check like this. I am fine if gluten has a 2 GiB limit., but we need to fix up the coalesce goal instead of blowing up. I am fine if this is a follow on issue as this feature is off by default.

revans2 · 2024-11-25T15:07:54Z

sql-plugin/src/main/scala/org/apache/spark/rapids/hybrid/HybridParquetScanRDD.scala

+  }
+}
+
+// In terms of CPU parquet reader, both hasNext and next might be time-consuming. So, it is


Generally in these cases we try to push the metric down then. We want to try and separate out the I/O time from the rest of it. I am fine if that is really hard to do, and we might want to do it as a follow on issue instead then.

revans2 · 2024-11-25T15:13:03Z

...gin/src/main/spark320/scala/com/nvidia/spark/rapids/shims/HybridFileSourceScanExecMeta.scala

+        case MapType(kt, vt, _) if kt.isInstanceOf[MapType] || vt.isInstanceOf[MapType] => false
+        // For the time being, BinaryType is not supported yet
+        case _: BinaryType => false
+        case _ => true


facebookincubator/velox#9560 I am not an expert, and I don't even know what version of velox we will end up using. It sounds like it is plugable. But according to this, even the latest version of velox cannot handle bytes/TINYINT. We are not looking for spaces in the names of columns, among other issues. I know that other implementations fall back for even more things. Should we be concerned about this?

revans2 · 2024-11-25T15:14:20Z

sql-plugin/src/main/spark320/scala/com/nvidia/spark/rapids/shims/ScanExecShims.scala

-      (fsse, conf, p, r) => new FileSourceScanExecMeta(fsse, conf, p, r)),
+      (fsse, conf, p, r) => {
+        // TODO: HybridScan supports DataSourceV2
+        // TODO: HybridScan only supports Spark 32X for now.


What happens if you try to enable it on anything else? Is the error message clear?

res-life requested review from jlowe and sperlingxx November 14, 2024 01:13

jlowe reviewed Nov 14, 2024

View reviewed changes

sameerz added the performance A performance related task/issue label Nov 16, 2024

revans2 reviewed Nov 20, 2024

View reviewed changes

res-life changed the base branch from branch-24.12 to branch-25.02 November 25, 2024 09:53

winningsix reviewed Nov 25, 2024

View reviewed changes

Chong Gao and others added 5 commits November 25, 2024 18:18

Merge C2C code to main

65de010

Signed-off-by: Chong Gao <[email protected]>

Update the dependencies in pom.xml

e6cede2

revert BD velox hdfs code

1e4fc13

fit codes into the new HybridScan hierarchy

46e19df

refine QueryPlan, RapidsMeta and test suites for HybridScan

4f2a4d6

res-life force-pushed the merge-c2c branch from 74b6075 to 4f2a4d6 Compare November 25, 2024 10:19

Integrate Hybrid plugin; update IT

4d52f90

res-life marked this pull request as ready for review November 25, 2024 10:25

res-life requested review from tgravescs, GaryShen2008, NvTimLiu and gerashegalov as code owners November 25, 2024 10:25

revans2 requested changes Nov 25, 2024

View reviewed changes

res-life marked this pull request as draft November 26, 2024 00:59

winningsix changed the title ~~Merge C2C code to main~~ Introduce hybrid (CPU) scan for Parquet read Nov 26, 2024

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Introduce hybrid (CPU) scan for Parquet read #11720

Introduce hybrid (CPU) scan for Parquet read #11720

res-life commented Nov 13, 2024 •

edited

Loading

res-life commented Nov 13, 2024 •

edited

Loading

jlowe left a comment

revans2 left a comment

revans2 Nov 20, 2024

revans2 Nov 20, 2024

revans2 Nov 20, 2024

winningsix Nov 25, 2024

winningsix Nov 25, 2024

res-life commented Nov 25, 2024

revans2 left a comment

revans2 Nov 25, 2024

revans2 Nov 25, 2024

revans2 Nov 25, 2024

revans2 Nov 25, 2024

revans2 Nov 25, 2024

revans2 Nov 25, 2024

revans2 Nov 25, 2024

revans2 Nov 25, 2024

revans2 Nov 25, 2024

revans2 Nov 25, 2024

Introduce hybrid (CPU) scan for Parquet read #11720

Are you sure you want to change the base?

Introduce hybrid (CPU) scan for Parquet read #11720

Conversation

res-life commented Nov 13, 2024 • edited Loading

res-life commented Nov 13, 2024 • edited Loading

jlowe left a comment

Choose a reason for hiding this comment

revans2 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

res-life commented Nov 25, 2024

revans2 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

res-life commented Nov 13, 2024 •

edited

Loading

res-life commented Nov 13, 2024 •

edited

Loading