anew commented on code in PR #57644:
URL: https://github.com/apache/spark/pull/57644#discussion_r3769497937
##########
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:
This should use the target AutoCDC flow’s effective resolver rather than the
Spark session resolver. Other AutoCDC paths derive case sensitivity from the
flows writing to the target because pipeline-level `SET
spark.sql.caseSensitive` can differ from the session
(`SchemaInferenceUtils.effectiveCaseSensitivity`,
`AutoCdcMergeFlow.effectiveResolver`, and the existing materialization resolver
below). With session case-insensitive but pipeline case-sensitive, a
user-declared `__SPARK_AUTOCDC_METADATA` can be treated as matching the
engine-owned lower-case metadata column, so materialization may skip appending
the actual column that the case-sensitive AutoCDC MERGE writes. The reverse
direction can append a duplicate under effective case-insensitive analysis. The
schema-validation path should use the same effective resolver too.
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/GraphValidations.scala:
##########
@@ -269,9 +271,13 @@ 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) {
+ // 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 (AutoCdcMergeFlow.stripReservedFields(inferredSchema, resolver) !=
Review Comment:
This also drops the declared reserved field before checking compatibility,
which means a user schema that declares `__spark_autocdc_metadata` with the
wrong type/nullability is accepted. Since `DatasetManager` then treats a
same-name reserved field as present and does not append the inferred
engine-owned field, the target can be created with a malformed metadata column
and fail later in the AutoCDC MERGE. If declaring the internal column remains
supported, I think we should either validate the declared reserved field
against the inferred field or always materialize the inferred engine-owned
shape.
--
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]
