[BUG] Wide schema read with operations falling back to the CPU show serious query slowdown

[kuhushukla]

Describe the bug
Queries with wide schemas show slowdown for basic SQL operations. This is especially worse if there are fallbacks along the way.

Steps/Code to reproduce bug
Write the test data:

import org.apache.spark.sql._
import org.apache.spark.sql.types._
import org.apache.spark.sql.functions._

val numColumns = 1000
val numRows = 1000000

val columnNames = (1 to numColumns).map(i => s"col$i")

val schema = StructType(columnNames.map(name => StructField(name, StringType, nullable = true)))

val rdd = spark.sparkContext.parallelize(1 to numRows).map { _ =>
  Row.fromSeq((1 to numColumns).map(_ => scala.util.Random.alphanumeric.take(10).mkString))
}

val df = spark.createDataFrame(rdd, schema)

df.printSchema()

df.write.format("avro").save("plarge_dataset.avro")

Run the query on this data:

import org.apache.spark.sql._
import org.apache.spark.sql.functions._
import org.apache.spark.sql.types._

def runProcessWithConfig(configValue: String): Double = {
  spark.conf.set("spark.rapids.sql.enabled", configValue)
  
  val startTime = System.nanoTime()
  
  val df = spark.read.format("avro").load("plarge_dataset.avro")
  val filteredDF = df.filter(col("col1").startsWith("A"))
  val appendSuffix = udf((s: String) => s + "-suffix")
  val projectedDF = filteredDF.withColumn("col2", appendSuffix(col("col2")))
  val unionDF = projectedDF.union(projectedDF)
  unionDF.write.mode("overwrite").parquet("output_dataset.parquet")
  
  val endTime = System.nanoTime()
  val duration = (endTime - startTime) / 1e9d
  println(s"Total time: $duration seconds with spark.rapids.sql.enabled = $configValue")
  
  duration
}

val durationFalse = runProcessWithConfig("false")
val durationTrue = runProcessWithConfig("true")

println(s"Time with spark.rapids.sql.enabled = false: $durationFalse seconds")
println(s"Time with spark.rapids.sql.enabled = true: $durationTrue seconds")

On my local machine with single RTX A5000 and spark in local mode the runtimes are as follows:

cala> println(s"Time with spark.rapids.sql.enabled = false: $durationFalse seconds")
Time with spark.rapids.sql.enabled = false: 24.504267984 seconds

scala> println(s"Time with spark.rapids.sql.enabled = true: $durationTrue seconds")
Time with spark.rapids.sql.enabled = true: 219.800524208 seconds

Expected behavior
Have a comparable if not better run time. Support performance with wide schemas . This case covers a non -shuffle case but also relates to #11127 which deals with wide schemas during shuffle.

Environment details (please complete the following information)

• Environment location: [Standalone, YARN, Kubernetes, Cloud(specify cloud provider)]
• Spark configuration settings related to the issue

Additional context
Add any other context about the problem here.

[abellina]

@kuhushukla do you know why avro fell back here?

[winningsix]

Similar question here about project fallback.

[revans2]

The project is falling back because it has a UDF in it. The avro is falling back because avro support is off by default. The CUDF code has not gotten a lot of love so we don't really trust it all that much right now.