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

MonotonicallyIncreasingID

2012-09-05admin阅读(1652)

MonotonicallyIncreasingID Nondeterministic Leaf Expression

MonotonicallyIncreasingID is a non-deterministic leaf expression that is the internal representation of the monotonically_increasing_id standard and SQL functions.

As a Nondeterministic expression, MonotonicallyIncreasingID requires explicit initialization (with the current partition index) before evaluating a value.

MonotonicallyIncreasingID generates Java source code (as ExprCode) for code-generated expression evaluation.



import org.apache.spark.sql.catalyst.expressions.MonotonicallyIncreasingID
val monotonicallyIncreasingID = MonotonicallyIncreasingID()

import org.apache.spark.sql.catalyst.expressions.codegen.{CodegenContext, ExprCode}
val ctx = new CodegenContext
// doGenCode is used when Expression.genCode is executed
val ExprCode(code, _, _) = monotonicallyIncreasingID.genCode(ctx)

// Helper methods
def trim(code: String): String = {
  code.trim.split("\n").map(_.trim).filter(line => line.nonEmpty).mkString("\n")
}
def prettyPrint(code: String) = println(trim(code))
// END: Helper methods

scala> println(trim(code))
final long value_0 = partitionMask + count_0;
count_0++;

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

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

val monotonicallyIncreasingID = MonotonicallyIncreasingID()

import org.apache.spark.sql.catalyst.expressions.codegen.{CodegenContext, ExprCode}

val ctx = new CodegenContext

// doGenCode is used when Expression.genCode is executed

val ExprCode(code, _, _) = monotonicallyIncreasingID.genCode(ctx)

// Helper methods

def trim(code: String): String = {

code.trim.split("\n").map(_.trim).filter(line => line.nonEmpty).mkString("\n")

}

def prettyPrint(code: String) = println(trim(code))

// END: Helper methods

scala> println(trim(code))

final long value_0 = partitionMask + count_0;

count_0++;

MonotonicallyIncreasingID uses LongType as the data type of the result of evaluating itself.

MonotonicallyIncreasingID is never nullable.

MonotonicallyIncreasingID uses monotonically_increasing_id for the user-facing name.

MonotonicallyIncreasingID uses monotonically_increasing_id() for the SQL representation.

MonotonicallyIncreasingID is created when monotonically_increasing_id standard function is used in a structured query.

MonotonicallyIncreasingID is registered as monotonically_increasing_id SQL function.

MonotonicallyIncreasingID takes no input parameters when created.

Table 1. MonotonicallyIncreasingID’s Internal Properties (e.g. Registries, Counters and Flags)
Name	Description
`count`	Number of evalInternal calls, i.e. the number of rows for which `MonotonicallyIncreasingID` was evaluated Initialized when `MonotonicallyIncreasingID` is requested to initialize and used to evaluate a value.
`partitionMask`	Current partition index shifted 33 bits left Initialized when `MonotonicallyIncreasingID` is requested to initialize and used to evaluate a value.

Generating Java Source Code (ExprCode) For Code-Generated Expression Evaluation — `doGenCode` Method



doGenCode(ctx: CodegenContext, ev: ExprCode): ExprCode

1

2

3

4

5

doGenCode(ctx: CodegenContext, ev: ExprCode): ExprCode

Note	`doGenCode` is part of Expression Contract to generate a Java source code (ExprCode) for code-generated expression evaluation.

doGenCode requests the CodegenContext to add a mutable state as count name and long Java type.

doGenCode requests the CodegenContext to add an immutable state (unless exists already) as partitionMask name and long Java type.

doGenCode requests the CodegenContext to addPartitionInitializationStatement with [countTerm] = 0L; statement.

doGenCode requests the CodegenContext to addPartitionInitializationStatement with [partitionMaskTerm] = ((long) partitionIndex) << 33; statement.

In the end, doGenCode returns the input ExprCode with the code as follows and isNull property disabled (false):



final [dataType] [value] = [partitionMaskTerm] + [countTerm];
      [countTerm]++;

1

2

3

4

5

6

final [dataType] [value] = [partitionMaskTerm] + [countTerm];

[countTerm]++;

Initializing Nondeterministic Expression — `initializeInternal` Method



initializeInternal(input: InternalRow): Long

1

2

3

4

5

initializeInternal(input: InternalRow): Long

Note	`initializeInternal` is part of Nondeterministic Contract to initialize a `Nondeterministic` expression.

initializeInternal simply sets the count to 0 and the partitionMask to partitionIndex.toLong << 33.



val partitionIndex = 1
val partitionMask = partitionIndex.toLong << 33
scala> println(partitionMask.toBinaryString)
1000000000000000000000000000000000

1

2

3

4

5

6

7

8

val partitionIndex = 1

val partitionMask = partitionIndex.toLong << 33

scala> println(partitionMask.toBinaryString)

1000000000000000000000000000000000

Evaluating Nondeterministic Expression — `evalInternal` Method



evalInternal(input: InternalRow): Long

1

2

3

4

5

evalInternal(input: InternalRow): Long

Note	`evalInternal` is part of Nondeterministic Contract to evaluate the value of a `Nondeterministic` expression.

evalInternal remembers the current value of the count and increments it.

In the end, evalInternal returns the sum of the current value of the partitionMask and the remembered value of the count.

Literal

2012-09-04admin阅读(1605)

Literal Leaf Expression

Literal is a leaf expression that is created to represent a Scala value of a specific type.

Literal is created when…MEFIXME

Table 1. Literal’s Properties
Property	Description
`foldable`	Enabled (i.e. `true`)
`nullable`	Enabled when value is `null`

Creating Literal Instance — `create` Object Method



create(v: Any, dataType: DataType): Literal

1

2

3

4

5

create(v: Any, dataType: DataType): Literal

create uses CatalystTypeConverters helper object to convert the input v Scala value to a Catalyst rows or types and creates a Literal (with the Catalyst value and the input DataType).

Note	`create` is used when…FIXME

Creating Literal Instance

Literal takes the following when created:

Scala value (of type Any)
DataType

Generating Java Source Code (ExprCode) For Code-Generated Expression Evaluation — `doGenCode` Method



doGenCode(ctx: CodegenContext, ev: ExprCode): ExprCode

1

2

3

4

5

doGenCode(ctx: CodegenContext, ev: ExprCode): ExprCode

Note	`doGenCode` is part of Expression Contract to generate a Java source code (ExprCode) for code-generated expression evaluation.

doGenCode…FIXME

ListQuery

2012-09-03admin阅读(1648)

ListQuery Subquery Expression

ListQuery is a SubqueryExpression that represents SQL’s IN predicate with a subquery, e.g. NOT? IN '(' query ')'.

ListQuery cannot be evaluated and produce a value given an internal row.

ListQuery is resolved when:

Children are resolved
Subquery logical plan is resolved
There is at least one child output attribute

Creating ListQuery Instance

ListQuery takes the following when created:

Subquery logical plan
Child expressions
Expression ID (as ExprId and defaults to a new ExprId)
Child output attributes

JsonTuple

2012-09-02admin阅读(1611)

JsonTuple Generator Expression

JsonTuple is…FIXME

JsonToStructs

2012-09-01admin阅读(1827)

JsonToStructs Unary Expression

JsonToStructs is a unary expression with timezone support and CodegenFallback.

JsonToStructs is created to represent from_json function.



import org.apache.spark.sql.functions.from_json
val jsonCol = from_json($"json", new StructType())

import org.apache.spark.sql.catalyst.expressions.JsonToStructs
val jsonExpr = jsonCol.expr.asInstanceOf[JsonToStructs]
scala> println(jsonExpr.numberedTreeString)
00 jsontostructs('json, None)
01 +- 'json

1

2

3

4

5

6

7

8

9

10

11

12

import org.apache.spark.sql.functions.from_json

val jsonCol = from_json($"json", new StructType())

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

val jsonExpr = jsonCol.expr.asInstanceOf[JsonToStructs]

scala> println(jsonExpr.numberedTreeString)

00 jsontostructs('json, None)

01 +- 'json

JsonToStructs is a ExpectsInputTypes expression.

Note	`JsonToStructs` uses JacksonParser in `FAILFAST` mode that simply fails early when a corrupted/malformed record is found (and hence does not support `columnNameOfCorruptRecord` JSON option).

Property Description

converter

Function that converts Seq[InternalRow] into…FIXME

nullable

Enabled (i.e. true)

parser

JacksonParser with rowSchema and JSON options

Note	JSON options are made up of the input options with mode option as `FAILFAST` and the input time zone as the default time zone.

rowSchema

StructType that…FIXME

schema when of type StructType
StructType of the elements in schema when of type ArrayType

Creating JsonToStructs Instance

JsonToStructs takes the following when created:

DataType
Options
Child expression
Optional time zone ID

JsonToStructs initializes the internal registries and counters.

Parsing Table Schema for String Literals — `validateSchemaLiteral` Method



validateSchemaLiteral(exp: Expression): StructType

1

2

3

4

5

validateSchemaLiteral(exp: Expression): StructType

validateSchemaLiteral requests CatalystSqlParser to parseTableSchema for Literal of StringType.

For any other non-StringType types, validateSchemaLiteral reports a AnalysisException:



Expected a string literal instead of [expression]

1

2

3

4

5

Expected a string literal instead of [expression]

InSubquery

2012-08-31admin阅读(1516)

InSubquery Expression

InSubquery is a ExecSubqueryExpression that…FIXME

InSubquery is created when…FIXME

`updateResult` Method



updateResult(): Unit

1

2

3

4

5

updateResult(): Unit

Note	`updateResult` is part of ExecSubqueryExpression Contract to…FIXME.

updateResult…FIXME

Creating InSubquery Instance

InSubquery takes the following when created:

Child expression
SubqueryExec physical operator
Expression ID (as ExprId)
result array (default: null)
updated flag (default: false)

InSet

2012-08-30admin阅读(1729)

InSet Predicate Expression

InSet is a predicate expression (i.e. the result data type is always boolean) that is an optimized variant of the In predicate expression.

Inline

2012-08-29admin阅读(1618)

Inline Generator Expression

Inline is a unary expression and CollectionGenerator.

Inline is created by inline and inline_outer standard functions.



// Query with inline function
val q = spark.range(1)
  .selectExpr("inline(array(struct(1, 'a'), struct(2, 'b')))")
val logicalPlan = q.queryExecution.analyzed
scala> println(logicalPlan.numberedTreeString)
00 Project [col1#61, col2#62]
01 +- Generate inline(array(named_struct(col1, 1, col2, a), named_struct(col1, 2, col2, b))), false, false, [col1#61, col2#62]
02    +- Range (0, 1, step=1, splits=Some(8))

// Query with inline_outer function
val q = spark.range(1)
  .selectExpr("inline_outer(array(struct(1, 'a'), struct(2, 'b')))")
val logicalPlan = q.queryExecution.analyzed
scala> println(logicalPlan.numberedTreeString)
00 Project [col1#69, col2#70]
01 +- Generate inline(array(named_struct(col1, 1, col2, a), named_struct(col1, 2, col2, b))), false, true, [col1#69, col2#70]
02    +- Range (0, 1, step=1, splits=Some(8))

import org.apache.spark.sql.catalyst.plans.logical.Generate
// get is safe since there is Generate logical operator
val generator = logicalPlan.collectFirst { case g: Generate => g.generator }.get
import org.apache.spark.sql.catalyst.expressions.Inline
val inline = generator.asInstanceOf[Inline]

// Inline Generator expression is also CollectionGenerator
scala> inline.collectionType.catalogString
res1: String = array<struct<col1:int,col2:string>>

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

// Query with inline function

val q = spark.range(1)

.selectExpr("inline(array(struct(1, 'a'), struct(2, 'b')))")

val logicalPlan = q.queryExecution.analyzed

scala> println(logicalPlan.numberedTreeString)

00 Project [col1#61, col2#62]

01 +- Generate inline(array(named_struct(col1, 1, col2, a), named_struct(col1, 2, col2, b))), false, false, [col1#61, col2#62]

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

// Query with inline_outer function

val q = spark.range(1)

.selectExpr("inline_outer(array(struct(1, 'a'), struct(2, 'b')))")

val logicalPlan = q.queryExecution.analyzed

scala> println(logicalPlan.numberedTreeString)

00 Project [col1#69, col2#70]

01 +- Generate inline(array(named_struct(col1, 1, col2, a), named_struct(col1, 2, col2, b))), false, true, [col1#69, col2#70]

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

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

// get is safe since there is Generate logical operator

val generator = logicalPlan.collectFirst { case g: Generate => g.generator }.get

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

val inline = generator.asInstanceOf[Inline]

// Inline Generator expression is also CollectionGenerator

scala> inline.collectionType.catalogString

res1: String = array<struct<col1:int,col2:string>>

ImperativeAggregate

2012-08-28admin阅读(1816)

ImperativeAggregate — Contract for Aggregate Function Expressions with Imperative Methods

ImperativeAggregate is the contract for aggregate functions that are expressed in terms of imperative initialize, update, and merge methods (that operate on Row-based aggregation buffers).

ImperativeAggregate is a Catalyst expression with CodegenFallback.

Table 1. ImperativeAggregate’s Direct Implementations
Name	Description
`HyperLogLogPlusPlus`
`PivotFirst`
ScalaUDAF
TypedImperativeAggregate

ImperativeAggregate Contract



package org.apache.spark.sql.catalyst.expressions.aggregate

abstract class ImperativeAggregate {
  def initialize(mutableAggBuffer: InternalRow): Unit
  val inputAggBufferOffset: Int
  def merge(mutableAggBuffer: InternalRow, inputAggBuffer: InternalRow): Unit
  val mutableAggBufferOffset: Int
  def update(mutableAggBuffer: InternalRow, inputRow: InternalRow): Unit
  def withNewInputAggBufferOffset(newInputAggBufferOffset: Int): ImperativeAggregate
  def withNewMutableAggBufferOffset(newMutableAggBufferOffset: Int): ImperativeAggregate
}

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

package org.apache.spark.sql.catalyst.expressions.aggregate

abstract class ImperativeAggregate {

def initialize(mutableAggBuffer: InternalRow): Unit

val inputAggBufferOffset: Int

def merge(mutableAggBuffer: InternalRow, inputAggBuffer: InternalRow): Unit

val mutableAggBufferOffset: Int

def update(mutableAggBuffer: InternalRow, inputRow: InternalRow): Unit

def withNewInputAggBufferOffset(newInputAggBufferOffset: Int): ImperativeAggregate

def withNewMutableAggBufferOffset(newMutableAggBufferOffset: Int): ImperativeAggregate

}

Table 2. ImperativeAggregate Contract
Method	Description
`initialize`	Used when: `AggregateProcessor` is initialized (for window aggregate functions) AggregationIterator, ObjectAggregationIterator, TungstenAggregationIterator (for aggregate functions)
`inputAggBufferOffset`
`merge`	Used when: `AggregationIterator` does generateProcessRow (for aggregate functions)
`mutableAggBufferOffset`
`update`	Used when: `AggregateProcessor` is updated (for window aggregate functions) AggregationIterator (for aggregate functions)
`withNewInputAggBufferOffset`
`withNewMutableAggBufferOffset`

GetStructField

2012-08-27admin阅读(1407)

GetStructField Unary Expression

GetStructField is a unary expression that…FIXME

GetStructField is created when…FIXME

Creating GetStructField Instance

GetStructField takes the following when created:

Child expression
Ordinal
Optional name

spark-sql 第34页

MonotonicallyIncreasingID

MonotonicallyIncreasingID Nondeterministic Leaf Expression

Generating Java Source Code (ExprCode) For Code-Generated Expression Evaluation — `doGenCode` Method

Initializing Nondeterministic Expression — `initializeInternal` Method

Evaluating Nondeterministic Expression — `evalInternal` Method

Literal

Literal Leaf Expression

Creating Literal Instance — `create` Object Method

Creating Literal Instance

Generating Java Source Code (ExprCode) For Code-Generated Expression Evaluation — `doGenCode` Method

ListQuery

ListQuery Subquery Expression

Creating ListQuery Instance

JsonTuple

JsonTuple Generator Expression

JsonToStructs

JsonToStructs Unary Expression

Creating JsonToStructs Instance

Parsing Table Schema for String Literals — `validateSchemaLiteral` Method

InSubquery

InSubquery Expression

`updateResult` Method

Creating InSubquery Instance

InSet

InSet Predicate Expression

Inline

Inline Generator Expression

ImperativeAggregate

ImperativeAggregate — Contract for Aggregate Function Expressions with Imperative Methods

ImperativeAggregate Contract

GetStructField

GetStructField Unary Expression

Creating GetStructField Instance

欢迎关注：spark技术分享

关注公众号：spark技术分享

QQ咨询

回顶部

spark-sql 第34页

MonotonicallyIncreasingID Nondeterministic Leaf Expression

Generating Java Source Code (ExprCode) For Code-Generated Expression Evaluation — doGenCode Method

Initializing Nondeterministic Expression — initializeInternal Method

Evaluating Nondeterministic Expression — evalInternal Method

Literal Leaf Expression

Creating Literal Instance — create Object Method

Creating Literal Instance

Generating Java Source Code (ExprCode) For Code-Generated Expression Evaluation — doGenCode Method

ListQuery Subquery Expression

Creating ListQuery Instance

JsonTuple Generator Expression

JsonToStructs Unary Expression

Creating JsonToStructs Instance

Parsing Table Schema for String Literals — validateSchemaLiteral Method

InSubquery Expression

updateResult Method

Creating InSubquery Instance

InSet Predicate Expression

Inline Generator Expression

ImperativeAggregate — Contract for Aggregate Function Expressions with Imperative Methods

ImperativeAggregate Contract

GetStructField Unary Expression

Creating GetStructField Instance

欢迎关注：spark技术分享

关注公众号：spark技术分享

QQ咨询

回顶部

Generating Java Source Code (ExprCode) For Code-Generated Expression Evaluation — `doGenCode` Method

Initializing Nondeterministic Expression — `initializeInternal` Method

Evaluating Nondeterministic Expression — `evalInternal` Method

Creating Literal Instance — `create` Object Method

Generating Java Source Code (ExprCode) For Code-Generated Expression Evaluation — `doGenCode` Method

Parsing Table Schema for String Literals — `validateSchemaLiteral` Method

`updateResult` Method