Skip to content
Closed
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,7 @@ import scala.collection.mutable

import org.apache.spark.sql.catalyst.SQLConfHelper
import org.apache.spark.sql.catalyst.analysis.TableOutputResolver.DefaultValueFillMode.{NONE, RECURSE}
import org.apache.spark.sql.catalyst.expressions.{Attribute, CreateNamedStruct, Expression, GetStructField, IsNull, Literal}
import org.apache.spark.sql.catalyst.expressions.{Attribute, CreateNamedStruct, Expression, GetStructField, Literal}
import org.apache.spark.sql.catalyst.plans.logical.Assignment
import org.apache.spark.sql.catalyst.types.DataTypeUtils
import org.apache.spark.sql.catalyst.util.CharVarcharUtils
Expand Down Expand Up @@ -55,7 +55,6 @@ object AssignmentUtils extends SQLConfHelper with CastSupport {
* (preserving existing fields).
* @param coerceNestedTypes whether to coerce nested types to match the target type
* for complex types
* @param missingSourcePaths paths that exist in target but not in source
* @return aligned update assignments that match table attributes
*/
def alignUpdateAssignments(
Expand All @@ -73,8 +72,7 @@ object AssignmentUtils extends SQLConfHelper with CastSupport {
assignments,
addError = err => errors += err,
colPath = Seq(attr.name),
coerceNestedTypes,
fromStar)
coerceNestedTypes)
}

if (errors.nonEmpty) {
Expand Down Expand Up @@ -158,8 +156,7 @@ object AssignmentUtils extends SQLConfHelper with CastSupport {
assignments: Seq[Assignment],
addError: String => Unit,
colPath: Seq[String],
coerceNestedTypes: Boolean = false,
updateStar: Boolean = false): Expression = {
coerceNestedTypes: Boolean = false): Expression = {

val (exactAssignments, otherAssignments) = assignments.partition { assignment =>
assignment.key.semanticEquals(colExpr)
Expand Down Expand Up @@ -197,7 +194,7 @@ object AssignmentUtils extends SQLConfHelper with CastSupport {
assignments: Seq[Assignment],
addError: String => Unit,
colPath: Seq[String],
coerceNestedTyptes: Boolean): Expression = {
coerceNestedTypes: Boolean): Expression = {

col.dataType match {
case structType: StructType =>
Expand All @@ -207,7 +204,7 @@ object AssignmentUtils extends SQLConfHelper with CastSupport {
}
val updatedFieldExprs = fieldAttrs.zip(fieldExprs).map { case (fieldAttr, fieldExpr) =>
applyAssignments(fieldAttr, fieldExpr, assignments, addError, colPath :+ fieldAttr.name,
coerceNestedTyptes)
coerceNestedTypes)
}
toNamedStruct(structType, updatedFieldExprs)

Expand All @@ -226,52 +223,6 @@ object AssignmentUtils extends SQLConfHelper with CastSupport {
CreateNamedStruct(namedStructExprs)
}

private def getMissingSourcePaths(targetType: StructType,
sourceType: DataType,
colPath: Seq[String],
addError: String => Unit): Seq[Seq[String]] = {
val nestedTargetPaths = DataTypeUtils.extractLeafFieldPaths(targetType, Seq.empty)
val nestedSourcePaths = sourceType match {
case sourceStructType: StructType =>
DataTypeUtils.extractLeafFieldPaths(sourceStructType, Seq.empty)
case _ =>
addError(s"Value for struct type: " +
s"${colPath.quoted} must be a struct but was ${sourceType.simpleString}")
Seq()
}
nestedSourcePaths.diff(nestedTargetPaths)
}

/**
* Creates a null check for a field at the given path within a struct expression.
* Navigates through the struct hierarchy following the path and returns an IsNull check
* for the final field.
*
* @param rootExpr the root expression to navigate from
* @param path the field path to navigate (sequence of field names)
* @return an IsNull expression checking if the field at the path is null
*/
private def createNullCheckForFieldPath(
rootExpr: Expression,
path: Seq[String]): Expression = {
var currentExpr: Expression = rootExpr
path.foreach { fieldName =>
currentExpr.dataType match {
case st: StructType =>
st.fields.find(f => conf.resolver(f.name, fieldName)) match {
case Some(field) =>
val fieldIndex = st.fieldIndex(field.name)
currentExpr = GetStructField(currentExpr, fieldIndex, Some(field.name))
case None =>
// Field not found, shouldn't happen
}
case _ =>
// Not a struct, shouldn't happen
}
}
IsNull(currentExpr)
}

/**
* Checks whether assignments are aligned and compatible with table columns.
*
Expand Down