[SPARK-57402][SQL][SDP] Register AUTO CDC SQL syntax with the dataflow graph

[anew]

What changes were proposed in this pull request? Hook up the two AUTO CDC SQL constructs introduced in SPARK-56249 to the Spark Declarative Pipelines dataflow graph in SqlGraphRegistrationContext:

1. CREATE STREAMING TABLE <name> FLOW AUTO CDC FROM <source> ..., parsed into CreateStreamingTableAutoCdc, now registers the streaming table and an AutoCdcFlow that targets it.
2. CREATE FLOW <name> AS AUTO CDC INTO <target> FROM <source> ..., parsed into a CreateFlowCommand wrapping an AutoCdcIntoCommand, now registers an AutoCdcFlow from the named flow into the target dataset.

A shared buildChangeArgs helper converts the parse-time expressions and unresolved attributes into the ChangeArgs consumed by AutoCdcFlow (keys, sequencing, delete condition, include/exclude column selection). SQL AUTO CDC only supports SCD Type 1, matching the Connect/proto path.

Why are the changes needed?

Before this change the AUTO CDC SQL syntax parsed successfully but was rejected during graph registration, so it could not be used to define pipeline datasets or flows.

Does this PR introduce any user-facing change?

Yes. AUTO CDC SQL statements now register datasets and flows in a pipeline.

How was this patch tested?

Added registration and end-to-end execution tests to SqlPipelineSuite covering both syntaxes, the optional clauses (APPLY AS DELETE WHEN, COLUMNS, COLUMNS * EXCEPT), and the multipart-flow-name error. Full SqlPipelineSuite passes (50 tests).

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

Generated-by: Opus 4.8

[szehon-ho]

some review comments

[szehon-ho]

If users write CREATE STREAMING TABLE target (id INT, name STRING) FLOW AUTO CDC ..., this stores the listed columns as the table’s full specified schema. But Auto CDC later appends _cdc_metadata to the flow schema, so validation can reject an otherwise natural data-column-only declaration. Could we either reject column lists for this SQL form, or interpret them as the data schema and add the metadata field internally before validation?

[anew]

This is a very interesting question. What if the given schema does not match the keys+columns? That should be rejected, I guess. A similar case is where the the streaming table is declared without a flow but with a schema. In that case we do not know(at parse time) whether the flow that is also declared on its own is an Auto CDC flow? It appears to me that need to allow specifying the schema, but interpret it as exclusive of the metadata column.

[anew]

Thanks for catching this — this matches where I landed too, and I went and investigated both the current behavior and the "user specifies data-columns-only" approach. Two findings worth sharing:

1. The two-statement form doesn't fail at validation today — it silently skips validation entirely. I expected this case (CREATE STREAMING TABLE target (id INT, name STRING) + separate CREATE FLOW ... AS AUTO CDC INTO target) to at least be caught by validateUserSpecifiedSchemas, but it isn't. That check keys the table lookup on the flow's identifier:

 flows.flatMap(f => table.get(f.identifier))

which only matches when the flow is an implicit/default flow (flow id == table id) — i.e. the combined form. For a named flow, table.get(f.identifier) is None, so the declared schema is never validated (its sibling validateFlowStreamingness correctly keys on the destination identifier). So the mismatch isn't even caught at graph-validation time for the two-statement pattern — it surfaces later as a mid-stream UNRESOLVED_COLUMN ... _cdc_metadata at materialization. That looks like a pre-existing bug independent of this PR, it pre-exists also for Python, and I think it deserves its own fix (keying on destinationIdentifier) + regression test.

2. How to fix it? I propose to make a quick fix for both SQL and Python to use the destinationIdentifier for lookup, in a separate PR. Then in a separate Jira, we can discuss whether we want to allow an explicit schema that omits the metadata column, in all cases. This would be beyond the scope of this PR because it changes existing, released behavior in a non-regressing way, but a change nonetheless.

What do you think, @szehon-ho ?

[anew]

I created SPARK-58116 for the quick fix.

[anew]

... and SPARK-58118 for the longer-term data-columns-only approach.

[szehon-ho]

This registers CreateStreamingTableAutoCdc in the pipeline graph path, but the normal Spark execution guard in SparkStrategies.Pipelines only rejects CreateFlowCommand and CreateStreamingTableAsSelect today. If someone runs this statement through regular spark.sql outside pipeline registration, it may miss the curated unsupported STREAMING TABLE error. Can we add a matching CreateStreamingTableAutoCdc case there?

[anew]

great catch!

[anew]

I addressed this, and in the process discovered that the same issue existed for a blank CREATE STREAMING TABLE (without a flow), and also for the CREATE FLOW AS AUTO CDC. The latter actually resulted in a CreateFlow command with a child AutoCdcIntoCommand that was a command in itself. I realized that this should not be a command (because it can never occur by itself, only within a CREATE FLOW), so I changed this to be named just AutoCdcInto and not extending Command anymore. Also added tests for all these cases. I think this is cleaner now.

[szehon-ho]

A few inline comments on SqlGraphRegistrationContext. Also: CreateStreamingTableAutoCdc in v2Commands.scala still has a scaladoc saying execution support is coming in SPARK-57402 — worth updating since this PR wires up registration (that file isn't in the diff so I couldn't anchor an inline comment there).

[szehon-ho]

If the user writes CREATE STREAMING TABLE target (id INT, name STRING) FLOW AUTO CDC ..., we store the column list as specifiedSchema. But AutoCdcMergeFlow always appends _cdc_metadata to the flow output schema, so GraphValidations.validateUserSpecifiedSchemas later rejects the pipeline with USER_SPECIFIED_AND_INFERRED_SCHEMA_NOT_COMPATIBLE (specified = data cols only, inferred = data cols + _cdc_metadata). The parser allows this form (AutoCdcParserSuite), but the new E2E tests avoid it by omitting the column list.

Would it be better to reject it here instead of letting registration succeed and failing at graph validation? For example:

if (cst.columns.nonEmpty) {
  throw SqlGraphElementRegistrationException(
    msg = "Explicit column lists are not supported for AUTO CDC streaming tables; " +
          "omit the column list and let schema be inferred from the source.",
    queryOrigin = queryOrigin)
}

That gives a clear, immediate error for the single-statement form. The tradeoff is it doesn't cover the two-statement pattern:

CREATE STREAMING TABLE target (id INT, name STRING);
CREATE FLOW f AS AUTO CDC INTO target ...

where CreateStreamingTableHandler registers the schema and a later CREATE FLOW attaches AUTO CDC — rejection would need to live in validation (or we'd need to reject column lists on all standalone CREATE STREAMING TABLE statements, which is too broad).

Longer term, treating specifiedSchema as data-columns-only for AUTO CDC targets (and appending _cdc_metadata during validation/materialization) is probably the right UX. But if we want a small fix for this PR, rejecting non-empty cst.columns here seems reasonable for the combined syntax — WDYT?

[anew]

In the existing Python API, there is no combined command - you have to create the streaming table and the flow separately. But if you create the table with a schema, then the flow schema must match it exactly. So it will fail if you don't specify the metadata column. The same will happen in SQL if you create the streaming table and the flow separately.

I am not opposed to restricting the combined syntax as you suggest, but it does not solve the problem for separate declarations.

Let me follow your suggestion for now and disallow a schema in this case. I will follow up with a separate Jira for the data-columns-only approach.

[szehon-ho]

AutoCdcParserSuite has a test that CLUSTER BY is honored at parse time — it lands in cst.partitioning as a ClusterByTransform. But this call routes all of partitioning through PartitionHelper, which only accepts IdentityTransform (PARTITIONED BY) and throws Invalid partitioning transform (ClusterByTransform(...)) for CLUSTER BY. So the parser test and pipeline registration disagree.

Options:

1. Split ClusterByTransform out of partitioning into clusterCols here (Connect already sets clusterCols from clustering columns).
2. Reject CLUSTER BY explicitly in this handler with a clear error, and adjust/remove the parser test expectation if we don't support it yet.

Same pattern exists in the other SQL table handlers (CreateStreamingTableHandler, etc.), but AUTO CDC is the one with a parser test claiming CLUSTER BY works.

[szehon-ho]

Nit: "Unable flow type..." → "Unknown flow type..." (or similar).