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

ColumnPruning

2013-03-14admin阅读(1532)

ColumnPruning Logical Optimization

ColumnPruning is a base logical optimization that FIXME.

ColumnPruning is part of the RewriteSubquery once-executed batch in the standard batches of the Catalyst Optimizer.

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

Example 1



val dataset = spark.range(10).withColumn("bucket", 'id % 3)

import org.apache.spark.sql.expressions.Window
val rankCol = rank over Window.partitionBy('bucket).orderBy('id) as "rank"

val ranked = dataset.withColumn("rank", rankCol)

scala> ranked.explain(true)
...
TRACE SparkOptimizer:
=== Applying Rule org.apache.spark.sql.catalyst.optimizer.ColumnPruning ===
 Project [id#73L, bucket#76L, rank#192]                                                                                                                              Project [id#73L, bucket#76L, rank#192]
!+- Project [id#73L, bucket#76L, rank#82, rank#82 AS rank#192]                                                                                                       +- Project [id#73L, bucket#76L, rank#82 AS rank#192]
    +- Window [rank(id#73L) windowspecdefinition(bucket#76L, id#73L ASC, ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS rank#82], [bucket#76L], [id#73L ASC]      +- Window [rank(id#73L) windowspecdefinition(bucket#76L, id#73L ASC, ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS rank#82], [bucket#76L], [id#73L ASC]
!      +- Project [id#73L, bucket#76L]                                                                                                                                     +- Project [id#73L, (id#73L % cast(3 as bigint)) AS bucket#76L]
!         +- Project [id#73L, (id#73L % cast(3 as bigint)) AS bucket#76L]                                                                                                     +- Range (0, 10, step=1, splits=Some(8))
!            +- Range (0, 10, step=1, splits=Some(8))
...
TRACE SparkOptimizer: Fixed point reached for batch Operator Optimizations after 2 iterations.
DEBUG SparkOptimizer:
=== Result of Batch Operator Optimizations ===
!Project [id#73L, bucket#76L, rank#192]                                                                                                                              Window [rank(id#73L) windowspecdefinition(bucket#76L, id#73L ASC, ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS rank#82], [bucket#76L], [id#73L ASC]
!+- Project [id#73L, bucket#76L, rank#82, rank#82 AS rank#192]                                                                                                       +- Project [id#73L, (id#73L % 3) AS bucket#76L]
!   +- Window [rank(id#73L) windowspecdefinition(bucket#76L, id#73L ASC, ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS rank#82], [bucket#76L], [id#73L ASC]      +- Range (0, 10, step=1, splits=Some(8))
!      +- Project [id#73L, bucket#76L]
!         +- Project [id#73L, (id#73L % cast(3 as bigint)) AS bucket#76L]
!            +- Range (0, 10, 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

25

26

27

28

29

30

31

32

val dataset = spark.range(10).withColumn("bucket", 'id % 3)

import org.apache.spark.sql.expressions.Window

val rankCol = rank over Window.partitionBy('bucket).orderBy('id) as "rank"

val ranked = dataset.withColumn("rank", rankCol)

scala> ranked.explain(true)

...

TRACE SparkOptimizer:

=== Applying Rule org.apache.spark.sql.catalyst.optimizer.ColumnPruning ===

Project [id#73L, bucket#76L, rank#192] Project [id#73L, bucket#76L, rank#192]

!+- Project [id#73L, bucket#76L, rank#82, rank#82 AS rank#192] +- Project [id#73L, bucket#76L, rank#82 AS rank#192]

+- Window [rank(id#73L) windowspecdefinition(bucket#76L, id#73L ASC, ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS rank#82], [bucket#76L], [id#73L ASC] +- Window [rank(id#73L) windowspecdefinition(bucket#76L, id#73L ASC, ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS rank#82], [bucket#76L], [id#73L ASC]

! +- Project [id#73L, bucket#76L] +- Project [id#73L, (id#73L % cast(3 as bigint)) AS bucket#76L]

! +- Project [id#73L, (id#73L % cast(3 as bigint)) AS bucket#76L] +- Range (0, 10, step=1, splits=Some(8))

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

...

TRACE SparkOptimizer: Fixed point reached for batch Operator Optimizations after 2 iterations.

DEBUG SparkOptimizer:

=== Result of Batch Operator Optimizations ===

!Project [id#73L, bucket#76L, rank#192] Window [rank(id#73L) windowspecdefinition(bucket#76L, id#73L ASC, ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS rank#82], [bucket#76L], [id#73L ASC]

!+- Project [id#73L, bucket#76L, rank#82, rank#82 AS rank#192] +- Project [id#73L, (id#73L % 3) AS bucket#76L]

! +- Window [rank(id#73L) windowspecdefinition(bucket#76L, id#73L ASC, ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS rank#82], [bucket#76L], [id#73L ASC] +- Range (0, 10, step=1, splits=Some(8))

! +- Project [id#73L, bucket#76L]

! +- Project [id#73L, (id#73L % cast(3 as bigint)) AS bucket#76L]

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

...

Example 2



// the business object
case class Person(id: Long, name: String, city: String)

// the dataset to query over
val dataset = Seq(Person(0, "Jacek", "Warsaw")).toDS

// the query
// Note that we work with names only (out of 3 attributes in Person)
val query = dataset.groupBy(upper('name) as 'name).count

scala> query.explain(extended = true)
...
TRACE SparkOptimizer:
=== Applying Rule org.apache.spark.sql.catalyst.optimizer.ColumnPruning ===
 Aggregate [upper(name#126)], [upper(name#126) AS name#160, count(1) AS count#166L]   Aggregate [upper(name#126)], [upper(name#126) AS name#160, count(1) AS count#166L]
!+- LocalRelation [id#125L, name#126, city#127]                                       +- Project [name#126]
!                                                                                        +- LocalRelation [id#125L, name#126, city#127]
...
== Parsed Logical Plan ==
'Aggregate [upper('name) AS name#160], [upper('name) AS name#160, count(1) AS count#166L]
+- LocalRelation [id#125L, name#126, city#127]

== Analyzed Logical Plan ==
name: string, count: bigint
Aggregate [upper(name#126)], [upper(name#126) AS name#160, count(1) AS count#166L]
+- LocalRelation [id#125L, name#126, city#127]

== Optimized Logical Plan ==
Aggregate [upper(name#126)], [upper(name#126) AS name#160, count(1) AS count#166L]
+- LocalRelation [name#126]

== Physical Plan ==
*HashAggregate(keys=[upper(name#126)#171], functions=[count(1)], output=[name#160, count#166L])
+- Exchange hashpartitioning(upper(name#126)#171, 200)
   +- *HashAggregate(keys=[upper(name#126) AS upper(name#126)#171], functions=[partial_count(1)], output=[upper(name#126)#171, count#173L])
      +- LocalTableScan [name#126]

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

// the business object

case class Person(id: Long, name: String, city: String)

// the dataset to query over

val dataset = Seq(Person(0, "Jacek", "Warsaw")).toDS

// the query

// Note that we work with names only (out of 3 attributes in Person)

val query = dataset.groupBy(upper('name) as 'name).count

scala> query.explain(extended = true)

...

TRACE SparkOptimizer:

=== Applying Rule org.apache.spark.sql.catalyst.optimizer.ColumnPruning ===

Aggregate [upper(name#126)], [upper(name#126) AS name#160, count(1) AS count#166L] Aggregate [upper(name#126)], [upper(name#126) AS name#160, count(1) AS count#166L]

!+- LocalRelation [id#125L, name#126, city#127] +- Project [name#126]

! +- LocalRelation [id#125L, name#126, city#127]

...

== Parsed Logical Plan ==

'Aggregate [upper('name) AS name#160], [upper('name) AS name#160, count(1) AS count#166L]

+- LocalRelation [id#125L, name#126, city#127]

== Analyzed Logical Plan ==

name: string, count: bigint

Aggregate [upper(name#126)], [upper(name#126) AS name#160, count(1) AS count#166L]

+- LocalRelation [id#125L, name#126, city#127]

== Optimized Logical Plan ==

Aggregate [upper(name#126)], [upper(name#126) AS name#160, count(1) AS count#166L]

+- LocalRelation [name#126]

== Physical Plan ==

*HashAggregate(keys=[upper(name#126)#171], functions=[count(1)], output=[name#160, count#166L])

+- Exchange hashpartitioning(upper(name#126)#171, 200)

+- *HashAggregate(keys=[upper(name#126) AS upper(name#126)#171], functions=[partial_count(1)], output=[upper(name#126)#171, count#173L])

+- LocalTableScan [name#126]

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…FIXME

CollapseWindow

2013-03-13admin阅读(1481)

CollapseWindow Logical Optimization

CollapseWindow is a base logical optimization that FIXME.

CollapseWindow is part of the Operator Optimization fixed-point batch in the standard batches of the Catalyst Optimizer.

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



// FIXME: DEMO
import org.apache.spark.sql.catalyst.optimizer.CollapseWindow

val logicalPlan = ???
val afterCollapseWindow = CollapseWindow(logicalPlan)

1

2

3

4

5

6

7

8

9

// FIXME: DEMO

import org.apache.spark.sql.catalyst.optimizer.CollapseWindow

val logicalPlan = ???

val afterCollapseWindow = CollapseWindow(logicalPlan)

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…FIXME

WindowsSubstitution

2013-03-12admin阅读(1587)

WindowsSubstitution Logical Evaluation Rule

WindowsSubstitution is a logical evaluation rule (i.e. Rule[LogicalPlan]) that the logical query plan analyzer uses to resolve (aka substitute) WithWindowDefinition unary logical operators with UnresolvedWindowExpression to their corresponding WindowExpression with resolved WindowSpecDefinition.

WindowsSubstitution is part of Substitution fixed-point batch of rules.

Note	It appears that `WindowsSubstitution` is exclusively used for pure SQL queries because WithWindowDefinition unary logical operator is created exclusively when `AstBuilder` parses window definitions.

If a window specification is not found, WindowsSubstitution fails analysis with the following error:



Window specification [windowName] is not defined in the WINDOW clause.

1

2

3

4

5

Window specification [windowName] is not defined in the WINDOW clause.

Note	The analysis failure is unlikely to happen given `AstBuilder` builds a lookup table of all the named window specifications defined in a SQL text and reports a `ParseException` when a `WindowSpecReference` is not available earlier.

For every WithWindowDefinition, WindowsSubstitution takes the child logical plan and transforms its UnresolvedWindowExpression expressions to be a WindowExpression with a window specification from the WINDOW clause (see WithWindowDefinition Example).

WindowFrameCoercion

2013-03-11admin阅读(1450)

WindowFrameCoercion Type Coercion Logical Rule

WindowFrameCoercion is a type coercion logical rule that cast the data types of the boundaries of a range window frame to the data type of the order specification in a WindowSpecDefinition in a logical plan.



import java.time.LocalDate
import java.sql.Timestamp
val sales = Seq(
  (Timestamp.valueOf(LocalDate.of(2018, 9, 1).atStartOfDay), 5),
  (Timestamp.valueOf(LocalDate.of(2018, 9, 2).atStartOfDay), 10),
  // Mind the 2-day gap
  (Timestamp.valueOf(LocalDate.of(2018, 9, 5).atStartOfDay), 5)
).toDF("time", "volume")
scala> sales.show
+-------------------+------+
|               time|volume|
+-------------------+------+
|2018-09-01 00:00:00|     5|
|2018-09-02 00:00:00|    10|
|2018-09-05 00:00:00|     5|
+-------------------+------+

scala> sales.printSchema
root
 |-- time: timestamp (nullable = true)
 |-- volume: integer (nullable = false)

// FIXME Use Catalyst DSL
// rangeBetween with column expressions
// data type of orderBy expression is date
// data types of range frame boundaries is interval
// WindowSpecDefinition(_, Seq(order), SpecifiedWindowFrame(RangeFrame, lower, upper))
import org.apache.spark.unsafe.types.CalendarInterval
val interval = lit(CalendarInterval.fromString("interval 1 days"))
import org.apache.spark.sql.expressions.Window
val windowSpec = Window.orderBy($"time").rangeBetween(currentRow(), interval)

val q = sales.select(
  $"time",
  (sum($"volume") over windowSpec) as "sum",
  (count($"volume") over windowSpec) as "count")
val plan = q.queryExecution.logical
scala> println(plan.numberedTreeString)
00 'Project [unresolvedalias('time, None), sum('volume) windowspecdefinition('time ASC NULLS FIRST, specifiedwindowframe(RangeFrame, currentrow$(), interval 1 days)) AS sum#156, count('volume) windowspecdefinition('time ASC NULLS FIRST, specifiedwindowframe(RangeFrame, currentrow$(), interval 1 days)) AS count#158]
01 +- AnalysisBarrier
02       +- Project [_1#129 AS time#132, _2#130 AS volume#133]
03          +- LocalRelation [_1#129, _2#130]

import spark.sessionState.analyzer.ResolveReferences
val planWithRefsResolved = ResolveReferences(plan)

import spark.sessionState.analyzer.ResolveAliases
val planWithAliasesResolved = ResolveReferences(planWithRefsResolved)

// FIXME Looks like nothing changes in the query plan with regard to WindowFrameCoercion

import org.apache.spark.sql.catalyst.analysis.TypeCoercion.WindowFrameCoercion
val afterWindowFrameCoercion = WindowFrameCoercion(planWithRefsResolved)
scala> println(afterWindowFrameCoercion.numberedTreeString)
00 'Project [unresolvedalias(time#132, None), sum(volume#133) windowspecdefinition(time#132 ASC NULLS FIRST, specifiedwindowframe(RangeFrame, currentrow$(), interval 1 days)) AS sum#156L, count(volume#133) windowspecdefinition(time#132 ASC NULLS FIRST, specifiedwindowframe(RangeFrame, currentrow$(), interval 1 days)) AS count#158L]
01 +- AnalysisBarrier
02       +- Project [_1#129 AS time#132, _2#130 AS volume#133]
03          +- LocalRelation [_1#129, _2#130]

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

import java.time.LocalDate

import java.sql.Timestamp

val sales = Seq(

(Timestamp.valueOf(LocalDate.of(2018, 9, 1).atStartOfDay), 5),

(Timestamp.valueOf(LocalDate.of(2018, 9, 2).atStartOfDay), 10),

// Mind the 2-day gap

(Timestamp.valueOf(LocalDate.of(2018, 9, 5).atStartOfDay), 5)

).toDF("time", "volume")

scala> sales.show

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

| time|volume|

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

|2018-09-01 00:00:00| 5|

|2018-09-02 00:00:00| 10|

|2018-09-05 00:00:00| 5|

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

scala> sales.printSchema

root

|-- time: timestamp (nullable = true)

|-- volume: integer (nullable = false)

// FIXME Use Catalyst DSL

// rangeBetween with column expressions

// data type of orderBy expression is date

// data types of range frame boundaries is interval

// WindowSpecDefinition(_, Seq(order), SpecifiedWindowFrame(RangeFrame, lower, upper))

import org.apache.spark.unsafe.types.CalendarInterval

val interval = lit(CalendarInterval.fromString("interval 1 days"))

import org.apache.spark.sql.expressions.Window

val windowSpec = Window.orderBy($"time").rangeBetween(currentRow(), interval)

val q = sales.select(

$"time",

(sum($"volume") over windowSpec) as "sum",

(count($"volume") over windowSpec) as "count")

val plan = q.queryExecution.logical

scala> println(plan.numberedTreeString)

00 'Project [unresolvedalias('time, None), sum('volume) windowspecdefinition('time ASC NULLS FIRST, specifiedwindowframe(RangeFrame, currentrow$(), interval 1 days)) AS sum#156, count('volume) windowspecdefinition('time ASC NULLS FIRST, specifiedwindowframe(RangeFrame, currentrow$(), interval 1 days)) AS count#158]

01 +- AnalysisBarrier

02 +- Project [_1#129 AS time#132, _2#130 AS volume#133]

03 +- LocalRelation [_1#129, _2#130]

import spark.sessionState.analyzer.ResolveReferences

val planWithRefsResolved = ResolveReferences(plan)

import spark.sessionState.analyzer.ResolveAliases

val planWithAliasesResolved = ResolveReferences(planWithRefsResolved)

// FIXME Looks like nothing changes in the query plan with regard to WindowFrameCoercion

import org.apache.spark.sql.catalyst.analysis.TypeCoercion.WindowFrameCoercion

val afterWindowFrameCoercion = WindowFrameCoercion(planWithRefsResolved)

scala> println(afterWindowFrameCoercion.numberedTreeString)

00 'Project [unresolvedalias(time#132, None), sum(volume#133) windowspecdefinition(time#132 ASC NULLS FIRST, specifiedwindowframe(RangeFrame, currentrow$(), interval 1 days)) AS sum#156L, count(volume#133) windowspecdefinition(time#132 ASC NULLS FIRST, specifiedwindowframe(RangeFrame, currentrow$(), interval 1 days)) AS count#158L]

01 +- AnalysisBarrier

02 +- Project [_1#129 AS time#132, _2#130 AS volume#133]

03 +- LocalRelation [_1#129, _2#130]



import java.time.LocalDate
import java.sql.Date
val sales = Seq(
  (Date.valueOf(LocalDate.of(2018, 9, 1)), 5),
  (Date.valueOf(LocalDate.of(2018, 9, 2)), 10),
  // Mind the 2-day gap
  (Date.valueOf(LocalDate.of(2018, 9, 5)), 5)
).toDF("time", "volume")
scala> sales.show
+----------+------+
|      time|volume|
+----------+------+
|2018-09-01|     5|
|2018-09-02|    10|
|2018-09-05|     5|
+----------+------+

scala> sales.printSchema
root
 |-- time: date (nullable = true)
 |-- volume: integer (nullable = false)

// FIXME Use Catalyst DSL
// rangeBetween with column expressions
// data type of orderBy expression is date
// WindowSpecDefinition(_, Seq(order), SpecifiedWindowFrame(RangeFrame, lower, upper))
import org.apache.spark.sql.expressions.Window
val windowSpec = Window.orderBy($"time").rangeBetween(currentRow(), lit(1))

val q = sales.select(
  $"time",
  (sum($"volume") over windowSpec) as "sum")
val plan = q.queryExecution.logical
scala> println(plan.numberedTreeString)
00 'Project [unresolvedalias('time, None), sum('volume) windowspecdefinition('time ASC NULLS FIRST, specifiedwindowframe(RangeFrame, currentrow$(), 1)) AS sum#238]
01 +- AnalysisBarrier
02       +- Project [_1#222 AS time#225, _2#223 AS volume#226]
03          +- LocalRelation [_1#222, _2#223]

import spark.sessionState.analyzer.ResolveReferences
val planWithRefsResolved = ResolveReferences(plan)

import spark.sessionState.analyzer.ResolveAliases
val planWithAliasesResolved = ResolveReferences(planWithRefsResolved)

// FIXME Looks like nothing changes in the query plan with regard to WindowFrameCoercion

import org.apache.spark.sql.catalyst.analysis.TypeCoercion.WindowFrameCoercion
val afterWindowFrameCoercion = WindowFrameCoercion(planWithAliasesResolved)
scala> println(afterWindowFrameCoercion.numberedTreeString)
00 'Project [unresolvedalias(time#132, None), sum(volume#133) windowspecdefinition(time#132 ASC NULLS FIRST, specifiedwindowframe(RangeFrame, currentrow$(), interval 1 days)) AS sum#156L, count(volume#133) windowspecdefinition(time#132 ASC NULLS FIRST, specifiedwindowframe(RangeFrame, currentrow$(), interval 1 days)) AS count#158L]
01 +- AnalysisBarrier
02       +- Project [_1#129 AS time#132, _2#130 AS volume#133]
03          +- LocalRelation [_1#129, _2#130]

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

import java.time.LocalDate

import java.sql.Date

val sales = Seq(

(Date.valueOf(LocalDate.of(2018, 9, 1)), 5),

(Date.valueOf(LocalDate.of(2018, 9, 2)), 10),

// Mind the 2-day gap

(Date.valueOf(LocalDate.of(2018, 9, 5)), 5)

).toDF("time", "volume")

scala> sales.show

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

| time|volume|

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

|2018-09-01| 5|

|2018-09-02| 10|

|2018-09-05| 5|

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

scala> sales.printSchema

root

|-- time: date (nullable = true)

|-- volume: integer (nullable = false)

// FIXME Use Catalyst DSL

// rangeBetween with column expressions

// data type of orderBy expression is date

// WindowSpecDefinition(_, Seq(order), SpecifiedWindowFrame(RangeFrame, lower, upper))

import org.apache.spark.sql.expressions.Window

val windowSpec = Window.orderBy($"time").rangeBetween(currentRow(), lit(1))

val q = sales.select(

$"time",

(sum($"volume") over windowSpec) as "sum")

val plan = q.queryExecution.logical

scala> println(plan.numberedTreeString)

00 'Project [unresolvedalias('time, None), sum('volume) windowspecdefinition('time ASC NULLS FIRST, specifiedwindowframe(RangeFrame, currentrow$(), 1)) AS sum#238]

01 +- AnalysisBarrier

02 +- Project [_1#222 AS time#225, _2#223 AS volume#226]

03 +- LocalRelation [_1#222, _2#223]

import spark.sessionState.analyzer.ResolveReferences

val planWithRefsResolved = ResolveReferences(plan)

import spark.sessionState.analyzer.ResolveAliases

val planWithAliasesResolved = ResolveReferences(planWithRefsResolved)

// FIXME Looks like nothing changes in the query plan with regard to WindowFrameCoercion

import org.apache.spark.sql.catalyst.analysis.TypeCoercion.WindowFrameCoercion

val afterWindowFrameCoercion = WindowFrameCoercion(planWithAliasesResolved)

scala> println(afterWindowFrameCoercion.numberedTreeString)

00 'Project [unresolvedalias(time#132, None), sum(volume#133) windowspecdefinition(time#132 ASC NULLS FIRST, specifiedwindowframe(RangeFrame, currentrow$(), interval 1 days)) AS sum#156L, count(volume#133) windowspecdefinition(time#132 ASC NULLS FIRST, specifiedwindowframe(RangeFrame, currentrow$(), interval 1 days)) AS count#158L]

01 +- AnalysisBarrier

02 +- Project [_1#129 AS time#132, _2#130 AS volume#133]

03 +- LocalRelation [_1#129, _2#130]

Coercing Types in Logical Plan — `coerceTypes` Method



coerceTypes(plan: LogicalPlan): LogicalPlan

1

2

3

4

5

coerceTypes(plan: LogicalPlan): LogicalPlan

Note	`coerceTypes` is part of the TypeCoercionRule Contract to coerce types in a logical plan.

coerceTypes traverses all Catalyst expressions (in the input LogicalPlan) and replaces the frameSpecification of every WindowSpecDefinition with a RangeFrame window frame and the single order specification expression resolved with the lower and upper window frame boundary expressions cast to the data type of the order specification expression.

`createBoundaryCast` Internal Method



createBoundaryCast(boundary: Expression, dt: DataType): Expression

1

2

3

4

5

createBoundaryCast(boundary: Expression, dt: DataType): Expression

createBoundaryCast returns a Catalyst expression per the input boundary Expression and the dt DataType (in the order of execution):

The input boundary expression if it is a SpecialFrameBoundary
The input boundary expression if the dt data type is DateType or TimestampType
Cast unary operator with the input boundary expression and the dt data type if the result type of the boundary expression is not the dt data type, but the result type can be cast to the dt data type
The input boundary expression

Note	`createBoundaryCast` is used exclusively when `WindowFrameCoercion` type coercion logical rule is requested to coerceTypes.

UpdateOuterReferences

2013-03-10admin阅读(1284)

UpdateOuterReferences Logical Rule

UpdateOuterReferences is…FIXME

Applying UpdateOuterReferences 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

TimeWindowing

2013-03-09admin阅读(1376)

TimeWindowing Logical Resolution Rule

TimeWindowing is a logical resolution rule that FIXME in a logical query plan.

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

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



// FIXME: DEMO

1

2

3

4

5

// FIXME: DEMO

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…FIXME

ResolveWindowOrder

2013-03-08admin阅读(2289)

ResolveWindowOrder Logical Resolution Rule

ResolveWindowOrder is…FIXME

ResolveWindowFrame

2013-03-07admin阅读(1609)

ResolveWindowFrame Logical Resolution Rule

ResolveWindowFrame is a logical resolution rule that the logical query plan analyzer uses to validate and resolve WindowExpression expressions in an entire logical query plan.

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

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

ResolveWindowFrame takes a logical plan and does the following:

Makes sure that the window frame of a WindowFunction is unspecified or matches the SpecifiedWindowFrame of the WindowSpecDefinition expression.

Reports a AnalysisException when the frames do not match:

Window Frame [f] must match the required frame [frame]

1
2
3
4
5

Window Frame [f] must match the required frame [frame]
Copies the frame specification of WindowFunction to WindowSpecDefinition
Creates a new SpecifiedWindowFrame for WindowExpression with the resolved Catalyst expression and UnspecifiedFrame

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



import import org.apache.spark.sql.expressions.Window
// cume_dist requires ordered windows
val q = spark.
  range(5).
  withColumn("cume_dist", cume_dist() over Window.orderBy("id"))
import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
val planBefore: LogicalPlan = q.queryExecution.logical

// Before ResolveWindowFrame
scala> println(planBefore.numberedTreeString)
00 'Project [*, cume_dist() windowspecdefinition('id ASC NULLS FIRST, UnspecifiedFrame) AS cume_dist#39]
01 +- Range (0, 5, step=1, splits=Some(8))

import spark.sessionState.analyzer.ResolveWindowFrame
val planAfter = ResolveWindowFrame.apply(plan)

// After ResolveWindowFrame
scala> println(planAfter.numberedTreeString)
00 'Project [*, cume_dist() windowspecdefinition('id ASC NULLS FIRST, RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS cume_dist#31]
01 +- Range (0, 5, 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

import import org.apache.spark.sql.expressions.Window

// cume_dist requires ordered windows

val q = spark.

range(5).

withColumn("cume_dist", cume_dist() over Window.orderBy("id"))

import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan

val planBefore: LogicalPlan = q.queryExecution.logical

// Before ResolveWindowFrame

scala> println(planBefore.numberedTreeString)

00 'Project [*, cume_dist() windowspecdefinition('id ASC NULLS FIRST, UnspecifiedFrame) AS cume_dist#39]

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

import spark.sessionState.analyzer.ResolveWindowFrame

val planAfter = ResolveWindowFrame.apply(plan)

// After ResolveWindowFrame

scala> println(planAfter.numberedTreeString)

00 'Project [*, cume_dist() windowspecdefinition('id ASC NULLS FIRST, RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS cume_dist#31]

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

Applying ResolveWindowFrame 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

ResolveSubquery

2013-03-06admin阅读(1563)

ResolveSubquery Logical Resolution Rule

ResolveSubquery is a logical resolution that resolves subquery expressions (ScalarSubquery, Exists and In) when transforming a logical plan with the following logical operators:

Filter operators with an Aggregate child operator
Unary operators with the children resolved

ResolveSubquery is part of Resolution fixed-point batch of rules of the Spark Analyzer.

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



// Use Catalyst DSL
import org.apache.spark.sql.catalyst.expressions._
val a = 'a.int

import org.apache.spark.sql.catalyst.plans.logical.LocalRelation
val rel = LocalRelation(a)

import org.apache.spark.sql.catalyst.expressions.Literal
val list = Seq[Literal](1)

// FIXME Use a correct query to demo ResolveSubquery
import org.apache.spark.sql.catalyst.plans.logical.Filter
import org.apache.spark.sql.catalyst.expressions.In
val plan = Filter(condition = In(value = a, list), child = rel)

scala> println(plan.numberedTreeString)
00 Filter a#9 IN (1)
01 +- LocalRelation <empty>, [a#9]

import spark.sessionState.analyzer.ResolveSubquery
val analyzedPlan = ResolveSubquery(plan)
scala> println(analyzedPlan.numberedTreeString)
00 Filter a#9 IN (1)
01 +- LocalRelation <empty>, [a#9]

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

// Use Catalyst DSL

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

val a = 'a.int

import org.apache.spark.sql.catalyst.plans.logical.LocalRelation

val rel = LocalRelation(a)

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

val list = Seq[Literal](1)

// FIXME Use a correct query to demo ResolveSubquery

import org.apache.spark.sql.catalyst.plans.logical.Filter

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

val plan = Filter(condition = In(value = a, list), child = rel)

scala> println(plan.numberedTreeString)

00 Filter a#9 IN (1)

01 +- LocalRelation <empty>, [a#9]

import spark.sessionState.analyzer.ResolveSubquery

val analyzedPlan = ResolveSubquery(plan)

scala> println(analyzedPlan.numberedTreeString)

00 Filter a#9 IN (1)

01 +- LocalRelation <empty>, [a#9]

Resolving Subquery Expressions (ScalarSubquery, Exists and In) — `resolveSubQueries` Internal Method



resolveSubQueries(plan: LogicalPlan, plans: Seq[LogicalPlan]): LogicalPlan

1

2

3

4

5

resolveSubQueries(plan: LogicalPlan, plans: Seq[LogicalPlan]): LogicalPlan

resolveSubQueries requests the input logical plan to transform expressions (down the operator tree) as follows:

For ScalarSubquery expressions with subquery plan not resolved and resolveSubQuery to create resolved ScalarSubquery expressions
For Exists expressions with subquery plan not resolved and resolveSubQuery to create resolved Exists expressions
For In expressions with ListQuery not resolved and resolveSubQuery to create resolved In expressions

Note	`resolveSubQueries` is used exclusively when `ResolveSubquery` is executed.

`resolveSubQuery` Internal Method



resolveSubQuery(
  e: SubqueryExpression,
  plans: Seq[LogicalPlan])(
  f: (LogicalPlan, Seq[Expression]) => SubqueryExpression): SubqueryExpression

1

2

3

4

5

6

7

8

resolveSubQuery(

e: SubqueryExpression,

plans: Seq[LogicalPlan])(

f: (LogicalPlan, Seq[Expression]) => SubqueryExpression): SubqueryExpression

resolveSubQuery…FIXME

Note	`resolveSubQuery` is used exclusively when `ResolveSubquery` is requested to resolve subquery expressions (ScalarSubquery, Exists and In).

Applying ResolveSubquery to Logical Plan (Executing ResolveSubquery) — `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 TreeNode, e.g. logical plan.

apply transforms the input logical plan as follows:

For Filter operators with an Aggregate operator (as the child operator) and the children resolved, apply resolves subquery expressions (ScalarSubquery, Exists and In) with the Filter operator and the plans with the Aggregate operator and its single child
For unary operators with the children resolved, apply resolves subquery expressions (ScalarSubquery, Exists and In) with the unary operator and its single child

ResolveSQLOnFile

2013-03-05admin阅读(1349)

ResolveSQLOnFile Logical Evaluation Rule for…FIXME

ResolveSQLOnFile is…FIXME

`maybeSQLFile` Internal Method



maybeSQLFile(u: UnresolvedRelation): Boolean

1

2

3

4

5

maybeSQLFile(u: UnresolvedRelation): Boolean

maybeSQLFile is enabled (i.e. true) where the following all hold:

FIXME

Note	`maybeSQLFile` is used exclusively when…FIXME

Applying 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

spark-sql 第15页

ColumnPruning

ColumnPruning Logical Optimization

Example 1

Example 2

Executing Rule — `apply` Method

CollapseWindow

CollapseWindow Logical Optimization

Executing Rule — `apply` Method

WindowsSubstitution

WindowsSubstitution Logical Evaluation Rule

WindowFrameCoercion

WindowFrameCoercion Type Coercion Logical Rule

Coercing Types in Logical Plan — `coerceTypes` Method

`createBoundaryCast` Internal Method

UpdateOuterReferences

UpdateOuterReferences Logical Rule

Applying UpdateOuterReferences to Logical Plan — `apply` Method

TimeWindowing

TimeWindowing Logical Resolution Rule

Executing Rule — `apply` Method

ResolveWindowOrder

ResolveWindowOrder Logical Resolution Rule

ResolveWindowFrame

ResolveWindowFrame Logical Resolution Rule

Applying ResolveWindowFrame to Logical Plan — `apply` Method

ResolveSubquery

ResolveSubquery Logical Resolution Rule

Resolving Subquery Expressions (ScalarSubquery, Exists and In) — `resolveSubQueries` Internal Method

`resolveSubQuery` Internal Method

Applying ResolveSubquery to Logical Plan (Executing ResolveSubquery) — `apply` Method

ResolveSQLOnFile

ResolveSQLOnFile Logical Evaluation Rule for…FIXME

`maybeSQLFile` Internal Method

Applying Rule to Logical Plan — `apply` Method

欢迎关注：spark技术分享

关注公众号：spark技术分享

QQ咨询

回顶部

spark-sql 第15页

ColumnPruning Logical Optimization

Example 1

Example 2

Executing Rule — apply Method

CollapseWindow Logical Optimization

Executing Rule — apply Method

WindowsSubstitution Logical Evaluation Rule

WindowFrameCoercion Type Coercion Logical Rule

Coercing Types in Logical Plan — coerceTypes Method

createBoundaryCast Internal Method

UpdateOuterReferences Logical Rule

Applying UpdateOuterReferences to Logical Plan — apply Method

TimeWindowing Logical Resolution Rule

Executing Rule — apply Method

ResolveWindowOrder Logical Resolution Rule

ResolveWindowFrame Logical Resolution Rule

Applying ResolveWindowFrame to Logical Plan — apply Method

ResolveSubquery Logical Resolution Rule

Resolving Subquery Expressions (ScalarSubquery, Exists and In) — resolveSubQueries Internal Method

resolveSubQuery Internal Method

Applying ResolveSubquery to Logical Plan (Executing ResolveSubquery) — apply Method

ResolveSQLOnFile Logical Evaluation Rule for…​FIXME

maybeSQLFile Internal Method

Applying Rule to Logical Plan — apply Method

欢迎关注：spark技术分享

关注公众号：spark技术分享

QQ咨询

回顶部

Executing Rule — `apply` Method

Executing Rule — `apply` Method

Coercing Types in Logical Plan — `coerceTypes` Method

`createBoundaryCast` Internal Method

Applying UpdateOuterReferences to Logical Plan — `apply` Method

Executing Rule — `apply` Method

Applying ResolveWindowFrame to Logical Plan — `apply` Method

Resolving Subquery Expressions (ScalarSubquery, Exists and In) — `resolveSubQueries` Internal Method

`resolveSubQuery` Internal Method

Applying ResolveSubquery to Logical Plan (Executing ResolveSubquery) — `apply` Method

ResolveSQLOnFile Logical Evaluation Rule for…FIXME

`maybeSQLFile` Internal Method

Applying Rule to Logical Plan — `apply` Method