spark-sql-spark技术分享-第16页

ResolveRelations Logical Resolution Rule — Resolving UnresolvedRelations With Tables in Catalog

ResolveRelations is a logical resolution rule that the logical query plan analyzer uses to resolve UnresolvedRelations (in a logical query plan), i.e.

Resolves UnresolvedRelation logical operators (in InsertIntoTable operators)
Other uses of UnresolvedRelation

Technically, ResolveRelations is just a Catalyst rule for transforming logical plans, i.e. Rule[LogicalPlan].

ResolveRelations is part of Resolution fixed-point batch of rules.



// Example: InsertIntoTable with UnresolvedRelation
import org.apache.spark.sql.catalyst.dsl.plans._
val plan = table("t1").insertInto(tableName = "t2", overwrite = true)
scala> println(plan.numberedTreeString)
00 'InsertIntoTable 'UnresolvedRelation `t2`, true, false
01 +- 'UnresolvedRelation `t1`

// Register the tables so the following resolution works
sql("CREATE TABLE IF NOT EXISTS t1(id long)")
sql("CREATE TABLE IF NOT EXISTS t2(id long)")

// ResolveRelations is a Scala object of the Analyzer class
// We need an instance of the Analyzer class to access it
import spark.sessionState.analyzer.ResolveRelations
val resolvedPlan = ResolveRelations(plan)
scala> println(resolvedPlan.numberedTreeString)
00 'InsertIntoTable 'UnresolvedRelation `t2`, true, false
01 +- 'SubqueryAlias t1
02    +- 'UnresolvedCatalogRelation `default`.`t1`, org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe

// Example: Other uses of UnresolvedRelation
// Use a temporary view
val v1 = spark.range(1).createOrReplaceTempView("v1")
scala> spark.catalog.listTables.filter($"name" === "v1").show
+----+--------+-----------+---------+-----------+
|name|database|description|tableType|isTemporary|
+----+--------+-----------+---------+-----------+
|  v1|    null|       null|TEMPORARY|       true|
+----+--------+-----------+---------+-----------+

import org.apache.spark.sql.catalyst.dsl.expressions._
val plan = table("v1").select(star())
scala> println(plan.numberedTreeString)
00 'Project [*]
01 +- 'UnresolvedRelation `v1`

val resolvedPlan = ResolveRelations(plan)
scala> println(resolvedPlan.numberedTreeString)
00 'Project [*]
01 +- SubqueryAlias v1
02    +- Range (0, 1, step=1, splits=Some(8))

// Example
import org.apache.spark.sql.catalyst.dsl.plans._
val plan = table(db = "db1", ref = "t1")
scala> println(plan.numberedTreeString)
00 'UnresolvedRelation `db1`.`t1`

// Register the database so the following resolution works
sql("CREATE DATABASE IF NOT EXISTS db1")

val resolvedPlan = ResolveRelations(plan)
scala> println(resolvedPlan.numberedTreeString)
00 'SubqueryAlias t1
01 +- 'UnresolvedCatalogRelation `db1`.`t1`, org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

47

48

49

50

51

52

53

54

55

56

57

58

59

// Example: InsertIntoTable with UnresolvedRelation

import org.apache.spark.sql.catalyst.dsl.plans._

val plan = table("t1").insertInto(tableName = "t2", overwrite = true)

scala> println(plan.numberedTreeString)

00 'InsertIntoTable 'UnresolvedRelation `t2`, true, false

01 +- 'UnresolvedRelation `t1`

// Register the tables so the following resolution works

sql("CREATE TABLE IF NOT EXISTS t1(id long)")

sql("CREATE TABLE IF NOT EXISTS t2(id long)")

// ResolveRelations is a Scala object of the Analyzer class

// We need an instance of the Analyzer class to access it

import spark.sessionState.analyzer.ResolveRelations

val resolvedPlan = ResolveRelations(plan)

scala> println(resolvedPlan.numberedTreeString)

00 'InsertIntoTable 'UnresolvedRelation `t2`, true, false

01 +- 'SubqueryAlias t1

02 +- 'UnresolvedCatalogRelation `default`.`t1`, org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe

// Example: Other uses of UnresolvedRelation

// Use a temporary view

val v1 = spark.range(1).createOrReplaceTempView("v1")

scala> spark.catalog.listTables.filter($"name" === "v1").show

+----+--------+-----------+---------+-----------+

+----+--------+-----------+---------+-----------+

+----+--------+-----------+---------+-----------+

import org.apache.spark.sql.catalyst.dsl.expressions._

val plan = table("v1").select(star())

scala> println(plan.numberedTreeString)

00 'Project [*]

01 +- 'UnresolvedRelation `v1`

val resolvedPlan = ResolveRelations(plan)

scala> println(resolvedPlan.numberedTreeString)

00 'Project [*]

01 +- SubqueryAlias v1

02 +- Range (0, 1, step=1, splits=Some(8))

// Example

import org.apache.spark.sql.catalyst.dsl.plans._

val plan = table(db = "db1", ref = "t1")

scala> println(plan.numberedTreeString)

00 'UnresolvedRelation `db1`.`t1`

// Register the database so the following resolution works

sql("CREATE DATABASE IF NOT EXISTS db1")

val resolvedPlan = ResolveRelations(plan)

scala> println(resolvedPlan.numberedTreeString)

00 'SubqueryAlias t1

01 +- 'UnresolvedCatalogRelation `db1`.`t1`, org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe

Applying ResolveRelations to Logical Plan — `apply` Method



apply(plan: LogicalPlan): LogicalPlan

1

2

3

4

5

apply(plan: LogicalPlan): LogicalPlan

Note	`apply` is part of Rule Contract to apply a rule to a logical plan.

apply…FIXME

Resolving Relation — `resolveRelation` Method



resolveRelation(plan: LogicalPlan): LogicalPlan

1

2

3

4

5

resolveRelation(plan: LogicalPlan): LogicalPlan

resolveRelation…FIXME

Note	`resolveRelation` is used when…FIXME

`isRunningDirectlyOnFiles` Internal Method



isRunningDirectlyOnFiles(table: TableIdentifier): Boolean

1

2

3

4

5

isRunningDirectlyOnFiles(table: TableIdentifier): Boolean

isRunningDirectlyOnFiles is enabled (i.e. true) when all of the following conditions hold:

The database of the input table is defined
spark.sql.runSQLOnFiles internal configuration property is enabled
The table is not a temporary table
The database or the table do not exist (in the SessionCatalog)

Note	`isRunningDirectlyOnFiles` is used exclusively when `ResolveRelations` resolves a relation (as a UnresolvedRelation leaf logical operator for a table reference).

Finding Table in Session-Scoped Catalog of Relational Entities — `lookupTableFromCatalog` Internal Method



lookupTableFromCatalog(
  u: UnresolvedRelation,
  defaultDatabase: Option[String] = None): LogicalPlan

1

2

3

4

5

6

7

lookupTableFromCatalog(

u: UnresolvedRelation,

defaultDatabase: Option[String] = None): LogicalPlan

lookupTableFromCatalog simply requests SessionCatalog to find the table in relational catalogs.

Note	`lookupTableFromCatalog` requests `Analyzer` for the current SessionCatalog.

Note	The table is described using TableIdentifier of the input `UnresolvedRelation`.

lookupTableFromCatalog fails the analysis phase (by reporting a AnalysisException) when the table or the table’s database cannot be found.

Note	`lookupTableFromCatalog` is used when `ResolveRelations` is executed (for InsertIntoTable with `UnresolvedRelation` operators) or resolves a relation (for “standalone” UnresolvedRelations).

ResolveReferences Logical Resolution Rule

ResolveReferences is a logical resolution rule that the logical query plan analyzer uses to resolve FIXME in a logical query plan, i.e.

Resolves…FIXME

Technically, ResolveReferences is just a Catalyst rule for transforming logical plans, i.e. Rule[LogicalPlan].

ResolveReferences is part of Resolution fixed-point batch of rules.



// FIXME: Use Catalyst DSL to create a logical plan
import org.apache.spark.sql.catalyst.dsl.plans._
val t1 = table("t1")

import org.apache.spark.sql.catalyst.dsl.expressions._
val logicalPlan = t1
  .select("a".attr, star())
  .groupBy(groupingExprs = "b".attr)(aggregateExprs = star())
scala> println(logicalPlan.numberedTreeString)
00 'Aggregate ['b], [*]
01 +- 'Project ['a, *]
02    +- 'UnresolvedRelation `t1`
// END FIXME

// Make the example repeatable
val table = "t1"
import org.apache.spark.sql.catalyst.TableIdentifier
spark.sessionState.catalog.dropTable(TableIdentifier(table), ignoreIfNotExists = true, purge = true)

Seq((0, "zero"), (1, "one")).toDF("id", "name").createOrReplaceTempView(table)
val query = spark.table("t1").select("id", "*").groupBy("id", "name").agg(col("*"))
val logicalPlan = query.queryExecution.logical
scala> println(logicalPlan.numberedTreeString)
00 'Aggregate [id#28, name#29], [id#28, name#29, *]
01 +- Project [id#28, id#28, name#29]
02    +- SubqueryAlias t1
03       +- Project [_1#25 AS id#28, _2#26 AS name#29]
04          +- LocalRelation [_1#25, _2#26]

import spark.sessionState.analyzer.ResolveReferences
val planWithRefsResolved = ResolveReferences(logicalPlan)
scala> println(planWithRefsResolved.numberedTreeString)
00 Aggregate [id#28, name#29], [id#28, name#29, id#28, id#28, name#29]
01 +- Project [id#28, id#28, name#29]
02    +- SubqueryAlias t1
03       +- Project [_1#25 AS id#28, _2#26 AS name#29]
04          +- LocalRelation [_1#25, _2#26]

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

// FIXME: Use Catalyst DSL to create a logical plan

import org.apache.spark.sql.catalyst.dsl.plans._

val t1 = table("t1")

import org.apache.spark.sql.catalyst.dsl.expressions._

val logicalPlan = t1

.select("a".attr, star())

.groupBy(groupingExprs = "b".attr)(aggregateExprs = star())

scala> println(logicalPlan.numberedTreeString)

00 'Aggregate ['b], [*]

01 +- 'Project ['a, *]

02 +- 'UnresolvedRelation `t1`

// END FIXME

// Make the example repeatable

val table = "t1"

import org.apache.spark.sql.catalyst.TableIdentifier

spark.sessionState.catalog.dropTable(TableIdentifier(table), ignoreIfNotExists = true, purge = true)

Seq((0, "zero"), (1, "one")).toDF("id", "name").createOrReplaceTempView(table)

val query = spark.table("t1").select("id", "*").groupBy("id", "name").agg(col("*"))

val logicalPlan = query.queryExecution.logical

scala> println(logicalPlan.numberedTreeString)

00 'Aggregate [id#28, name#29], [id#28, name#29, *]

01 +- Project [id#28, id#28, name#29]

02 +- SubqueryAlias t1

03 +- Project [_1#25 AS id#28, _2#26 AS name#29]

04 +- LocalRelation [_1#25, _2#26]

import spark.sessionState.analyzer.ResolveReferences

val planWithRefsResolved = ResolveReferences(logicalPlan)

scala> println(planWithRefsResolved.numberedTreeString)

00 Aggregate [id#28, name#29], [id#28, name#29, id#28, id#28, name#29]

01 +- Project [id#28, id#28, name#29]

02 +- SubqueryAlias t1

03 +- Project [_1#25 AS id#28, _2#26 AS name#29]

04 +- LocalRelation [_1#25, _2#26]

Resolving Expressions of Logical Plan — `resolve` Internal Method



resolve(e: Expression, q: LogicalPlan): Expression

1

2

3

4

5

resolve(e: Expression, q: LogicalPlan): Expression

resolve resolves the input expression per type:

UnresolvedAttribute expressions
UnresolvedExtractValue expressions
All other expressions

Note	`resolve` is used exclusively when `ResolveReferences` is requested to resolve reference expressions in a logical query plan.

Resolving Reference Expressions In Logical Query Plan (Applying ResolveReferences to Logical Plan) — `apply` Method



apply(plan: LogicalPlan): LogicalPlan

1

2

3

4

5

apply(plan: LogicalPlan): LogicalPlan

Note	`apply` is part of Rule Contract to apply a rule to a logical plan.

apply resolves the following logical operators:

Project logical operator with a Star expression to…FIXME
Aggregate logical operator with a Star expression to…FIXME
ScriptTransformation logical operator with a Star expression to…FIXME
Generate logical operator with a Star expression to…FIXME
Join logical operator with duplicateResolved…FIXME
Intersect logical operator with duplicateResolved…FIXME
Except logical operator with duplicateResolved…FIXME
Sort logical operator unresolved with child operators resolved…FIXME
Generate logical operator resolved…FIXME
Generate logical operator unresolved…FIXME

In the end, apply resolves the expressions of the input logical operator.

apply skips logical operators that:

Use UnresolvedDeserializer expressions
Have child operators unresolved

Expanding Star Expressions — `buildExpandedProjectList` Internal Method



buildExpandedProjectList(
  exprs: Seq[NamedExpression],
  child: LogicalPlan): Seq[NamedExpression]

1

2

3

4

5

6

7

buildExpandedProjectList(

exprs: Seq[NamedExpression],

child: LogicalPlan): Seq[NamedExpression]

buildExpandedProjectList expands (converts) Star expressions in the input named expressions recursively (down the expression tree) per expression:

For a Star expression, buildExpandedProjectList requests it to expand given the input child logical plan
For a UnresolvedAlias expression with a Star child expression, buildExpandedProjectList requests it to expand given the input child logical plan (similarly to a Star expression alone in the above case)
For exprs with Star expressions down the expression tree, buildExpandedProjectList expandStarExpression passing the input exprs and child

Note	`buildExpandedProjectList` is used when `ResolveReferences` is requested to resolve reference expressions (in `Project` and `Aggregate` operators with `Star` expressions).

`expandStarExpression` Method



expandStarExpression(expr: Expression, child: LogicalPlan): Expression

1

2

3

4

5

expandStarExpression(expr: Expression, child: LogicalPlan): Expression

expandStarExpression expands (transforms) the following expressions in the input expr expression:

For UnresolvedFunction expressions with Star child expressions, expandStarExpression requests the Star expressions to expand given the input child logical plan and the resolver.



// Using Catalyst DSL to create a logical plan with a function with Star child expression
import org.apache.spark.sql.catalyst.dsl.plans._
val t1 = table("t1")

import org.apache.spark.sql.catalyst.dsl.expressions._
val f1 = 'f1.function(star())

val plan = t1.select(f1)
scala> println(plan.numberedTreeString)
00 'Project [unresolvedalias('f1(*), None)]
01 +- 'UnresolvedRelation `t1`

// CAUTION: FIXME How to demo that the plan gets resolved using ResolveReferences.expandStarExpression?

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

// Using Catalyst DSL to create a logical plan with a function with Star child expression

import org.apache.spark.sql.catalyst.dsl.plans._

val t1 = table("t1")

import org.apache.spark.sql.catalyst.dsl.expressions._

val f1 = 'f1.function(star())

val plan = t1.select(f1)

scala> println(plan.numberedTreeString)

00 'Project [unresolvedalias('f1(*), None)]

01 +- 'UnresolvedRelation `t1`

// CAUTION: FIXME How to demo that the plan gets resolved using ResolveReferences.expandStarExpression?

For CreateNamedStruct expressions with Star child expressions among the values, expandStarExpression…FIXME
For CreateArray expressions with Star child expressions, expandStarExpression…FIXME
For Murmur3Hash expressions with Star child expressions, expandStarExpression…FIXME

For any other uses of Star expressions, expandStarExpression fails analysis with a AnalysisException:



Invalid usage of '*' in expression '[exprName]'

1

2

3

4

5

Invalid usage of '*' in expression '[exprName]'

Note	`expandStarExpression` is used exclusively when `ResolveReferences` is requested to expand Star expressions (in `Project` and `Aggregate` operators).

`dedupRight` Internal Method



dedupRight(left: LogicalPlan, right: LogicalPlan): LogicalPlan

1

2

3

4

5

dedupRight(left: LogicalPlan, right: LogicalPlan): LogicalPlan

dedupRight…FIXME

Note	`dedupRight` is used when…FIXME

`dedupOuterReferencesInSubquery` Internal Method



dedupOuterReferencesInSubquery(
  plan: LogicalPlan,
  attrMap: AttributeMap[Attribute]): LogicalPlan

1

2

3

4

5

6

7

dedupOuterReferencesInSubquery(

plan: LogicalPlan,

attrMap: AttributeMap[Attribute]): LogicalPlan

dedupOuterReferencesInSubquery…FIXME

Note	`dedupOuterReferencesInSubquery` is used when…FIXME

ResolveOrdinalInOrderByAndGroupBy

2013-03-02admin阅读(1414)

ResolveOrdinalInOrderByAndGroupBy Logical Resolution Rule

ResolveOrdinalInOrderByAndGroupBy is a logical resolution rule that converts ordinal positions in Sort and Aggregate logical operators with corresponding expressions in a logical query plan.

ResolveOrdinalInOrderByAndGroupBy is part of the Resolution fixed-point batch in the standard batches of the Analyzer.

ResolveOrdinalInOrderByAndGroupBy is simply a Catalyst rule for transforming logical plans, i.e. Rule[LogicalPlan].

ResolveOrdinalInOrderByAndGroupBy takes no arguments when created.



// FIXME: DEMO
val rule = spark.sessionState.analyzer.ResolveOrdinalInOrderByAndGroupBy

val plan = ???
val planResolved = rule(plan)
scala> println(planResolved.numberedTreeString)
00 'UnresolvedRelation `t1`

1

2

3

4

5

6

7

8

9

10

11

// FIXME: DEMO

val rule = spark.sessionState.analyzer.ResolveOrdinalInOrderByAndGroupBy

val plan = ???

val planResolved = rule(plan)

scala> println(planResolved.numberedTreeString)

00 'UnresolvedRelation `t1`

Executing Rule — `apply` Method



apply(plan: LogicalPlan): LogicalPlan

1

2

3

4

5

apply(plan: LogicalPlan): LogicalPlan

Note	`apply` is part of the Rule Contract to execute (apply) a rule on a TreeNode (e.g. LogicalPlan).

apply walks the logical plan from children up the tree and looks for Sort and Aggregate logical operators with UnresolvedOrdinal leaf expressions (in ordering and grouping expressions, respectively).

For a Sort logical operator with UnresolvedOrdinal expressions, apply replaces all the SortOrder expressions (with UnresolvedOrdinal child expressions) with SortOrder expressions and the expression at the index - 1 position in the output schema of the child logical operator.

For a Aggregate logical operator with UnresolvedOrdinal expressions, apply replaces all the expressions (with UnresolvedOrdinal child expressions) with the expression at the index - 1 position in the aggregate named expressions of the current Aggregate logical operator.

apply throws a AnalysisException (and hence fails an analysis) if the ordinal is outside the range:



ORDER BY position [index] is not in select list (valid range is [1, [output.size]])
GROUP BY position [index] is not in select list (valid range is [1, [aggs.size]])

1

2

3

4

5

6

ORDER BY position [index] is not in select list (valid range is [1, [output.size]])

GROUP BY position [index] is not in select list (valid range is [1, [aggs.size]])

ResolveMissingReferences

2013-03-01admin阅读(1768)

ResolveMissingReferences

ResolveMissingReferences is…FIXME

`resolveExprsAndAddMissingAttrs` Internal Method



resolveExprsAndAddMissingAttrs(
  exprs: Seq[Expression],
  plan: LogicalPlan): (Seq[Expression], LogicalPlan)

1

2

3

4

5

6

7

resolveExprsAndAddMissingAttrs(

exprs: Seq[Expression],

plan: LogicalPlan): (Seq[Expression], LogicalPlan)

resolveExprsAndAddMissingAttrs…FIXME

Note	`resolveExprsAndAddMissingAttrs` is used when…FIXME

ResolveInlineTables

2013-02-28admin阅读(1547)

ResolveInlineTables Logical Resolution Rule

ResolveInlineTables is a logical resolution rule that resolves (replaces) UnresolvedInlineTable operators to LocalRelations in a logical query plan.

ResolveInlineTables is part of the Resolution fixed-point batch in the standard batches of the Analyzer.

ResolveInlineTables is simply a Catalyst rule for transforming logical plans, i.e. Rule[LogicalPlan].

ResolveInlineTables takes a SQLConf when created.



val q = sql("VALUES 1, 2, 3")
val plan = q.queryExecution.logical
scala> println(plan.numberedTreeString)
00 'UnresolvedInlineTable [col1], [List(1), List(2), List(3)]

scala> :type spark
org.apache.spark.sql.SparkSession

scala> :type spark.sessionState.conf
org.apache.spark.sql.internal.SQLConf

import org.apache.spark.sql.catalyst.analysis.ResolveInlineTables
val rule = ResolveInlineTables(spark.sessionState.conf)

val planAfterResolveInlineTables = rule(plan)
scala> println(planAfterResolveInlineTables.numberedTreeString)
00 LocalRelation [col1#2]

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

val q = sql("VALUES 1, 2, 3")

val plan = q.queryExecution.logical

scala> println(plan.numberedTreeString)

00 'UnresolvedInlineTable [col1], [List(1), List(2), List(3)]

scala> :type spark

org.apache.spark.sql.SparkSession

scala> :type spark.sessionState.conf

org.apache.spark.sql.internal.SQLConf

import org.apache.spark.sql.catalyst.analysis.ResolveInlineTables

val rule = ResolveInlineTables(spark.sessionState.conf)

val planAfterResolveInlineTables = rule(plan)

scala> println(planAfterResolveInlineTables.numberedTreeString)

00 LocalRelation [col1#2]

Executing Rule — `apply` Method



apply(plan: LogicalPlan): LogicalPlan

1

2

3

4

5

apply(plan: LogicalPlan): LogicalPlan

Note	`apply` is part of the Rule Contract to execute (apply) a rule on a TreeNode (e.g. LogicalPlan).

apply simply searches the input plan up to find UnresolvedInlineTable logical operators with rows expressions resolved.

For such a UnresolvedInlineTable logical operator, apply validateInputDimension and validateInputEvaluable.

In the end, apply converts the UnresolvedInlineTable to a LocalRelation.

`validateInputDimension` Internal Method



validateInputDimension(table: UnresolvedInlineTable): Unit

1

2

3

4

5

validateInputDimension(table: UnresolvedInlineTable): Unit

validateInputDimension…FIXME

Note	`validateInputDimension` is used exclusively when `ResolveInlineTables` logical resolution rule is executed.

`validateInputEvaluable` Internal Method



validateInputEvaluable(table: UnresolvedInlineTable): Unit

1

2

3

4

5

validateInputEvaluable(table: UnresolvedInlineTable): Unit

validateInputEvaluable…FIXME

Note	`validateInputEvaluable` is used exclusively when `ResolveInlineTables` logical resolution rule is executed.

Converting UnresolvedInlineTable to LocalRelation — `convert` Internal Method



convert(table: UnresolvedInlineTable): LocalRelation

1

2

3

4

5

convert(table: UnresolvedInlineTable): LocalRelation

convert…FIXME

Note	`convert` is used exclusively when `ResolveInlineTables` logical resolution rule is executed.

ResolveHiveSerdeTable

2013-02-27admin阅读(1103)

ResolveHiveSerdeTable Logical Resolution Rule

ResolveHiveSerdeTable is a logical resolution rule (i.e. Rule[LogicalPlan]) that the Hive-specific logical query plan analyzer uses to resolve the metadata of a hive table for CreateTable logical operators.

ResolveHiveSerdeTable is part of additional rules in Resolution fixed-point batch of rules.



// FIXME Example of ResolveHiveSerdeTable

1

2

3

4

5

// FIXME Example of ResolveHiveSerdeTable

Applying ResolveHiveSerdeTable Rule to Logical Plan — `apply` Method



apply(plan: LogicalPlan): LogicalPlan

1

2

3

4

5

apply(plan: LogicalPlan): LogicalPlan

Note	`apply` is part of Rule Contract to apply a rule to a logical plan.

apply…FIXME

ResolveFunctions

2013-02-26admin阅读(1568)

ResolveFunctions Logical Resolution Rule — Resolving grouping__id UnresolvedAttribute, UnresolvedGenerator And UnresolvedFunction Expressions

ResolveFunctions is a logical resolution rule that the logical query plan analyzer uses to resolve grouping__id UnresolvedAttribute, UnresolvedGenerator and UnresolvedFunction expressions in an entire logical query plan.

Technically, ResolveReferences is just a Catalyst rule for transforming logical plans, i.e. Rule[LogicalPlan].

ResolveFunctions is part of Resolution fixed-point batch of rules.

Note	`ResolveFunctions` is a Scala object inside Analyzer class.



import spark.sessionState.analyzer.ResolveFunctions

// Example: UnresolvedAttribute with VirtualColumn.hiveGroupingIdName (grouping__id) => Alias
import org.apache.spark.sql.catalyst.expressions.VirtualColumn
import org.apache.spark.sql.catalyst.analysis.UnresolvedAttribute
val groupingIdAttr = UnresolvedAttribute(VirtualColumn.hiveGroupingIdName)
scala> println(groupingIdAttr.sql)
`grouping__id`

// Using Catalyst DSL to create a logical plan with grouping__id
import org.apache.spark.sql.catalyst.dsl.plans._
val t1 = table("t1")
val plan = t1.select(groupingIdAttr)
scala> println(plan.numberedTreeString)
00 'Project ['grouping__id]
01 +- 'UnresolvedRelation `t1`

val resolvedPlan = ResolveFunctions(plan)
scala> println(resolvedPlan.numberedTreeString)
00 'Project [grouping_id() AS grouping__id#0]
01 +- 'UnresolvedRelation `t1`

import org.apache.spark.sql.catalyst.expressions.Alias
val alias = resolvedPlan.expressions.head.asInstanceOf[Alias]
scala> println(alias.sql)
grouping_id() AS `grouping__id`

// Example: UnresolvedGenerator => a) Generator or b) analysis failure
// Register a function so a function resolution works
import org.apache.spark.sql.catalyst.FunctionIdentifier
import org.apache.spark.sql.catalyst.catalog.CatalogFunction
val f1 = CatalogFunction(FunctionIdentifier(funcName = "f1"), "java.lang.String", resources = Nil)
import org.apache.spark.sql.catalyst.expressions.{Expression, Stack}
// FIXME What happens when looking up a function with the functionBuilder None in registerFunction?
// Using Stack as ResolveFunctions requires that the function to be resolved is a Generator
// You could roll your own, but that's a demo, isn't it? (don't get too carried away)
spark.sessionState.catalog.registerFunction(
  funcDefinition = f1,
  overrideIfExists = true,
  functionBuilder = Some((children: Seq[Expression]) => Stack(children = Nil)))

import org.apache.spark.sql.catalyst.analysis.UnresolvedGenerator
import org.apache.spark.sql.catalyst.FunctionIdentifier
val ungen = UnresolvedGenerator(name = FunctionIdentifier("f1"), children = Seq.empty)
val plan = t1.select(ungen)
scala> println(plan.numberedTreeString)
00 'Project [unresolvedalias('f1(), None)]
01 +- 'UnresolvedRelation `t1`

val resolvedPlan = ResolveFunctions(plan)
scala> println(resolvedPlan.numberedTreeString)
00 'Project [unresolvedalias(stack(), None)]
01 +- 'UnresolvedRelation `t1`

CAUTION: FIXME

// Example: UnresolvedFunction => a) AggregateWindowFunction with(out) isDistinct, b) AggregateFunction, c) other with(out) isDistinct
val plan = ???
val resolvedPlan = ResolveFunctions(plan)

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

47

48

49

50

51

52

53

54

55

56

57

58

59

60

61

62

63

import spark.sessionState.analyzer.ResolveFunctions

// Example: UnresolvedAttribute with VirtualColumn.hiveGroupingIdName (grouping__id) => Alias

import org.apache.spark.sql.catalyst.expressions.VirtualColumn

import org.apache.spark.sql.catalyst.analysis.UnresolvedAttribute

val groupingIdAttr = UnresolvedAttribute(VirtualColumn.hiveGroupingIdName)

scala> println(groupingIdAttr.sql)

`grouping__id`

// Using Catalyst DSL to create a logical plan with grouping__id

import org.apache.spark.sql.catalyst.dsl.plans._

val t1 = table("t1")

val plan = t1.select(groupingIdAttr)

scala> println(plan.numberedTreeString)

00 'Project ['grouping__id]

01 +- 'UnresolvedRelation `t1`

val resolvedPlan = ResolveFunctions(plan)

scala> println(resolvedPlan.numberedTreeString)

00 'Project [grouping_id() AS grouping__id#0]

01 +- 'UnresolvedRelation `t1`

import org.apache.spark.sql.catalyst.expressions.Alias

val alias = resolvedPlan.expressions.head.asInstanceOf[Alias]

scala> println(alias.sql)

grouping_id() AS `grouping__id`

// Example: UnresolvedGenerator => a) Generator or b) analysis failure

// Register a function so a function resolution works

import org.apache.spark.sql.catalyst.FunctionIdentifier

import org.apache.spark.sql.catalyst.catalog.CatalogFunction

val f1 = CatalogFunction(FunctionIdentifier(funcName = "f1"), "java.lang.String", resources = Nil)

import org.apache.spark.sql.catalyst.expressions.{Expression, Stack}

// FIXME What happens when looking up a function with the functionBuilder None in registerFunction?

// Using Stack as ResolveFunctions requires that the function to be resolved is a Generator

// You could roll your own, but that's a demo, isn't it? (don't get too carried away)

spark.sessionState.catalog.registerFunction(

funcDefinition = f1,

overrideIfExists = true,

functionBuilder = Some((children: Seq[Expression]) => Stack(children = Nil)))

import org.apache.spark.sql.catalyst.analysis.UnresolvedGenerator

import org.apache.spark.sql.catalyst.FunctionIdentifier

val ungen = UnresolvedGenerator(name = FunctionIdentifier("f1"), children = Seq.empty)

val plan = t1.select(ungen)

scala> println(plan.numberedTreeString)

00 'Project [unresolvedalias('f1(), None)]

01 +- 'UnresolvedRelation `t1`

val resolvedPlan = ResolveFunctions(plan)

scala> println(resolvedPlan.numberedTreeString)

00 'Project [unresolvedalias(stack(), None)]

01 +- 'UnresolvedRelation `t1`

CAUTION: FIXME

// Example: UnresolvedFunction => a) AggregateWindowFunction with(out) isDistinct, b) AggregateFunction, c) other with(out) isDistinct

val plan = ???

val resolvedPlan = ResolveFunctions(plan)

Resolving grouping__id UnresolvedAttribute, UnresolvedGenerator and UnresolvedFunction Expressions In Entire Query Plan (Applying ResolveFunctions to Logical Plan) — `apply` Method



apply(plan: LogicalPlan): LogicalPlan

1

2

3

4

5

apply(plan: LogicalPlan): LogicalPlan

Note	`apply` is part of Rule Contract to apply a rule to a logical plan.

apply takes a logical plan and transforms each expression (for every logical operator found in the query plan) as follows:

For UnresolvedAttributes with names as groupingid, apply creates a Alias (with a GroupingID child expression and groupingid name).

That case seems mostly for compatibility with Hive as grouping__id attribute name is used by Hive.
For UnresolvedGenerators, apply requests the SessionCatalog to find a Generator function by name.

If some other non-generator function is found for the name, apply fails the analysis phase by reporting an AnalysisException:

[name] is expected to be a generator. However, its class is [className], which is not a generator.

1
2
3
4
5

[name] is expected to be a generator. However, its class is [className], which is not a generator.
For UnresolvedFunctions, apply requests the SessionCatalog to find a function by name.
AggregateWindowFunctions are returned directly or apply fails the analysis phase by reporting an AnalysisException when the UnresolvedFunction has isDistinct flag enabled.

[name] does not support the modifier DISTINCT

1
2
3
4
5

[name] does not support the modifier DISTINCT
AggregateFunctions are wrapped in a AggregateExpression (with Complete aggregate mode)
All other functions are returned directly or apply fails the analysis phase by reporting an AnalysisException when the UnresolvedFunction has isDistinct flag enabled.

[name] does not support the modifier DISTINCT

1
2
3
4
5

[name] does not support the modifier DISTINCT

apply skips unresolved expressions.

ResolveCreateNamedStruct

2013-02-25admin阅读(1471)

ResolveCreateNamedStruct Logical Resolution Rule — Resolving NamePlaceholders In CreateNamedStruct Expressions

ResolveCreateNamedStruct is a logical resolution rule that replaces NamePlaceholders with Literals for the names in CreateNamedStruct expressions in an entire logical query plan.

ResolveCreateNamedStruct is part of the Resolution fixed-point batch in the standard batches of the Analyzer.

ResolveCreateNamedStruct is simply a Catalyst rule for transforming logical plans, i.e. Rule[LogicalPlan].



scala> :type spark
org.apache.spark.sql.SparkSession

val q = spark.range(1).select(struct($"id"))
val logicalPlan = q.queryExecution.logical
scala> println(logicalPlan.numberedTreeString)
00 'Project [unresolvedalias(named_struct(NamePlaceholder, 'id), None)]
01 +- AnalysisBarrier
02       +- Range (0, 1, step=1, splits=Some(8))

// Let's resolve references first
import spark.sessionState.analyzer.ResolveReferences
val planWithRefsResolved = ResolveReferences(logicalPlan)

import org.apache.spark.sql.catalyst.analysis.ResolveCreateNamedStruct
val afterResolveCreateNamedStruct = ResolveCreateNamedStruct(planWithRefsResolved)
scala> println(afterResolveCreateNamedStruct.numberedTreeString)
00 'Project [unresolvedalias(named_struct(id, id#4L), None)]
01 +- AnalysisBarrier
02       +- Range (0, 1, step=1, splits=Some(8))

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

scala> :type spark

org.apache.spark.sql.SparkSession

val q = spark.range(1).select(struct($"id"))

val logicalPlan = q.queryExecution.logical

scala> println(logicalPlan.numberedTreeString)

00 'Project [unresolvedalias(named_struct(NamePlaceholder, 'id), None)]

01 +- AnalysisBarrier

02 +- Range (0, 1, step=1, splits=Some(8))

// Let's resolve references first

import spark.sessionState.analyzer.ResolveReferences

val planWithRefsResolved = ResolveReferences(logicalPlan)

import org.apache.spark.sql.catalyst.analysis.ResolveCreateNamedStruct

val afterResolveCreateNamedStruct = ResolveCreateNamedStruct(planWithRefsResolved)

scala> println(afterResolveCreateNamedStruct.numberedTreeString)

00 'Project [unresolvedalias(named_struct(id, id#4L), None)]

01 +- AnalysisBarrier

02 +- Range (0, 1, step=1, splits=Some(8))

Executing Rule — `apply` Method



apply(plan: LogicalPlan): LogicalPlan

1

2

3

4

5

apply(plan: LogicalPlan): LogicalPlan

Note	`apply` is part of the Rule Contract to execute (apply) a rule on a TreeNode (e.g. LogicalPlan).

apply traverses all Catalyst expressions (in the input LogicalPlan) that are CreateNamedStruct expressions which are not resolved yet and replaces NamePlaceholders with Literal expressions.

In other words, apply finds unresolved CreateNamedStruct expressions with NamePlaceholder expressions in the children and replaces them with the name of corresponding NamedExpression, but only if the NamedExpression is resolved.

In the end, apply creates a CreateNamedStruct with new children.

ResolveCoalesceHints

2013-02-24admin阅读(1555)

ResolveCoalesceHints Logical Resolution Rule — Resolving UnresolvedHint Operators with COALESCE and REPARTITION Hints

ResolveCoalesceHints is a logical resolution rule that the Spark Analyzer uses to resolve UnresolvedHint logical operators with COALESCE or REPARTITION hints (case-insensitive) to ResolvedHint operators.

COALESCE or REPARTITION hints expect a partition number as the only parameter.

Technically, ResolveCoalesceHints is a Catalyst rule for transforming logical plans, i.e. Rule[LogicalPlan].

ResolveCoalesceHints is part of Hints fixed-point batch of rules (that is executed before any other rule).

ResolveCoalesceHints takes no input parameters when created.


Example: Using COALESCE Hint

// Use Catalyst DSL to create a logical plan
import org.apache.spark.sql.catalyst.dsl.plans._
val plan = table("t1").hint(name = "COALESCE", 3)
scala> println(plan.numberedTreeString)
00 'UnresolvedHint COALESCE, [3]
01 +- 'UnresolvedRelation `t1`

import org.apache.spark.sql.catalyst.analysis.ResolveHints.ResolveCoalesceHints
val analyzedPlan = ResolveCoalesceHints(plan)
scala> println(analyzedPlan.numberedTreeString)
00 'Repartition 3, false
01 +- 'UnresolvedRelation `t1`

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

Example: Using COALESCE Hint

// Use Catalyst DSL to create a logical plan

import org.apache.spark.sql.catalyst.dsl.plans._

val plan = table("t1").hint(name = "COALESCE", 3)

scala> println(plan.numberedTreeString)

00 'UnresolvedHint COALESCE, [3]

01 +- 'UnresolvedRelation `t1`

import org.apache.spark.sql.catalyst.analysis.ResolveHints.ResolveCoalesceHints

val analyzedPlan = ResolveCoalesceHints(plan)

scala> println(analyzedPlan.numberedTreeString)

00 'Repartition 3, false

01 +- 'UnresolvedRelation `t1`


Example: Using REPARTITION Hint

// Use Catalyst DSL to create a logical plan
import org.apache.spark.sql.catalyst.dsl.plans._
val plan = table("t1").hint(name = "REPARTITION", 3)
scala> println(plan.numberedTreeString)
00 'UnresolvedHint REPARTITION, [3]
01 +- 'UnresolvedRelation `t1`

import org.apache.spark.sql.catalyst.analysis.ResolveHints.ResolveCoalesceHints
val analyzedPlan = ResolveCoalesceHints(plan)
scala> println(analyzedPlan.numberedTreeString)
00 'Repartition 3, true
01 +- 'UnresolvedRelation `t1`

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

Example: Using REPARTITION Hint

// Use Catalyst DSL to create a logical plan

import org.apache.spark.sql.catalyst.dsl.plans._

val plan = table("t1").hint(name = "REPARTITION", 3)

scala> println(plan.numberedTreeString)

00 'UnresolvedHint REPARTITION, [3]

01 +- 'UnresolvedRelation `t1`

import org.apache.spark.sql.catalyst.analysis.ResolveHints.ResolveCoalesceHints

val analyzedPlan = ResolveCoalesceHints(plan)

scala> println(analyzedPlan.numberedTreeString)

00 'Repartition 3, true

01 +- 'UnresolvedRelation `t1`


Example: Using COALESCE Hint in SQL

val q = sql("SELECT /*+ COALESCE(10) */ * FROM VALUES 1 t(id)")
val plan = q.queryExecution.logical
scala> println(plan.numberedTreeString)
00 'UnresolvedHint COALESCE, [10]
01 +- 'Project [*]
02    +- 'SubqueryAlias `t`
03       +- 'UnresolvedInlineTable [id], [List(1)]

import org.apache.spark.sql.catalyst.analysis.ResolveHints.ResolveCoalesceHints
val analyzedPlan = ResolveCoalesceHints(plan)
scala> println(analyzedPlan.numberedTreeString)
00 'Repartition 10, false
01 +- 'Project [*]
02    +- 'SubqueryAlias `t`
03       +- 'UnresolvedInlineTable [id], [List(1)]

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

Example: Using COALESCE Hint in SQL

val q = sql("SELECT /*+ COALESCE(10) */ * FROM VALUES 1 t(id)")

val plan = q.queryExecution.logical

scala> println(plan.numberedTreeString)

00 'UnresolvedHint COALESCE, [10]

01 +- 'Project [*]

02 +- 'SubqueryAlias `t`

03 +- 'UnresolvedInlineTable [id], [List(1)]

import org.apache.spark.sql.catalyst.analysis.ResolveHints.ResolveCoalesceHints

val analyzedPlan = ResolveCoalesceHints(plan)

scala> println(analyzedPlan.numberedTreeString)

00 'Repartition 10, false

01 +- 'Project [*]

02 +- 'SubqueryAlias `t`

03 +- 'UnresolvedInlineTable [id], [List(1)]

ResolveBroadcastHints

2013-02-23admin阅读(1641)

ResolveBroadcastHints Logical Resolution Rule — Resolving UnresolvedHint Operators with BROADCAST, BROADCASTJOIN and MAPJOIN Hint Names

ResolveBroadcastHints is a logical resolution rule that the Spark Analyzer uses to resolve UnresolvedHint logical operators with BROADCAST, BROADCASTJOIN or MAPJOIN hints (case-insensitive) to ResolvedHint operators.

Technically, ResolveBroadcastHints is a Catalyst rule for transforming logical plans, i.e. Rule[LogicalPlan].

ResolveBroadcastHints is part of Hints fixed-point batch of rules (that is executed before any other rule).

ResolveBroadcastHints takes a SQLConf when created.



// Use Catalyst DSL to create a logical plan
import org.apache.spark.sql.catalyst.dsl.plans._
val plan = table("t1").join(table("t2")).hint(name = "broadcast", "t1", "table2")
scala> println(plan.numberedTreeString)
00 'UnresolvedHint broadcast, [t1, table2]
01 +- 'Join Inner
02    :- 'UnresolvedRelation `t1`
03    +- 'UnresolvedRelation `t2`

import org.apache.spark.sql.catalyst.analysis.ResolveHints.ResolveBroadcastHints
val resolver = new ResolveBroadcastHints(spark.sessionState.conf)
val analyzedPlan = resolver(plan)
scala> println(analyzedPlan.numberedTreeString)
00 'Join Inner
01 :- 'ResolvedHint (broadcast)
02 :  +- 'UnresolvedRelation `t1`
03 +- 'UnresolvedRelation `t2`

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

// Use Catalyst DSL to create a logical plan

import org.apache.spark.sql.catalyst.dsl.plans._

val plan = table("t1").join(table("t2")).hint(name = "broadcast", "t1", "table2")

scala> println(plan.numberedTreeString)

00 'UnresolvedHint broadcast, [t1, table2]

01 +- 'Join Inner

02 :- 'UnresolvedRelation `t1`

03 +- 'UnresolvedRelation `t2`

import org.apache.spark.sql.catalyst.analysis.ResolveHints.ResolveBroadcastHints

val resolver = new ResolveBroadcastHints(spark.sessionState.conf)

val analyzedPlan = resolver(plan)

scala> println(analyzedPlan.numberedTreeString)

00 'Join Inner

01 :- 'ResolvedHint (broadcast)

02 : +- 'UnresolvedRelation `t1`

03 +- 'UnresolvedRelation `t2`

Resolving UnresolvedHint with BROADCAST, BROADCASTJOIN or MAPJOIN Hint Names (Applying ResolveBroadcastHints to Logical Plan) — `apply` Method



apply(plan: LogicalPlan): LogicalPlan

1

2

3

4

5

apply(plan: LogicalPlan): LogicalPlan

Note	`apply` is part of Rule Contract to apply a rule to a logical plan.

apply transforms UnresolvedHint operators into ResolvedHint for the hint names as BROADCAST, BROADCASTJOIN or MAPJOIN (case-insensitive).

For UnresolvedHints with no parameters, apply marks the entire child logical plan as eligible for broadcast, i.e. creates a ResolvedHint with the child operator and HintInfo with broadcast flag on.

For UnresolvedHints with parameters defined, apply considers the parameters the names of the tables to apply broadcast hint to.

Note	The table names can be of `String` or `UnresolvedAttribute` types.

apply reports an AnalysisException for the parameters that are not of String or UnresolvedAttribute types.



org.apache.spark.sql.AnalysisException: Broadcast hint parameter should be an identifier or string but was [unsupported] ([className]

1

2

3

4

5

org.apache.spark.sql.AnalysisException: Broadcast hint parameter should be an identifier or string but was [unsupported] ([className]



// Use Catalyst DSL to create a logical plan
import org.apache.spark.sql.catalyst.dsl.plans._
// !!! IT WON'T WORK !!!
// 1 is not a table name or of type `UnresolvedAttribute`
val plan = table("t1").hint(name = "broadcast", 1)

scala> println(plan.numberedTreeString)
00 'UnresolvedHint broadcast, [1]
01 +- 'UnresolvedRelation `t1`

// Resolve hints
import org.apache.spark.sql.catalyst.analysis.ResolveHints
val broadcastHintResolver = new ResolveHints.ResolveBroadcastHints(spark.sessionState.conf)
scala> broadcastHintResolver(plan)
org.apache.spark.sql.AnalysisException: Broadcast hint parameter should be an identifier or string but was 1 (class java.lang.Integer;
  at org.apache.spark.sql.catalyst.analysis.ResolveHints$ResolveBroadcastHints$$anonfun$apply$1$$anonfun$applyOrElse$1.apply(ResolveHints.scala:98)
  at org.apache.spark.sql.catalyst.analysis.ResolveHints$ResolveBroadcastHints$$anonfun$apply$1$$anonfun$applyOrElse$1.apply(ResolveHints.scala:95)
  at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
  at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
  at scala.collection.IndexedSeqOptimized$class.foreach(IndexedSeqOptimized.scala:33)
  at scala.collection.mutable.WrappedArray.foreach(WrappedArray.scala:35)
  at scala.collection.TraversableLike$class.map(TraversableLike.scala:234)
  at scala.collection.AbstractTraversable.map(Traversable.scala:104)
  at org.apache.spark.sql.catalyst.analysis.ResolveHints$ResolveBroadcastHints$$anonfun$apply$1.applyOrElse(ResolveHints.scala:95)
  at org.apache.spark.sql.catalyst.analysis.ResolveHints$ResolveBroadcastHints$$anonfun$apply$1.applyOrElse(ResolveHints.scala:88)
  at org.apache.spark.sql.catalyst.trees.TreeNode$$anonfun$transformUp$1.apply(TreeNode.scala:289)
  at org.apache.spark.sql.catalyst.trees.TreeNode$$anonfun$transformUp$1.apply(TreeNode.scala:289)
  at org.apache.spark.sql.catalyst.trees.CurrentOrigin$.withOrigin(TreeNode.scala:70)
  at org.apache.spark.sql.catalyst.trees.TreeNode.transformUp(TreeNode.scala:288)
  at org.apache.spark.sql.catalyst.analysis.ResolveHints$ResolveBroadcastHints.apply(ResolveHints.scala:88)
  ... 51 elided

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

// Use Catalyst DSL to create a logical plan

import org.apache.spark.sql.catalyst.dsl.plans._

// !!! IT WON'T WORK !!!

// 1 is not a table name or of type `UnresolvedAttribute`

val plan = table("t1").hint(name = "broadcast", 1)

scala> println(plan.numberedTreeString)

00 'UnresolvedHint broadcast, [1]

01 +- 'UnresolvedRelation `t1`

// Resolve hints

import org.apache.spark.sql.catalyst.analysis.ResolveHints

val broadcastHintResolver = new ResolveHints.ResolveBroadcastHints(spark.sessionState.conf)

scala> broadcastHintResolver(plan)

org.apache.spark.sql.AnalysisException: Broadcast hint parameter should be an identifier or string but was 1 (class java.lang.Integer;

at org.apache.spark.sql.catalyst.analysis.ResolveHints$ResolveBroadcastHints$$anonfun$apply$1$$anonfun$applyOrElse$1.apply(ResolveHints.scala:98)

at org.apache.spark.sql.catalyst.analysis.ResolveHints$ResolveBroadcastHints$$anonfun$apply$1$$anonfun$applyOrElse$1.apply(ResolveHints.scala:95)

at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)

at scala.collection.IndexedSeqOptimized$class.foreach(IndexedSeqOptimized.scala:33)

at scala.collection.mutable.WrappedArray.foreach(WrappedArray.scala:35)

at scala.collection.TraversableLike$class.map(TraversableLike.scala:234)

at scala.collection.AbstractTraversable.map(Traversable.scala:104)

at org.apache.spark.sql.catalyst.analysis.ResolveHints$ResolveBroadcastHints$$anonfun$apply$1.applyOrElse(ResolveHints.scala:95)

at org.apache.spark.sql.catalyst.analysis.ResolveHints$ResolveBroadcastHints$$anonfun$apply$1.applyOrElse(ResolveHints.scala:88)

at org.apache.spark.sql.catalyst.trees.TreeNode$$anonfun$transformUp$1.apply(TreeNode.scala:289)

at org.apache.spark.sql.catalyst.trees.CurrentOrigin$.withOrigin(TreeNode.scala:70)

at org.apache.spark.sql.catalyst.trees.TreeNode.transformUp(TreeNode.scala:288)

at org.apache.spark.sql.catalyst.analysis.ResolveHints$ResolveBroadcastHints.apply(ResolveHints.scala:88)

... 51 elided

`applyBroadcastHint` Internal Method



applyBroadcastHint(plan: LogicalPlan, toBroadcast: Set[String]): LogicalPlan

1

2

3

4

5

applyBroadcastHint(plan: LogicalPlan, toBroadcast: Set[String]): LogicalPlan

applyBroadcastHint…FIXME

Note	`applyBroadcastHint` is used exclusively when `ResolveBroadcastHints` is requested to execute.

spark-sql 第16页

ResolveRelations Logical Resolution Rule — Resolving UnresolvedRelations With Tables in Catalog

Applying ResolveRelations to Logical Plan — apply Method

Resolving Relation — resolveRelation Method

isRunningDirectlyOnFiles Internal Method

Finding Table in Session-Scoped Catalog of Relational Entities — lookupTableFromCatalog Internal Method

ResolveReferences Logical Resolution Rule

Resolving Expressions of Logical Plan — resolve Internal Method

Resolving Reference Expressions In Logical Query Plan (Applying ResolveReferences to Logical Plan) — apply Method

Expanding Star Expressions — buildExpandedProjectList Internal Method

expandStarExpression Method

dedupRight Internal Method

dedupOuterReferencesInSubquery Internal Method

ResolveOrdinalInOrderByAndGroupBy Logical Resolution Rule

Executing Rule — apply Method

ResolveMissingReferences

resolveExprsAndAddMissingAttrs Internal Method

ResolveInlineTables Logical Resolution Rule

Executing Rule — apply Method

validateInputDimension Internal Method

validateInputEvaluable Internal Method

Converting UnresolvedInlineTable to LocalRelation — convert Internal Method

ResolveHiveSerdeTable Logical Resolution Rule

Applying ResolveHiveSerdeTable Rule to Logical Plan — apply Method

ResolveFunctions Logical Resolution Rule — Resolving grouping__id UnresolvedAttribute, UnresolvedGenerator And UnresolvedFunction Expressions

Resolving grouping__id UnresolvedAttribute, UnresolvedGenerator and UnresolvedFunction Expressions In Entire Query Plan (Applying ResolveFunctions to Logical Plan) — apply Method

ResolveCreateNamedStruct Logical Resolution Rule — Resolving NamePlaceholders In CreateNamedStruct Expressions

Executing Rule — apply Method

ResolveCoalesceHints Logical Resolution Rule — Resolving UnresolvedHint Operators with COALESCE and REPARTITION Hints

ResolveBroadcastHints Logical Resolution Rule — Resolving UnresolvedHint Operators with BROADCAST, BROADCASTJOIN and MAPJOIN Hint Names

Resolving UnresolvedHint with BROADCAST, BROADCASTJOIN or MAPJOIN Hint Names (Applying ResolveBroadcastHints to Logical Plan) — apply Method

applyBroadcastHint Internal Method

欢迎关注：spark技术分享

关注公众号：spark技术分享

QQ咨询

回顶部

Applying ResolveRelations to Logical Plan — `apply` Method

Resolving Relation — `resolveRelation` Method

`isRunningDirectlyOnFiles` Internal Method

Finding Table in Session-Scoped Catalog of Relational Entities — `lookupTableFromCatalog` Internal Method

Resolving Expressions of Logical Plan — `resolve` Internal Method

Resolving Reference Expressions In Logical Query Plan (Applying ResolveReferences to Logical Plan) — `apply` Method

Expanding Star Expressions — `buildExpandedProjectList` Internal Method

`expandStarExpression` Method

`dedupRight` Internal Method

`dedupOuterReferencesInSubquery` Internal Method

Executing Rule — `apply` Method

`resolveExprsAndAddMissingAttrs` Internal Method

Executing Rule — `apply` Method

`validateInputDimension` Internal Method

`validateInputEvaluable` Internal Method

Converting UnresolvedInlineTable to LocalRelation — `convert` Internal Method

Applying ResolveHiveSerdeTable Rule to Logical Plan — `apply` Method

Resolving grouping__id UnresolvedAttribute, UnresolvedGenerator and UnresolvedFunction Expressions In Entire Query Plan (Applying ResolveFunctions to Logical Plan) — `apply` Method

Executing Rule — `apply` Method

Resolving UnresolvedHint with BROADCAST, BROADCASTJOIN or MAPJOIN Hint Names (Applying ResolveBroadcastHints to Logical Plan) — `apply` Method

`applyBroadcastHint` Internal Method