[SPARK-58116][SDP] Validate user-specified schema for tables fed by a named flow

[anew]

What changes were proposed in this pull request?

GraphValidations.validateUserSpecifiedSchemas looked up a table by the incoming flow's own identifier (table.get(f.identifier)). That only matches when the flow is an implicit/default flow whose identifier equals its destination table's. For a named flow (e.g. CREATE FLOW <name> AS AUTO CDC INTO <target>, or any explicitly-named flow) the lookup returned None, so the table's declared schema was never validated against the inferred schema.

This changes the lookup to key on f.destinationIdentifier (with .distinct, since multiple flows can share a destination), matching the sibling validateFlowStreamingness validation.

Why are the changes needed?

An incompatible user-declared schema on a named-flow table went undetected at graph-validation time and surfaced only as a confusing mid-stream runtime failure at materialization.

Does this introduce any user-facing change?

Yes: an incompatible user-specified schema on a table fed by a named flow now fails with USER_SPECIFIED_AND_INFERRED_SCHEMA_NOT_COMPATIBLE at validation time, as it already did for the implicit-flow form.

How was this patch tested?

New UserSpecifiedSchemaValidationSuite covers validateUserSpecifiedSchemas across the flow-identity axis (implicit flow, where the flow identifier equals the destination table's, vs. named flow, where it differs) and the schema-compatibility axis, for both plain and AUTO CDC flows:

• Plain flow (inferred schema == the source's data columns), implicit and named: a compatible declared schema is accepted; one missing a data column is rejected.
• AUTO CDC flow (inferred schema == the data columns plus the appended reserved __spark_autocdc_metadata column), implicit and named: a data-only declared schema is rejected (it omits the metadata column); one that also includes the metadata column is accepted.

The named-flow cases fail without the fix (the table lookup returns None, so validation is skipped). The AUTO CDC positive cases derive the expected schema from the resolved graph's inferred schema rather than hardcoding the reserved metadata struct. Also ran ConnectValid/Invalid, SqlPipeline, and AutoCdc pipeline suites green.

Was this patch authored or co-authored using generative AI tooling?

Generated-by: Claude Opus 4.8

[szehon-ho]

flowsTo is already grouped by destinationIdentifier and its keys are distinct, so we can iterate it directly and drop .distinct. This also mirrors the sibling validateFlowStreamingness, which walks flowsTo, keeping the two validations structurally aligned.

Suggested change

	// Look up the destination table of each flow, not the table matching the flow's own
	// identifier. The two coincide only for an implicit/default flow (whose identifier equals its
	// destination table's); for a named flow (e.g. `CREATE FLOW <name> AS AUTO CDC INTO <target>`)
	// they differ, and keying on the flow identifier would silently skip validation. `distinct`
	// collapses the multiple flows that can share a single destination.
	flows.flatMap(f => table.get(f.destinationIdentifier)).distinct.foreach { t: TableElement =>
	flowsTo.keys.flatMap(table.get).foreach { t: TableElement =>

[anew]

yes good catch

[szehon-ho]

Since the fix is that the error is now attributed to the right table for named flows, assert the message actually names the target too, not just the error condition. targetIdentifier is already in scope, so no signature change is needed.

Suggested change

	private def assertSchemaIncompatible(graph: DataflowGraph): Unit = {
	val ex = intercept[AnalysisException](graph.validate())
	assert(ex.getCondition == "USER_SPECIFIED_AND_INFERRED_SCHEMA_NOT_COMPATIBLE")
	}
	private def assertSchemaIncompatible(graph: DataflowGraph): Unit = {
	val ex = intercept[AnalysisException](graph.validate())
	assert(ex.getCondition == "USER_SPECIFIED_AND_INFERRED_SCHEMA_NOT_COMPATIBLE")
	assert(ex.getMessage.contains(targetIdentifier.unquotedString))
	}

[anew]

ok that makes sense

[szehon-ho]

nit: can we remove this comment 'flowsTo' is grouped by destination... mirroring 'validateFlowStreamingness'

(seems too low level)

[anew]

ok

[szehon-ho]

LGTM. Nice catch on this latent bug: keying the table lookup on the flow's own identifier meant named flows (e.g. CREATE FLOW <name> AS AUTO CDC INTO <target>) silently skipped user-specified schema validation, since the flow and destination identifiers only coincide for implicit flows. Keying on the destination via flowsTo fixes it and aligns with the sibling validateFlowStreamingness. The test matrix over the flow-identity and schema-compatibility axes (plain and AUTO CDC) is thorough.

[szehon-ho]

One more thing worth calling out about the latent bug: keying on f.identifier didn't only skip validation for named flows, it could validate the wrong table. If a named flow's identifier happened to collide with some other table's identifier, the old table.get(f.identifier) would pull up that unrelated table and validate it, while the flow's actual destination table went unchecked. Keying on the destination identifier removes that misattribution too.