[SPARK-54867][SS] Introduce NamedStreamingRelation wrapper for source identification during analysis

### What changes were proposed in this pull request?

This PR introduces infrastructure for tracking and propagating source identifying names through query analysis for streaming queries. It adds:

1. **StreamingSourceIdentifyingName** - A sealed trait hierarchy representing the naming state of streaming sources:
   - `UserProvided(name)` - Explicitly set via `.name()` API
   - `FlowAssigned(name)` - Assigned by external flow systems (e.g., SDP)
   - `Unassigned` - No name assigned yet (to be auto-generated)

2. **NamedStreamingRelation** - A transparent wrapper node that:
   - Carries source identifying names through the analyzer phase
   - Extends `UnaryNode` for transparent interaction with analyzer rules
   - Stays unresolved until explicitly unwrapped by a future `NameStreamingSources` analyzer rule
   - Provides `withUserProvidedName()` to attach user-specified names

3. **NAMED_STREAMING_RELATION** tree pattern for efficient pattern matching

### Why are the changes needed?

Streaming sources need stable, predictable names for:
- **Checkpoint location stability** - Schema evolution and offset tracking require consistent source identification
- **Schema lookup at specific offsets** - Analysis-time operations need to reference sources by name
- **Flow integration** - SDP and similar systems need per-source metadata paths

By introducing this wrapper during analysis (rather than at execution planning), we enable these capabilities while maintaining a clean separation between parsing, analysis, and execution phases.

### Does this PR introduce _any_ user-facing change?

No.

### How was this patch tested?

New unit tests in `NamedStreamingRelationSuite` covering:
- Source name state transitions (Unassigned → UserProvided)
- Output delegation to child plan
- Tree pattern registration
- Resolved state behavior
- String representation

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

No.

Closes #53639 from ericm-db/named-streaming-relation.

Authored-by: ericm-db <[email protected]>
Signed-off-by: Anish Shrigondekar <[email protected]>

Author: ericm-db

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/NamedStreamingRelation.scala

@@ -0,0 +1,85 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis
+
+import org.apache.spark.sql.catalyst.expressions.Attribute
+import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, UnaryNode}
+import org.apache.spark.sql.catalyst.streaming.{StreamingSourceIdentifyingName, UserProvided}
+import org.apache.spark.sql.catalyst.trees.TreePattern.{NAMED_STREAMING_RELATION, TreePattern}
+
+/**
+ * A wrapper for streaming relations that carries a source identifying name through analysis.
+ *
+ * This node is introduced during query parsing/resolution and is removed by the
+ * [[NameStreamingSources]] analyzer rule. It serves to:
+ * 1. Track user-provided source names from `.name()` API
+ * 2. Track flow-assigned names from SDP context
+ * 3. Ensure all sources have names before execution planning
+ *
+ * By extending [[UnaryNode]], this wrapper is transparent to analyzer rules - they naturally
+ * descend into the child plan via `mapChildren`, resolve it, and the wrapper persists with the
+ * updated child. This eliminates the need for explicit handling in most analyzer rules.
+ *
+ * The naming happens in the analyzer (before execution) to enable:
+ * - Schema lookup at specific offsets during analysis
+ * - Stable checkpoint locations for source evolution
+ * - SDP flow integration with per-source metadata paths
+ *
+ * @param child The underlying streaming relation (UnresolvedDataSource, etc.)
+ * @param sourceIdentifyingName The source identifying name (UserProvided, FlowAssigned,
+ *                              or Unassigned)
+ */
+case class NamedStreamingRelation(
+    child: LogicalPlan,
+    sourceIdentifyingName: StreamingSourceIdentifyingName)
+  extends UnaryNode {
+
+  override def isStreaming: Boolean = true
+
+  // Delegate output to child for transparent wrapping
+  override def output: Seq[Attribute] = child.output
+
+  // Keep unresolved until NameStreamingSources explicitly unwraps this node.
+  // This ensures the wrapper persists through analysis until we're ready to
+  // propagate the sourceIdentifyingName to the underlying StreamingRelationV2.
+  override lazy val resolved: Boolean = false
+
+  override protected def withNewChildInternal(newChild: LogicalPlan): NamedStreamingRelation =
+    copy(child = newChild)
+
+  /**
+   * Attaches a user-provided name from the `.name()` API.
+   * If nameOpt is None, returns this node unchanged.
+   *
+   * @param nameOpt The user-provided source name
+   * @return A new NamedStreamingRelation with the user name attached
+   */
+  def withUserProvidedName(nameOpt: Option[String]): NamedStreamingRelation = {
+    nameOpt.map { n =>
+      copy(sourceIdentifyingName = UserProvided(n))
+    }.getOrElse {
+      this
+    }
+  }
+
+  override val nodePatterns: Seq[TreePattern] = Seq(NAMED_STREAMING_RELATION)
+
+  override def toString: String = {
+    s"NamedStreamingRelation($child, $sourceIdentifyingName)"
+  }
+}

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/streaming/StreamingSourceIdentifyingName.scala

@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.streaming
+
+/**
+ * Represents the identifying name state for a streaming source during query analysis.
+ *
+ * Source names can be:
+ * - User-provided via the `.name()` API
+ * - Flow-assigned by external systems (e.g., SDP)
+ * - Unassigned, to be auto-generated during analysis
+ */
+sealed trait StreamingSourceIdentifyingName {
+  override def toString: String = this match {
+    case UserProvided(name) => s"""name="$name""""
+    case FlowAssigned(name) => s"""name="$name""""
+    case Unassigned => "name=<Unassigned>"
+  }
+}
+
+/**
+ * A source name explicitly provided by the user via the `.name()` API.
+ * Takes highest precedence.
+ */
+case class UserProvided(name: String) extends StreamingSourceIdentifyingName
+
+/**
+ * A source name assigned by an external flow system (e.g., SDP).
+ * Used when the source is part of a managed pipeline.
+ */
+case class FlowAssigned(name: String) extends StreamingSourceIdentifyingName
+
+/**
+ * No name has been assigned yet. The analyzer will auto-generate one if needed.
+ */
+case object Unassigned extends StreamingSourceIdentifyingName

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/trees/TreePatterns.scala

@@ -179,6 +179,7 @@ object TreePattern extends Enumeration  {
   val WITH_WINDOW_DEFINITION: Value = Value
 
   // Unresolved Plan patterns (Alphabetically ordered)
+  val NAMED_STREAMING_RELATION: Value = Value
   val PLAN_WITH_UNRESOLVED_IDENTIFIER: Value = Value
   val UNRESOLVED_EVENT_TIME_WATERMARK: Value = Value
   val UNRESOLVED_HAVING: Value = Value

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/NamedStreamingRelationSuite.scala

@@ -0,0 +1,120 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis
+
+import org.apache.spark.SparkFunSuite
+import org.apache.spark.sql.catalyst.expressions.AttributeReference
+import org.apache.spark.sql.catalyst.plans.logical.LocalRelation
+import org.apache.spark.sql.catalyst.streaming.{FlowAssigned, Unassigned, UserProvided}
+import org.apache.spark.sql.catalyst.trees.TreePattern
+import org.apache.spark.sql.types.IntegerType
+
+/**
+ * Unit tests for the NamedStreamingRelation wrapper node.
+ */
+class NamedStreamingRelationSuite extends SparkFunSuite {
+
+  private def createMockPlan(): LocalRelation = {
+    LocalRelation(AttributeReference("id", IntegerType)())
+  }
+
+  test("Unassigned sourceIdentifyingName") {
+    val plan = createMockPlan()
+    val wrapper = NamedStreamingRelation(plan, Unassigned)
+
+    assert(wrapper.child eq plan)
+    assert(wrapper.sourceIdentifyingName == Unassigned)
+  }
+
+  test("withUserProvidedName sets UserProvided name") {
+    val plan = createMockPlan()
+    val wrapper = NamedStreamingRelation(plan, Unassigned)
+    val named = wrapper.withUserProvidedName(Some("my_source"))
+
+    assert(named.sourceIdentifyingName == UserProvided("my_source"))
+  }
+
+  test("withUserProvidedName(None) returns same instance") {
+    val plan = createMockPlan()
+    val wrapper = NamedStreamingRelation(plan, Unassigned)
+    val result = wrapper.withUserProvidedName(None)
+
+    assert(result eq wrapper)
+  }
+
+  test("isStreaming returns true") {
+    val plan = createMockPlan()
+    val wrapper = NamedStreamingRelation(plan, Unassigned)
+
+    assert(wrapper.isStreaming)
+  }
+
+  test("has NAMED_STREAMING_RELATION tree pattern") {
+    val plan = createMockPlan()
+    val wrapper = NamedStreamingRelation(plan, Unassigned)
+
+    assert(wrapper.nodePatterns.contains(TreePattern.NAMED_STREAMING_RELATION))
+  }
+
+  test("resolved is false until unwrapped") {
+    val plan = createMockPlan()
+    val wrapper = NamedStreamingRelation(plan, UserProvided("test"))
+
+    // Even with a named child, wrapper stays unresolved
+    assert(!wrapper.resolved)
+  }
+
+  test("output delegates to child") {
+    val plan = createMockPlan()
+    val wrapper = NamedStreamingRelation(plan, Unassigned)
+
+    assert(wrapper.output == plan.output)
+  }
+
+  test("pattern matching on sourceIdentifyingName variants") {
+    val plan = createMockPlan()
+
+    val userProvided = NamedStreamingRelation(plan, UserProvided("test"))
+    val flowAssigned = NamedStreamingRelation(plan, FlowAssigned("0"))
+    val unassigned = NamedStreamingRelation(plan, Unassigned)
+
+    def extractName(wrapper: NamedStreamingRelation): Option[String] = {
+      wrapper.sourceIdentifyingName match {
+        case UserProvided(n) => Some(n)
+        case FlowAssigned(n) => Some(n)
+        case Unassigned => None
+      }
+    }
+
+    assert(extractName(userProvided).contains("test"))
+    assert(extractName(flowAssigned).contains("0"))
+    assert(extractName(unassigned).isEmpty)
+  }
+
+  test("toString includes sourceIdentifyingName") {
+    val plan = createMockPlan()
+
+    val userNamed = NamedStreamingRelation(plan, UserProvided("my_source"))
+    val flowNamed = NamedStreamingRelation(plan, FlowAssigned("0"))
+    val unnamed = NamedStreamingRelation(plan, Unassigned)
+
+    assert(userNamed.toString.contains("name=\"my_source\""))
+    assert(flowNamed.toString.contains("name=\"0\""))
+    assert(unnamed.toString.contains("name=<Unassigned>"))
+  }
+}