Fix index-based access to the head elements (#2192)

Signed-off-by: Gera Shegalov <gera@apache.org> Fix simple collection operations. Contributes to #2109 - replace map with foreach when the return value is not used - forall where appropriate - access head elements without specifying indices. - simplify index iteration
NVIDIA · Apr 21, 2021 · 7e66f53 · 7e66f53
1 parent d0964b5
commit 7e66f53
Show file tree

Hide file tree

Showing 28 changed files with 129 additions and 120 deletions.
diff --git a/integration_tests/src/test/scala/com/nvidia/spark/rapids/StringFunctionSuite.scala b/integration_tests/src/test/scala/com/nvidia/spark/rapids/StringFunctionSuite.scala
@@ -208,7 +208,7 @@ class StringOperatorsDiagnostics extends SparkQueryCompareTestSuite {
 
     println("\u001b[1;36mSummary of diffs:\u001b[0m")
     println("\u001b[1;36mCodepoint:\u001b[0m ")    
-    for (i <- 0 until fromCpu.length) { 
+    for (i <- fromCpu.indices) {
       if (fromCpu(i) != fromGpu(i)) { 
         val codepoint = TestCodepoints.validCodepointIndices(i)
         print(f"$codepoint%5d, ")  
@@ -219,7 +219,7 @@ class StringOperatorsDiagnostics extends SparkQueryCompareTestSuite {
     println("\u001b[1;36mDetails:")
     println("Codepoint       CPU               GPU")
     println("single -> single mappings\u001b[0m");
-    for (i <- 0 until fromCpu.length) {
+    for (i <- fromCpu.indices) {
       if (fromCpu(i) != fromGpu(i) && fromCpu(i).getString(0).length == 1) {
         val codepoint = TestCodepoints.validCodepointIndices(i)
 
@@ -230,7 +230,7 @@ class StringOperatorsDiagnostics extends SparkQueryCompareTestSuite {
       }
     }
     println("\u001b[1;36msingle -> multi mappings\u001b[0m");
-    for (i <- 0 until fromCpu.length) {
+    for (i <- fromCpu.indices) {
       if (fromCpu(i) != fromGpu(i) && fromCpu(i).getString(0).length > 1) {
         var cpu_str = fromCpu(i).getString(0)
         var gpu_str = fromGpu(i).getString(0)
@@ -264,7 +264,7 @@ class StringOperatorsDiagnostics extends SparkQueryCompareTestSuite {
 
     // upper results
     val (fromCpuUpper, fromGpuUpper) = generateResults(upper)
-    for (i <- 0 until fromCpuUpper.length) {
+    for (i <- fromCpuUpper.indices) {
       if (fromCpuUpper(i) != fromGpuUpper(i) && fromGpuUpper(i).getString(0).length == 1) {
         val codepoint = TestCodepoints.validCodepointIndices(i)
 
@@ -276,7 +276,7 @@ class StringOperatorsDiagnostics extends SparkQueryCompareTestSuite {
 
     // lower results
     val (fromCpuLower, fromGpuLower) = generateResults(lower)
-    for (i <- 0 until fromCpuLower.length) {
+    for (i <- fromCpuLower.indices) {
       if (fromCpuLower(i) != fromGpuLower(i) && fromGpuLower(i).getString(0).length == 1) {
         val codepoint = TestCodepoints.validCodepointIndices(i)
 

diff --git a/...k300/src/main/scala/com/nvidia/spark/rapids/shims/spark300/GpuBroadcastHashJoinExec.scala b/...k300/src/main/scala/com/nvidia/spark/rapids/shims/spark300/GpuBroadcastHashJoinExec.scala
@@ -51,10 +51,10 @@ class GpuBroadcastHashJoinMeta(
 
   override def tagPlanForGpu(): Unit = {
     GpuHashJoin.tagJoin(this, join.joinType, join.leftKeys, join.rightKeys, join.condition)
-
+    val Seq(leftChild, rightChild) = childPlans
     val buildSide = join.buildSide match {
-      case BuildLeft => childPlans(0)
-      case BuildRight => childPlans(1)
+      case BuildLeft => leftChild
+      case BuildRight => rightChild
     }
 
     if (!buildSide.canThisBeReplaced) {
@@ -67,8 +67,7 @@ class GpuBroadcastHashJoinMeta(
   }
 
   override def convertToGpu(): GpuExec = {
-    val left = childPlans(0).convertIfNeeded()
-    val right = childPlans(1).convertIfNeeded()
+    val Seq(left, right) = childPlans.map(_.convertIfNeeded())
     // The broadcast part of this must be a BroadcastExchangeExec
     val buildSide = join.buildSide match {
       case BuildLeft => left

diff --git a/...rk300/src/main/scala/com/nvidia/spark/rapids/shims/spark300/GpuShuffledHashJoinExec.scala b/...rk300/src/main/scala/com/nvidia/spark/rapids/shims/spark300/GpuShuffledHashJoinExec.scala
@@ -56,16 +56,18 @@ class GpuShuffledHashJoinMeta(
     GpuHashJoin.tagJoin(this, join.joinType, join.leftKeys, join.rightKeys, join.condition)
   }
 
-  override def convertToGpu(): GpuExec =
+  override def convertToGpu(): GpuExec = {
+    val Seq(left, right) = childPlans.map(_.convertIfNeeded)
     GpuShuffledHashJoinExec(
       leftKeys.map(_.convertToGpu()),
       rightKeys.map(_.convertToGpu()),
       join.joinType,
       GpuJoinUtils.getGpuBuildSide(join.buildSide),
       condition.map(_.convertToGpu()),
-      childPlans(0).convertIfNeeded(),
-      childPlans(1).convertIfNeeded(),
+      left,
+      right,
       isSkewJoin = false)
+  }
 }
 
 case class GpuShuffledHashJoinExec(

diff --git a/...spark300/src/main/scala/com/nvidia/spark/rapids/shims/spark300/GpuSortMergeJoinExec.scala b/...spark300/src/main/scala/com/nvidia/spark/rapids/shims/spark300/GpuSortMergeJoinExec.scala
@@ -77,14 +77,15 @@ class GpuSortMergeJoinMeta(
     } else {
       throw new IllegalStateException(s"Cannot build either side for ${join.joinType} join")
     }
+    val Seq(left, right) = childPlans.map(_.convertIfNeeded())
     GpuShuffledHashJoinExec(
       leftKeys.map(_.convertToGpu()),
       rightKeys.map(_.convertToGpu()),
       join.joinType,
       GpuJoinUtils.getGpuBuildSide(buildSide),
       condition.map(_.convertToGpu()),
-      childPlans(0).convertIfNeeded(),
-      childPlans(1).convertIfNeeded(),
+      left,
+      right,
       join.isSkewJoin)
   }
 

diff --git a/...k301/src/main/scala/com/nvidia/spark/rapids/shims/spark301/GpuBroadcastHashJoinExec.scala b/...k301/src/main/scala/com/nvidia/spark/rapids/shims/spark301/GpuBroadcastHashJoinExec.scala
@@ -49,10 +49,10 @@ class GpuBroadcastHashJoinMeta(
 
   override def tagPlanForGpu(): Unit = {
     GpuHashJoin.tagJoin(this, join.joinType, join.leftKeys, join.rightKeys, join.condition)
-
+    val Seq(leftChild, rightChild) = childPlans
     val buildSide = join.buildSide match {
-      case BuildLeft => childPlans(0)
-      case BuildRight => childPlans(1)
+      case BuildLeft => leftChild
+      case BuildRight => rightChild
     }
 
     if (!canBuildSideBeReplaced(buildSide)) {
@@ -65,8 +65,7 @@ class GpuBroadcastHashJoinMeta(
   }
 
   override def convertToGpu(): GpuExec = {
-    val left = childPlans(0).convertIfNeeded()
-    val right = childPlans(1).convertIfNeeded()
+    val Seq(left, right) = childPlans.map(_.convertIfNeeded())
     // The broadcast part of this must be a BroadcastExchangeExec
     val buildSide = join.buildSide match {
       case BuildLeft => left

diff --git a/shims/spark301/src/main/scala/com/nvidia/spark/rapids/shims/spark301/Spark301Shims.scala b/shims/spark301/src/main/scala/com/nvidia/spark/rapids/shims/spark301/Spark301Shims.scala
@@ -68,15 +68,15 @@ class Spark301Shims extends Spark300Shims {
         Seq(ParamCheck("input", TypeSig.commonCudfTypes + TypeSig.NULL, TypeSig.all))),
       (a, conf, p, r) => new ExprMeta[First](a, conf, p, r) {
         override def convertToGpu(): GpuExpression =
-          GpuFirst(childExprs(0).convertToGpu(), a.ignoreNulls)
+          GpuFirst(childExprs.head.convertToGpu(), a.ignoreNulls)
       }),
     GpuOverrides.expr[Last](
       "last aggregate operator",
       ExprChecks.aggNotWindow(TypeSig.commonCudfTypes + TypeSig.NULL, TypeSig.all,
         Seq(ParamCheck("input", TypeSig.commonCudfTypes + TypeSig.NULL, TypeSig.all))),
       (a, conf, p, r) => new ExprMeta[Last](a, conf, p, r) {
         override def convertToGpu(): GpuExpression =
-          GpuLast(childExprs(0).convertToGpu(), a.ignoreNulls)
+          GpuLast(childExprs.head.convertToGpu(), a.ignoreNulls)
       })
   ).map(r => (r.getClassFor.asSubclass(classOf[Expression]), r)).toMap
 

diff --git a/...db/src/main/scala/com/nvidia/spark/rapids/shims/spark301db/GpuBroadcastHashJoinExec.scala b/...db/src/main/scala/com/nvidia/spark/rapids/shims/spark301db/GpuBroadcastHashJoinExec.scala
@@ -48,10 +48,10 @@ class GpuBroadcastHashJoinMeta(
 
   override def tagPlanForGpu(): Unit = {
     GpuHashJoin.tagJoin(this, join.joinType, join.leftKeys, join.rightKeys, join.condition)
-
+    val Seq(leftChild, rightChild) = childPlans
     val buildSide = join.buildSide match {
-      case BuildLeft => childPlans(0)
-      case BuildRight => childPlans(1)
+      case BuildLeft => leftChild
+      case BuildRight => rightChild
     }
 
     if (!canBuildSideBeReplaced(buildSide)) {
@@ -64,8 +64,7 @@ class GpuBroadcastHashJoinMeta(
   }
 
   override def convertToGpu(): GpuExec = {
-    val left = childPlans(0).convertIfNeeded()
-    val right = childPlans(1).convertIfNeeded()
+    val Seq(left, right) = childPlans.map(_.convertIfNeeded())
     // The broadcast part of this must be a BroadcastExchangeExec
     val buildSide = join.buildSide match {
       case BuildLeft => left

diff --git a/...1db/src/main/scala/com/nvidia/spark/rapids/shims/spark301db/GpuShuffledHashJoinExec.scala b/...1db/src/main/scala/com/nvidia/spark/rapids/shims/spark301db/GpuShuffledHashJoinExec.scala
@@ -57,15 +57,17 @@ class GpuShuffledHashJoinMeta(
     GpuHashJoin.tagJoin(this, join.joinType, join.leftKeys, join.rightKeys, join.condition)
   }
 
-  override def convertToGpu(): GpuExec =
+  override def convertToGpu(): GpuExec = {
+    val Seq(leftChild, rightChild) = childPlans.map(_.convertIfNeeded())
     GpuShuffledHashJoinExec(
       leftKeys.map(_.convertToGpu()),
       rightKeys.map(_.convertToGpu()),
       join.joinType,
       GpuJoinUtils.getGpuBuildSide(join.buildSide),
       condition.map(_.convertToGpu()),
-      childPlans(0).convertIfNeeded(),
-      childPlans(1).convertIfNeeded())
+      leftChild,
+      rightChild)
+  }
 }
 
 case class GpuShuffledHashJoinExec(

diff --git a/...k301db/src/main/scala/com/nvidia/spark/rapids/shims/spark301db/GpuSortMergeJoinExec.scala b/...k301db/src/main/scala/com/nvidia/spark/rapids/shims/spark301db/GpuSortMergeJoinExec.scala
@@ -78,14 +78,15 @@ class GpuSortMergeJoinMeta(
     } else {
       throw new IllegalStateException(s"Cannot build either side for ${join.joinType} join")
     }
+    val Seq(leftChild, rightChild) = childPlans.map(_.convertIfNeeded())
     GpuShuffledHashJoinExec(
       leftKeys.map(_.convertToGpu()),
       rightKeys.map(_.convertToGpu()),
       join.joinType,
       GpuJoinUtils.getGpuBuildSide(buildSide),
       condition.map(_.convertToGpu()),
-      childPlans(0).convertIfNeeded(),
-      childPlans(1).convertIfNeeded())
+      leftChild,
+      rightChild)
   }
 
   /**

diff --git a/...k311/src/main/scala/com/nvidia/spark/rapids/shims/spark311/GpuBroadcastHashJoinExec.scala b/...k311/src/main/scala/com/nvidia/spark/rapids/shims/spark311/GpuBroadcastHashJoinExec.scala
@@ -53,10 +53,10 @@ class GpuBroadcastHashJoinMeta(
 
   override def tagPlanForGpu(): Unit = {
     GpuHashJoin.tagJoin(this, join.joinType, join.leftKeys, join.rightKeys, join.condition)
-
+    val Seq(leftChild, rightChild) = childPlans
     val buildSide = join.buildSide match {
-      case BuildLeft => childPlans(0)
-      case BuildRight => childPlans(1)
+      case BuildLeft => leftChild
+      case BuildRight => rightChild
     }
 
     if (!canBuildSideBeReplaced(buildSide)) {
@@ -69,8 +69,7 @@ class GpuBroadcastHashJoinMeta(
   }
 
   override def convertToGpu(): GpuExec = {
-    val left = childPlans(0).convertIfNeeded()
-    val right = childPlans(1).convertIfNeeded()
+    val Seq(left, right) = childPlans.map(_.convertIfNeeded())
     // The broadcast part of this must be a BroadcastExchangeExec
     val buildSide = join.buildSide match {
       case BuildLeft => left

diff --git a/...rk311/src/main/scala/com/nvidia/spark/rapids/shims/spark311/GpuShuffledHashJoinExec.scala b/...rk311/src/main/scala/com/nvidia/spark/rapids/shims/spark311/GpuShuffledHashJoinExec.scala
@@ -57,16 +57,18 @@ class GpuShuffledHashJoinMeta(
     GpuHashJoin.tagJoin(this, join.joinType, join.leftKeys, join.rightKeys, join.condition)
   }
 
-  override def convertToGpu(): GpuExec =
+  override def convertToGpu(): GpuExec = {
+    val Seq(left, right) = childPlans.map(_.convertIfNeeded())
     GpuShuffledHashJoinExec(
       leftKeys.map(_.convertToGpu()),
       rightKeys.map(_.convertToGpu()),
       join.joinType,
       GpuJoinUtils.getGpuBuildSide(join.buildSide),
       condition.map(_.convertToGpu()),
-      childPlans(0).convertIfNeeded(),
-      childPlans(1).convertIfNeeded(),
+      left,
+      right,
       isSkewJoin = false)
+  }
 }
 
 case class GpuShuffledHashJoinExec(

diff --git a/...spark311/src/main/scala/com/nvidia/spark/rapids/shims/spark311/GpuSortMergeJoinExec.scala b/...spark311/src/main/scala/com/nvidia/spark/rapids/shims/spark311/GpuSortMergeJoinExec.scala
@@ -78,14 +78,15 @@ class GpuSortMergeJoinMeta(
     } else {
       throw new IllegalStateException(s"Cannot build either side for ${join.joinType} join")
     }
+    val Seq(left, right) = childPlans.map(_.convertIfNeeded())
     GpuShuffledHashJoinExec(
       leftKeys.map(_.convertToGpu()),
       rightKeys.map(_.convertToGpu()),
       join.joinType,
       GpuJoinUtils.getGpuBuildSide(buildSide),
       condition.map(_.convertToGpu()),
-      childPlans(0).convertIfNeeded(),
-      childPlans(1).convertIfNeeded(),
+      left,
+      right,
       join.isSkewJoin)
   }
 

diff --git a/shims/spark311/src/main/scala/com/nvidia/spark/rapids/shims/spark311/Spark311Shims.scala b/shims/spark311/src/main/scala/com/nvidia/spark/rapids/shims/spark311/Spark311Shims.scala
@@ -191,10 +191,8 @@ class Spark311Shims extends Spark301Shims {
           }
         }
         override def convertToGpu(): GpuExpression = {
-          GpuStringReplace(
-            childExprs(0).convertToGpu(),
-            childExprs(1).convertToGpu(),
-            childExprs(2).convertToGpu())
+          val Seq(child0, child1, child2) = childExprs.map(_.convertToGpu())
+          GpuStringReplace(child0, child1, child2)
         }
       }),
     // Spark 3.1.1-specific LEAD expression, using custom OffsetWindowFunctionMeta.

diff --git a/sql-plugin/src/main/scala/com/nvidia/spark/rapids/DateUtils.scala b/sql-plugin/src/main/scala/com/nvidia/spark/rapids/DateUtils.scala
@@ -148,7 +148,7 @@ object DateUtils {
     var sb = new StringBuilder()
     var index = 0;
     val patterns = new ListBuffer[FormatKeywordToReplace]
-    format.map(character => {
+    format.foreach(character => {
       // We are checking to see if this char is a part of a previously read pattern
       // or start of a new one.
       if (sb.isEmpty || sb.last == character) {

diff --git a/sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuExpressions.scala b/sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuExpressions.scala
@@ -272,46 +272,47 @@ trait GpuTernaryExpression extends TernaryExpression with GpuExpression {
   def doColumnar(numRows: Int, val0: Scalar, val1: Scalar, val2: Scalar): ColumnVector
 
   override def columnarEval(batch: ColumnarBatch): Any = {
-    withResourceIfAllowed(children(0).columnarEval(batch)) { val0 =>
-      withResourceIfAllowed(children(1).columnarEval(batch)) { val1 =>
-        withResourceIfAllowed(children(2).columnarEval(batch)) { val2 =>
+    val Seq(child0, child1, child2) = children
+    withResourceIfAllowed(child0.columnarEval(batch)) { val0 =>
+      withResourceIfAllowed(child1.columnarEval(batch)) { val1 =>
+        withResourceIfAllowed(child2.columnarEval(batch)) { val2 =>
           (val0, val1, val2) match {
             case (v0: GpuColumnVector, v1: GpuColumnVector, v2: GpuColumnVector) =>
               doColumnar(v0, v1, v2)
             case (v0, v1: GpuColumnVector, v2: GpuColumnVector) =>
-              withResource(GpuScalar.from(v0, children(0).dataType)) { scalar0 =>
+              withResource(GpuScalar.from(v0, child0.dataType)) { scalar0 =>
                 GpuColumnVector.from(doColumnar(scalar0, v1, v2), dataType)
               }
             case (v0: GpuColumnVector, v1, v2: GpuColumnVector) =>
-              withResource(GpuScalar.from(v1, children(1).dataType)) { scalar1 =>
+              withResource(GpuScalar.from(v1, child1.dataType)) { scalar1 =>
                 GpuColumnVector.from(doColumnar(v0, scalar1, v2), dataType)
               }
             case (v0: GpuColumnVector, v1: GpuColumnVector, v2) =>
-              withResource(GpuScalar.from(v2, children(2).dataType)) { scalar2 =>
+              withResource(GpuScalar.from(v2, child2.dataType)) { scalar2 =>
                 GpuColumnVector.from(doColumnar(v0, v1, scalar2), dataType)
               }
             case (v0, v1, v2: GpuColumnVector) =>
-              withResource(GpuScalar.from(v0, children(0).dataType)) { scalar0 =>
-                withResource(GpuScalar.from(v1, children(1).dataType)) { scalar1 =>
+              withResource(GpuScalar.from(v0, child0.dataType)) { scalar0 =>
+                withResource(GpuScalar.from(v1, child1.dataType)) { scalar1 =>
                   GpuColumnVector.from(doColumnar(scalar0, scalar1, v2), dataType)
                 }
               }
             case (v0, v1: GpuColumnVector, v2) =>
-              withResource(GpuScalar.from(v0, children(0).dataType)) { scalar0 =>
-                withResource(GpuScalar.from(v2, children(2).dataType)) { scalar2 =>
+              withResource(GpuScalar.from(v0, child0.dataType)) { scalar0 =>
+                withResource(GpuScalar.from(v2, child2.dataType)) { scalar2 =>
                   GpuColumnVector.from(doColumnar(scalar0, v1, scalar2), dataType)
                 }
               }
             case (v0: GpuColumnVector, v1, v2) =>
-              withResource(GpuScalar.from(v1, children(1).dataType)) { scalar1 =>
-                withResource(GpuScalar.from(v2, children(2).dataType)) { scalar2 =>
+              withResource(GpuScalar.from(v1, child1.dataType)) { scalar1 =>
+                withResource(GpuScalar.from(v2, child2.dataType)) { scalar2 =>
                   GpuColumnVector.from(doColumnar(v0, scalar1, scalar2), dataType)
                 }
               }
             case (v0, v1, v2) if v0 != null && v1 != null && v2 != null =>
-              withResource(GpuScalar.from(v0, children(0).dataType)) { v0Scalar =>
-                withResource(GpuScalar.from(v1, children(1).dataType)) { v1Scalar =>
-                  withResource(GpuScalar.from(v2, children(2).dataType)) { v2Scalar =>
+              withResource(GpuScalar.from(v0, child0.dataType)) { v0Scalar =>
+                withResource(GpuScalar.from(v1, child1.dataType)) { v1Scalar =>
+                  withResource(GpuScalar.from(v2, child2.dataType)) { v2Scalar =>
                     GpuColumnVector.from(doColumnar(batch.numRows(), v0Scalar, v1Scalar, v2Scalar),
                       dataType)
                   }