dtenedor commented on code in PR #40652:
URL: https://github.com/apache/spark/pull/40652#discussion_r1158755699
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveDefaultColumns.scala:
##########
@@ -271,32 +271,37 @@ case class ResolveDefaultColumns(catalog: SessionCatalog)
extends Rule[LogicalPl
/**
* Updates an inline table to generate missing default column values.
*/
- private def addMissingDefaultValuesForInsertFromInlineTable(
+ def addMissingDefaultValuesForInsertFromInlineTable(
node: LogicalPlan,
insertTableSchemaWithoutPartitionColumns: StructType,
numUserSpecifiedColumns: Int): LogicalPlan = {
val schema = insertTableSchemaWithoutPartitionColumns
- val newDefaultExpressions: Seq[Expression] =
+ val newDefaultExpressions: Seq[UnresolvedAttribute] =
getDefaultExpressionsForInsert(schema, numUserSpecifiedColumns)
- val newNames: Seq[String] = if (numUserSpecifiedColumns > 0) {
- schema.fields.drop(numUserSpecifiedColumns).map(_.name)
- } else {
- schema.fields.map(_.name)
- }
+ val newNames: Seq[String] = schema.fields.map(_.name)
node match {
case _ if newDefaultExpressions.isEmpty => node
case table: UnresolvedInlineTable =>
table.copy(
- names = table.names ++ newNames,
+ names = newNames,
rows = table.rows.map { row => row ++ newDefaultExpressions })
- case local: LocalRelation =>
- // Note that we have consumed a LocalRelation but return an
UnresolvedInlineTable, because
- // addMissingDefaultValuesForInsertFromProject must replace unresolved
DEFAULT references.
- UnresolvedInlineTable(
- local.output.map(_.name) ++ newNames,
- local.data.map { row =>
- val colTypes = StructType(local.output.map(col =>
StructField(col.name, col.dataType)))
- row.toSeq(colTypes).map(Literal(_)) ++ newDefaultExpressions
+ case local: LocalRelation
+ if (numUserSpecifiedColumns > 0 && node.output.size <=
numUserSpecifiedColumns) ||
Review Comment:
You're right, this was confusing. I replaced this with a simpler way to add
this check at the end of the `getNewDefaultExpressionsForInsert` method instead
(the last five lines are new):
```
val numDefaultExpressionsToAdd =
getStructFieldsForDefaultExpressions(remainingFields).size
// Limit the number of new DEFAULT expressions to the difference of the
number
// of columns in the target table and the number of provided values in the
source
// relation. This clamps the total final number of provided values to the
number
// of columns in the target table.
.min(insertTableSchemaWithoutPartitionColumns.size - numProvidedValues)
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
