[SPARK-51290][SQL] Enable filling default values in DSv2 writes #50044
Conversation
| @@ -3534,7 +3534,8 @@ class Analyzer(override val catalogManager: CatalogManager) extends RuleExecutor | |||
| TableOutputResolver.suitableForByNameCheck(v2Write.isByName, | |||
| expected = v2Write.table.output, queryOutput = v2Write.query.output) | |||
| val projection = TableOutputResolver.resolveOutputColumns( | |||
| v2Write.table.name, v2Write.table.output, v2Write.query, v2Write.isByName, conf) | |||
| v2Write.table.name, v2Write.table.output, v2Write.query, v2Write.isByName, conf, | |||
| supportColDefaultValue = true) | |||
There was a problem hiding this comment.
I don't think there is value in validating if the catalog defines SUPPORT_COLUMN_DEFAULT_VALUE in capabilities during writes. If a connector includes default value metadata in its schema, it should be enough to fill default values. The flag exists for ALTER and CREATE/REPLACE statements.
There was a problem hiding this comment.
Yea true, Spark fills the default values during table writing and it works for all catalogs.
There was a problem hiding this comment.
You mean supportColDefaultValue is true or false doesn't matter for v2 here?
There was a problem hiding this comment.
Oh, I see. You mean to check for the flag SUPPORT_COLUMN_DEFAULT_VALUE here for the catalog.
There was a problem hiding this comment.
Correct, I don't see value in checking SUPPORT_COLUMN_DEFAULT_VALUE here.
| @@ -718,6 +724,11 @@ private class BufferedRowsReader( | |||
| schema: StructType, | |||
| row: InternalRow): Any = { | |||
| val index = schema.fieldIndex(field.name) | |||
|
|
|||
| if (index >= row.numFields) { | |||
There was a problem hiding this comment.
This is needed for support for adding columns with default values to the end.
There was a problem hiding this comment.
This is method extractFieldValue. Looks like it is only used by get. Why this is for adding columns?
There was a problem hiding this comment.
This is needed to read data inserted prior to adding columns to the schema. If that happens, there would be extra columns in the schema and we have to default new columns using the existence default value.
| @@ -423,8 +423,8 @@ abstract class V2WriteAnalysisSuiteBase extends AnalysisTest { | |||
| assertNotResolved(parsedPlan) | |||
| assertAnalysisErrorCondition( | |||
| parsedPlan, | |||
| expectedErrorCondition = "INCOMPATIBLE_DATA_FOR_TABLE.CANNOT_FIND_DATA", | |||
| expectedMessageParameters = Map("tableName" -> "`table-name`", "colName" -> "`x`") | |||
| expectedErrorCondition = "INCOMPATIBLE_DATA_FOR_TABLE.EXTRA_COLUMNS", | |||
There was a problem hiding this comment.
This is because of spark.sql.defaultColumn.useNullsForMissingDefaultValues and is aligned with V1 writes.
| @@ -328,7 +336,7 @@ trait AlterTableTests extends SharedSparkSession with QueryErrorsBase { | |||
| } | |||
|
|
|||
| test("SPARK-39383 DEFAULT columns on V2 data sources with ALTER TABLE ADD/ALTER COLUMN") { | |||
| withSQLConf(SQLConf.DEFAULT_COLUMN_ALLOWED_PROVIDERS.key -> s"$v2Format, ") { | |||
| withSQLConf(SQLConf.DEFAULT_COLUMN_ALLOWED_PROVIDERS.key -> s"$v2Format,$catalog") { | |||
There was a problem hiding this comment.
How does this conf affect the testing v2 in-memory catalog? I thought it's only for v1 file source.
There was a problem hiding this comment.
I think that it is because https://github.com/apache/spark/pull/50044/files#r1968081855
There was a problem hiding this comment.
@viirya is correct. We previously passed "" as provider in the in-memory connector, which required workarounds like this. No longer needed as we pass the catalog name as provider. Simplifies testing.
| @@ -122,7 +122,7 @@ class BasicInMemoryTableCatalog extends TableCatalog { | |||
| override def alterTable(ident: Identifier, changes: TableChange*): Table = { | |||
| val table = loadTable(ident).asInstanceOf[InMemoryTable] | |||
| val properties = CatalogV2Util.applyPropertiesChanges(table.properties, changes) | |||
| val schema = CatalogV2Util.applySchemaChanges(table.schema, changes, None, "ALTER TABLE") | |||
| val schema = CatalogV2Util.applySchemaChanges(table.schema, changes, Some(name), "ALTER TABLE") | |||
There was a problem hiding this comment.
Do we need to add memory table provider to DEFAULT_COLUMN_ALLOWED_PROVIDERS?
There was a problem hiding this comment.
Oh, nvm, the name is given when initializing the catalog.
What changes were proposed in this pull request?
This PR enables filling default values in DSv2 writes.
Why are the changes needed?
These changes are needed for proper support of default values for DSv2 connectors.
Does this PR introduce any user-facing change?
Users will be able to omit columns with default values. There is no impact to existing jobs.
How was this patch tested?
This patch comes with tests.
Was this patch authored or co-authored using generative AI tooling?
No.