support explode outer

integration_tests/src/main/python/generate_expr_test.py

@@ -69,6 +69,27 @@ def test_explode_nested_array_data(spark_tmp_path, data_gen):
             'a', 'explode(b) as c').selectExpr('a', 'explode(c)'),
         conf=conf_to_enforce_split_input)
 
+#sort locally because of https://github.com/NVIDIA/spark-rapids/issues/84
+# After 3.1.0 is the min spark version we can drop this
+@ignore_order(local=True)
+@pytest.mark.parametrize('data_gen', all_gen, ids=idfn)
+def test_explode_outer_array_data(spark_tmp_path, data_gen):
+    data_gen = [int_gen, ArrayGen(data_gen)]
+    assert_gpu_and_cpu_are_equal_collect(
+        lambda spark: two_col_df(spark, *data_gen).selectExpr('a', 'explode_outer(b)'),
+        conf=conf_to_enforce_split_input)
+
+#sort locally because of https://github.com/NVIDIA/spark-rapids/issues/84
+# After 3.1.0 is the min spark version we can drop this
+@ignore_order(local=True)
+@pytest.mark.parametrize('data_gen', all_gen, ids=idfn)
+def test_explode_outer_nested_array_data(spark_tmp_path, data_gen):
+    data_gen = [int_gen, ArrayGen(ArrayGen(data_gen))]
+    assert_gpu_and_cpu_are_equal_collect(
+        lambda spark: two_col_df(spark, *data_gen).selectExpr(
+            'a', 'explode_outer(b) as c').selectExpr('a', 'explode_outer(c)'),
+        conf=conf_to_enforce_split_input)
+
 
 #sort locally because of https://github.com/NVIDIA/spark-rapids/issues/84
 # After 3.1.0 is the min spark version we can drop this
@@ -108,3 +129,24 @@ def test_posexplode_nested_array_data(spark_tmp_path, data_gen):
         lambda spark: two_col_df(spark, *data_gen).selectExpr(
             'a', 'posexplode(b) as (pos, c)').selectExpr('a', 'pos', 'posexplode(c)'),
         conf=conf_to_enforce_split_input)
+
+#sort locally because of https://github.com/NVIDIA/spark-rapids/issues/84
+# After 3.1.0 is the min spark version we can drop this
+@ignore_order(local=True)
+@pytest.mark.parametrize('data_gen', all_gen, ids=idfn)
+def test_posexplode_outer_array_data(spark_tmp_path, data_gen):
+    data_gen = [int_gen, ArrayGen(data_gen)]
+    assert_gpu_and_cpu_are_equal_collect(
+        lambda spark: two_col_df(spark, *data_gen).selectExpr('a', 'posexplode_outer(b)'),
+        conf=conf_to_enforce_split_input)
+
+#sort locally because of https://github.com/NVIDIA/spark-rapids/issues/84
+# After 3.1.0 is the min spark version we can drop this
+@ignore_order(local=True)
+@pytest.mark.parametrize('data_gen', all_gen, ids=idfn)
+def test_posexplode_nested_outer_array_data(spark_tmp_path, data_gen):
+    data_gen = [int_gen, ArrayGen(ArrayGen(data_gen))]
+    assert_gpu_and_cpu_are_equal_collect(
+        lambda spark: two_col_df(spark, *data_gen).selectExpr(
+            'a', 'posexplode_outer(b) as (pos, c)').selectExpr('a', 'pos', 'posexplode_outer(c)'),
+        conf=conf_to_enforce_split_input)

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

@@ -342,8 +342,10 @@ case class GpuExplode(child: Expression) extends GpuExplodeBase {
     require(inputBatch.numCols() - 1 == generatorOffset,
       "Internal Error GpuExplode supports one and only one input attribute.")
     val schema = resultSchema(GpuColumnVector.extractTypes(inputBatch), generatorOffset)
+    val explodeFun = (t: Table) =>
+      if (outer) t.explodeOuter(generatorOffset) else t.explode(generatorOffset)
     withResource(GpuColumnVector.from(inputBatch)) { table =>
-      withResource(table.explode(generatorOffset)) { exploded =>
+      withResource(explodeFun(table)) { exploded =>
         GpuColumnVector.from(exploded, schema)
       }
     }
@@ -362,8 +364,10 @@ case class GpuPosExplode(child: Expression) extends GpuExplodeBase {
       "Internal Error GpuPosExplode supports one and only one input attribute.")
     val schema = resultSchema(
       GpuColumnVector.extractTypes(inputBatch), generatorOffset, includePos = true)
+    val explodePosFun = (t: Table) =>
+      if (outer) t.explodeOuterPosition(generatorOffset) else t.explodePosition(generatorOffset)
     withResource(GpuColumnVector.from(inputBatch)) { table =>
-      withResource(table.explodePosition(generatorOffset)) { exploded =>
+      withResource(explodePosFun(table)) { exploded =>
         GpuColumnVector.from(exploded, schema)
       }
     }

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

@@ -2393,8 +2393,7 @@ object GpuOverrides {
           GpuMakeDecimal(child, a.precision, a.scale, a.nullOnOverflow)
       }),
     expr[Explode](
-      "Given an input array produces a sequence of rows for each value in the array. "
-        + "Explode with outer Generate is not supported under GPU runtime." ,
+      "Given an input array produces a sequence of rows for each value in the array.",
       ExprChecks.unaryProject(
         // Here is a walk-around representation, since multi-level nested type is not supported yet.
         // related issue: https://github.com/NVIDIA/spark-rapids/issues/1901
@@ -2405,11 +2404,11 @@ object GpuOverrides {
           TypeSig.commonCudfTypes + TypeSig.DECIMAL + TypeSig.NULL + TypeSig.ARRAY),
         (TypeSig.ARRAY + TypeSig.MAP).nested(TypeSig.all)),
       (a, conf, p, r) => new GeneratorExprMeta[Explode](a, conf, p, r) {
+        override val supportOuter: Boolean = true
         override def convertToGpu(): GpuExpression = GpuExplode(childExprs(0).convertToGpu())
       }),
     expr[PosExplode](
-      "Given an input array produces a sequence of rows for each value in the array. "
-        + "PosExplode with outer Generate is not supported under GPU runtime." ,
+      "Given an input array produces a sequence of rows for each value in the array.",
       ExprChecks.unaryProject(
         // Here is a walk-around representation, since multi-level nested type is not supported yet.
         // related issue: https://github.com/NVIDIA/spark-rapids/issues/1901
@@ -2421,6 +2420,7 @@ object GpuOverrides {
         TypeSig.ARRAY.nested(
           TypeSig.commonCudfTypes + TypeSig.DECIMAL + TypeSig.NULL + TypeSig.ARRAY)),
       (a, conf, p, r) => new GeneratorExprMeta[PosExplode](a, conf, p, r) {
+        override val supportOuter: Boolean = true
         override def convertToGpu(): GpuExpression = GpuPosExplode(childExprs(0).convertToGpu())
       }),
     expr[CollectList](