naveenp2708 commented on code in PR #57644:
URL: https://github.com/apache/spark/pull/57644#discussion_r4371787336
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/AutoCdcReservedNames.scala:
##########
@@ -40,4 +45,280 @@ private[pipelines] object AutoCdcReservedNames {
* Note that the schema of the CDC metadata column however can and does
differ on the SCD-type.
*/
val cdcMetadataColName: String = s"${prefix}metadata"
+
+ // A field is engine-owned when its name carries the reserved prefix.
Matching goes through the
+ // caller's `resolver` so a user-declared column is still recognized as
reserved under
+ // case-insensitive analysis, and the length guard keeps `substring` safe
for shorter names.
+ private def isReservedFieldName(name: String, resolver: Resolver): Boolean =
+ name.length >= prefix.length && resolver(name.substring(0, prefix.length),
prefix)
+
+ /** The engine-owned reserved AUTO CDC column(s) present in `schema`, if
any. */
+ private[pipelines] def reservedFields(schema: StructType, resolver:
Resolver): Seq[StructField] =
+ schema.fields.toSeq.filter(f => isReservedFieldName(f.name, resolver))
+
+ /** `schema` with the engine-owned reserved AUTO CDC column(s) removed. */
+ private[pipelines] def stripReservedFields(schema: StructType, resolver:
Resolver): StructType =
+ StructType(schema.fields.filterNot(f => isReservedFieldName(f.name,
resolver)))
+
+ /**
+ * The schema a materialized AUTO CDC target carries when the user declares
a schema: the user's
+ * columns with the engine-owned type/nullability substituted into any
declared reserved column,
+ * followed by the engine-owned reserved column(s) the declaration omitted.
`flowInferredSchema`
+ * must be flow-derived (no user-declared columns merged in), so its
reserved fields are exactly
+ * the engine-produced ones and never a user column that happens to carry
the prefix.
+ *
+ * A declared reserved field keeps its own spelling, position, and metadata
(e.g. a `COMMENT`);
+ * only its type and nullability are replaced with the engine's. That
matters on the upgrade path:
+ * an already-materialized
+ * target may carry the reserved column under a different casing (e.g. an
older definition
+ * declared it upper-cased under case-insensitive analysis), `evolveTable`
merges that existing
+ * spelling in, and this schema -- which also feeds the analysis-time read
path -- must agree with
+ * it so a downstream `SELECT *` plans against the column the target
actually has.
+ *
+ * The engine-owned reserved fields are made nullable so a target created
from a declared schema
+ * matches one created from an omitted schema (that path materializes the
inferred schema
+ * `asNullable`), avoiding a spurious nullability diff when a declaration is
added or removed.
+ */
+ private[pipelines] def appendEngineOwnedReservedFields(
+ declaredSchema: StructType,
+ flowInferredSchema: StructType,
+ resolver: Resolver): StructType = {
+ val engineReserved = StructType(reservedFields(flowInferredSchema,
resolver)).asNullable.fields
+
+ // A declaration carries at most one field matching each engine-owned
reserved column;
+ // `validateDeclaredReservedFields` rejects a case-insensitive duplicate
upstream, so this guard
+ // is defense-in-depth, provably unreachable from a validated declaration.
+ engineReserved.foreach { ef =>
+ val matches = declaredSchema.fields.filter(df => resolver(df.name,
ef.name))
+ if (matches.length > 1) {
+ throw SparkException.internalError(
+ s"Ambiguous reserved AUTO CDC field:
${matches.map(_.name).mkString(", ")} all match " +
+ s"the engine-owned column ${ef.name}.")
+ }
+ }
+
+ // Rebuild the declared schema in place: a declared reserved field keeps
its spelling and
+ // position but takes the engine-owned type/nullability.
`validateDeclaredReservedFields`
+ // rejects a reserved-prefixed declared field with no engine counterpart
upstream, so
+ // `case None => df` is unreachable from a validated declaration.
+ val usedEngineNames = scala.collection.mutable.Set.empty[String]
+ val rebuilt = declaredSchema.fields.map { df =>
+ if (isReservedFieldName(df.name, resolver)) {
+ engineReserved.find(ef => resolver(df.name, ef.name)) match {
+ case Some(ef) =>
+ usedEngineNames += ef.name
+ reconcileReservedField(df, ef, resolver)
+ case None => df
+ }
+ } else {
+ df
+ }
+ }
+
+ // Append engine-owned reserved fields the declaration omitted, in
canonical form.
+ val appended = engineReserved.filterNot(ef =>
usedEngineNames.contains(ef.name))
+ StructType(rebuilt ++ appended)
+ }
+
+ /**
+ * Reject a declared AUTO CDC target schema whose reserved metadata
column(s) are shaped in a way
+ * reconciliation cannot honor, so `appendEngineOwnedReservedFields`, the
read path, and the
+ * casing-drift check can assume a well-formed declaration instead of a
guard elsewhere.
+ * Run this from schema validation
(`GraphValidations.validateUserSpecifiedSchemas`) before
+ * materialization. Recursively through structs and array-of-struct elements
it throws a
+ * user-facing `AnalysisException` when a declared reserved-prefixed field
has no engine-owned
+ * counterpart (a typo or an extra reserved-namespace column), when two
declared fields resolve to
+ * one engine field, or when a declared nested field under a reserved column
is not one the engine
+ * produces.
+ */
+ private[pipelines] def validateDeclaredReservedFields(
+ tableName: String,
+ declaredSchema: StructType,
+ flowInferredSchema: StructType,
+ resolver: Resolver): Unit =
+ validateReservedLevel(
+ tableName,
+ reservedColumn = None,
+ reservedFields(declaredSchema, resolver),
+ reservedFields(flowInferredSchema, resolver),
+ resolver)
+
+ // Validate one level of reserved fields against the engine's.
`reservedColumn` is the enclosing
+ // reserved column's (possibly dotted) name for nested-field messages, None
at the top level. At
+ // the top level `declared`/`engine` are the reserved-prefixed columns;
within a reserved struct
+ // every field is engine-owned, so the recursion passes all of a matched
struct's fields.
+ private def validateReservedLevel(
+ tableName: String,
+ reservedColumn: Option[String],
+ declared: Seq[StructField],
+ engine: Seq[StructField],
+ resolver: Resolver): Unit = {
+ declared.filterNot(df => engine.exists(ef => resolver(df.name,
ef.name))).foreach { df =>
+ reservedColumn match {
+ case None =>
+ throw new AnalysisException(
+ "AUTOCDC_INVALID_RESERVED_SCHEMA.UNKNOWN_COLUMN",
+ Map("tableName" -> tableName, "columnName" -> df.name,
+ "reservedColumnNamePrefix" -> prefix))
+ case Some(col) =>
+ throw new AnalysisException(
+ "AUTOCDC_INVALID_RESERVED_SCHEMA.UNKNOWN_NESTED_FIELD",
+ Map("tableName" -> tableName, "reservedColumnName" -> col,
"fieldName" -> df.name))
+ }
+ }
+ engine.foreach { ef =>
+ val matches = declared.filter(df => resolver(df.name, ef.name))
+ if (matches.length > 1) {
+ val names = matches.map(f => s"`${f.name}`").mkString(", ")
+ reservedColumn match {
+ case None =>
+ throw new AnalysisException(
+ "AUTOCDC_INVALID_RESERVED_SCHEMA.DUPLICATE_COLUMN",
+ Map("tableName" -> tableName, "columnNames" -> names,
"engineColumnName" -> ef.name))
+ case Some(col) =>
+ throw new AnalysisException(
+ "AUTOCDC_INVALID_RESERVED_SCHEMA.DUPLICATE_NESTED_FIELD",
+ Map("tableName" -> tableName, "reservedColumnName" -> col,
"fieldNames" -> names,
+ "engineFieldName" -> ef.name))
+ }
+ }
+ }
+ declared.foreach { df =>
+ engine.find(ef => resolver(df.name, ef.name)).foreach { ef =>
+ val col = Some(reservedColumn.map(c =>
s"$c.${df.name}").getOrElse(df.name))
+ (df.dataType, ef.dataType) match {
+ case (ds: StructType, es: StructType) =>
+ validateReservedLevel(tableName, col, ds.fields.toSeq,
es.fields.toSeq, resolver)
+ case (da: ArrayType, ea: ArrayType) =>
+ (da.elementType, ea.elementType) match {
+ case (ds: StructType, es: StructType) =>
+ validateReservedLevel(tableName, col, ds.fields.toSeq,
es.fields.toSeq, resolver)
+ case _ =>
Review Comment:
Good catch @szehon-ho , thanks. Fixed in 58059a5 - validateReservedLevel now
recurses through the array element types too, same as reconcileReservedType
already does, so a bogus field inside ARRAY<ARRAY<STRUCT<...>>> gets rejected
instead of slipping through and getting reconciled into the target. Added a
regression test for that shape (the array(array(named_struct(‘ordinal’,
version))) one you gave). Also backfilled a couple of tests the last commit was
missing - a COMMENT on a declared reserved column surviving materialization,
and the omitted nested fields getting appended.
--
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]
