Skip to content

Commit 62c57b4

Browse files
committed
[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>
1 parent a8916b6 commit 62c57b4

1 file changed

Lines changed: 42 additions & 50 deletions

File tree

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

Lines changed: 42 additions & 50 deletions
Original file line numberDiff line numberDiff line change
@@ -93,31 +93,8 @@ class ResolveDataSource(sparkSession: SparkSession) extends Rule[LogicalPlan] {
9393
ds match {
9494
// file source v2 does not support streaming yet.
9595
case provider: TableProvider if !provider.isInstanceOf[FileDataSourceV2] =>
96-
val sessionOptions = DataSourceV2Utils.extractSessionConfigs(
97-
source = provider, conf = sparkSession.sessionState.conf)
98-
val finalOptions =
99-
sessionOptions.filter { case (k, _) => !optionsWithPath.contains(k) } ++
100-
optionsWithPath.originalMap
101-
val dsOptions = new CaseInsensitiveStringMap(finalOptions.asJava)
102-
provider match {
103-
case p: PythonDataSourceV2 => p.setShortName(source)
104-
case _ =>
105-
}
106-
val table =
107-
DataSourceV2Utils.getTableFromProvider(provider, dsOptions, userSpecifiedSchema)
108-
import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Implicits._
109-
table match {
110-
case _: SupportsRead if table.supportsAny(MICRO_BATCH_READ, CONTINUOUS_READ) =>
111-
import org.apache.spark.sql.connector.catalog.CatalogV2Implicits._
112-
StreamingRelationV2(
113-
Some(provider), source, table, dsOptions,
114-
table.columns.toOutputAttributes, None, None, v1Relation,
115-
v1DataSource.streamingSourceIdentifyingName)
116-
117-
// fallback to v1
118-
// TODO (SPARK-27483): we should move this fallback logic to an analyzer rule.
119-
case _ => StreamingRelation(v1DataSource)
120-
}
96+
resolveV2StreamingRelation(
97+
provider, source, optionsWithPath, userSpecifiedSchema, v1Relation, v1DataSource)
12198

12299
case _ =>
123100
// Code path for data source v1.
@@ -154,31 +131,8 @@ class ResolveDataSource(sparkSession: SparkSession) extends Rule[LogicalPlan] {
154131
ds match {
155132
// file source v2 does not support streaming yet.
156133
case provider: TableProvider if !provider.isInstanceOf[FileDataSourceV2] =>
157-
val sessionOptions = DataSourceV2Utils.extractSessionConfigs(
158-
source = provider, conf = sparkSession.sessionState.conf)
159-
val finalOptions =
160-
sessionOptions.filter { case (k, _) => !optionsWithPath.contains(k) } ++
161-
optionsWithPath.originalMap
162-
val dsOptions = new CaseInsensitiveStringMap(finalOptions.asJava)
163-
provider match {
164-
case p: PythonDataSourceV2 => p.setShortName(source)
165-
case _ =>
166-
}
167-
val table =
168-
DataSourceV2Utils.getTableFromProvider(provider, dsOptions, userSpecifiedSchema)
169-
import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Implicits._
170-
table match {
171-
case _: SupportsRead if table.supportsAny(MICRO_BATCH_READ, CONTINUOUS_READ) =>
172-
import org.apache.spark.sql.connector.catalog.CatalogV2Implicits._
173-
StreamingRelationV2(
174-
Some(provider), source, table, dsOptions,
175-
table.columns.toOutputAttributes, None, None, v1Relation,
176-
v1DataSource.streamingSourceIdentifyingName)
177-
178-
// fallback to v1
179-
// TODO (SPARK-27483): we should move this fallback logic to an analyzer rule.
180-
case _ => StreamingRelation(v1DataSource)
181-
}
134+
resolveV2StreamingRelation(
135+
provider, source, optionsWithPath, userSpecifiedSchema, v1Relation, v1DataSource)
182136

183137
case _ =>
184138
// Code path for data source v1.
@@ -187,6 +141,44 @@ class ResolveDataSource(sparkSession: SparkSession) extends Rule[LogicalPlan] {
187141

188142
}
189143

144+
/**
145+
* Builds the streaming relation for a v2 `TableProvider`: a `StreamingRelationV2` if the
146+
* table supports micro-batch or continuous reads, else a v1 `StreamingRelation` fallback.
147+
*/
148+
private def resolveV2StreamingRelation(
149+
provider: TableProvider,
150+
source: String,
151+
optionsWithPath: CaseInsensitiveMap[String],
152+
userSpecifiedSchema: Option[StructType],
153+
v1Relation: Option[LogicalPlan],
154+
v1DataSource: DataSource): LogicalPlan = {
155+
val sessionOptions = DataSourceV2Utils.extractSessionConfigs(
156+
source = provider, conf = sparkSession.sessionState.conf)
157+
val finalOptions =
158+
sessionOptions.filter { case (k, _) => !optionsWithPath.contains(k) } ++
159+
optionsWithPath.originalMap
160+
val dsOptions = new CaseInsensitiveStringMap(finalOptions.asJava)
161+
provider match {
162+
case p: PythonDataSourceV2 => p.setShortName(source)
163+
case _ =>
164+
}
165+
val table =
166+
DataSourceV2Utils.getTableFromProvider(provider, dsOptions, userSpecifiedSchema)
167+
import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Implicits._
168+
table match {
169+
case _: SupportsRead if table.supportsAny(MICRO_BATCH_READ, CONTINUOUS_READ) =>
170+
import org.apache.spark.sql.connector.catalog.CatalogV2Implicits._
171+
StreamingRelationV2(
172+
Some(provider), source, table, dsOptions,
173+
table.columns.toOutputAttributes, None, None, v1Relation,
174+
v1DataSource.streamingSourceIdentifyingName)
175+
176+
// fallback to v1
177+
// TODO (SPARK-27483): we should move this fallback logic to an analyzer rule.
178+
case _ => StreamingRelation(v1DataSource)
179+
}
180+
}
181+
190182
private def loadV1BatchSource(
191183
source: String,
192184
userSpecifiedSchema: Option[StructType],

0 commit comments

Comments
 (0)