naveenp2708 commented on code in PR #57644:
URL: https://github.com/apache/spark/pull/57644#discussion_r3772866357
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DatasetManager.scala:
##########
@@ -317,9 +317,21 @@ 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. Matching goes
+ // through the session resolver so a reserved column the user declared
in a different case
+ // is treated as present (not re-appended as a duplicate) under
case-insensitive analysis.
+ val resolver = context.spark.sessionState.conf.resolver
Review Comment:
Fixed. Both the materialization path (DatasetManager) and the
schema-validation path (GraphValidations) now take the resolver from the flow's
effective case sensitivity via SchemaInferenceUtils.effectiveCaseSensitivity
instead of the session resolver, so a pipeline-level spark.sql.caseSensitive
that differs from the session is honored the same way the rest of AUTO CDC does
--
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]
