spark技术分享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.
|
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 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.
| Name | Description |
|---|---|
|
|
Number of evalInternal calls, i.e. the number of rows for which Initialized when |
|
|
Current partition index shifted 33 bits left Initialized when |
Generating Java Source Code (ExprCode) For Code-Generated Expression Evaluation — doGenCode Method
|
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):
|
1 2 3 4 5 6 |
final [dataType] [value] = [partitionMaskTerm] + [countTerm]; [countTerm]++; |
Initializing Nondeterministic Expression — initializeInternal Method
|
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.
|
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
|
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.
