Update SortMergeJoinExec.scala

[BoleynSu]

fix a bug in outputOrdering

What changes were proposed in this pull request?

Change case Inner to case _: InnerLike so that Cross will be handled properly.

How was this patch tested?

No unit tests are needed.

Please review http://spark.apache.org/contributing.html before opening a pull request.

[srowen]

You didn't read the link above, I take it?
http://spark.apache.org/contributing.html

[AmplabJenkins]

Can one of the admins verify this patch?

[gatorsmile]

Thanks for fixing this. Please follow the contribution guideline.

Also, you need to add a test case. You can follow what we did in this PR: #17339

[BoleynSu]

A test case to make the existing code fail.
@srowen I am sorry that this pull request is not well formatted but I just want to help.

import org.apache.spark.sql.SparkSession

object Test extends App {
  val spark = SparkSession.builder().master("local").appName("test").getOrCreate()
  import spark.sqlContext.implicits._
  case class T(i: Int)
  spark.sparkContext.parallelize(List(T(1), T(3), T(3))).toDF.createOrReplaceTempView("T")
  val in = "select distinct a.i + 1,a.* from T a cross join T t where a.i > 1 and t.i = a.i group by a.i having a.i > 2"
  val sql = spark.sql(in)
  sql.queryExecution.executedPlan.children.map { x =>
    x.children.map { x =>
      x.children.map { x =>
        x.children.map { x =>
          x.children.map { x =>
            x.children.map { x =>
              println(x.outputOrdering)
            }
          }
        }
      }
    }
  }
}

[gatorsmile]

@BoleynSu Do you want to continue the PR? or you want us to take it over?

[BoleynSu]

@gatorsmile I am not familiar with the PR process, it is great that you can take it over. Thanks.

[gatorsmile]

@BoleynSu Sure, I can do it. Will give all the credits to you. Please continue to help us report new issues or fixes. Thanks!

[hvanhovell]

Can someone explain to me what is being fixed here? The other InnerLike variant, Cross, does not get planned using a SortMergeJoin.

[BoleynSu]

I think we can get a SortMergeJoin paln with Cross, e.g. select distinct a.i + 1,a.* from T a cross join T t where a.i > 1 and t.i = a.i group by a.i having a.i > 2.

[gatorsmile]

Even worse, this could cause an exception

val df = Seq((1, 1)).toDF("i", "j")
df.createOrReplaceTempView("T")
withSQLConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1") {
  sql("select * from (select a.i from T a cross join T t where t.i = a.i) as t1 " +
    "cross join T t2 where t2.i = t1.i").explain(true)
}

It will return the following error:

SortMergeJoinExec should not take Cross as the JoinType
java.lang.IllegalArgumentException: SortMergeJoinExec should not take Cross as the JoinType
	at org.apache.spark.sql.execution.joins.SortMergeJoinExec.outputOrdering(SortMergeJoinExec.scala:100)
	at org.apache.spark.sql.execution.ProjectExec

We need to backport it to 2.2

[hvanhovell]

Yeah that makes sense