anew commented on code in PR #57644:
URL: https://github.com/apache/spark/pull/57644#discussion_r3762119317
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/GraphValidations.scala:
##########
@@ -270,8 +270,12 @@ trait GraphValidations extends Logging {
t.specifiedSchema.foreach { ss =>
// Check the inferred schema matches the specified schema. Used to
catch errors where the
- // inferred user-facing schema has columns that are not in the
specified one.
- if (inferredSchema != ss) {
+ // inferred user-facing schema has columns that are not in the
specified one. The reserved
+ // AUTO CDC metadata column(s) are engine-owned, so a declared schema
that differs from the
+ // inferred schema only by omitting them is accepted; the engine
appends them to the
+ // effective table schema at materialization.
+ if (inferredSchema != ss &&
Review Comment:
this works right now because there is exactly one reserved field. If, in the
future, we add another one, there could be a situation where the user declares
one out of two, and then both conditions won't be met.
it would be better to compare the version of both schemas that has the
reserved fields removed.
##########
sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/UserSpecifiedSchemaValidationSuite.scala:
##########
@@ -182,14 +183,24 @@ class UserSpecifiedSchemaValidationSuite extends
PipelineTest with SharedSparkSe
// AUTO CDC flows: the inferred schema appends a reserved metadata column to
the data columns.
- test("data-only user-specified schema is rejected for an implicit AUTO CDC
flow") {
- // Schema lists the data columns but omits the appended metadata column,
so it is incompatible.
- assertSchemaIncompatible(autoCdcGraph(flowName = "target", declaredSchema
= Some(dataSchema)))
+ test("SPARK-58118: data-only user-specified schema is accepted for an
implicit AUTO CDC flow") {
Review Comment:
minor: I don't think we need the Jira IDs in all these test names.
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/Flow.scala:
##########
@@ -494,3 +494,14 @@ class AutoCdcMergeFlow(
}
}
+
+object AutoCdcMergeFlow {
+
+ /** The engine-owned reserved AUTO CDC column(s) present in `schema`, if
any. */
Review Comment:
are these two methods only called on the system-inferred schema? Or also on
the user-declared schema? If so, they need to honor the case sensitivity.
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DatasetManager.scala:
##########
@@ -317,9 +317,19 @@ object DatasetManager extends Logging {
val (catalog, identifier) =
PipelinesCatalogUtils.resolveTableCatalog(context.spark,
table.identifier)
- val outputSchema = table.specifiedSchema.getOrElse(
- inferredSchemas(table.identifier).asNullable
- )
+ val outputSchema = table.specifiedSchema match {
+ case Some(ss) =>
+ // The user schema describes the logical table; the engine owns the
reserved AUTO CDC
+ // metadata column(s). Append any that the incoming flows produce but
the user omitted,
+ // so the created table matches what the AUTO CDC MERGE writes at
runtime.
+ val specifiedNames = ss.fieldNames.toSet
+ val omittedReservedFields = AutoCdcMergeFlow
+ .reservedFields(inferredSchemas(table.identifier))
+ .filterNot(f => specifiedNames.contains(f.name))
Review Comment:
does this need to account for the case sensitivity conf? If the user
specified the meta fields in different case?
--
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]
