Upmerge spark2 directory to the latest 22.02 changes

scripts/rundiffspark2.sh

@@ -481,5 +481,15 @@ sed -n '/object GpuFileSourceScanExec/,/^}/{/^}/!p}'  ../sql-plugin/src/main/sca
 diff $tmp_dir/GpuFileSourceScanExec_new.out $tmp_dir/GpuFileSourceScanExec_old.out > $tmp_dir/GpuFileSourceScanExec.newdiff || true
 diff -c spark2diffs/GpuFileSourceScanExec.diff $tmp_dir/GpuFileSourceScanExec.newdiff
 
+sed -n '/object GpuReadJsonFileFormat/,/^}/{/^}/!p}'  ../spark2-sql-plugin/src/main/scala/org/apache/spark/sql/catalyst/json/rapids/GpuReadJsonFileFormat.scala > $tmp_dir/GpuReadJsonFileFormat_new.out
+sed -n '/object GpuReadJsonFileFormat/,/^}/{/^}/!p}'  ../sql-plugin/src/main/scala/org/apache/spark/sql/catalyst/json/rapids/GpuReadJsonFileFormat.scala > $tmp_dir/GpuReadJsonFileFormat_old.out
+diff $tmp_dir/GpuReadJsonFileFormat_new.out $tmp_dir/GpuReadJsonFileFormat_old.out > $tmp_dir/GpuReadJsonFileFormat.newdiff || true
+diff -c spark2diffs/GpuReadJsonFileFormat.diff $tmp_dir/GpuReadJsonFileFormat.newdiff
+
+sed -n '/object GpuJsonScan/,/^}/{/^}/!p}'  ../spark2-sql-plugin/src/main/scala/org/apache/spark/sql/catalyst/json/rapids/GpuJsonScan.scala > $tmp_dir/GpuJsonScan_new.out
+sed -n '/object GpuJsonScan/,/^}/{/^}/!p}'  ../sql-plugin/src/main/scala/org/apache/spark/sql/catalyst/json/rapids/GpuJsonScan.scala > $tmp_dir/GpuJsonScan_old.out
+diff $tmp_dir/GpuJsonScan_new.out $tmp_dir/GpuJsonScan_old.out > $tmp_dir/GpuJsonScan.newdiff || true
+diff -c spark2diffs/GpuJsonScan.diff $tmp_dir/GpuJsonScan.newdiff
+
 echo "Done running Diffs of spark2.x files"
 rm -r $tmp_dir

scripts/spark2diffs/FileSourceScanExec.diff

@@ -1,21 +1,30 @@
-6a7,24
+6a7,33
 >           // 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 = wrapped.partitionFilters.map { filter =>
->             filter.transformDown {
->               case dpe @ DynamicPruningExpression(inSub: InSubqueryExec)
->                 if inSub.plan.isInstanceOf[SubqueryBroadcastExec] =>
-> 
->                 val subBcMeta = GpuOverrides.wrapAndTagPlan(inSub.plan, conf)
->                 subBcMeta.tagForExplain()
->                 val gpuSubBroadcast = subBcMeta.convertIfNeeded().asInstanceOf[BaseSubqueryExec]
->                 dpe.copy(inSub.copy(plan = gpuSubBroadcast))
+>           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
+>                   }
+>               }
 >             }
 >           }
 > 
-10a29
+10a38
 > 

scripts/spark2diffs/GpuFileSourceScanExec.diff

@@ -1,10 +1,11 @@
-10a11,20
+11a12,22
 > 
 >   def convertFileFormat(format: FileFormat): FileFormat = {
 >     format match {
 >       case _: CSVFileFormat => new GpuReadCSVFileFormat
 >       case f if GpuOrcFileFormat.isSparkOrcFormat(f) => new GpuReadOrcFileFormat
 >       case _: ParquetFileFormat => new GpuReadParquetFileFormat
+>       case _: JsonFileFormat => new GpuReadJsonFileFormat
 >       case f =>
 >         throw new IllegalArgumentException(s"${f.getClass.getCanonicalName} is not supported")
 >     }

scripts/spark2diffs/GpuJsonScan.diff

@@ -0,0 +1,43 @@
+3,16d2
+<   def dateFormatInRead(fileOptions: Serializable): Option[String] = {
+<     fileOptions match {
+<       case jsonOpts: JSONOptions => Option(jsonOpts.dateFormat.getPattern)
+<       case _ => throw new RuntimeException("Wrong file options.")
+<     }
+<   }
+< 
+<   def timestampFormatInRead(fileOptions: Serializable): Option[String] = {
+<     fileOptions match {
+<       case jsonOpts: JSONOptions => Option(jsonOpts.timestampFormat.getPattern)
+<       case _ => throw new RuntimeException("Wrong file options.")
+<     }
+<   }
+< 
+41a28,37
+>   def tagSupport(scanMeta: ScanMeta[JsonScan]) : Unit = {
+>     val scan = scanMeta.wrapped
+>     tagSupport(
+>       scan.sparkSession,
+>       scan.dataSchema,
+>       scan.readDataSchema,
+>       scan.options.asScala.toMap,
+>       scanMeta)
+>   }
+> 
+47c43
+<       meta: RapidsMeta[_, _]): Unit = {
+---
+>       meta: RapidsMeta[_, _, _]): Unit = {
+109c105
+<       dateFormatInRead(parsedOptions).foreach { dateFormat =>
+---
+>       ShimLoader.getSparkShims.dateFormatInRead(parsedOptions).foreach { dateFormat =>
+117,118c113
+<       // Spark 2.x doesn't have zoneId, so use timeZone and then to id
+<       if (!TypeChecks.areTimestampsSupported(parsedOptions.timeZone.toZoneId)) {
+---
+>       if (!TypeChecks.areTimestampsSupported(parsedOptions.zoneId)) {
+121c116
+<       timestampFormatInRead(parsedOptions).foreach { tsFormat =>
+---
+>       ShimLoader.getSparkShims.timestampFormatInRead(parsedOptions).foreach { tsFormat =>

scripts/spark2diffs/GpuOrcFileFormat.diff

@@ -6,7 +6,7 @@
 <                     schema: StructType): Unit = {
 ---
 >                     schema: StructType): Option[GpuOrcFileFormat] = {
-79a80,84
+75a76,80
 >     if (meta.canThisBeReplaced) {
 >       Some(new GpuOrcFileFormat)
 >     } else {