Skip to content

[SPARK-25121][SQL] Supports multi-part relation names for join strategy hint resolution #27935

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
maropu wants to merge 9 commits into from
Closed

[SPARK-25121][SQL] Supports multi-part relation names for join strategy hint resolution #27935

Show file tree
Hide file tree
Changes from 8 commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
52089fc
Fix
maropu Mar 17, 2020
ac19ea1
Fix
maropu Mar 18, 2020
70c994a
Fix
maropu Mar 18, 2020
fbbe824
Fix
maropu Mar 19, 2020
2444a20
Fix
maropu Mar 19, 2020
f07613d
Fix
maropu Mar 19, 2020
6528dad
Fix
maropu Mar 19, 2020
c517c1f
Fix
maropu Mar 19, 2020
61e4a95
Address Dongjoon reviews
maropu Mar 19, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 4 additions & 3 deletions sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/HintErrorLogger.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,15 +24,16 @@ import org.apache.spark.sql.catalyst.plans.logical.{HintErrorHandler, HintInfo}
* The hint error handler that logs warnings for each hint error.
*/
object HintErrorLogger extends HintErrorHandler with Logging {
import org.apache.spark.sql.connector.catalog.CatalogV2Implicits._

override def hintNotRecognized(name: String, parameters: Seq[Any]): Unit = {
logWarning(s"Unrecognized hint: ${hintToPrettyString(name, parameters)}")
}

override def hintRelationsNotFound(
name: String, parameters: Seq[Any], invalidRelations: Set[String]): Unit = {
invalidRelations.foreach { n =>
logWarning(s"Count not find relation '$n' specified in hint " +
name: String, parameters: Seq[Any], invalidRelations: Set[Seq[String]]): Unit = {
invalidRelations.foreach { ident =>
logWarning(s"Count not find relation '${ident.quoted}' specified in hint " +
s"'${hintToPrettyString(name, parameters)}'.")
}
}
Expand Down
70 changes: 50 additions & 20 deletions sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveHints.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -64,31 +64,58 @@ object ResolveHints {
_.toUpperCase(Locale.ROOT)).contains(hintName.toUpperCase(Locale.ROOT))))
}

// This method checks if given multi-part identifiers are matched with each other.
// The [[ResolveJoinStrategyHints]] rule is applied before the resolution batch
// in the analyzer and we cannot semantically compare them at this stage.
// Therefore, we follow a simple rule; they match if an identifier in a hint
// is a tail of an identifier in a relation. This process is independent of a session
// catalog (`currentDb` in [[SessionCatalog]]) and it just compares them literally.
//
// For example,
// * in a query `SELECT /*+ BROADCAST(t) */ * FROM db1.t JOIN t`,
// the broadcast hint will match both tables, `db1.t` and `t`.

@dongjoon-hyun dongjoon-hyun Mar 19, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more. 

- the broadcast hint will match both tables, `db1.t` and `t`.
+ the broadcast hint will match both tables, `db1.t` and `t`, even when the current db is `db2`.

// * in a query `SELECT /*+ BROADCAST(default.t) */ * FROM default.t JOIN t`,
// the broadcast hint will match the left-side table only, `default.t`.
private def matchedIdentifier(identInHint: Seq[String], identInQuery: Seq[String]): Boolean = {
if (identInHint.length <= identInQuery.length) {
identInHint.zip(identInQuery.takeRight(identInHint.length))
.forall { case (i1, i2) => resolver(i1, i2) }
} else {
false
}
}

private def extractIdentifier(r: SubqueryAlias): Seq[String] = {
r.identifier.qualifier :+ r.identifier.name
}

private def applyJoinStrategyHint(
plan: LogicalPlan,
relations: mutable.HashSet[String],
relationsInHint: Set[Seq[String]],
relationsInHintWithMatch: mutable.HashSet[Seq[String]],
hintName: String): LogicalPlan = {
// Whether to continue recursing down the tree
var recurse = true

def matchedIdentifierInHint(identInQuery: Seq[String]): Boolean = {
relationsInHint.find(matchedIdentifier(_, identInQuery))
.map(relationsInHintWithMatch.add).nonEmpty
}

val newNode = CurrentOrigin.withOrigin(plan.origin) {
plan match {
case ResolvedHint(u @ UnresolvedRelation(ident), hint)
if relations.exists(resolver(_, ident.last)) =>
relations.remove(ident.last)
Comment thread
cloud-fan marked this conversation as resolved.
if matchedIdentifierInHint(ident) =>
ResolvedHint(u, createHintInfo(hintName).merge(hint, hintErrorHandler))

case ResolvedHint(r: SubqueryAlias, hint)
if relations.exists(resolver(_, r.alias)) =>
relations.remove(r.alias)
if matchedIdentifierInHint(extractIdentifier(r)) =>
ResolvedHint(r, createHintInfo(hintName).merge(hint, hintErrorHandler))

case u @ UnresolvedRelation(ident) if relations.exists(resolver(_, ident.last)) =>
relations.remove(ident.last)
case UnresolvedRelation(ident) if matchedIdentifierInHint(ident) =>
ResolvedHint(plan, createHintInfo(hintName))

case r: SubqueryAlias if relations.exists(resolver(_, r.alias)) =>
relations.remove(r.alias)
case r: SubqueryAlias if matchedIdentifierInHint(extractIdentifier(r)) =>
ResolvedHint(plan, createHintInfo(hintName))

case _: ResolvedHint | _: View | _: With | _: SubqueryAlias =>
Expand All @@ -107,7 +134,9 @@ object ResolveHints {
}

if ((plan fastEquals newNode) && recurse) {
newNode.mapChildren(child => applyJoinStrategyHint(child, relations, hintName))
newNode.mapChildren { child =>
applyJoinStrategyHint(child, relationsInHint, relationsInHintWithMatch, hintName)
}
} else {
newNode
}
Expand All @@ -120,17 +149,19 @@ object ResolveHints {
ResolvedHint(h.child, createHintInfo(h.name))
} else {
// Otherwise, find within the subtree query plans to apply the hint.
val relationNames = h.parameters.map {
case tableName: String => tableName
case tableId: UnresolvedAttribute => tableId.name
val relationNamesInHint = h.parameters.map {
case tableName: String => UnresolvedAttribute.parseAttributeName(tableName)
case tableId: UnresolvedAttribute => tableId.nameParts
case unsupported => throw new AnalysisException("Join strategy hint parameter " +
s"should be an identifier or string but was $unsupported (${unsupported.getClass}")
}
val relationNameSet = new mutable.HashSet[String]
relationNames.foreach(relationNameSet.add)

val applied = applyJoinStrategyHint(h.child, relationNameSet, h.name)
hintErrorHandler.hintRelationsNotFound(h.name, h.parameters, relationNameSet.toSet)
}.toSet
val relationsInHintWithMatch = new mutable.HashSet[Seq[String]]
val applied = applyJoinStrategyHint(
h.child, relationNamesInHint, relationsInHintWithMatch, h.name)

// Filters unmatched relation identifiers in the hint
val unmatchedIdents = relationNamesInHint -- relationsInHintWithMatch
hintErrorHandler.hintRelationsNotFound(h.name, h.parameters, unmatchedIdents)
applied
}
}
Expand Down Expand Up @@ -246,5 +277,4 @@ object ResolveHints {
h.child
}
}

}
2 changes: 1 addition & 1 deletion sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/package.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -196,7 +196,7 @@ package object expressions {
// For example, consider an example where "cat" is the catalog name, "db1" is the database
// name, "a" is the table name and "b" is the column name and "c" is the struct field name.
// If the name parts is cat.db1.a.b.c, then Attribute will match
// Attribute(b, qualifier("cat", "db1, "a")) and List("c") will be the second element
// Attribute(b, qualifier("cat", "db1", "a")) and List("c") will be the second element
var matches: (Seq[Attribute], Seq[String]) = nameParts match {
case catalogPart +: dbPart +: tblPart +: name +: nestedFields =>
val key = (catalogPart.toLowerCase(Locale.ROOT), dbPart.toLowerCase(Locale.ROOT),
Expand Down
3 changes: 2 additions & 1 deletion sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/hints.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -186,7 +186,8 @@ trait HintErrorHandler {
* @param parameters the hint parameters
* @param invalidRelations the set of relation names that cannot be associated
*/
def hintRelationsNotFound(name: String, parameters: Seq[Any], invalidRelations: Set[String]): Unit
def hintRelationsNotFound(
name: String, parameters: Seq[Any], invalidRelations: Set[Seq[String]]): Unit

/**
* Callback for a join hint specified on a relation that is not part of a join.
Expand Down
2 changes: 2 additions & 0 deletions sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/AnalysisTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,6 +45,8 @@ trait AnalysisTest extends PlanTest {
catalog.createTempView("TaBlE", TestRelations.testRelation, overrideIfExists = true)
catalog.createTempView("TaBlE2", TestRelations.testRelation2, overrideIfExists = true)
catalog.createTempView("TaBlE3", TestRelations.testRelation3, overrideIfExists = true)
catalog.createGlobalTempView("TaBlE4", TestRelations.testRelation4, overrideIfExists = true)
catalog.createGlobalTempView("TaBlE5", TestRelations.testRelation5, overrideIfExists = true)
new Analyzer(catalog, conf) {
override val extendedResolutionRules = EliminateSubqueryAliases +: extendedAnalysisRules
}
Expand Down
48 changes: 48 additions & 0 deletions sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/ResolveHintsSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -241,4 +241,52 @@ class ResolveHintsSuite extends AnalysisTest {
Project(testRelation.output, testRelation),
caseSensitive = false)
}

test("Supports multi-part table names for broadcast hint resolution") {
// local temp table (single-part identifier case)
checkAnalysis(
UnresolvedHint("MAPJOIN", Seq("table", "table2"),
table("table").join(table("table2"))),

@dongjoon-hyun dongjoon-hyun Mar 19, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The following will be better because this is a caseSensitive = false test case.

- table("table").join(table("table2"))),
+ table("TaBlE").join(table("TaBlE2"))),

@maropu maropu Mar 19, 2020

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yea, yes!

Join(
ResolvedHint(testRelation, HintInfo(strategy = Some(BROADCAST))),
ResolvedHint(testRelation2, HintInfo(strategy = Some(BROADCAST))),
Inner,
None,
JoinHint.NONE),
caseSensitive = false)

checkAnalysis(
UnresolvedHint("MAPJOIN", Seq("TaBlE", "table2"),
table("TaBlE").join(table("TaBlE2"))),
Join(
ResolvedHint(testRelation, HintInfo(strategy = Some(BROADCAST))),
testRelation2,
Inner,
None,
JoinHint.NONE),
caseSensitive = true)

// global temp table (multi-part identifier case)
checkAnalysis(
UnresolvedHint("MAPJOIN", Seq("GlOBal_TeMP.table4", "table5"),
table("global_temp", "table4").join(table("global_temp", "table5"))),
Join(
ResolvedHint(testRelation4, HintInfo(strategy = Some(BROADCAST))),
ResolvedHint(testRelation5, HintInfo(strategy = Some(BROADCAST))),
Inner,
None,
JoinHint.NONE),
caseSensitive = false)

checkAnalysis(
UnresolvedHint("MAPJOIN", Seq("global_temp.TaBlE4", "table5"),
table("global_temp", "TaBlE4").join(table("global_temp", "TaBlE5"))),
Join(
ResolvedHint(testRelation4, HintInfo(strategy = Some(BROADCAST))),
testRelation5,
Inner,
None,
JoinHint.NONE),
caseSensitive = true)
}
}
2 changes: 2 additions & 0 deletions sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/TestRelations.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,6 +44,8 @@ object TestRelations {
AttributeReference("g", StringType)(),
AttributeReference("h", MapType(IntegerType, IntegerType))())

val testRelation5 = LocalRelation(AttributeReference("i", StringType)())

val nestedRelation = LocalRelation(
AttributeReference("top", StructType(
StructField("duplicateField", StringType) ::
Expand Down
100 changes: 99 additions & 1 deletion sql/core/src/test/scala/org/apache/spark/sql/DataFrameJoinSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,10 +17,13 @@

package org.apache.spark.sql

import org.apache.spark.sql.catalyst.TableIdentifier
import org.apache.spark.sql.catalyst.plans.{Inner, InnerLike, LeftOuter, RightOuter}
import org.apache.spark.sql.catalyst.plans.logical.{Filter, Join, LogicalPlan, Project}
import org.apache.spark.sql.catalyst.plans.logical.{BROADCAST, Filter, HintInfo, Join, JoinHint, LogicalPlan, Project}
import org.apache.spark.sql.execution.FileSourceScanExec
import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
import org.apache.spark.sql.execution.datasources.LogicalRelation
import org.apache.spark.sql.execution.exchange.BroadcastExchangeExec
import org.apache.spark.sql.execution.joins.BroadcastHashJoinExec
import org.apache.spark.sql.functions._
import org.apache.spark.sql.internal.SQLConf
Expand Down Expand Up @@ -322,4 +325,99 @@ class DataFrameJoinSuite extends QueryTest
}
}
}

test("Supports multi-part names for broadcast hint resolution") {
val (table1Name, table2Name) = ("t1", "t2")

withTempDatabase { dbName =>
withTable(table1Name, table2Name) {
withSQLConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1") {
spark.range(50).write.saveAsTable(s"$dbName.$table1Name")
spark.range(100).write.saveAsTable(s"$dbName.$table2Name")

// First, makes sure a join is not broadcastable
val plan = sql(s"SELECT * FROM $dbName.$table1Name, $dbName.$table2Name " +
s"WHERE $table1Name.id = $table2Name.id")
.queryExecution.executedPlan
assert(plan.collect { case p: BroadcastHashJoinExec => p }.isEmpty)

@dongjoon-hyun dongjoon-hyun Mar 19, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

checkIfHintNotApplied?

-   .queryExecution.executedPlan
- assert(plan.collect { case p: BroadcastHashJoinExec => p }.isEmpty)
+ checkIfHintNotApplied(plan)

@maropu maropu Mar 19, 2020

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ah... I forgot to remove the old tests... I can remove it. Thanks!


def checkIfHintApplied(df: DataFrame): Unit = {
val sparkPlan = df.queryExecution.executedPlan
val broadcastHashJoins = sparkPlan.collect { case p: BroadcastHashJoinExec => p }
assert(broadcastHashJoins.size == 1)
val broadcastExchanges = broadcastHashJoins.head.collect {
case p: BroadcastExchangeExec => p
}
assert(broadcastExchanges.size == 1)
val tables = broadcastExchanges.head.collect {
case FileSourceScanExec(_, _, _, _, _, _, Some(tableIdent)) => tableIdent
}
assert(tables.size == 1)
assert(tables.head === TableIdentifier(table1Name, Some(dbName)))
}

def checkIfHintNotApplied(df: DataFrame): Unit = {
val sparkPlan = df.queryExecution.executedPlan
val broadcastHashJoins = sparkPlan.collect { case p: BroadcastHashJoinExec => p }
assert(broadcastHashJoins.isEmpty)
}

def sqlTemplate(tableName: String, hintTableName: String): DataFrame = {
sql(s"SELECT /*+ BROADCASTJOIN($hintTableName) */ * " +
s"FROM $tableName, $dbName.$table2Name " +
s"WHERE $tableName.id = $table2Name.id")
}

def dfTemplate(tableName: String, hintTableName: String): DataFrame = {
spark.table(tableName).join(spark.table(s"$dbName.$table2Name"), "id")
.hint("broadcast", hintTableName)
}

sql(s"USE $dbName")

checkIfHintApplied(sqlTemplate(table1Name, table1Name))
checkIfHintApplied(sqlTemplate(s"$dbName.$table1Name", s"$dbName.$table1Name"))
checkIfHintApplied(sqlTemplate(s"$dbName.$table1Name", table1Name))
checkIfHintNotApplied(sqlTemplate(table1Name, s"$dbName.$table1Name"))
checkIfHintNotApplied(sqlTemplate(s"$dbName.$table1Name", s"$dbName.$table1Name.id"))

@dongjoon-hyun dongjoon-hyun Mar 19, 2020 •
edited
Loading

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is $dbName.$table1Name.id used as a negative value for hintTableName? It's a little confusing because there is a catalog concept. For three fields, can we use catalog instead of id?

@maropu maropu Mar 19, 2020

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

ok, I will modify id -> spark_catalog.


checkIfHintApplied(dfTemplate(table1Name, table1Name))
checkIfHintApplied(dfTemplate(s"$dbName.$table1Name", s"$dbName.$table1Name"))
checkIfHintApplied(dfTemplate(s"$dbName.$table1Name", table1Name))
checkIfHintApplied(dfTemplate(table1Name, s"$dbName.$table1Name"))
checkIfHintNotApplied(dfTemplate(s"$dbName.$table1Name", s"$dbName.$table1Name.id"))

withTempView("tv") {
sql(s"CREATE TEMPORARY VIEW tv AS SELECT * FROM $dbName.$table1Name")
checkIfHintApplied(sqlTemplate("tv", "tv"))
checkIfHintNotApplied(sqlTemplate("tv", "default.tv"))

@dongjoon-hyun dongjoon-hyun Mar 19, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This might be misleading. Technically, this is the same with sqlTemplate("tv", "non_exist") because we cannot use a database qualifier for Temporary View.

scala> sql("select * from default.tv").show
org.apache.spark.sql.AnalysisException: Table or view not found: default.tv; line 1 pos 14;

@maropu maropu Mar 19, 2020

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yea, but I think a query with a hint having a non-existent relation identifier should work?;


scala> sql("create table t1 (id int)")
scala> sql("create table t2 (id int)")
scala> sql("create temporary view tv as select * from t1")
scala> sql("SELECT /*+ BROADCASTJOIN(default.non_exist) */ * FROM tv, t2 WHERE tv.id = t2.id").explain(true)
20/03/20 07:34:02 WARN HintErrorLogger: Count not find relation 'default.non_exist' specified in hint 'BROADCASTJOIN(default.non_exist)'.
== Parsed Logical Plan ==
'UnresolvedHint BROADCASTJOIN, ['default.non_exist]
+- 'Project [*]
   +- 'Filter ('tv.id = 't2.id)
      +- 'Join Inner
         :- 'UnresolvedRelation [tv]
         +- 'UnresolvedRelation [t2]

== Analyzed Logical Plan ==
id: int, id: int
Project [id#0, id#1]
+- Filter (id#0 = id#1)
   +- Join Inner
      :- SubqueryAlias tv
      :  +- Project [id#0]
      :     +- SubqueryAlias spark_catalog.default.t1
      :        +- Relation[id#0] parquet
      +- SubqueryAlias spark_catalog.default.t2
         +- Relation[id#1] parquet

@maropu maropu Mar 19, 2020

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I will modify the tests a little.

}
}
}
}
}

test("The same table name exists in two databases for broadcast hint resolution") {
val (db1Name, db2Name) = ("db1", "db2")

withDatabase(db1Name, db2Name) {
withTable("t") {
withSQLConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1") {
sql(s"CREATE DATABASE $db1Name")
sql(s"CREATE DATABASE $db2Name")
spark.range(1).write.saveAsTable(s"$db1Name.t")
spark.range(1).write.saveAsTable(s"$db2Name.t")

// Checks if a broadcast hint applied in both sides
val statement = s"SELECT /*+ BROADCASTJOIN(t) */ * FROM $db1Name.t, $db2Name.t " +
s"WHERE $db1Name.t.id = $db2Name.t.id"
sql(statement).queryExecution.optimizedPlan match {
case Join(_, _, _, _, JoinHint(Some(HintInfo(Some(BROADCAST))),
Some(HintInfo(Some(BROADCAST))))) =>
case _ => fail("broadcast hint not found in both tables")
}
}
}
}
}
}
24 changes: 23 additions & 1 deletion sql/core/src/test/scala/org/apache/spark/sql/execution/GlobalTempViewSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,8 @@ package org.apache.spark.sql.execution
import org.apache.spark.sql.{AnalysisException, QueryTest, Row}
import org.apache.spark.sql.catalog.Table
import org.apache.spark.sql.catalyst.TableIdentifier
import org.apache.spark.sql.catalyst.analysis.NoSuchTableException
import org.apache.spark.sql.catalyst.plans.logical.{BROADCAST, HintInfo, Join, JoinHint}
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.test.SharedSparkSession
import org.apache.spark.sql.types.StructType

Expand Down Expand Up @@ -170,4 +171,25 @@ class GlobalTempViewSuite extends QueryTest with SharedSparkSession {
isTemporary = true).toString)
}
}

test("broadcast hint on global temp view") {
withGlobalTempView("v1") {
spark.range(10).createGlobalTempView("v1")
withTempView("v2") {
spark.range(10).createTempView("v2")

withSQLConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1") {
Seq(
"SELECT /*+ MAPJOIN(v1) */ * FROM global_temp.v1, v2 WHERE v1.id = v2.id",
"SELECT /*+ MAPJOIN(global_temp.v1) */ * FROM global_temp.v1, v2 WHERE v1.id = v2.id"
).foreach { statement =>
sql(statement).queryExecution.optimizedPlan match {
case Join(_, _, _, _, JoinHint(Some(HintInfo(Some(BROADCAST))), None)) =>
case _ => fail("broadcast hint not found in a left-side table")
}
}
}
}
}
}
}