to_json - array with single map

[Feng-Jiang28]

The same issue in: when there is only a single map in the array, rapids throws an exception:
#10920
CPU:

scala> import org.apache.spark.sql.{SparkSession, Row}
import org.apache.spark.sql.{SparkSession, Row}

scala> import org.apache.spark.sql.functions._
import org.apache.spark.sql.functions._

scala> import org.apache.spark.sql.types._
import org.apache.spark.sql.types._

scala> import org.apache.spark.sql.catalyst.util.ArrayBasedMapData
import org.apache.spark.sql.catalyst.util.ArrayBasedMapData

scala> import org.apache.spark.unsafe.types.UTF8String
import org.apache.spark.unsafe.types.UTF8String

scala> 

scala> val spark = SparkSession.builder.appName("Test").getOrCreate()
24/05/28 14:56:32 WARN Utils: Service 'SparkUI' could not bind on port 4040. Attempting port 4041.
spark: org.apache.spark.sql.SparkSession = org.apache.spark.sql.SparkSession@286ced2a

scala> import spark.implicits._
import spark.implicits._

scala> val mapType = MapType(StringType, IntegerType)
mapType: org.apache.spark.sql.types.MapType = MapType(StringType,IntegerType,true)

scala> val inputSchema = ArrayType(mapType)
inputSchema: org.apache.spark.sql.types.ArrayType = ArrayType(MapType(StringType,IntegerType,true),true)

scala> val data = Seq(
     |   Row(Seq(Map("a" -> 1)))
     | )
data: Seq[org.apache.spark.sql.Row] = List([List(Map(a -> 1))])

scala> val df = spark.createDataFrame(
     |   spark.sparkContext.parallelize(data),
     |   StructType(Seq(StructField("value", inputSchema)))
     | )
df: org.apache.spark.sql.DataFrame = [value: array<map<string,int>>]

scala> val jsonDF = df.select(to_json(col("value")).alias("json_string"))
jsonDF: org.apache.spark.sql.DataFrame = [json_string: string]

scala> jsonDF.show(false)
+-----------+
|json_string|
+-----------+
|[{"a":1}]  |
+-----------+

GPU:

$ $SPARK_HOME/bin/spark-shell --master local[*] --jars ${SPARK_RAPIDS_PLUGIN_JAR} 
--conf spark.plugins=com.nvidia.spark.SQLPlugin 
--conf spark.rapids.sql.enabled=true 
--conf spark.rapids.sql.explain=ALL --driver-java-options '-ea -Duser.timezone=UTC ' 
--conf spark.rapids.sql.expression.JsonTuple=true 
--conf spark.rapids.sql.expression.GetJsonObject=true 
--conf spark.rapids.sql.expression.JsonToStructs=true 
--conf spark.rapids.sql.expression.StructsToJson=true

scala> import org.apache.spark.sql.{SparkSession, Row}
import org.apache.spark.sql.{SparkSession, Row}

scala> import org.apache.spark.sql.functions._
import org.apache.spark.sql.functions._

scala> import org.apache.spark.sql.types._
import org.apache.spark.sql.types._

scala> import org.apache.spark.sql.catalyst.util.ArrayBasedMapData
import org.apache.spark.sql.catalyst.util.ArrayBasedMapData

scala> import org.apache.spark.unsafe.types.UTF8String
import org.apache.spark.unsafe.types.UTF8String

scala> 

scala> val spark = SparkSession.builder.appName("Test").getOrCreate()
24/05/28 06:59:25 WARN SparkSession: Using an existing Spark session; only runtime SQL configurations will take effect.
spark: org.apache.spark.sql.SparkSession = org.apache.spark.sql.SparkSession@4effe36a

scala> import spark.implicits._
import spark.implicits._

scala> val mapType = MapType(StringType, IntegerType)
mapType: org.apache.spark.sql.types.MapType = MapType(StringType,IntegerType,true)

scala> val inputSchema = ArrayType(mapType)
inputSchema: org.apache.spark.sql.types.ArrayType = ArrayType(MapType(StringType,IntegerType,true),true)

scala> val data = Seq(
     |   Row(Seq(Map("a" -> 1)))
     | )
data: Seq[org.apache.spark.sql.Row] = List([List(Map(a -> 1))])

scala> val df = spark.createDataFrame(
     |   spark.sparkContext.parallelize(data),
     |   StructType(Seq(StructField("value", inputSchema)))
     | )
df: org.apache.spark.sql.DataFrame = [value: array<map<string,int>>]

scala> val jsonDF = df.select(to_json(col("value")).alias("json_string"))
jsonDF: org.apache.spark.sql.DataFrame = [json_string: string]

scala> jsonDF.show(false)
24/05/28 06:59:28 WARN GpuOverrides: 
!Exec <CollectLimitExec> cannot run on GPU because the Exec CollectLimitExec has been disabled, and is disabled by default because Collect Limit replacement can be slower on the GPU, if huge number of rows in a batch it could help by limiting the number of rows transferred from GPU to CPU. Set spark.rapids.sql.exec.CollectLimitExec to true if you wish to enable it
  @Partitioning <SinglePartition$> could run on GPU
  *Exec <ProjectExec> will run on GPU
    *Expression <Alias> to_json(value#1, Some(UTC)) AS json_string#6 will run on GPU
      *Expression <StructsToJson> to_json(value#1, Some(UTC)) will run on GPU
    ! <RDDScanExec> cannot run on GPU because GPU does not currently support the operator class org.apache.spark.sql.execution.RDDScanExec
      @Expression <AttributeReference> value#1 could run on GPU

24/05/28 06:59:29 ERROR Executor: Exception in task 6.0 in stage 2.0 (TID 11)
java.lang.ClassCastException: org.apache.spark.sql.types.ArrayType cannot be cast to org.apache.spark.sql.types.StructType
	at org.apache.spark.sql.rapids.GpuStructsToJson.doColumnar(GpuStructsToJson.scala:86)
	at com.nvidia.spark.rapids.GpuUnaryExpression.doItColumnar(GpuExpressions.scala:250)
	at com.nvidia.spark.rapids.GpuUnaryExpression.$anonfun$columnarEval$1(GpuExpressions.scala:261)
	at com.nvidia.spark.rapids.Arm$.withResource(Arm.scala:30)
	at com.nvidia.spark.rapids.GpuUnaryExpression.columnarEval(GpuExpressions.scala:260)
	at com.nvidia.spark.rapids.RapidsPluginImplicits$ReallyAGpuExpression.columnarEval(implicits.scala:35)
	at com.nvidia.spark.rapids.GpuAlias.columnarEval(namedExpressions.scala:110)
	at com.nvidia.spark.rapids.RapidsPluginImplicits$ReallyAGpuExpression.columnarEval(implicits.scala:35)
	at com.nvidia.spark.rapids.GpuProjectExec$.$anonfun$project$1(basicPhysicalOperators.scala:110)
	at com.nvidia.spark.rapids.RapidsPluginImplicits$MapsSafely.$anonfun$safeMap$1(implicits.scala:221)
	at com.nvidia.spark.rapids.RapidsPluginImplicits$MapsSafely.$anonfun$safeMap$1$adapted(implicits.scala:218)
	at scala.collection.immutable.List.foreach(List.scala:431)
	at com.nvidia.spark.rapids.RapidsPluginImplicits$MapsSafely.safeMap(implicits.scala:218)
	at com.nvidia.spark.rapids.RapidsPluginImplicits$AutoCloseableProducingSeq.safeMap(implicits.scala:253)
	at com.nvidia.spark.rapids.GpuProjectExec$.project(basicPhysicalOperators.scala:110)
	at com.nvidia.spark.rapids.GpuTieredProject.$anonfun$project$2(basicPhysicalOperators.scala:619)
	at com.nvidia.spark.rapids.Arm$.withResource(Arm.scala:30)
	at com.nvidia.spark.rapids.GpuTieredProject.recurse$2(basicPhysicalOperators.scala:618)
	at com.nvidia.spark.rapids.GpuTieredProject.project(basicPhysicalOperators.scala:631)
	at com.nvidia.spark.rapids.GpuTieredProject.$anonfun$projectWithRetrySingleBatchInternal$5(basicPhysicalOperators.scala:567)