Stack is…FIXME
spark技术分享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…FIXME
SortOrder is a unary expression that represents the following operators in a logical plan:
AstBuilder is requested to parse ORDER BY or SORT BY sort specifications
Column.asc, Column.asc_nulls_first, Column.asc_nulls_last, Column.desc, Column.desc_nulls_first, and Column.desc_nulls_last operators are used
SortOrder is used to specify the output data ordering requirements of a physical operator.
SortOrder is an unevaluable expression and cannot be evaluated (i.e. produce a value given an internal row).
|
Note
|
An unevaluable expression cannot be evaluated to produce a value (neither in interpreted nor code-generated expression evaluations) and has to be resolved (replaced) to some other expressions or logical operators at analysis or optimization phases or they fail analysis. |
SortOrder is never foldable (as an unevaluable expression with no evaluation).
|
Tip
|
Use asc, asc_nullsLast, desc or desc_nullsFirst operators from the Catalyst DSL to create a SortOrder expression, e.g. for testing or Spark SQL internals exploration.
|
|
Note
|
Dataset.repartitionByRange, Dataset.sortWithinPartitions, Dataset.sort and WindowSpec.orderBy default to Ascending sort direction. |
apply Factory Method|
1 2 3 4 5 6 7 8 |
apply( child: Expression, direction: SortDirection, sameOrderExpressions: Set[Expression] = Set.empty): SortOrder |
apply is a convenience method to create a SortOrder with the defaultNullOrdering of the SortDirection.
|
Note
|
apply is used exclusively in window function.
|
asc, asc_nullsLast, desc and desc_nullsFirst Operators|
1 2 3 4 5 6 7 8 |
asc: SortOrder asc_nullsLast: SortOrder desc: SortOrder desc_nullsFirst: SortOrder |
asc, asc_nullsLast, desc and desc_nullsFirst create a SortOrder expression with the Ascending or Descending sort direction, respectively.
|
1 2 3 4 5 6 7 8 |
import org.apache.spark.sql.catalyst.dsl.expressions._ val sortNullsLast = 'id.asc_nullsLast scala> println(sortNullsLast.sql) `id` ASC NULLS LAST |
SortOrder takes the following when created:
Child expression
“Same Order” expressions
SortDirection is the base of sort directions.
| Method | Description | ||
|---|---|---|---|
|
|
Used when…FIXME |
||
|
|
Used when…FIXME |
There are two sorting directions available, i.e. Ascending and Descending.
SizeBasedWindowFunction is the extension of the AggregateWindowFunction Contract for window functions that require the size of the current window for calculation.
|
1 2 3 4 5 6 7 8 9 |
package org.apache.spark.sql.catalyst.expressions trait SizeBasedWindowFunction extends AggregateWindowFunction { // No required properties (vals and methods) that have no implementation } |
| Property | Description |
|---|---|
|
|
Size of the current window as a AttributeReference expression with |
| SizeBasedWindowFunction | Description |
|---|---|
|
Window function expression for cume_dist standard function (Dataset API) and cume_dist SQL function |
|
|
NTile |
|
|
PercentRank |
SimpleTypedAggregateExpression is…FIXME
SimpleTypedAggregateExpression is created when…FIXME
| Name | Description |
|---|---|
|
|
|
|
|
SimpleTypedAggregateExpression takes the following when created:
Optional input deserializer expression
Optional schema for the input
Buffer serializer (as a collection of named expressions)
Buffer deserializer expression
Output serializer (as a collection of expressions)
ScalaUDAF is a Catalyst expression adapter to manage the lifecycle of UserDefinedAggregateFunction and hook it in Spark SQL’s Catalyst execution path.
ScalaUDAF is created when:
UserDefinedAggregateFunction creates a Column for a user-defined aggregate function using all and distinct values (to use the UDAF in Dataset operators)
UDFRegistration is requested to register a user-defined aggregate function (to use the UDAF in SQL mode)
ScalaUDAF is a ImperativeAggregate.
| Method Name | Behaviour |
|---|---|
|
Requests UserDefinedAggregateFunction to initialize |
|
|
Requests UserDefinedAggregateFunction to merge |
|
|
Requests UserDefinedAggregateFunction to update |
When evaluated, ScalaUDAF…FIXME
ScalaUDAF has no representation in SQL.
| Name | Description |
|---|---|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
Copy of aggBufferAttributes |
|
|
|
|
|
Always enabled (i.e. |
| Name | Description |
|---|---|
|
Used when…FIXME |
|
|
Used when…FIXME |
|
|
Used when…FIXME |
|
|
Used when…FIXME |
ScalaUDAF takes the following when created:
Children Catalyst expressions
ScalaUDAF initializes the internal registries and counters.
initialize Method|
1 2 3 4 5 |
initialize(buffer: InternalRow): Unit |
initialize sets the input buffer internal binary row as underlyingBuffer of MutableAggregationBufferImpl and requests the UserDefinedAggregateFunction to initialize (with the MutableAggregationBufferImpl).
|
Note
|
initialize is part of ImperativeAggregate Contract.
|
update Method|
1 2 3 4 5 |
update(mutableAggBuffer: InternalRow, inputRow: InternalRow): Unit |
update sets the input buffer internal binary row as underlyingBuffer of MutableAggregationBufferImpl and requests the UserDefinedAggregateFunction to update.
|
Note
|
update uses inputProjection on the input input and converts it using inputToScalaConverters.
|
|
Note
|
update is part of ImperativeAggregate Contract.
|
merge Method|
1 2 3 4 5 |
merge(buffer1: InternalRow, buffer2: InternalRow): Unit |
merge first sets:
underlyingBuffer of MutableAggregationBufferImpl to the input buffer1
underlyingInputBuffer of InputAggregationBuffer to the input buffer2
merge then requests the UserDefinedAggregateFunction to merge (passing in the MutableAggregationBufferImpl and InputAggregationBuffer).
|
Note
|
merge is part of ImperativeAggregate Contract.
|
ScalaUDF is a Catalyst expression to manage the lifecycle of a user-defined function (and hook it in to Spark SQL’s Catalyst execution path).
ScalaUDF is a ImplicitCastInputTypes and UserDefinedExpression.
ScalaUDF has no representation in SQL.
ScalaUDF is created when:
UserDefinedFunction is executed
UDFRegistration is requested to register a Scala function as a user-defined function (in FunctionRegistry)
|
1 2 3 4 5 6 7 8 9 10 11 12 |
val lengthUDF = udf { s: String => s.length }.withName("lengthUDF") val c = lengthUDF($"name") scala> println(c.expr.treeString) UDF:lengthUDF('name) +- 'name import org.apache.spark.sql.catalyst.expressions.ScalaUDF val scalaUDF = c.expr.asInstanceOf[ScalaUDF] |
|
Note
|
Spark SQL Analyzer uses HandleNullInputsForUDF logical evaluation rule to…FIXME |
|
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 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 |
// Defining a zero-argument UDF val myUDF = udf { () => "Hello World" } // "Execute" the UDF // Attach it to an "execution environment", i.e. a Dataset // by specifying zero columns to execute on (since the UDF is no-arg) import org.apache.spark.sql.catalyst.expressions.ScalaUDF val scalaUDF = myUDF().expr.asInstanceOf[ScalaUDF] scala> scalaUDF.resolved res1: Boolean = true // Execute the UDF (on every row in a Dataset) // We simulate it relying on the EmptyRow that is the default InternalRow of eval scala> scalaUDF.eval() res2: Any = Hello World // Defining a UDF of one input parameter val hello = udf { s: String => s"Hello $s" } // Binding the hello UDF to a column name import org.apache.spark.sql.catalyst.expressions.ScalaUDF val helloScalaUDF = hello($"name").expr.asInstanceOf[ScalaUDF] scala> helloScalaUDF.resolved res3: Boolean = false // Resolve helloScalaUDF, i.e. the only `name` column reference scala> helloScalaUDF.children res4: Seq[org.apache.spark.sql.catalyst.expressions.Expression] = ArrayBuffer('name) // The column is free (i.e. not bound to a Dataset) // Define a Dataset that becomes the rows for the UDF val names = Seq("Jacek", "Agata").toDF("name") scala> println(names.queryExecution.analyzed.numberedTreeString) 00 Project [value#1 AS name#3] 01 +- LocalRelation [value#1] // Resolve the references using the Dataset val plan = names.queryExecution.analyzed val resolver = spark.sessionState.analyzer.resolver import org.apache.spark.sql.catalyst.analysis.UnresolvedAttribute val resolvedUDF = helloScalaUDF.transformUp { case a @ UnresolvedAttribute(names) => // we're in controlled environment // so get is safe plan.resolve(names, resolver).get } scala> resolvedUDF.resolved res6: Boolean = true scala> println(resolvedUDF.numberedTreeString) 00 UDF(name#3) 01 +- name#3: string import org.apache.spark.sql.catalyst.expressions.BindReferences val attrs = names.queryExecution.sparkPlan.output val boundUDF = BindReferences.bindReference(resolvedUDF, attrs) // Create an internal binary row, i.e. InternalRow import org.apache.spark.sql.catalyst.encoders.ExpressionEncoder val stringEncoder = ExpressionEncoder[String] val row = stringEncoder.toRow("world") // YAY! It works! scala> boundUDF.eval(row) res8: Any = Hello world // Just to show the regular execution path // i.e. how to execute a UDF in a context of a Dataset val q = names.select(hello($"name")) scala> q.show +-----------+ | UDF(name)| +-----------+ |Hello Jacek| |Hello Agata| +-----------+ |
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…FIXME
eval Method|
1 2 3 4 5 |
eval(input: InternalRow): Any |
|
Note
|
eval is part of Expression Contract for the interpreted (non-code-generated) expression evaluation, i.e. evaluating a Catalyst expression to a JVM object for a given internal binary row.
|
eval executes the Scala function on the input internal row.
ScalaUDF takes the following when created:
Output data type
Child Catalyst expressions
Input data types (if available)
ScalaUDF initializes the internal registries and counters.
ScalarSubquery is an ExecSubqueryExpression that can give exactly one value (i.e. the value of executing SubqueryExec subquery that can result in a single row and a single column or null if no row were computed).
|
Important
|
Spark SQL uses the name of ScalarSubquery twice to represent an ExecSubqueryExpression (this page) and a SubqueryExpression. It is confusing and you should not be anymore.
|
ScalarSubquery is created exclusively when PlanSubqueries physical optimization is executed (and plans a ScalarSubquery expression).
|
1 2 3 4 5 6 7 8 9 10 |
// FIXME DEMO import org.apache.spark.sql.execution.PlanSubqueries val spark = ... val planSubqueries = PlanSubqueries(spark) val plan = ... val executedPlan = planSubqueries(plan) |
ScalarSubquery expression cannot be evaluated, i.e. produce a value given an internal row.
ScalarSubquery uses…FIXME…for the data type.
| Name | Description |
|---|---|
|
|
The value of the single column in a single row after collecting the rows from executing the subquery plan or |
|
|
Flag that says whether |
updateResult Method|
1 2 3 4 5 |
updateResult(): Unit |
|
Note
|
updateResult is part of ExecSubqueryExpression Contract to fill an Catalyst expression with a collected result from executing a subquery plan.
|
updateResult requests SubqueryExec physical plan to execute and collect internal rows.
updateResult sets result to the value of the only column of the single row or null if no row were collected.
In the end, updateResult marks the ScalarSubquery instance as updated.
updateResult reports a RuntimeException when there are more than 1 rows in the result.
|
1 2 3 4 5 6 |
more than one row returned by a subquery used as an expression: [plan] |
updateResult reports an AssertionError when the number of fields is not exactly 1.
|
1 2 3 4 5 |
Expects 1 field, but got [numFields] something went wrong in analysis |
eval Method|
1 2 3 4 5 |
eval(input: InternalRow): Any |
|
Note
|
eval is part of Expression Contract for the interpreted (non-code-generated) expression evaluation, i.e. evaluating a Catalyst expression to a JVM object for a given internal binary row.
|
eval simply returns result value.
eval reports an IllegalArgumentException if the ScalarSubquery expression has not been updated yet.
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 first makes sure that the updated flag is on (true). If not, doGenCode throws an IllegalArgumentException exception with the following message:
|
1 2 3 4 5 |
requirement failed: [this] has not finished |
doGenCode then creates a Literal (for the result and the dataType) and simply requests it to generate a Java source code.
ScalarSubquery is a SubqueryExpression that returns a single row and a single column only.
ScalarSubquery represents a structured query that can be used as a “column”.
|
Important
|
Spark SQL uses the name of ScalarSubquery twice to represent a SubqueryExpression (this page) and an ExecSubqueryExpression. You’ve been warned.
|
ScalarSubquery is created exclusively when AstBuilder is requested to parse a subquery expression.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 |
// FIXME DEMO // Borrowed from ExpressionParserSuite.scala // ScalarSubquery(table("tbl").select('max.function('val))) > 'current) val sql = "(select max(val) from tbl) > current" // 'a === ScalarSubquery(table("s").select('b)) val sql = "a = (select b from s)" // Borrowed from PlanParserSuite.scala // table("t").select(ScalarSubquery(table("s").select('max.function('b))).as("ss")) val sql = "select (select max(b) from s) ss from t" // table("t").where('a === ScalarSubquery(table("s").select('b))).select(star()) val sql = "select * from t where a = (select b from s)" // table("t").groupBy('g)('g).where('a > ScalarSubquery(table("s").select('b))) val sql = "select g from t group by g having a > (select b from s)" |
ScalarSubquery takes the following when created:
Subquery logical plan
Child expressions (default: no children)
Expression ID (as ExprId and defaults to a new ExprId)
RuntimeReplaceable is the marker contract for unary expressions that are replaced by Catalyst Optimizer with their child expression (that can then be evaluated).
|
Note
|
Catalyst Optimizer uses ReplaceExpressions logical optimization to replace RuntimeReplaceable expressions.
|
RuntimeReplaceable contract allows for expression aliases, i.e. expressions that are fairly complex in the inside than on the outside, and is used to provide compatibility with other SQL databases by supporting SQL functions with their more complex Catalyst expressions (that are already supported by Spark SQL).
|
Note
|
RuntimeReplaceables are tied up to their SQL functions in FunctionRegistry. |
RuntimeReplaceable expressions cannot be evaluated (i.e. produce a value given an internal row) and therefore have to be replaced in the query execution pipeline.
|
1 2 3 4 5 6 7 8 9 10 |
package org.apache.spark.sql.catalyst.expressions trait RuntimeReplaceable extends UnaryExpression with Unevaluable { // as a marker contract it only marks a class // no methods are required } |
|
Note
|
To make sure the explain plan and expression SQL works correctly, a RuntimeReplaceable implementation should override flatArguments and sql methods.
|
| RuntimeReplaceable | Standard Function | SQL Function |
|---|---|---|
