org.apache.spark.sql.catalyst.analysis.resolver
ExpressionIdAssigner
Companion object ExpressionIdAssigner
class ExpressionIdAssigner extends AnyRef
ExpressionIdAssigner is used by the ExpressionResolver to assign unique expression IDs to NamedExpressions (AttributeReferences and Aliases). This is necessary to ensure that Optimizer performs its work correctly and does not produce correctness issues.
The framework works the following way:
- Each leaf operator must have unique output IDs (even if it's the same table, view, or CTE).
- The AttributeReferences get propagated "upwards" through the operator tree with their IDs preserved.
- Each Alias gets assigned a new unique ID and it sticks with it after it gets converted to an AttributeReference when it is outputted from the operator that produced it.
- Any operator may have AttributeReferences with the same IDs in its output given it is the same attribute. Thus, **no multi-child operator may have children with conflicting AttributeReference IDs**. In other words, two subtrees must not output the AttributeReferences with the same IDs, since relations, views and CTEs all output unique attributes, and Aliases get assigned new IDs as well. ExpressionIdAssigner.assertOutputsHaveNoConflictingExpressionIds is used to assert this invariant.
For SQL queries, this framework provides correctness just by reallocating relation outputs and by validating the invariants mentioned above. Reallocation is done in Resolver.handleLeafOperator. If all the relations (even if it's the same table) have unique output IDs, the expression ID assignment will be correct, because there are no duplicate IDs in a pure unresolved tree. The old ID -> new ID mapping is not needed in this case. For example, consider this query:
SELECT * FROM t AS t1 CROSS JOIN t AS t2 ON t1.col1 = t2.col1
The analyzed plan should be:
Project [col1#0, col2#1, col1#2, col2#3] +- Join Cross, (col1#0 = col1#2) :- SubqueryAlias t1 : +- Relation t[col1#0,col2#1] parquet +- SubqueryAlias t2 +- Relation t[col1#2,col2#3] parquet
and not:
Project [col1#0, col2#1, col1#0, col2#1] +- Join Cross, (col1#0 = col1#0) :- SubqueryAlias t1 : +- Relation t[col1#0,col2#1] parquet +- SubqueryAlias t2 +- Relation t[col1#0,col2#1] parquet
Because in the latter case the join condition is always true.
For DataFrame programs we need the full power of ExpressionIdAssigner, and old ID -> new ID mapping comes in handy, because DataFrame programs pass _partially_ resolved plans to the Resolver, which may consist of duplicate subtrees, and thus will have already assigned expression IDs. These already resolved duplicate subtrees with assigned IDs will conflict. Hence, we need to reallocate all the leaf node outputs _and_ remap old IDs to the new ones. Also, DataFrame programs may introduce the same Aliases in different parts of the query plan, so we just reallocate all the Aliases.
For example, consider this DataFrame program:
spark.range(0, 10).select($"id").write.format("parquet").saveAsTable("t") val alias = ($"id" + 1).as("id") spark.table("t").select(alias).select(alias)
The analyzed plan should be:
Project [(id#6L + cast(1 as bigint)) AS id#13L] +- Project [(id#4L + cast(1 as bigint)) AS id#6L] +- SubqueryAlias spark_catalog.default.t +- Relation spark_catalog.default.t[id#4L] parquet
and not:
Project [(id#6L + cast(1 as bigint)) AS id#6L] +- Project [(id#4L + cast(1 as bigint)) AS id#6L] +- SubqueryAlias spark_catalog.default.t +- Relation spark_catalog.default.t[id#4L] parquet
Because the latter case will confuse the Optimizer and the top Project will be eliminated leading to incorrect result.
There's an important caveat here: the leftmost branch of a logical plan tree. In this branch we need to preserve the expression IDs wherever possible because DataFrames may reference each other using their attributes. This also makes sense for performance reasons.
Consider this example:
val df1 = spark.range(0, 10).select($"id") val df2 = spark.range(5, 15).select($"id") df1.union(df2).filter(df1("id") === 5)
In this example df("id") references lower id attribute by expression ID, so union must not
reassign expression IDs in df1 (left child). Referencing df2 (right child) is not supported
in Spark.
The ExpressionIdAssigner covers both SQL and DataFrame scenarios with single approach and is integrated in the single-pass analysis framework.
The ExpressionIdAssigner is used in the following way:
- When the Resolver traverses the tree downwards prior to starting bottom-up analysis,
we build the mappingStack by calling withNewMapping (i.e. mappingStack.push)
for every child of a multi-child operator, so we have a separate stack entry (separate
mapping) for each branch. This way sibling branches' mappings are isolated from each other and
attribute IDs are reused only within the same branch. Initially we push
None, because the mapping needs to be initialized later with the correct output of a resolved operator. - When the bottom-up analysis starts, we assign IDs to all the NamedExpressions which are present in operators starting from the LeafNodes using mapExpression. createMapping is called right after each LeafNode is resolved, and first remapped attributes come from that LeafNode. This is done in Resolver.handleLeafOperator for each logical plan tree branch except the leftmost.
- Once the child branch is resolved, withNewMapping ends by calling mappingStack.pop.
- After the multi-child operator is resolved, we call createMapping to initialize the mapping with attributes _chosen_ (e.g. Union.mergeChildOutputs) by that operator's resolution algorithm and remap _old_ expression IDs to those chosen attributes.
- Continue remapping expressions until we reach the root of the operator tree.
- Alphabetic
- By Inheritance
- ExpressionIdAssigner
- AnyRef
- Any
- Hide All
- Show All
- Public
- Protected
Instance Constructors
- new ExpressionIdAssigner()
Value Members
- final def !=(arg0: Any): Boolean
- Definition Classes
- AnyRef → Any
- final def ##: Int
- Definition Classes
- AnyRef → Any
- final def ==(arg0: Any): Boolean
- Definition Classes
- AnyRef → Any
- final def asInstanceOf[T0]: T0
- Definition Classes
- Any
- def clone(): AnyRef
- Attributes
- protected[lang]
- Definition Classes
- AnyRef
- Annotations
- @throws(classOf[java.lang.CloneNotSupportedException]) @IntrinsicCandidate() @native()
- def createMapping(newOutput: Seq[Attribute] = Seq.empty, oldOutput: Option[Seq[Attribute]] = None): Unit
Create mapping with the given
newOutputthat rewrites theoldOutput.Create mapping with the given
newOutputthat rewrites theoldOutput. This is used by the Resolver after the multi-child operator is resolved to fill the current mapping with the attributes _chosen_ by that operator's resolution algorithm and remap _old_ expression IDs to those chosen attributes. It's also used by the ExpressionResolver right before remapping the attributes of a LeafNode.oldOutputis present for already resolved subtrees (e.g. DataFrames), but for SQL queries is will beNone, because that logical plan is analyzed for the first time. - final def eq(arg0: AnyRef): Boolean
- Definition Classes
- AnyRef
- def equals(arg0: AnyRef): Boolean
- Definition Classes
- AnyRef → Any
- final def getClass(): Class[_ <: AnyRef]
- Definition Classes
- AnyRef → Any
- Annotations
- @IntrinsicCandidate() @native()
- def hashCode(): Int
- Definition Classes
- AnyRef → Any
- Annotations
- @IntrinsicCandidate() @native()
- final def isInstanceOf[T0]: Boolean
- Definition Classes
- Any
- def isLeftmostBranch: Boolean
Returns
trueif the current logical plan branch is the leftmost branch.Returns
trueif the current logical plan branch is the leftmost branch. This is important in the context of preserving expression IDs in DataFrames. See class doc for more details. - def mapExpression(originalExpression: NamedExpression): NamedExpression
Assign a correct ID to the given originalExpression and return a new instance of that expression, or return a corresponding new instance of the same attribute, that was previously reallocated and is present in the current mappingStack entry.
Assign a correct ID to the given originalExpression and return a new instance of that expression, or return a corresponding new instance of the same attribute, that was previously reallocated and is present in the current mappingStack entry.
For Aliases: Try to preserve them if we are in the leftmost logical plan tree branch and unless they conflict. Conflicting Alias IDs are never acceptable. Otherwise, reallocate with a new ID and return that instance.
For AttributeReferences: If the attribute is present in the current mappingStack entry, return that instance, otherwise reallocate with a new ID and return that instance. The mapping is done both from the original expression ID _and_ from the new expression ID - this way we are able to replace old references to that attribute in the current operator branch, and preserve already reallocated attributes to make this call idempotent.
When remapping the provided expressions, we don't replace them with the previously seen attributes, but replace their IDs (NamedExpression.withExprId). This is done to preserve the properties of attributes at a certain point in the query plan. Examples where it's important:
1) Preserve the name case. In Spark the "requested" name takes precedence over the "original" name:
-- The output schema is [col1, COL1] SELECT col1, COL1 FROM VALUES (1);2) Preserve the metadata:
// Metadata "m1" remains, "m2" gets overwritten by the specified schema, "m3" is newly added. val metadata1 = new MetadataBuilder().putString("m1", "1").putString("m2", "2").build() val metadata2 = new MetadataBuilder().putString("m2", "3").putString("m3", "4").build() val schema = new StructType().add("a", IntegerType, nullable = true, metadata = metadata2) val df = spark.sql("SELECT col1 FROM VALUES (1)").select(col("col1").as("a", metadata1)).to(schema)
- final def ne(arg0: AnyRef): Boolean
- Definition Classes
- AnyRef
- final def notify(): Unit
- Definition Classes
- AnyRef
- Annotations
- @IntrinsicCandidate() @native()
- final def notifyAll(): Unit
- Definition Classes
- AnyRef
- Annotations
- @IntrinsicCandidate() @native()
- final def synchronized[T0](arg0: => T0): T0
- Definition Classes
- AnyRef
- def toString(): String
- Definition Classes
- AnyRef → Any
- final def wait(arg0: Long, arg1: Int): Unit
- Definition Classes
- AnyRef
- Annotations
- @throws(classOf[java.lang.InterruptedException])
- final def wait(arg0: Long): Unit
- Definition Classes
- AnyRef
- Annotations
- @throws(classOf[java.lang.InterruptedException]) @native()
- final def wait(): Unit
- Definition Classes
- AnyRef
- Annotations
- @throws(classOf[java.lang.InterruptedException])
- def withNewMapping[R](isLeftmostChild: Boolean = false)(body: => R): R
A RAII-wrapper for mappingStack.push and mappingStack.pop.
A RAII-wrapper for mappingStack.push and mappingStack.pop. Resolver uses this for every child of a multi-child operator to ensure that each operator branch uses an isolated expression ID mapping.
- isLeftmostChild
whether the current child is the leftmost child of the operator that is being resolved. This is used to determine whether the new stack entry is gonna be in the leftmost logical plan branch. It's
falseby default, because it's safer to remap attributes than to leave duplicates (to prevent correctness issues).
Deprecated Value Members
- def finalize(): Unit
- Attributes
- protected[lang]
- Definition Classes
- AnyRef
- Annotations
- @throws(classOf[java.lang.Throwable]) @Deprecated
- Deprecated
(Since version 9)