naveenp2708 commented on code in PR #57644:
URL: https://github.com/apache/spark/pull/57644#discussion_r3799195384
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DatasetManager.scala:
##########
@@ -317,9 +318,23 @@ 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). Drop whatever reserved column(s) the user
declared and append the
+ // engine-owned shape from the inferred schema, so the created table
always has exactly the
+ // reserved column(s) the AUTO CDC MERGE writes, even if the user
declared one with a
+ // different type or nullability. Matching goes through the flow's
effective resolver (the
+ // same one the rest of AUTO CDC uses, which honors a case-sensitivity
conf set on the flow,
+ // not just the session).
+ val resolver = SchemaInferenceUtils.resolverFor(
+ effectiveCaseSensitivityFor(resolvedDataflowGraph, table.identifier,
context))
Review Comment:
Done. Hoisted effectiveCaseSensitive and effectiveResolver above
outputSchema and reused them, so the flows only get walked once.
--
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]
