Andreas Neumann created SPARK-58116:
---------------------------------------
Summary: User-specified schema validation is skipped for pipeline
tables fed by a named flow
Key: SPARK-58116
URL: https://issues.apache.org/jira/browse/SPARK-58116
Project: Spark
Issue Type: Test
Components: Declarative Pipelines
Affects Versions: 4.2.0
Reporter: Andreas Neumann
`GraphValidations.validateUserSpecifiedSchemas` reconciles a pipeline table's
user-declared schema against the schema inferred from its incoming flows,
throwing `USER_SPECIFIED_AND_INFERRED_SCHEMA_NOT_COMPATIBLE` on a mismatch. The
validation iterates:
flows.flatMap(f => table.get(f.identifier))
i.e. it keys the table lookup on the *flow's own identifier*. A table is
therefore only found when the flow is an *implicit/default* flow whose
identifier equals the table identifier (the `CREATE STREAMING TABLE ... FLOW
AUTO CDC ...` / default-query form). For a *named* flow — e.g. the separate
`CREATE FLOW <name> AS AUTO CDC INTO <target>` form, or any explicitly-named
`CREATE FLOW ... INSERT INTO`, or a Python `apply_changes` / append flow whose
name differs from the target — `table.get(f.identifier)` returns `None`, and
the table's declared schema is never validated.
This is not SQL-specific: the validation runs on the shared dataflow graph, so
it affects the **Python** decorator API (`@dp.append_flow` /
`create_auto_cdc_flow` and friends) identically.
In Python there is no combined single-statement form — the streaming table and
its flow are always declared separately, so a Python-declared streaming table
with an explicit schema fed by a named flow almost *always* takes the
un-validated path.
The sibling validation `validateFlowStreamingness` in the same file correctly
keys on the *destination* identifier (`table.get(destTableIdentifier)`), which
is the intended semantics here, too.
Consequence: an incompatible user-declared schema on a named-flow table is not
caught at graph-validation time. It then flows into materialization
(`DatasetManager.materializeTable` uses `table.specifiedSchema` verbatim as the
output schema), so the mismatch surfaces — if at all — as a confusing
mid-stream runtime failure instead of the intended clean analysis-time error.
**Repro (SQL):** `CREATE STREAMING TABLE t (id INT, ...)` + `CREATE FLOW f AS
AUTO CDC INTO t ...` where the declared columns disagree with the flow output →
no `USER_SPECIFIED_AND_INFERRED_...` error at validation (contrast with the
combined implicit-flow form, which does error).
**Repro (Python):** declare a streaming table with an explicit `schema=` and
attach a separate named flow (`@dp.append_flow` / AUTO CDC) whose output
disagrees with that schema → same missing validation.
**Fix:** key the table lookup on `f.destinationIdentifier` (and de-dup, since
multiple flows can share a destination). The fix is in the shared
graph-validation code (`GraphValidations`), which runs on the dataflow graph
after either the SQL or the Python front end builds it, so a single change
covers both paths. A graph-level regression test (registering the table and a
named flow directly, the layer both front ends target) exercises both the
implicit- and named-flow forms.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
