spark-test commented on code in PR #57644:
URL: https://github.com/apache/spark/pull/57644#discussion_r4365140833
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/elements.scala:
##########
@@ -183,10 +184,23 @@ case class VirtualTableInput(
*/
def load(asStreaming: Boolean): DataFrame = {
val deducedSchema = specifiedSchema match {
- // If the user specified a schema, use it directly.
+ // For a declared AUTO CDC target, materialization reconciles the user's
reserved column(s)
+ // in place and appends any the declaration omitted (see
DatasetManager.materializeTable).
+ // Produce that same schema here so a same-graph downstream consumer
plans against the columns
+ // the target is actually created with, rather than the declaration
verbatim.
+ case Some(ss) if
availableFlows.exists(GraphElementTypeUtils.isAutoCdcFlow) =>
+ val flowInferredSchema = SchemaInferenceUtils.inferSchemaFromFlows(
+ tableIdentifier = identifier,
+ flows = availableFlows,
+ userSpecifiedSchema = None,
+ sessionCaseSensitive = sessionCaseSensitive)
+ val resolver = SchemaInferenceUtils.resolverFor(
+ SchemaInferenceUtils.effectiveCaseSensitivity(
+ identifier, availableFlows, sessionCaseSensitive))
+ AutoCdcReservedNames.appendEngineOwnedReservedFields(ss,
flowInferredSchema, resolver)
Review Comment:
**Nit (P3):** `VirtualTableInput.load` passes the raw declared schema to
`appendEngineOwnedReservedFields`, and `load` runs while downstream flows are
analyzed: `FlowAnalysis.readGraphInput` calls it inside the reading flow's
`Try(FlowAnalysis.analyze(...))`. `DataflowGraph.validationFailure` then
reports flow-analysis failures (`validateSuccessfulFlowAnalysis`) before
`validateUserSpecifiedSchemas`, which is where 742a80b added
`validateDeclaredReservedFields`. Nothing at registration rejects
case-duplicate columns either: `SqlGraphRegistrationContext` builds the schema
with `StructType(cst.columns.map(_.toV1Column))`.
As a result, under case-insensitive analysis, a target declared with both
`__spark_autocdc_metadata` and `__SPARK_AUTOCDC_METADATA`, plus `CREATE
MATERIALIZED VIEW copy AS SELECT * FROM target` in the same graph, still fails
with the helper's `SparkException.internalError` ("Ambiguous reserved AUTO CDC
field ...") attributed to `copy`, instead of
`AUTOCDC_INVALID_RESERVED_SCHEMA.DUPLICATE_COLUMN` on `target`. The new
DUPLICATE_COLUMN test passes only because its graph has no reader of the
target. Under a case-sensitive flow the two names are distinct columns and
nothing fails. The comments saying the read path only sees validated
declarations (`AutoCdcReservedNames.scala:88-90` and `124-126`,
`GraphValidations.scala:294-296`) do not hold for this route.
Suggested fix: have `appendEngineOwnedReservedFields` take the table name
and run `validateDeclaredReservedFields` before reconciling (or call the
validator from `load`), so every route raises the user-facing error, then
update those comments. The validation-time half of the earlier comment on
`GraphValidations.scala` is addressed by `validateDeclaredReservedFields`.
This extends existing discussions
[1](https://github.com/apache/spark/pull/57644#issuecomment-6369698193),
[2](https://github.com/apache/spark/pull/57644#issuecomment-6373631540).
**Verification:**
- **Regression:** A pipeline whose AUTO CDC target declares both case
variants of the reserved metadata column under case-insensitive analysis, and
whose graph also contains a dataset that reads that target, fails with
AUTOCDC_INVALID_RESERVED_SCHEMA.DUPLICATE_COLUMN naming the target rather than
with INTERNAL_ERROR.
- **Compatibility:** These all keep their current validation outcome and
their current materialized and analysis-time schemas: data-only declarations,
declarations that list the reserved column once, case-sensitive flows with a
distinct differently cased user column, and tables not written by AUTO CDC.
##########
sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/UserSpecifiedSchemaValidationSuite.scala:
##########
@@ -206,4 +247,79 @@ class UserSpecifiedSchemaValidationSuite extends
PipelineTest with SharedSparkSe
declaredSchema = Some(autoCdcInferredSchema("auto_cdc_flow"))).validate(
spark.sessionState.conf.caseSensitiveAnalysis)
}
+
+ test("a user-specified schema that declares the reserved metadata column
with the wrong " +
+ "type is rejected for an AUTO CDC flow") {
+ // Declaring the engine-owned metadata column is allowed, but only with
the shape the engine
+ // produces. A conflicting type is caught when the declared schema is
merged with the inferred
+ // one, before the AUTO CDC MERGE ever runs, so the target can never be
created with a
+ // malformed metadata column.
+ val wrongTypeMetadata = StructType(
+ dataSchema.fields :+
StructField(AutoCdcReservedNames.cdcMetadataColName, StringType))
+ assertSchemaIncompatible(
+ autoCdcGraph(flowName = "auto_cdc_flow", declaredSchema =
Some(wrongTypeMetadata)))
+ }
+
+ test("data-only user-specified schema is accepted for an SCD2 AUTO CDC
flow") {
+ // An SCD2 flow's inferred schema is the data columns plus the SCD2
interval bounds
+ // __START_AT / __END_AT plus the reserved metadata column. The interval
bounds are part of the
+ // SCD2 contract and stay user-visible; only the prefixed metadata column
is engine-reserved,
+ // so a declared schema that omits just that column is accepted.
+ val declaredWithoutReserved = StructType(
+ autoCdcInferredSchema("target", ScdType.Type2)
+ .filterNot(_.name.startsWith(AutoCdcReservedNames.prefix)))
+ autoCdcGraph("target", Some(declaredWithoutReserved), ScdType.Type2)
+ .validate(spark.sessionState.conf.caseSensitiveAnalysis)
+ }
+
+ /** The engine-owned reserved metadata field from the inferred AUTO CDC
schema. */
+ private def metadataField: StructField =
+ autoCdcInferredSchema("target").fields
+ .find(_.name == AutoCdcReservedNames.cdcMetadataColName).get
+
+ test("a declared schema with an unknown reserved-prefixed column is rejected
for an AUTO CDC " +
+ "flow") {
+ // A reserved-prefixed column the engine never produces (a typo, or an
extra reserved-namespace
+ // column) is rejected up front rather than surviving reconciliation into
the target.
+ val schema = StructType(
+ dataSchema.fields :+ StructField(s"${AutoCdcReservedNames.prefix}bogus",
IntegerType))
+ val ex = intercept[AnalysisException](
+ validateGraph(autoCdcGraph("auto_cdc_flow", Some(schema))))
+ assert(ex.getCondition == "AUTOCDC_INVALID_RESERVED_SCHEMA.UNKNOWN_COLUMN")
Review Comment:
**Nit (P3):** 742a80b adds four behaviors that no test would catch
regressing:
- `validateReservedLevel`'s `ArrayType` arm
(`AutoCdcReservedNames.scala:192-197`): the nested rejection tests here
sequence by `col("version")` (LONG), so the metadata struct never contains an
array. Removing the arm would let an unknown nested field inside an
`ARRAY<STRUCT<...>>` sequencing field reach the target.
- `reconcileReservedField` building from the declared field so a `COMMENT`
survives (`AutoCdcReservedNames.scala:212-218`): no test declares metadata on a
reserved column, so reverting to `engine.copy(name = declared.name)` stays
green.
- `reconcileReservedStruct` appending the engine nested fields a declaration
omits (`AutoCdcReservedNames.scala:258`): every test declares the full metadata
struct or adds a field to it. A declaration listing only `deleteSequence`
passes validation, so dropping the append would create a target without
`upsertSequence` and fail only at MERGE time.
- These four `AUTOCDC_INVALID_RESERVED_SCHEMA` tests check only
`ex.getCondition`, not the parameters `validateReservedLevel` builds
(`tableName`, the dot-qualified `reservedColumnName`, the backtick-quoted
`columnNames`, and the engine field names). `checkError`, or the
`checkErrorInPipelineFailure` pattern the drift tests use, would pin them.
**Verification:**
- **Regression:** A declared metadata column whose array-of-struct
sequencing field contains an unknown nested field, or two nested fields
differing only in case, is rejected with the matching nested-field error, and
the reported reserved column path names the sequencing field.
- **Regression:** A declared reserved metadata column carrying a COMMENT
keeps that comment in the created target and in the schema a same-graph
consumer is planned against, while still taking the engine's type and
nullability.
- **Regression:** A declared metadata column that lists only some engine
nested fields materializes with all engine nested fields, and the AUTO CDC
MERGE writes rows successfully.
- **Behavior:** Each reserved-schema rejection reports the expected message
parameters, including the table name, the offending column or nested field
names, and the dot-qualified reserved column path.
--
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]
