[SPARK-57402][SQL][FOLLOWUP] Allow pipeline definitions in batch operation checks

[cloud-fan]

What changes were proposed in this pull request?

Followup to #57176.

Exclude streaming-table and flow definition plans from the generic batch streaming-source check.
Add Catalyst tests covering all three definition forms and document why the check applies only to
root pipeline commands while excluding materialized views.

Why are the changes needed?

UnsupportedOperationChecker says streaming sources are allowed within Spark Declarative Pipeline definitions, but currently rejects those definition plans before the pipelines subsystem can interpret them.

Does this PR introduce any user-facing change?

No. This fixes validation consistency for pipeline-definition logical plans introduced on the unreleased master branch.

How was this patch tested?

Ran before the latest review follow-up:

./build/sbt 'catalyst/testOnly org.apache.spark.sql.catalyst.analysis.UnsupportedOperationsSuite'

All 218 tests passed. The latest follow-up adds the third AUTO CDC regression case; the suite has
not been rerun locally since that addition.

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

Generated-by: Codex (GPT-5)

[viirya]

Nice cleanup — the early-return correctly hands the error back to SparkStrategies.Pipelines so users get the curated unsupportedCreatePipeline* message instead of the generic writeStream one. These three plans are always at the root of the plan tree (they're Commands), and the strategy layer also only matches the root, so checking just the top-level plan is the right scope. A few things worth addressing:

Missing test for CreateStreamingTableAutoCdc. The production isPipelineDefinition lists three forms, but the two new tests only cover CreateStreamingTableAsSelect and CreateFlowCommand. The AUTO CDC form has no coverage, so if it's ever dropped from the list there's no failing test to catch it. Could we add an assertSupportedInBatchPlan case for CreateStreamingTableAutoCdc too? (The PR description says "both definition forms," but there are actually three.)

A comment on isPipelineDefinition would help. The helper carries a few non-obvious decisions: why we return early, why only the top-level plan is checked, and — most importantly — why the list is exactly these three. In particular it excludes CreateMaterializedViewAsSelect, which I assume is intentional (a materialized view is a batch query, so an MV over a streaming source should be rejected). Spelling that out would save the next maintainer the round-trip.

Asymmetry with SparkStrategies.Pipelines worth noting. The strategy layer matches CreateFlowCommand + CreateMaterializedViewAsSelect + CreateStreamingTableAsSelect (but not CreateStreamingTableAutoCdc), while this helper lists CreateStreamingTableAsSelect + CreateStreamingTableAutoCdc + CreateFlowCommand (but not the MV). Each list is missing a different member. The MV omission here looks deliberate as noted above; the CreateStreamingTableAutoCdc gap in the strategy layer is the pre-existing hole flagged during #57176 review — probably out of scope here, but since this PR is tidying up the handling of the same plan family, it'd be good to mention it in the description (and align the two lists or file a follow-up).

I also considered whether keying off the CreatePipelineDataset trait would be cleaner than enumerating classes, but the trait also covers CreateMaterializedViewAsSelect and the bare CreateStreamingTable, which shouldn't necessarily be skipped — so the explicit list is the safer, more intentional choice. Just worth a comment to make that reasoning visible.

[viirya]

Thanks for the update — both points are well addressed.

The comment on isPipelineDefinition captures exactly the reasoning that was missing: root-plan scope, the hand-off to SparkStrategies.Pipelines for the dedicated errors, and why materialized views are deliberately left out (batch queries that should still reject streaming sources here).

The new CreateStreamingTableAutoCdc test closes the coverage gap — streamRelation sits in the source (right child) position, so it genuinely exercises the skip path and would fail if the production fix were removed.

The SparkStrategies.Pipelines list still doesn't handle CreateStreamingTableAutoCdc, but that's the pre-existing gap from #57176 and out of scope here — no need to hold this PR for it.

LGTM.