[
https://issues.apache.org/jira/browse/SPARK-58116?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Szehon Ho reassigned SPARK-58116:
---------------------------------
Assignee: Andreas Neumann
> 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
> Assignee: Andreas Neumann
> Priority: Major
> Labels: pull-request-available
>
> `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]
