This is an automated email from the ASF dual-hosted git repository. gurwls223 pushed a commit to branch branch-3.5 in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-3.5 by this push: new 4f869ce6f67 [SPARK-44132][SQL] Materialize `Stream` of join column names to avoid codegen failure 4f869ce6f67 is described below commit 4f869ce6f67c4bb2444c351632adaf9c78c5ec1b Author: Steven Aerts <steven.ae...@gmail.com> AuthorDate: Tue Aug 8 08:09:05 2023 +0900 [SPARK-44132][SQL] Materialize `Stream` of join column names to avoid codegen failure ### What changes were proposed in this pull request? Materialize passed join columns as an `IndexedSeq` before passing it to the lower layers. ### Why are the changes needed? When nesting multiple full outer joins using column names which are a `Stream`, the code generator will generate faulty code resulting in a NPE or bad `UnsafeRow` access at runtime. See the 2 added test cases. ### Why are the changes needed? Otherwise the code will crash, see the 2 added test cases. Which show an NPE and a bad `UnsafeRow` access. ### Does this PR introduce _any_ user-facing change? No, only bug fix. ### How was this patch tested? A reproduction scenario was created and added to the code base. Closes #41712 from steven-aerts/SPARK-44132-fix. Authored-by: Steven Aerts <steven.ae...@gmail.com> Signed-off-by: Hyukjin Kwon <gurwls...@apache.org> (cherry picked from commit 8911578020f8a2428b12dd72cb0ed4b7d747d835) Signed-off-by: Hyukjin Kwon <gurwls...@apache.org> --- .../main/scala/org/apache/spark/sql/Dataset.scala | 2 +- .../sql/execution/joins/JoinCodegenSupport.scala | 2 +- .../test/scala/org/apache/spark/sql/JoinSuite.scala | 20 ++++++++++++++++++++ 3 files changed, 22 insertions(+), 2 deletions(-) diff --git a/sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala b/sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala index 61c83829d20..eda017937d9 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala @@ -1092,7 +1092,7 @@ class Dataset[T] private[sql]( Join( joined.left, joined.right, - UsingJoin(JoinType(joinType), usingColumns), + UsingJoin(JoinType(joinType), usingColumns.toIndexedSeq), None, JoinHint.NONE) } diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/JoinCodegenSupport.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/JoinCodegenSupport.scala index a7d1edefcd6..6496f9a0006 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/JoinCodegenSupport.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/JoinCodegenSupport.scala @@ -79,7 +79,7 @@ trait JoinCodegenSupport extends CodegenSupport with BaseJoinExec { setDefaultValue: Boolean): Seq[ExprCode] = { ctx.currentVars = null ctx.INPUT_ROW = row - plan.output.zipWithIndex.map { case (a, i) => + plan.output.toIndexedSeq.zipWithIndex.map { case (a, i) => val ev = BoundReference(i, a.dataType, a.nullable).genCode(ctx) if (setDefaultValue) { // the variables are needed even there is no matched rows diff --git a/sql/core/src/test/scala/org/apache/spark/sql/JoinSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/JoinSuite.scala index 7f358723eeb..14f1fb27906 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/JoinSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/JoinSuite.scala @@ -1709,4 +1709,24 @@ class JoinSuite extends QueryTest with SharedSparkSession with AdaptiveSparkPlan checkAnswer(sql(query), expected) } } + + test("SPARK-44132: FULL OUTER JOIN by streamed column name fails with NPE") { + val dsA = Seq((1, "a")).toDF("id", "c1") + val dsB = Seq((2, "b")).toDF("id", "c2") + val dsC = Seq((3, "c")).toDF("id", "c3") + val joined = dsA.join(dsB, Stream("id"), "full_outer").join(dsC, Stream("id"), "full_outer") + + val expected = Seq(Row(1, "a", null, null), Row(2, null, "b", null), Row(3, null, null, "c")) + + checkAnswer(joined, expected) + } + + test("SPARK-44132: FULL OUTER JOIN by streamed column name fails with invalid access") { + val ds = Seq((1, "a")).toDF("id", "c1") + val joined = ds.join(ds, Stream("id"), "full_outer").join(ds, Stream("id"), "full_outer") + + val expected = Seq(Row(1, "a", "a", "a")) + + checkAnswer(joined, expected) + } } --------------------------------------------------------------------- To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org For additional commands, e-mail: commits-h...@spark.apache.org