[SPARK-26352][SQL] join reorder should not change the order of output attributes

[rednaxelafx]

What changes were proposed in this pull request?

The optimizer rule org.apache.spark.sql.catalyst.optimizer.ReorderJoin performs join reordering on inner joins. This was introduced from SPARK-12032 (#10073) in 2015-12.

After it had reordered the joins, though, it didn't check whether or not the output attribute order is still the same as before. Thus, it's possible to have a mismatch between the reordered output attributes order vs the schema that a DataFrame thinks it has.
The same problem exists in the CBO version of join reordering (CostBasedJoinReorder) too.

This can be demonstrated with the example:

spark.sql("create table table_a (x int, y int) using parquet")
spark.sql("create table table_b (i int, j int) using parquet")
spark.sql("create table table_c (a int, b int) using parquet")
val df = spark.sql("""
  with df1 as (select * from table_a cross join table_b)
  select * from df1 join table_c on a = x and b = i
""")

here's what the DataFrame thinks:

scala> df.printSchema
root
 |-- x: integer (nullable = true)
 |-- y: integer (nullable = true)
 |-- i: integer (nullable = true)
 |-- j: integer (nullable = true)
 |-- a: integer (nullable = true)
 |-- b: integer (nullable = true)

here's what the optimized plan thinks, after join reordering:

scala> df.queryExecution.optimizedPlan.output.foreach(a => println(s"|-- ${a.name}: ${a.dataType.typeName}"))
|-- x: integer
|-- y: integer
|-- a: integer
|-- b: integer
|-- i: integer
|-- j: integer

If we exclude the ReorderJoin rule (using Spark 2.4's optimizer rule exclusion feature), it's back to normal:

scala> spark.conf.set("spark.sql.optimizer.excludedRules", "org.apache.spark.sql.catalyst.optimizer.ReorderJoin")

scala> val df = spark.sql("with df1 as (select * from table_a cross join table_b) select * from df1 join table_c on a = x and b = i")
df: org.apache.spark.sql.DataFrame = [x: int, y: int ... 4 more fields]

scala> df.queryExecution.optimizedPlan.output.foreach(a => println(s"|-- ${a.name}: ${a.dataType.typeName}"))
|-- x: integer
|-- y: integer
|-- i: integer
|-- j: integer
|-- a: integer
|-- b: integer

Note that this output attribute ordering problem leads to data corruption, and can manifest itself in various symptoms:

• Silently corrupting data, if the reordered columns happen to either have matching types or have sufficiently-compatible types (e.g. all fixed length primitive types are considered as "sufficiently compatible" in an UnsafeRow), then only the resulting data is going to be wrong but it might not trigger any alarms immediately. Or
• Weird Java-level exceptions like java.lang.NegativeArraySizeException, or even SIGSEGVs.

How was this patch tested?

Added new unit test in JoinReorderSuite and new end-to-end test in JoinSuite.
Also made JoinReorderSuite and StarJoinReorderSuite assert more strongly on maintaining output attribute order.

[gatorsmile]

scala> spark.sql("with tmp as (select * from tab1 cross join tab2) select * from tmp join tab3 on a = x and b = i").show()
+---+---+---+---+---+---+
|  x|  y|  i|  j|  a|  b|
+---+---+---+---+---+---+
|  1|100| 42|200|  1| 42|
+---+---+---+---+---+---+

This still passes without code changes, right?

[gatorsmile]

Maybe the command lines work but the test case does not work since we turn off ConvertToLocalRelation in the test environment?

[rednaxelafx]

Yeah, before sending this PR I've actually explicitly run both test cases that I added with and without the fix, and verified that without the fix it'd fail the test.

The reason why you "think" it works in your environment is because you used show, which after @maropu 's earlier PR will introduce a Project at the top, which happens to cure the issue.
If you just do a collect you'll see the problem still exists:

scala> df.show
+---+---+---+---+---+---+
|  x|  y|  i|  j|  a|  b|
+---+---+---+---+---+---+
|  1|100| 42|200|  1| 42|
+---+---+---+---+---+---+

scala> df.collect
res5: Array[org.apache.spark.sql.Row] = Array([1,100,1,42,42,200])

[cloud-fan]

just for curiosity, we only need this for top-level join? I feel it's ok to change the columnar order for intermedia joins.

[rednaxelafx]

That's right, only the top-level really needs to maintain the appearance. But this is the easiest to implement (the change is local to the rule where order could have changed, so this projection is easier to understand than adding it elsewhere), and it doesn't affect the final result because other optimizer rules are actually going to get rid of the extra intermediate projections.

e.g. if on top of the df, we do an extra operation:

df.groupBy('a, 'b).agg(first('i), first('j), first('x), first('y))

you're going to see that the extra Project gets optimized away in:

=== Result of Batch Operator Optimization before Inferring Filters ===
 Aggregate [a#65, b#66], [a#65, b#66, first(i#63, false) AS first(i, false)#121, first(j#64, false) AS first(j, false)#122, first(x#61, false) AS first(x, false)#123, first(y#62, false) AS first(y, false)#124]   Aggregate [a#65, b#66], [a#65, b#66, first(i#63, false) AS first(i, false)#121, first(j#64, false) AS first(j, false)#122, first(x#61, false) AS first(x, false)#123, first(y#62, false) AS first(y, false)#124]
!+- Project [x#61, y#62, i#63, j#64, a#65, b#66]                                                                                                                                                                    +- Join Cross, (b#66 = i#63)
!   +- Join Inner, ((a#65 = x#61) && (b#66 = i#63))                                                                                                                                                                    :- Join Inner, (a#65 = x#61)
!      :- Project [x#61, y#62, i#63, j#64]                                                                                                                                                                             :  :- Relation[x#61,y#62] parquet
!      :  +- Join Cross                                                                                                                                                                                                :  +- Relation[a#65,b#66] parquet
!      :     :- Relation[x#61,y#62] parquet                                                                                                                                                                            +- Relation[i#63,j#64] parquet
!      :     +- Relation[i#63,j#64] parquet                                                                                                                                                                         
!      +- Relation[a#65,b#66] parquet

(a few other rules may also remove the extra Project)

[cloud-fan]

can we reproduce the bug with temp view?

[rednaxelafx]

That's more prone to be subject to ConvertToLocalRelation, i.e. you can't copy-and-paste the test case into a Spark shell and expect the same behavior. Would you still prefer to use temp view instead?

[cloud-fan]

let's keep it as it is

[cloud-fan]

good catch! LGTM

[SparkQA]

Test build #100057 has finished for PR 23303 at commit f0de32f.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[rednaxelafx]

@cloud-fan @gatorsmile @hvanhovell I've updated the PR with a fix to the same problem in CBO join reorder, and also fixed some test cases that have been wrong for a long time (the expected plans were in fact wrong).

PTAL, thanks!

[cloud-fan]

is this an improvement? Maybe we should do it in another PR, so that this PR can be backported to old branches.

[rednaxelafx]

This isn't really an improvement. It's here mainly to help pass existing tests -- I'm adding new projections to fix the output attribute order problem, but these extra projections in the middle would make (expected) query plans look pretty ugly.

So consider this is two things that can cancel each other out:

1. Add projections to fix output attribute order;
2. If one of these extra projections is in the middle, get rid of it.

The cleanup (2) is only meant to clean extra projections created because of (1). Both (1) and (2) are in this PR so I don't consider this as a performance improvement over existing stuff.

[SparkQA]

Test build #100085 has finished for PR 23303 at commit 0f81933.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[gatorsmile]

@rednaxelafx Can we separate this to two PRs? One is for the non-CBO reordering. Another is for the CBO reordering?

[rednaxelafx]

@gatorsmile that was my original intent. I only wanted to fix the non-CBO one in this PR originally, but after reading through the join reorder related test suite I found that most of them have both the non-CBO and the CBO one in the same Optimize. The test cases are going to fail one way or the other if we only fix one of them. Sadly we'll have to fix both in the same PR.

The feature fix itself is really small, but granted it's in a very dense piece of code. The majority of this PR is actually about fixing the test cases....a lot of them were wrong to begin with (had incorrect output attribute order in the expected plans)

[SparkQA]

Test build #100110 has started for PR 23303 at commit ae15739.

[rednaxelafx]

jenkins retest this please

[SparkQA]

Test build #100117 has finished for PR 23303 at commit ae15739.

• This patch fails PySpark unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[rednaxelafx]

jenkins retest this please

[SparkQA]

Test build #100140 has finished for PR 23303 at commit ae15739.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[rednaxelafx]

As per offline discussion with @cloud-fan and @hvanhovell , I've updated the PR and moved the logic that eliminates redundant projections from the CBO-based join reorder to RemoveRedundantProject.

Could you please take another look? Thanks!
cc @gatorsmile

[SparkQA]

Test build #100162 has finished for PR 23303 at commit 7ef2fa3.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[rednaxelafx]

Oops, the relaxed transformation in RemoveRedundantProject is too wide and it's wrong now. Finding alternatives now.

[SparkQA]

Test build #100173 has finished for PR 23303 at commit 385cbd3.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #100176 has finished for PR 23303 at commit a7fb7a6.

• This patch fails due to an unknown error code, -9.
• This patch merges cleanly.
• This patch adds no public classes.

[rednaxelafx]

jenkind retest this please

[SparkQA]

Test build #100183 has finished for PR 23303 at commit a7fb7a6.

• This patch fails PySpark unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[cloud-fan]

This is too risky, are there other ways to work around it? Or can we accept sub-optimal plans?

[SparkQA]

Test build #100211 has finished for PR 23303 at commit 9451b68.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[cloud-fan]

thanks, merging to master/2.4/2.3!