[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 both definition forms.

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:

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

All 218 tests passed. The two new regression tests fail with the generic streaming-source error when the production fix is removed.

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.