[SPARK-59362][SQL] Extract the duplicated TableProvider resolution block in ResolveDataSource

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

`ResolveDataSource` contained an identical ~25-line `TableProvider` resolution block in two `resolveOperatorsUp` cases -- the `NamedStreamingRelation`-wrapped case and the no-name streaming `UnresolvedDataSource` case. This extracts that block into a single private helper, `resolveV2StreamingRelation`, and calls it from both sites.

The helper builds a `StreamingRelationV2` when the provider's table supports micro-batch or continuous reads, and otherwise falls back to a v1 `StreamingRelation` -- exactly as the inline code did.

### Why are the changes needed?

The two blocks were byte-for-byte identical; extracting them keeps the v2-provider streaming-resolution logic in one place. Behavior is unchanged -- the helper body is the original code verbatim, and both call sites pass the same values.

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

No.

### How was this patch tested?

Existing tests; `build/sbt sql/compile` passes. This is a behavior-preserving refactor (verbatim extraction of an identical block).

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

Generated-by: Claude Code (Opus 4.8)

Closes #58653 from uros-b/dedup-resolve-datasource-block.

Authored-by: Uros <221401595+uros-b@users.noreply.github.com>
Signed-off-by: Uros Bojanic <221401595+uros-b@users.noreply.github.com>
(cherry picked from commit 62c57b4)
Signed-off-by: Uros Bojanic <221401595+uros-b@users.noreply.github.com>

Author: uros-b

sql/core/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveDataSource.scala

@@ -93,31 +93,8 @@ class ResolveDataSource(sparkSession: SparkSession) extends Rule[LogicalPlan] {
       ds match {
         // file source v2 does not support streaming yet.
         case provider: TableProvider if !provider.isInstanceOf[FileDataSourceV2] =>
-          val sessionOptions = DataSourceV2Utils.extractSessionConfigs(
-            source = provider, conf = sparkSession.sessionState.conf)
-          val finalOptions =
-            sessionOptions.filter { case (k, _) => !optionsWithPath.contains(k) } ++
-            optionsWithPath.originalMap
-          val dsOptions = new CaseInsensitiveStringMap(finalOptions.asJava)
-          provider match {
-            case p: PythonDataSourceV2 => p.setShortName(source)
-            case _ =>
-          }
-          val table =
-            DataSourceV2Utils.getTableFromProvider(provider, dsOptions, userSpecifiedSchema)
-          import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Implicits._
-          table match {
-            case _: SupportsRead if table.supportsAny(MICRO_BATCH_READ, CONTINUOUS_READ) =>
-              import org.apache.spark.sql.connector.catalog.CatalogV2Implicits._
-              StreamingRelationV2(
-                  Some(provider), source, table, dsOptions,
-                  table.columns.toOutputAttributes, None, None, v1Relation,
-                  v1DataSource.streamingSourceIdentifyingName)
-
-            // fallback to v1
-            // TODO (SPARK-27483): we should move this fallback logic to an analyzer rule.
-            case _ => StreamingRelation(v1DataSource)
-          }
+          resolveV2StreamingRelation(
+            provider, source, optionsWithPath, userSpecifiedSchema, v1Relation, v1DataSource)
 
         case _ =>
           // Code path for data source v1.
@@ -154,31 +131,8 @@ class ResolveDataSource(sparkSession: SparkSession) extends Rule[LogicalPlan] {
       ds match {
         // file source v2 does not support streaming yet.
         case provider: TableProvider if !provider.isInstanceOf[FileDataSourceV2] =>
-          val sessionOptions = DataSourceV2Utils.extractSessionConfigs(
-            source = provider, conf = sparkSession.sessionState.conf)
-          val finalOptions =
-            sessionOptions.filter { case (k, _) => !optionsWithPath.contains(k) } ++
-            optionsWithPath.originalMap
-          val dsOptions = new CaseInsensitiveStringMap(finalOptions.asJava)
-          provider match {
-            case p: PythonDataSourceV2 => p.setShortName(source)
-            case _ =>
-          }
-          val table =
-            DataSourceV2Utils.getTableFromProvider(provider, dsOptions, userSpecifiedSchema)
-          import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Implicits._
-          table match {
-            case _: SupportsRead if table.supportsAny(MICRO_BATCH_READ, CONTINUOUS_READ) =>
-              import org.apache.spark.sql.connector.catalog.CatalogV2Implicits._
-              StreamingRelationV2(
-                  Some(provider), source, table, dsOptions,
-                  table.columns.toOutputAttributes, None, None, v1Relation,
-                  v1DataSource.streamingSourceIdentifyingName)
-
-            // fallback to v1
-            // TODO (SPARK-27483): we should move this fallback logic to an analyzer rule.
-            case _ => StreamingRelation(v1DataSource)
-          }
+          resolveV2StreamingRelation(
+            provider, source, optionsWithPath, userSpecifiedSchema, v1Relation, v1DataSource)
 
         case _ =>
           // Code path for data source v1.
@@ -187,6 +141,44 @@ class ResolveDataSource(sparkSession: SparkSession) extends Rule[LogicalPlan] {
 
   }
 
+  /**
+   * Builds the streaming relation for a v2 `TableProvider`: a `StreamingRelationV2` if the
+   * table supports micro-batch or continuous reads, else a v1 `StreamingRelation` fallback.
+   */
+  private def resolveV2StreamingRelation(
+      provider: TableProvider,
+      source: String,
+      optionsWithPath: CaseInsensitiveMap[String],
+      userSpecifiedSchema: Option[StructType],
+      v1Relation: Option[LogicalPlan],
+      v1DataSource: DataSource): LogicalPlan = {
+    val sessionOptions = DataSourceV2Utils.extractSessionConfigs(
+      source = provider, conf = sparkSession.sessionState.conf)
+    val finalOptions =
+      sessionOptions.filter { case (k, _) => !optionsWithPath.contains(k) } ++
+      optionsWithPath.originalMap
+    val dsOptions = new CaseInsensitiveStringMap(finalOptions.asJava)
+    provider match {
+      case p: PythonDataSourceV2 => p.setShortName(source)
+      case _ =>
+    }
+    val table =
+      DataSourceV2Utils.getTableFromProvider(provider, dsOptions, userSpecifiedSchema)
+    import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Implicits._
+    table match {
+      case _: SupportsRead if table.supportsAny(MICRO_BATCH_READ, CONTINUOUS_READ) =>
+        import org.apache.spark.sql.connector.catalog.CatalogV2Implicits._
+        StreamingRelationV2(
+            Some(provider), source, table, dsOptions,
+            table.columns.toOutputAttributes, None, None, v1Relation,
+            v1DataSource.streamingSourceIdentifyingName)
+
+      // fallback to v1
+      // TODO (SPARK-27483): we should move this fallback logic to an analyzer rule.
+      case _ => StreamingRelation(v1DataSource)
+    }
+  }
+
   private def loadV1BatchSource(
       source: String,
       userSpecifiedSchema: Option[StructType],