naveenp2708 commented on code in PR #57644:
URL: https://github.com/apache/spark/pull/57644#discussion_r3799221432
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/GraphValidations.scala:
##########
@@ -258,20 +259,30 @@ trait GraphValidations extends Logging {
// table's); for a named flow (e.g. `CREATE FLOW <name> AS AUTO CDC INTO
<target>`) they
// differ, and keying on the flow identifier would silently skip
validation.
flowsTo.keys.flatMap(table.get).foreach { t: TableElement =>
+ val flows = flowsTo(t.identifier).map(f => resolvedFlow(f.identifier))
// The output inferred schema of a table is the declared schema merged
with the
// schema of all incoming flows. This must be equivalent to the declared
schema.
val inferredSchema = SchemaInferenceUtils
.inferSchemaFromFlows(
tableIdentifier = t.identifier,
- flowsTo(t.identifier).map(f => resolvedFlow(f.identifier)),
+ flows,
userSpecifiedSchema = t.specifiedSchema,
sessionCaseSensitive = sessionCaseSensitive
)
+ // Match reserved columns with the flow's effective case sensitivity,
not just the session:
+ // a case-sensitivity conf set on the flow overrides the session, the
same way the rest of
+ // AUTO CDC derives its resolver.
+ val resolver = SchemaInferenceUtils.resolverFor(
+ SchemaInferenceUtils.effectiveCaseSensitivity(t.identifier, flows,
sessionCaseSensitive))
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) {
+ // Check the specified schema matches the inferred schema once the
engine-owned reserved
+ // AUTO CDC metadata column(s) are set aside on both sides. The user
may omit them (the
+ // engine appends them at materialization) or declare them; comparing
both schemas with the
+ // reserved columns removed accepts either while still catching a
genuine mismatch in the
+ // remaining columns, and stays correct if more than one reserved
column is ever added.
+ if (AutoCdcReservedNames.stripReservedFields(inferredSchema, resolver)
!=
+ AutoCdcReservedNames.stripReservedFields(ss, resolver)) {
val datasetType = GraphElementTypeUtils
.getDatasetTypeForMaterializedViewOrStreamingTable(
flowsTo(t.identifier).map(f => resolvedFlow(f.identifier))
Review Comment:
Done, using that flows val now.
--
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]
