[SPARK-59395][SDP] Expose deterministic flow execution attribution

[marcuslin123]

What changes were proposed in this pull request?

This PR adds public Spark Declarative Pipelines metadata for correlating a logical flow and an
individual flow execution attempt with the Spark SQL executions and jobs that it launches.

Each flow execution receives a unique UUID. While batch and streaming flows start their Spark
work, the implementation exposes the canonical flow identifier and execution UUID as Spark local
properties and adds a deterministic execution tag. Batch execution also carries the submitting
thread's local properties into the worker thread and restores the worker state afterward.

Why are the changes needed?

Spark Declarative Pipelines currently does not provide a deterministic way for external observers
to attribute Spark SQL executions and jobs to a specific flow attempt. Consumers otherwise need to
infer attribution from text or timestamps, which is ambiguous when flows or attempts overlap.

Does this PR introduce any user-facing change?

Yes. Listeners and event-log consumers can use PipelineExecutionMetadata to read the logical flow
identifier and unique flow execution ID from Spark job properties, then correlate each job with its
Spark SQL execution ID. The execution ID is also exposed as a Spark job tag and is available on
SparkListenerSQLExecutionStart.

How was this patch tested?

Added focused tests for metadata naming, scoped restoration, failure cleanup, batch attribution,
multiple batch attempts, reused worker-thread isolation, streaming attribution, and exact
job-to-SQL-execution correlation.

Ran:

build/sbt pipelines/Test/compile pipelines/scalastyle pipelines/Test/scalastyle \
  'pipelines/testOnly org.apache.spark.sql.pipelines.PipelineExecutionMetadataSuite' \
  'pipelines/testOnly org.apache.spark.sql.pipelines.graph.GraphExecutionSuite' \
  'pipelines/testOnly org.apache.spark.sql.pipelines.graph.TriggeredGraphExecutionSuite' \
  'pipelines/testOnly org.apache.spark.sql.pipelines.graph.SinkExecutionSuite'

All 28 tests passed, and both production and test Scalastyle checks completed with no errors or
warnings.

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

Generated-by: OpenAI Codex (GPT-5), used for code and test assistance under author direction and
review.