This is an automated email from the ASF dual-hosted git repository.
szehon-ho pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 0ac43804a051 [SPARK-58517][SDP] Honor spark.sql.caseSensitive in table
schema evolution
0ac43804a051 is described below
commit 0ac43804a0510092632ec0164f1e80f71e08ede8
Author: Andreas Neumann <[email protected]>
AuthorDate: Thu Aug 6 11:29:12 2026 -0700
[SPARK-58517][SDP] Honor spark.sql.caseSensitive in table schema evolution
### What changes were proposed in this pull request?
Declarative Pipelines' table schema evolution in
`DatasetManager.evolveTable` compared schemas **case-sensitively regardless of
the session's `spark.sql.caseSensitive`**. Under the default (case-insensitive)
session, a target column `value` and an incoming column `Value` were treated as
distinct, so the target was evolved to carry *both* columns -- a schema that is
internally inconsistent under the resolver the rest of the engine uses.
The root cause is on two utilities on the evolution path, each taking the
case-sensitive default and never forwarding the session conf:
- `SchemaMergingUtils.mergeSchemas` called `StructType.merge(tableSchema,
dataSchema)` positionally, leaving `StructType.merge`'s `caseSensitive: Boolean
= true` parameter at its default.
- `SchemaInferenceUtils.diffSchemas` built `name -> field` maps and
computed `targetFields.keySet.diff(currentFields.keySet)` -- an exact-string
set difference, also case-sensitive.
This PR threads case-sensitivity through both:
- `mergeSchemas` and `diffSchemas` gain a `caseSensitive: Boolean = true`
parameter. The default preserves existing behavior for the schema-*inference*
callers (`inferSchemaFromFlows`, `DataflowGraph.inferredSchema`), which
intentionally merge case-sensitively.
- `DatasetManager.evolveTable` forwards
`context.spark.sessionState.conf.caseSensitiveAnalysis` at both call sites (the
target table and the AutoCDC auxiliary table).
- With case-insensitivity, `StructType.merge` folds the incoming `Value`
onto the existing `value` (the left field's name/position wins), so no
duplicate column is added; `diffSchemas` matches on the normalized name while
emitting the column's already-persisted name, so a case-only difference is an
in-place no-op rather than a spurious drop-then-add, and no rename is produced.
This is not AutoCDC-specific -- `SchemaMergingUtils` /
`SchemaInferenceUtils` are general pipeline-table utilities, so the fix covers
any SDP table that evolves its schema under a case-insensitive session.
### Why are the changes needed?
It is a correctness bug. Evolving a table under the default
case-insensitive session with a column that differs only in case from an
existing one produced a target schema with two columns (`value` and `Value`)
that the engine's own case-insensitive resolver cannot tell apart. The corrupt
schema then failed downstream where that resolver runs:
- AutoCDC SCD2: `COLUMN_ALREADY_EXISTS`, raised by
`ResolveUnion.checkColumnNames` over the `unionByName` in
`Scd2ForeachBatchHandler.reconcileMicrobatch` (the affected-rows union reads
the now-two-column target back).
- AutoCDC SCD1: `AMBIGUOUS_REFERENCE`, deeper in the MERGE plan (no
reconcile union).
The same user mistake thus surfaced as two different, confusing error
conditions -- both downstream symptoms of the corrupt evolved schema rather
than the root cause.
### Does this PR introduce _any_ user-facing change?
Yes, a bug fix (only within the unreleased `master` / `branch-4.x`; the
buggy behavior was never in a release). Under case-insensitive resolution
(Spark's default), a pipeline that emits a column differing only in case from
an existing target column now evolves the table as a no-op (the incoming value
is written to the existing column) instead of corrupting the schema and failing
with `COLUMN_ALREADY_EXISTS` (SCD2) or `AMBIGUOUS_REFERENCE` (SCD1). Under
`spark.sql.caseSensitive=tru [...]
### How was this patch tested?
Added and updated tests:
- `SchemaInferenceUtilsSuite` -- new unit tests for `diffSchemas`:
case-only difference is distinct under case-sensitive, a no-op under
case-insensitive, and a case-insensitive type change addresses the current
(persisted) column name.
- `MaterializeTablesSuite` -- an end-to-end pair (non-AutoCDC path)
asserting no `alterTable` is issued and the persisted column keeps its
original case under case-insensitive resolution, versus a column being added
under case-sensitive.
- `AutoCdcScd1SchemaEvolutionSuite` -- rewrote the case-only-difference
test from asserting the old `AMBIGUOUS_REFERENCE` failure to asserting the
fixed no-op-merge success; refreshed the suite header and a now-stale comment
in `AutoCdcScd1KeyDriftSuite`.
Ran the full `pipelines` module test suite. All pass except two failures
(`TriggeredGraphExecutionSuite`: "stream failure on deletes and updates",
"stopping a pipeline mid-execution") that I confirmed are pre-existing on clean
`master` (they fail identically with this change stashed), so this PR
introduces no regressions.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 4.8, Codex gpt-5.5
Closes #57722 from anew/spark-58517-schema-evolution-case-sensitivity.
Authored-by: Andreas Neumann <[email protected]>
Signed-off-by: Szehon Ho <[email protected]>
(cherry picked from commit ef128e33efb625d83f670fa607fb6626dc9d84ec)
Signed-off-by: Szehon Ho <[email protected]>
---
.../src/main/resources/error/error-conditions.json | 8 +
.../connect/pipelines/PythonPipelineSuite.scala | 55 +--
.../SparkDeclarativePipelinesServerSuite.scala | 7 +-
.../spark/sql/pipelines/autocdc/ChangeArgs.scala | 5 +-
.../pipelines/graph/AutoCdcAuxiliaryTable.scala | 8 +-
.../graph/CoreDataflowNodeProcessor.scala | 11 +-
.../spark/sql/pipelines/graph/DataflowGraph.scala | 62 ++--
.../spark/sql/pipelines/graph/DatasetManager.scala | 82 ++++-
.../apache/spark/sql/pipelines/graph/Flow.scala | 21 +-
.../spark/sql/pipelines/graph/FlowExecution.scala | 14 +-
.../spark/sql/pipelines/graph/GraphErrors.scala | 29 ++
.../sql/pipelines/graph/GraphValidations.scala | 6 +-
.../sql/pipelines/graph/PipelineExecution.scala | 3 +-
.../spark/sql/pipelines/graph/elements.scala | 8 +-
.../sql/pipelines/util/SchemaInferenceUtils.scala | 151 +++++++-
.../sql/pipelines/util/SchemaMergingUtils.scala | 23 +-
.../sql/pipelines/autocdc/AutoCdcFlowSuite.scala | 5 +-
.../pipelines/graph/AutoCdcConfigDriftSuite.scala | 23 ++
.../graph/AutoCdcScd1AuxiliaryTableSpecSuite.scala | 4 +-
.../pipelines/graph/AutoCdcScd1KeyDriftSuite.scala | 50 ++-
.../graph/AutoCdcScd1SchemaEvolutionSuite.scala | 56 ++-
.../graph/AutoCdcScd2AuxiliaryTableSpecSuite.scala | 7 +-
.../graph/ConnectInvalidPipelineSuite.scala | 53 +--
.../graph/ConnectValidPipelineSuite.scala | 4 +-
.../pipelines/graph/MaterializeTablesSuite.scala | 408 ++++++++++++++++++++-
.../sql/pipelines/graph/SqlPipelineSuite.scala | 32 +-
.../graph/TriggeredGraphExecutionSuite.scala | 4 +-
.../graph/UserSpecifiedSchemaValidationSuite.scala | 68 +++-
.../pipelines/util/SchemaInferenceUtilsSuite.scala | 130 ++++++-
.../utils/TestGraphRegistrationContext.scala | 8 +-
30 files changed, 1133 insertions(+), 212 deletions(-)
diff --git a/common/utils/src/main/resources/error/error-conditions.json
b/common/utils/src/main/resources/error/error-conditions.json
index 867e2d9dae30..a575551475be 100644
--- a/common/utils/src/main/resources/error/error-conditions.json
+++ b/common/utils/src/main/resources/error/error-conditions.json
@@ -1247,6 +1247,14 @@
],
"sqlState" : "KD009"
},
+ "CONFLICTING_PIPELINE_FLOW_CASE_SENSITIVITY" : {
+ "message" : [
+ "The flows writing to table <tableName> do not agree on '<configKey>':
<flowConfigurations>.",
+ "The effective value determines whether column names that differ only in
case identify the same column, so the table's schema would depend on the order
the flows are evaluated in.",
+ "Set '<configKey>' to the same value for every flow writing to this
table."
+ ],
+ "sqlState" : "42KD9"
+ },
"CONNECT" : {
"message" : [
"Generic Spark Connect error."
diff --git
a/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/pipelines/PythonPipelineSuite.scala
b/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/pipelines/PythonPipelineSuite.scala
index 17ce101bf181..ba44051ee692 100644
---
a/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/pipelines/PythonPipelineSuite.scala
+++
b/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/pipelines/PythonPipelineSuite.scala
@@ -137,14 +137,16 @@ class PythonPipelineSuite
TableIdentifier(catalog = Option("spark_catalog"), database =
Option("default"), table = name)
}
+ private def sessionCaseSensitive: Boolean =
spark.sessionState.conf.caseSensitiveAnalysis
+
test("basic") {
val graph = buildGraph("""
|@dp.table
|def table1():
| return spark.readStream.format("rate").load()
|""".stripMargin)
- .resolve()
- .validate()
+ .resolve(sessionCaseSensitive)
+ .validate(sessionCaseSensitive)
assert(graph.flows.size == 1)
assert(graph.tables.size == 1)
}
@@ -302,7 +304,7 @@ class PythonPipelineSuite
|def a():
| return spark.range(5)
|""".stripMargin)
- val resolvedGraph = graph.resolve().validate()
+ val resolvedGraph =
graph.resolve(sessionCaseSensitive).validate(sessionCaseSensitive)
assert(resolvedGraph.tables.size == 4)
assert(resolvedGraph.resolvedFlows.size == 4)
}
@@ -316,7 +318,7 @@ class PythonPipelineSuite
|@dp.append_flow(target = "a")
|def supplement():
| return spark.readStream.format("rate").load()
- |""".stripMargin).resolve().validate()
+
|""".stripMargin).resolve(sessionCaseSensitive).validate(sessionCaseSensitive)
assert(graph.tables.map(_.identifier.table).toSet == Set("a"))
assert(graph.resolvedFlows.size == 2)
@@ -377,7 +379,7 @@ class PythonPipelineSuite
|@dp.table
|def d():
| return spark.sql("SELECT * FROM STREAM src")
- |""".stripMargin).resolve().validate()
+
|""".stripMargin).resolve(sessionCaseSensitive).validate(sessionCaseSensitive)
assert(
graph.table.keySet == Set(
@@ -424,7 +426,7 @@ class PythonPipelineSuite
|@dp.table
|def e():
| return spark.sql("SELECT * FROM STREAM spark_catalog.default.src")
- |""".stripMargin).resolve().validate()
+
|""".stripMargin).resolve(sessionCaseSensitive).validate(sessionCaseSensitive)
assert(
graph.tables.map(_.identifier).toSet == Set(
@@ -466,7 +468,7 @@ class PythonPipelineSuite
|@dp.table
|def e():
| return spark.sql("SELECT * FROM STREAM src")
- |""".stripMargin).resolve()
+ |""".stripMargin).resolve(sessionCaseSensitive)
assert(graph.resolutionFailedFlows.size == 5)
graph.resolutionFailedFlows.foreach { flow =>
@@ -496,7 +498,7 @@ class PythonPipelineSuite
|@dp.table
|def e():
| return spark.sql("SELECT * FROM STREAM spark_catalog.default.src")
- |""".stripMargin).resolve()
+ |""".stripMargin).resolve(sessionCaseSensitive)
assert(graph.resolutionFailedFlows.size == 5)
graph.resolutionFailedFlows.foreach { flow =>
assert(flow.failure.head.getMessage.contains("[TABLE_OR_VIEW_NOT_FOUND]"))
@@ -517,7 +519,7 @@ class PythonPipelineSuite
|@dp.materialized_view
|def mv_from_read_table_df():
| return read_table_df
- |""".stripMargin).resolve().validate()
+
|""".stripMargin).resolve(sessionCaseSensitive).validate(sessionCaseSensitive)
assert(
graph.resolvedFlows.map(_.identifier).toSet == Set(
@@ -541,7 +543,7 @@ class PythonPipelineSuite
|def mv_from_read_table_df():
| return read_table_df
|
- |""".stripMargin).resolve().validate()
+
|""".stripMargin).resolve(sessionCaseSensitive).validate(sessionCaseSensitive)
assert(
graph.resolvedFlows.map(_.identifier).toSet == Set(
graphIdentifier("mv_from_read_table_df"),
@@ -608,7 +610,7 @@ class PythonPipelineSuite
|@dp.table(name = "schema_b.st_2")
|def irrelevant_3():
| return spark.readStream.format("rate").load()
- |""".stripMargin).resolve()
+ |""".stripMargin).resolve(sessionCaseSensitive)
// validate these dataset are properly fully qualified
assert(
@@ -659,7 +661,7 @@ class PythonPipelineSuite
|@dp.table(name = "some_catalog.some_schema.st")
|def irrelevant_2():
| return spark.readStream.format("rate").load()
- |""".stripMargin).resolve()
+ |""".stripMargin).resolve(sessionCaseSensitive)
}
assert(graphTry.isSuccess)
assert(
@@ -689,7 +691,7 @@ class PythonPipelineSuite
|@dp.temporary_view(name= "view_3")
|def irrelevant_2():
| return spark.read.table("view_1")
- |""".stripMargin).resolve()
+ |""".stripMargin).resolve(sessionCaseSensitive)
// views are temporary views, so they're not fully qualified.
assert(
Set("view_1", "view_2", "view_3").subsetOf(
@@ -725,7 +727,7 @@ class PythonPipelineSuite
|@dp.append_flow(target = "default.a")
|def supplement():
| return spark.readStream.format("rate").load()
- |""".stripMargin).resolve().validate()
+
|""".stripMargin).resolve(sessionCaseSensitive).validate(sessionCaseSensitive)
assert(graph.tables.map(_.identifier) == Seq(graphIdentifier("a")))
assert(
@@ -894,8 +896,8 @@ class PythonPipelineSuite
|def table_with_string_schema():
| return spark.range(5).withColumn("name", lit("test"))
|""".stripMargin)
- .resolve()
- .validate()
+ .resolve(sessionCaseSensitive)
+ .validate(sessionCaseSensitive)
assert(graph.flows.size == 1)
assert(graph.tables.size == 1)
@@ -917,8 +919,8 @@ class PythonPipelineSuite
|def table_with_struct_schema():
| return spark.range(5).withColumn("name", lit("test"))
|""".stripMargin)
- .resolve()
- .validate()
+ .resolve(sessionCaseSensitive)
+ .validate(sessionCaseSensitive)
assert(graph.flows.size == 1)
assert(graph.tables.size == 1)
@@ -936,9 +938,9 @@ class PythonPipelineSuite
|def table_with_wrong_schema():
| return spark.range(5).withColumn("wrong_column", lit("test"))
|""".stripMargin)
- .resolve()
+ .resolve(sessionCaseSensitive)
- val ex = intercept[AnalysisException] { graph.validate() }
+ val ex = intercept[AnalysisException] {
graph.validate(sessionCaseSensitive) }
assert(ex.getMessage.contains("has a user-specified schema that is
incompatible"))
assert(ex.getMessage.contains("table_with_wrong_schema"))
}
@@ -955,9 +957,9 @@ class PythonPipelineSuite
|def table_with_wrong_struct_schema():
| return spark.range(5).withColumn("different_column", lit("test"))
|""".stripMargin)
- .resolve()
+ .resolve(sessionCaseSensitive)
- val ex = intercept[AnalysisException] { graph.validate() }
+ val ex = intercept[AnalysisException] {
graph.validate(sessionCaseSensitive) }
assert(ex.getMessage.contains("has a user-specified schema that is
incompatible"))
assert(ex.getMessage.contains("table_with_wrong_struct_schema"))
}
@@ -1205,7 +1207,7 @@ class PythonPipelineSuite
| keys = ["value"],
| sequence_by = "timestamp",
|)
- |""".stripMargin).resolve()
+ |""".stripMargin).resolve(sessionCaseSensitive)
val resolvedFlow = graph.resolvedFlow(graphIdentifier("target"))
assert(resolvedFlow.inputs == Set(graphIdentifier("src")))
@@ -1234,7 +1236,8 @@ class PythonPipelineSuite
|""".stripMargin,
defaultCatalog = Some("my_catalog"),
defaultDatabase = Some("my_db"),
- setupSql = Some("CREATE NAMESPACE IF NOT EXISTS
my_catalog.my_db")).resolve()
+ setupSql = Some("CREATE NAMESPACE IF NOT EXISTS my_catalog.my_db"))
+ .resolve(sessionCaseSensitive)
val resolvedFlow =
graph.resolvedFlow(TableIdentifier("target", Some("my_db"),
Some("my_catalog")))
@@ -1257,7 +1260,7 @@ class PythonPipelineSuite
| keys = ["value"],
| sequence_by = "timestamp",
|)
- |""".stripMargin).resolve()
+ |""".stripMargin).resolve(sessionCaseSensitive)
val targetIdent = TableIdentifier("target", Some("some_schema"),
Some("some_catalog"))
val srcIdent = TableIdentifier("src", Some("some_schema"),
Some("some_catalog"))
@@ -1329,7 +1332,7 @@ class PythonPipelineSuite
| apply_as_deletes = "value % 2 = 0",
| column_list = ["value", "timestamp"],
|)
- |""".stripMargin).resolve().validate()
+
|""".stripMargin).resolve(sessionCaseSensitive).validate(sessionCaseSensitive)
val resolvedFlow = graph.resolvedFlow(graphIdentifier("target"))
assert(resolvedFlow.isInstanceOf[AutoCdcMergeFlow])
diff --git
a/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/pipelines/SparkDeclarativePipelinesServerSuite.scala
b/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/pipelines/SparkDeclarativePipelinesServerSuite.scala
index 7ff8504a3e6a..ec06d15ad8db 100644
---
a/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/pipelines/SparkDeclarativePipelinesServerSuite.scala
+++
b/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/pipelines/SparkDeclarativePipelinesServerSuite.scala
@@ -30,6 +30,9 @@ import org.apache.spark.sql.connect.service.{SessionKey,
SparkConnectService}
class SparkDeclarativePipelinesServerSuite
extends SparkDeclarativePipelinesServerTest
with Logging {
+
+ private def sessionCaseSensitive: Boolean =
spark.sessionState.conf.caseSensitiveAnalysis
+
test("CreateDataflowGraph request creates a new graph") {
withRawBlockingStub { implicit stub =>
assert(Option(createDataflowGraph(stub)).isDefined)
@@ -265,7 +268,7 @@ class SparkDeclarativePipelinesServerSuite
val definition =
getDefaultSessionHolder.dataflowGraphRegistry.getDataflowGraphOrThrow(graphId)
- val graph = definition.toDataflowGraph.resolve()
+ val graph = definition.toDataflowGraph.resolve(sessionCaseSensitive)
assert(graph.flows.size == 3)
assert(graph.tables.size == 2)
@@ -312,7 +315,7 @@ class SparkDeclarativePipelinesServerSuite
registerPipelineOutputs(pipeline)
val graph = definition.toDataflowGraph
- .resolve()
+ .resolve(sessionCaseSensitive)
assert(graph.flows.size == 3)
assert(graph.tables.size == 2)
diff --git
a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/ChangeArgs.scala
b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/ChangeArgs.scala
index a8c3485da31c..fe28decab357 100644
---
a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/ChangeArgs.scala
+++
b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/ChangeArgs.scala
@@ -74,9 +74,8 @@ object ColumnSelection {
* @param columnSelection The user-provided selection. `None` is a no-op and
returns `schema`
* unchanged.
* @param resolver Determines whether two column names are considered
equal. Callers
- * should pass the session resolver, e.g.
- * `session.sessionState.conf.resolver`, so column
matching stays
- * consistent with `spark.sql.caseSensitive`.
+ * should pass the resolver for the effective
`spark.sql.caseSensitive`
+ * of the operation being validated.
*/
def applyToSchema(
schemaName: String,
diff --git
a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/AutoCdcAuxiliaryTable.scala
b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/AutoCdcAuxiliaryTable.scala
index 9251501887a3..da5b7eb8ad12 100644
---
a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/AutoCdcAuxiliaryTable.scala
+++
b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/AutoCdcAuxiliaryTable.scala
@@ -143,7 +143,7 @@ object AutoCdcAuxiliaryTable {
): AuxiliaryTableSpec = {
val scd1AuxiliaryTableIdentifier = identifier(targetTable.identifier)
- val resolver = inputAutoCdcFlow.df.sparkSession.sessionState.conf.resolver
+ val resolver = inputAutoCdcFlow.effectiveResolver
val autoCdcKeyColumnNames = inputAutoCdcFlow.changeArgs.keys.map(_.name)
// The auxiliary table should derive its schema from the exact same
key/CDC metadata column
@@ -217,7 +217,7 @@ object AutoCdcAuxiliaryTable {
): AuxiliaryTableSpec = {
val scd2AuxiliaryTableIdentifier = identifier(targetTable.identifier)
- val resolver = inputAutoCdcFlow.df.sparkSession.sessionState.conf.resolver
+ val resolver = inputAutoCdcFlow.effectiveResolver
val autoCdcKeyColumnNames = inputAutoCdcFlow.changeArgs.keys.map(_.name)
// Resolve the key fields from the (evolved) target schema, exactly as
SCD1 does, so the
@@ -283,7 +283,7 @@ object AutoCdcAuxiliaryTable {
*
* @param targetTableSchema the AutoCDC target's evolved schema to resolve
against
* @param fieldName the column name to resolve
- * @param resolver the session resolver used for case-sensitivity-aware
field lookups
+ * @param resolver the effective resolver used for case-sensitivity-aware
field lookups
* @param targetTableIdentifier the AutoCDC target's identifier, named in
the error message
* @param autoCdcFlowIdentifier the AutoCDC flow writing to the target,
named in the error message
* @return the matching field
@@ -415,7 +415,7 @@ object AutoCdcAuxiliaryTable {
* @param expectedScdType the SCD type of the incoming AutoCDC flow, which
determines which inner
* `_cdc_metadata` field carries the recorded
sequencing type.
* @param expectedSequencingType the resolved sequencing type of the
incoming AutoCDC flow.
- * @param resolver the session resolver, used to match the reserved column
and inner field names
+ * @param resolver the effective resolver, used to match the reserved column
and inner field names
* the same case-aware way as every other schema lookup in
this file.
*/
private[graph] def validateNoTargetSequencingTypeDrift(
diff --git
a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/CoreDataflowNodeProcessor.scala
b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/CoreDataflowNodeProcessor.scala
index 66f2995ee02d..affa4725a2bf 100644
---
a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/CoreDataflowNodeProcessor.scala
+++
b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/CoreDataflowNodeProcessor.scala
@@ -32,9 +32,9 @@ import
org.apache.spark.sql.pipelines.graph.DataflowGraphTransformer.{
* Processor that is responsible for analyzing each flow and sort the nodes in
* topological order
*/
-class CoreDataflowNodeProcessor(rawGraph: DataflowGraph) {
+class CoreDataflowNodeProcessor(rawGraph: DataflowGraph, sessionCaseSensitive:
Boolean) {
- private val flowResolver = new FlowResolver(rawGraph)
+ private val flowResolver = new FlowResolver(rawGraph, sessionCaseSensitive)
// Map of input identifier to resolved [[Input]].
private val resolvedInputs = new ConcurrentHashMap[TableIdentifier, Input]()
@@ -86,7 +86,8 @@ class CoreDataflowNodeProcessor(rawGraph: DataflowGraph) {
identifier = table.identifier,
specifiedSchema = table.specifiedSchema,
incomingFlowIdentifiers = flowsToTable.map(_.identifier).toSet,
- availableFlows = resolvedFlowsToTable
+ availableFlows = resolvedFlowsToTable,
+ sessionCaseSensitive = sessionCaseSensitive
)
resolvedInputs.put(table.identifier, virtualTableInput)
Seq(table)
@@ -110,7 +111,7 @@ class CoreDataflowNodeProcessor(rawGraph: DataflowGraph) {
}
}
-private class FlowResolver(rawGraph: DataflowGraph) {
+private class FlowResolver(rawGraph: DataflowGraph, sessionCaseSensitive:
Boolean) {
/** Helper used to track which confs were set by which flows. */
private case class FlowConf(key: String, value: String, flowIdentifier:
TableIdentifier)
@@ -203,7 +204,7 @@ private class FlowResolver(rawGraph: DataflowGraph) {
flow: UnresolvedFlow,
funcResult: FlowFunctionResult): ResolvedFlow = {
flow match {
- case acf: AutoCdcFlow => new AutoCdcMergeFlow(acf, funcResult)
+ case acf: AutoCdcFlow => new AutoCdcMergeFlow(acf, funcResult,
sessionCaseSensitive)
case utf: UntypedFlow => transformUntypedFlowToResolvedFlow(utf,
funcResult)
}
}
diff --git
a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DataflowGraph.scala
b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DataflowGraph.scala
index c5210976d3f9..cd1dbf63ccdf 100644
---
a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DataflowGraph.scala
+++
b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DataflowGraph.scala
@@ -22,7 +22,7 @@ import scala.util.Try
import org.apache.spark.sql.AnalysisException
import org.apache.spark.sql.catalyst.TableIdentifier
import org.apache.spark.sql.pipelines.graph.DataflowGraph.mapUnique
-import org.apache.spark.sql.pipelines.util.SchemaMergingUtils
+import org.apache.spark.sql.pipelines.util.SchemaInferenceUtils
import org.apache.spark.sql.types.StructType
/**
@@ -146,7 +146,9 @@ case class DataflowGraph(
* upstream flows
* @return The reanalyzed flow
*/
- protected[graph] def reanalyzeFlow(srcFlow: Flow): ResolvedFlow = {
+ protected[graph] def reanalyzeFlow(
+ srcFlow: Flow,
+ sessionCaseSensitive: Boolean): ResolvedFlow = {
val upstreamDatasetIdentifiers = dfsInternal(
flowNodes(srcFlow.identifier).output,
downstream = false,
@@ -164,35 +166,46 @@ case class DataflowGraph(
tables = table.get(srcFlow.destinationIdentifier).toSeq,
sinks = sink.get(srcFlow.destinationIdentifier).toSeq
)
- subgraph.resolve().resolvedFlow(srcFlow.identifier)
+ subgraph.resolve(sessionCaseSensitive).resolvedFlow(srcFlow.identifier)
}
/**
* A map of the inferred schema of each table, computed by merging the
analyzed schemas
* of all flows writing to that table.
+ *
+ * The merge honors the effective `spark.sql.caseSensitive` of the flows
writing to each table:
+ * under case-insensitive analysis two flows emitting column names that
differ only in case
+ * contribute a single column rather than both, which would otherwise
produce a target schema the
+ * engine's own resolver cannot disambiguate. Which spelling survives such a
fold is fixed by
+ * [[SchemaInferenceUtils.inferSchemaFromFlows]], which merges in sorted
flow identifier order.
*/
- lazy val inferredSchema: Map[TableIdentifier, StructType] = {
- flowsTo.view.mapValues { flows =>
- flows
- .map { flow =>
- resolvedFlow(flow.identifier).schema
- }
- .reduce(SchemaMergingUtils.mergeSchemas)
- }.toMap
+ def inferSchemas(sessionCaseSensitive: Boolean): Map[TableIdentifier,
StructType] = {
+ flowsTo.map { case (destinationIdentifier, flows) =>
+ val resolvedFlows = flows.map { flow =>
+ resolvedFlow(flow.identifier)
+ }
+ destinationIdentifier -> SchemaInferenceUtils.inferSchemaFromFlows(
+ tableIdentifier = destinationIdentifier,
+ flows = resolvedFlows,
+ userSpecifiedSchema = None,
+ sessionCaseSensitive = sessionCaseSensitive)
+ }
}
/**
* The internal auxiliary tables owned by each destination [[Table]],
derived from the resolved
- * flows writing to it and that destination's [[inferredSchema]]. Keyed by
the destination's
- * identifier; only destinations that actually require auxiliary tables
appear. Today only
- * AutoCDC flow destination tables have an auxiliary table, and exactly one.
+ * flows writing to it and the destination schemas inferred from them. Keyed
by the destination's
+ * identifier; only destinations that actually require auxiliary tables
appear. Today only AutoCDC
+ * flow destination tables have an auxiliary table, and exactly one.
*
* Auxiliary tables are deliberately NOT part of the logical graph (they are
never resolved,
* connected, or exposed as [[Input]]s); this is purely a derived view used
during dataset
* materialization to create/evolve them alongside their owning table. The
derivation is pure and
* performs no catalog access.
*/
- lazy val auxiliaryTableSpecs: Map[TableIdentifier, AuxiliaryTableSpec] = {
+ def auxiliaryTableSpecs(
+ inferredSchemas: Map[TableIdentifier, StructType]
+ ): Map[TableIdentifier, AuxiliaryTableSpec] = {
resolvedFlowsTo.flatMap { case (destinationTableIdentifier,
flowsToDestinationTable) =>
table.get(destinationTableIdentifier).flatMap { destinationTable =>
flowsToDestinationTable
@@ -203,7 +216,7 @@ case class DataflowGraph(
.map { autoCdcFlow =>
val spec = AutoCdcAuxiliaryTable.buildAuxiliaryTableSpecFor(
targetTable = destinationTable,
- targetTableSchema = inferredSchema(destinationTableIdentifier),
+ targetTableSchema = inferredSchemas(destinationTableIdentifier),
inputAutoCdcFlow = autoCdcFlow
)
destinationTableIdentifier -> spec
@@ -213,22 +226,22 @@ case class DataflowGraph(
}
/** Ensure that the [[DataflowGraph]] is valid and throws errors if not. */
- def validate(): DataflowGraph = {
- validationFailure.toOption match {
+ def validate(sessionCaseSensitive: Boolean): DataflowGraph = {
+ validationFailure(sessionCaseSensitive).toOption match {
case Some(exception) => throw exception
case None => this
}
}
/**
- * Validate the current [[DataflowGraph]] and cache the validation failure.
+ * Validate the current [[DataflowGraph]] and return the validation failure,
if one exists.
*
* To add more validations, add them in a helper function that throws an
exception if the
* validation fails, and invoke the helper function here.
*/
- private lazy val validationFailure: Try[Throwable] = Try {
+ private def validationFailure(sessionCaseSensitive: Boolean): Try[Throwable]
= Try {
validateSuccessfulFlowAnalysis()
- validateUserSpecifiedSchemas()
+ validateUserSpecifiedSchemas(sessionCaseSensitive)
// Connecting the graph sorts it topologically
validateGraphIsTopologicallySorted()
validateMultiQueryTables()
@@ -236,7 +249,6 @@ case class DataflowGraph(
validateEveryDatasetHasFlow()
validateTablesAreResettable()
validateFlowStreamingness()
- inferredSchema
}.failed
/**
@@ -259,10 +271,12 @@ case class DataflowGraph(
def resolved: Boolean =
flows.forall(f => resolvedFlow.contains(f.identifier))
- def resolve(): DataflowGraph =
+ def resolve(sessionCaseSensitive: Boolean): DataflowGraph =
DataflowGraphTransformer.withDataflowGraphTransformer(this) { transformer
=>
val coreDataflowNodeProcessor =
- new CoreDataflowNodeProcessor(rawGraph = this)
+ new CoreDataflowNodeProcessor(
+ rawGraph = this,
+ sessionCaseSensitive = sessionCaseSensitive)
transformer
.transformDownNodes(coreDataflowNodeProcessor.processNode)
.getDataflowGraph
diff --git
a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DatasetManager.scala
b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DatasetManager.scala
index cf5f811d3461..61d188f302e1 100644
---
a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DatasetManager.scala
+++
b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DatasetManager.scala
@@ -24,7 +24,7 @@ import org.apache.spark.SparkException
import org.apache.spark.internal.{Logging, LogKeys}
import org.apache.spark.sql.AnalysisException
import org.apache.spark.sql.catalyst.TableIdentifier
-import org.apache.spark.sql.catalyst.analysis.{NoSuchTableException,
PersistedView}
+import org.apache.spark.sql.catalyst.analysis.{NoSuchTableException,
PersistedView, Resolver}
import org.apache.spark.sql.classic.SparkSession
import org.apache.spark.sql.connector.catalog.{
CatalogV2Util,
@@ -39,9 +39,12 @@ import
org.apache.spark.sql.connector.catalog.CatalogV2Util.v2ColumnsToStructTyp
import org.apache.spark.sql.connector.expressions.{ClusterByTransform,
Expressions, Transform}
import org.apache.spark.sql.execution.command.CreateViewCommand
import org.apache.spark.sql.pipelines.graph.QueryOrigin.ExceptionHelpers
-import org.apache.spark.sql.pipelines.util.PipelinesCatalogUtils
+import org.apache.spark.sql.pipelines.util.{
+ PipelinesCatalogUtils,
+ SchemaInferenceUtils,
+ SchemaMergingUtils
+}
import org.apache.spark.sql.pipelines.util.SchemaInferenceUtils.diffSchemas
-import org.apache.spark.sql.pipelines.util.SchemaMergingUtils
import org.apache.spark.sql.types.StructType
/**
@@ -97,6 +100,9 @@ object DatasetManager extends Logging {
val tablesToMaterialize = {
tablesToMatz(resolvedDataflowGraph).map(t => t.table.identifier ->
t).toMap
}
+ val sessionCaseSensitive =
context.spark.sessionState.conf.caseSensitiveAnalysis
+ val inferredSchemas =
resolvedDataflowGraph.inferSchemas(sessionCaseSensitive)
+ val auxiliaryTableSpecs =
resolvedDataflowGraph.auxiliaryTableSpecs(inferredSchemas)
// materialized [[DataflowGraph]] where each table has been materialized
and each table
// has metadata (e.g., normalized table storage path) populated
@@ -112,8 +118,7 @@ object DatasetManager extends Logging {
// materializeAuxiliaryTable (which uses it to decide
evolve-vs-create). Nothing
// between them mutates the auxiliary table, so a single load
is safe and avoids a
// redundant catalog round trip.
- val auxiliaryTableSpecOpt =
-
resolvedDataflowGraph.auxiliaryTableSpecs.get(table.identifier)
+ val auxiliaryTableSpecOpt =
auxiliaryTableSpecs.get(table.identifier)
val existingAuxiliaryTable = auxiliaryTableSpecOpt.flatMap {
spec =>
val (auxCatalog, auxId) =
PipelinesCatalogUtils.resolveTableCatalog(context.spark,
spec.identifier)
@@ -122,6 +127,7 @@ object DatasetManager extends Logging {
val (tableWithMaterializationMetadata, catalogTableEntity) =
materializeTable(
resolvedDataflowGraph = resolvedDataflowGraph,
table = table,
+ inferredSchemas = inferredSchemas,
isFullRefresh = isFullRefresh,
auxiliaryTableSpecOpt = auxiliaryTableSpecOpt,
existingAuxiliaryTable = existingAuxiliaryTable,
@@ -149,6 +155,10 @@ object DatasetManager extends Logging {
auxiliaryTableSpec = auxiliaryTableSpec,
isFullRefresh = isFullRefresh,
existingAuxiliaryTable = existingAuxiliaryTable,
+ // The auxiliary schema is derived from its target's, so
it evolves under the
+ // target's effective case sensitivity.
+ caseSensitive = effectiveCaseSensitivityFor(
+ resolvedDataflowGraph, table.identifier, context),
context = context
)
}
@@ -282,8 +292,9 @@ object DatasetManager extends Logging {
/**
* Materializes a table in the catalog. This method will create or update
the table in the
* catalog based on the given table and context.
- * @param resolvedDataflowGraph The resolved [[DataflowGraph]] used to infer
the table schema.
+ * @param resolvedDataflowGraph The resolved [[DataflowGraph]] used for
table metadata.
* @param table The table to be materialized.
+ * @param inferredSchemas The schemas inferred from the resolved graph,
keyed by table.
* @param isFullRefresh Whether this table should be full refreshed or not.
* @param auxiliaryTableSpecOpt The spec for the auxiliary table (if this
table has one)
* @param existingAuxiliaryTable The already-loaded auxiliary table for this
target (if it has one
@@ -296,6 +307,7 @@ object DatasetManager extends Logging {
private def materializeTable(
resolvedDataflowGraph: DataflowGraph,
table: Table,
+ inferredSchemas: Map[TableIdentifier, StructType],
isFullRefresh: Boolean,
auxiliaryTableSpecOpt: Option[AuxiliaryTableSpec],
existingAuxiliaryTable: Option[V2Table],
@@ -306,7 +318,7 @@ object DatasetManager extends Logging {
PipelinesCatalogUtils.resolveTableCatalog(context.spark,
table.identifier)
val outputSchema = table.specifiedSchema.getOrElse(
- resolvedDataflowGraph.inferredSchema(table.identifier).asNullable
+ inferredSchemas(table.identifier).asNullable
)
val mergedProperties = resolveTableProperties(table, identifier)
val partitioning =
table.partitionCols.toSeq.flatten.map(Expressions.identity)
@@ -354,6 +366,9 @@ object DatasetManager extends Logging {
val autoCdcAuxTableSpecOpt = auxiliaryTableSpecOpt.collect {
case autoCdcSpec: AutoCdcAuxiliaryTableSpec => autoCdcSpec
}
+ val effectiveCaseSensitive = effectiveCaseSensitivityFor(
+ resolvedDataflowGraph, table.identifier, context)
+ val effectiveResolver =
SchemaInferenceUtils.resolverFor(effectiveCaseSensitive)
// For an incrementally-updated AutoCDC target, validate that the AutoCDC
configuration recorded
// on the auxiliary table has not drifted, BEFORE anything is created or
evolved this run. These
@@ -365,7 +380,7 @@ object DatasetManager extends Logging {
// Running here turns that into one clear drift error (remedy: full
refresh).
if (isTableIncrementallyUpdated) {
autoCdcAuxTableSpecOpt.foreach {
- validateNoAutoCdcAuxConfigDrift(_, existingAuxiliaryTable, context)
+ validateNoAutoCdcAuxConfigDrift(_, existingAuxiliaryTable,
effectiveResolver)
}
}
@@ -387,7 +402,7 @@ object DatasetManager extends Logging {
targetTableIdentifier = autoCdcSpec.targetTableIdentifier,
expectedScdType = autoCdcSpec.expectedScdType,
expectedSequencingType = autoCdcSpec.expectedSequencingType,
- resolver = context.spark.sessionState.conf.resolver
+ resolver = effectiveResolver
)
}
}
@@ -397,7 +412,8 @@ object DatasetManager extends Logging {
existingTable = existingTable,
desiredSchema = outputSchema,
properties = mergedProperties,
- mergeWithExistingSchema = isTableIncrementallyUpdated
+ mergeWithExistingSchema = isTableIncrementallyUpdated,
+ caseSensitive = effectiveCaseSensitive
)
case None =>
createTable(
@@ -455,12 +471,15 @@ object DatasetManager extends Logging {
* @param existingAuxiliaryTable the already-loaded auxiliary table (if it
exists), loaded once by
* the caller and shared with the config-drift
validation in
* [[materializeTable]] rather than re-loaded
here.
+ * @param caseSensitive the effective case sensitivity of the flows writing
to the auxiliary
+ * table's TARGET, whose schema the auxiliary schema is
derived from.
* @param context the context for the pipeline update.
*/
private def materializeAuxiliaryTable(
auxiliaryTableSpec: AuxiliaryTableSpec,
isFullRefresh: Boolean,
existingAuxiliaryTable: Option[V2Table],
+ caseSensitive: Boolean,
context: PipelineUpdateContext): Unit = {
// Get the DSv2 catalog handler and identifier for the aux table.
val (catalog, auxiliaryTableIdentifier) =
@@ -511,7 +530,8 @@ object DatasetManager extends Logging {
existingTable = existingAuxTable,
desiredSchema = auxiliaryTableSpec.schema,
properties = auxiliaryTableSpec.properties,
- mergeWithExistingSchema = true
+ mergeWithExistingSchema = true,
+ caseSensitive = caseSensitive
)
case None =>
createTable(
@@ -541,13 +561,12 @@ object DatasetManager extends Logging {
* @param existingAuxiliaryTableOpt the already-loaded auxiliary table (if
it exists), shared with
* the caller and
[[materializeAuxiliaryTable]] to avoid a
* redundant load.
- * @param context the context for the pipeline update.
+ * @param resolver the effective resolver of the flows writing to the
AutoCDC target.
*/
private def validateNoAutoCdcAuxConfigDrift(
autoCdcSpec: AutoCdcAuxiliaryTableSpec,
existingAuxiliaryTableOpt: Option[V2Table],
- context: PipelineUpdateContext): Unit = {
- val resolver = context.spark.sessionState.conf.resolver
+ resolver: Resolver): Unit = {
existingAuxiliaryTableOpt.foreach { existingAuxiliaryTable =>
AutoCdcAuxiliaryTable.validateNoKeyColumnDrift(
existingAuxiliaryTable = existingAuxiliaryTable,
@@ -569,6 +588,23 @@ object DatasetManager extends Logging {
}
}
+ /**
+ * The effective `spark.sql.caseSensitive` for schema evolution of
`tableIdentifier`, read from
+ * the flows writing to it rather than from the session, so evolution stays
consistent with the
+ * flows whose schemas it is evolving (a pipeline-level `SET` never reaches
the session). Fails if
+ * those flows disagree; see
[[SchemaInferenceUtils.effectiveCaseSensitivity]].
+ */
+ private def effectiveCaseSensitivityFor(
+ resolvedDataflowGraph: DataflowGraph,
+ tableIdentifier: TableIdentifier,
+ context: PipelineUpdateContext): Boolean = {
+ SchemaInferenceUtils.effectiveCaseSensitivity(
+ tableIdentifier = tableIdentifier,
+ flows = resolvedDataflowGraph.flowsTo.getOrElse(tableIdentifier,
Seq.empty),
+ sessionCaseSensitive =
context.spark.sessionState.conf.caseSensitiveAnalysis
+ )
+ }
+
/**
* Loads the table at `identifier` from `catalog`, or `None` if it does not
exist. A single
* `loadTable` guarded by a `NoSuchTableException` catch, rather than a
`tableExists` +
@@ -626,6 +662,15 @@ object DatasetManager extends Logging {
* @param mergeWithExistingSchema whether the effective schema is the merge
of the existing and
* desired schemas (additive evolution)
rather than the desired
* schema as-is.
+ * @param caseSensitive whether the additive schema merge treats
field names differing
+ * only in case as distinct columns. Callers
should pass the
+ * effective `spark.sql.caseSensitive` used
to resolve the schema
+ * being evolved. When `false`, an incoming
column differing from
+ * an existing one only in case is folded
onto it rather than
+ * added as a duplicate. Only affects the
merge (i.e.
+ * `mergeWithExistingSchema = true`); the
subsequent diff always
+ * keys columns on their exact names, so a
case-only rename on a
+ * non-merging path stays an explicit
drop-then-add.
*/
private def evolveTable(
catalog: TableCatalog,
@@ -633,13 +678,18 @@ object DatasetManager extends Logging {
existingTable: V2Table,
desiredSchema: StructType,
properties: Map[String, String],
- mergeWithExistingSchema: Boolean): Unit = {
+ mergeWithExistingSchema: Boolean,
+ caseSensitive: Boolean): Unit = {
val currentSchema = v2ColumnsToStructType(existingTable.columns())
val targetSchema = if (mergeWithExistingSchema) {
- SchemaMergingUtils.mergeSchemas(currentSchema, desiredSchema)
+ SchemaMergingUtils.mergeSchemas(currentSchema, desiredSchema,
caseSensitive)
} else {
desiredSchema
}
+ // `diffSchemas` keys column identity on exact field names. On the
incremental path the merge
+ // above has already folded a case-only-differing incoming field onto the
persisted one. On the
+ // non-merging paths (materialized views, full refresh), `targetSchema` is
the declared schema
+ // as-is, where exact-name matching keeps a case-only rename visible as a
schema change.
val columnChanges = diffSchemas(currentSchema, targetSchema)
val existingProperties = existingTable.properties()
diff --git
a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/Flow.scala
b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/Flow.scala
index 856fd6a3923c..a8be0a4c2ec3 100644
---
a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/Flow.scala
+++
b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/Flow.scala
@@ -23,6 +23,7 @@ import org.apache.spark.SparkException
import org.apache.spark.internal.Logging
import org.apache.spark.sql.{functions => F, AnalysisException, Column}
import org.apache.spark.sql.catalyst.{AliasIdentifier, TableIdentifier}
+import org.apache.spark.sql.catalyst.analysis.Resolver
import org.apache.spark.sql.classic.DataFrame
import org.apache.spark.sql.pipelines.autocdc.{
AutoCdcReservedNames,
@@ -33,6 +34,7 @@ import org.apache.spark.sql.pipelines.autocdc.{
Scd2BatchProcessor,
ScdType
}
+import org.apache.spark.sql.pipelines.util.SchemaInferenceUtils
import org.apache.spark.sql.types.{DataType, StructField, StructType}
/**
@@ -250,8 +252,15 @@ class AppendOnceFlow(
*/
class AutoCdcMergeFlow(
val flow: AutoCdcFlow,
- val funcResult: FlowFunctionResult
+ val funcResult: FlowFunctionResult,
+ sessionCaseSensitive: Boolean
) extends ResolvedFlow {
+ private[graph] val effectiveResolver: Resolver =
SchemaInferenceUtils.resolverFor(
+ SchemaInferenceUtils.effectiveCaseSensitivity(
+ tableIdentifier = destinationIdentifier,
+ flows = Seq(this),
+ sessionCaseSensitive = sessionCaseSensitive))
+
requireReservedPrefixAbsentInSourceColumns()
requireReservedFrameworkColumnsAbsentInSourceColumns()
@@ -263,7 +272,7 @@ class AutoCdcMergeFlow(
schemaName = "changeDataFeed",
schema = df.schema,
columnSelection = changeArgs.columnSelection,
- resolver = spark.sessionState.conf.resolver
+ resolver = effectiveResolver
)
// AutoCDC flows require all key columns to be present in the
user-selected source schema,
// so that they survive into the target table where SCD reconciliation
needs them.
@@ -303,7 +312,7 @@ class AutoCdcMergeFlow(
Some(Scd2BatchProcessor.computeTrackedHistoryColumns(
schema = userSelectedSchema,
changeArgs = changeArgs,
- resolver = spark.sessionState.conf.resolver))
+ resolver = effectiveResolver))
case ScdType.Type1 => None
}
@@ -405,7 +414,7 @@ class AutoCdcMergeFlow(
* names that use the reserved Spark AutoCDC prefix.
*/
private def requireReservedPrefixAbsentInSourceColumns(): Unit = {
- val resolver = spark.sessionState.conf.resolver
+ val resolver = effectiveResolver
val reservedPrefix = AutoCdcReservedNames.prefix
def nameContainsReservedPrefix(name: String): Boolean = {
@@ -437,7 +446,7 @@ class AutoCdcMergeFlow(
* during preprocessing. No-op for SCD1, which has no such columns.
*/
private def requireReservedFrameworkColumnsAbsentInSourceColumns(): Unit = {
- val resolver = spark.sessionState.conf.resolver
+ val resolver = effectiveResolver
val reservedPrefix = AutoCdcReservedNames.prefix
// Only the non-prefixed reserved names need checking here; prefixed ones
are already rejected
@@ -469,7 +478,7 @@ class AutoCdcMergeFlow(
* Validate all keys specified in changeArgs are actually present in the
user-selected schema.
*/
private def requireKeysPresentInSelectedSchema(selectedSchema: StructType):
Unit = {
- val resolver = spark.sessionState.conf.resolver
+ val resolver = effectiveResolver
changeArgs.keys
.find(key => !selectedSchema.fieldNames.exists(name => resolver(name,
key.name)))
diff --git
a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/FlowExecution.scala
b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/FlowExecution.scala
index c36d824bdf4a..c665c344c53b 100644
---
a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/FlowExecution.scala
+++
b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/FlowExecution.scala
@@ -105,6 +105,10 @@ trait FlowExecution {
/** Context about this pipeline update. */
def updateContext: PipelineUpdateContext
+ /** The session's `spark.sql.caseSensitive` fallback for resolving this
flow. */
+ protected def sessionCaseSensitive: Boolean =
+ spark.sessionState.conf.caseSensitiveAnalysis
+
/** The thread execution context for the current `FlowExecution`. */
implicit val executionContext: ExecutionContext = {
ExecutionContext.fromExecutor(FlowExecution.threadPool)
@@ -232,7 +236,7 @@ class StreamingTableWrite(
override def getOrigin: QueryOrigin = flow.origin
def startStream(): StreamingQuery = {
- val data = graph.reanalyzeFlow(flow).df
+ val data = graph.reanalyzeFlow(flow, sessionCaseSensitive).df
val dataStreamWriter = data
.writeStream
.queryName(displayName)
@@ -260,7 +264,7 @@ class BatchTableWrite(
def executeInternal(): Future[Unit] = {
SparkSessionUtils.withSqlConf(spark, sqlConf.toList: _*) {
updateContext.flowProgressEventLogger.recordRunning(flow = flow)
- val data = graph.reanalyzeFlow(flow).df
+ val data = graph.reanalyzeFlow(flow, sessionCaseSensitive).df
Future {
val dataFrameWriter = data.write
destination.format.foreach(dataFrameWriter.format)
@@ -298,7 +302,7 @@ class SinkWrite(
override def getOrigin: QueryOrigin = flow.origin
def startStream(): StreamingQuery = {
- val data = graph.reanalyzeFlow(flow).df
+ val data = graph.reanalyzeFlow(flow, sessionCaseSensitive).df
data.writeStream
.queryName(displayName)
.option("checkpointLocation", checkpointPath)
@@ -328,7 +332,7 @@ class Scd1MergeStreamingWrite(
override def getOrigin: QueryOrigin = flow.origin
override def startStream(): StreamingQuery = {
- val sourceChangeDataFeed = graph.reanalyzeFlow(flow).df
+ val sourceChangeDataFeed = graph.reanalyzeFlow(flow,
sessionCaseSensitive).df
// The auxiliary table is created and evolved during dataset
materialization (see
// [[DatasetManager]]), so it already exists by the time this flow
executes; resolve its
@@ -373,7 +377,7 @@ class Scd2MergeStreamingWrite(
override def getOrigin: QueryOrigin = flow.origin
override def startStream(): StreamingQuery = {
- val sourceChangeDataFeed = graph.reanalyzeFlow(flow).df
+ val sourceChangeDataFeed = graph.reanalyzeFlow(flow,
sessionCaseSensitive).df
// The auxiliary table is created and evolved during dataset
materialization (see
// [[DatasetManager]]), so it already exists by the time this flow
executes; resolve its
diff --git
a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/GraphErrors.scala
b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/GraphErrors.scala
index c835665a0f38..7bc26e6288c1 100644
---
a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/GraphErrors.scala
+++
b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/GraphErrors.scala
@@ -109,4 +109,33 @@ object GraphErrors {
cause = Option(cause.orNull)
)
}
+
+ /**
+ * Throws if the flows writing to one table disagree on a configuration
whose value determines
+ * how the table's schema is derived, so that the resulting schema would
otherwise depend on the
+ * order the flows happen to be evaluated in.
+ *
+ * @param tableIdentifier the destination table the conflicting flows write
to
+ * @param configKey the configuration the flows disagree on
+ * @param valuesByFlow the distinct values, each with the flows that
declared it
+ */
+ def conflictingFlowConfigurationError(
+ tableIdentifier: TableIdentifier,
+ configKey: String,
+ valuesByFlow: Map[String, Seq[TableIdentifier]]): AnalysisException = {
+ val rendered = valuesByFlow.toSeq
+ .sortBy(_._1)
+ .map { case (value, flows) =>
+ s"$value (${flows.map(_.unquotedString).sorted.mkString(", ")})"
+ }
+ .mkString("; ")
+ new AnalysisException(
+ errorClass = "CONFLICTING_PIPELINE_FLOW_CASE_SENSITIVITY",
+ messageParameters = Map(
+ "tableName" -> tableIdentifier.unquotedString,
+ "configKey" -> configKey,
+ "flowConfigurations" -> rendered
+ )
+ )
+ }
}
diff --git
a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/GraphValidations.scala
b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/GraphValidations.scala
index ebe90b677aa2..5b4eaba4e709 100644
---
a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/GraphValidations.scala
+++
b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/GraphValidations.scala
@@ -252,7 +252,7 @@ trait GraphValidations extends Logging {
}
}
- protected def validateUserSpecifiedSchemas(): Unit = {
+ protected def validateUserSpecifiedSchemas(sessionCaseSensitive: Boolean):
Unit = {
// Look up tables by their destination identifier, not by the flow's own
identifier. The two
// coincide only for an implicit/default flow (whose identifier equals its
destination
// table's); for a named flow (e.g. `CREATE FLOW <name> AS AUTO CDC INTO
<target>`) they
@@ -262,8 +262,10 @@ trait GraphValidations extends Logging {
// schema of all incoming flows. This must be equivalent to the declared
schema.
val inferredSchema = SchemaInferenceUtils
.inferSchemaFromFlows(
+ tableIdentifier = t.identifier,
flowsTo(t.identifier).map(f => resolvedFlow(f.identifier)),
- userSpecifiedSchema = t.specifiedSchema
+ userSpecifiedSchema = t.specifiedSchema,
+ sessionCaseSensitive = sessionCaseSensitive
)
t.specifiedSchema.foreach { ss =>
diff --git
a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/PipelineExecution.scala
b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/PipelineExecution.scala
index d35d701d44e5..724ccc56e35c 100644
---
a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/PipelineExecution.scala
+++
b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/PipelineExecution.scala
@@ -110,7 +110,8 @@ class PipelineExecution(context: PipelineUpdateContext) {
private def resolveGraph(): DataflowGraph = {
try {
- context.unresolvedGraph.resolve().validate()
+ val sessionCaseSensitive =
context.spark.sessionState.conf.caseSensitiveAnalysis
+
context.unresolvedGraph.resolve(sessionCaseSensitive).validate(sessionCaseSensitive)
} catch {
case e: UnresolvedPipelineException =>
handleInvalidPipeline(e)
diff --git
a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/elements.scala
b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/elements.scala
index 885755fd78ec..baef25176e9a 100644
---
a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/elements.scala
+++
b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/elements.scala
@@ -160,12 +160,14 @@ case class Table(
* @param identifier The identifier of the parent table.
* @param specifiedSchema The user-specified schema for the parent table.
* @param incomingFlowIdentifiers The identifiers of all flows that write to
the parent table.
+ * @param sessionCaseSensitive The session's `spark.sql.caseSensitive`
fallback.
* @param availableFlows All resolved flows that write to the parent table.
*/
case class VirtualTableInput(
identifier: TableIdentifier,
specifiedSchema: Option[StructType],
incomingFlowIdentifiers: Set[TableIdentifier],
+ sessionCaseSensitive: Boolean,
availableFlows: Seq[ResolvedFlow] = Nil
) extends TableElement with Input
with Logging {
@@ -186,7 +188,11 @@ case class VirtualTableInput(
// Otherwise infer the schema from a combination of the incoming flows
and the
// user-specified schema, if provided.
case _ =>
- SchemaInferenceUtils.inferSchemaFromFlows(availableFlows,
specifiedSchema)
+ SchemaInferenceUtils.inferSchemaFromFlows(
+ tableIdentifier = identifier,
+ flows = availableFlows,
+ userSpecifiedSchema = specifiedSchema,
+ sessionCaseSensitive = sessionCaseSensitive)
}
// Produce either a streaming or batch dataframe, depending on whether
this is a virtual
diff --git
a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/util/SchemaInferenceUtils.scala
b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/util/SchemaInferenceUtils.scala
index 4777772342d7..10b3202a61af 100644
---
a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/util/SchemaInferenceUtils.scala
+++
b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/util/SchemaInferenceUtils.scala
@@ -20,14 +20,89 @@ package org.apache.spark.sql.pipelines.util
import scala.util.control.NonFatal
import org.apache.spark.sql.catalyst.TableIdentifier
+import org.apache.spark.sql.catalyst.analysis.{
+ caseInsensitiveResolution,
+ caseSensitiveResolution,
+ Resolver
+}
import org.apache.spark.sql.connector.catalog.TableChange
+import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.pipelines.common.DatasetType
-import org.apache.spark.sql.pipelines.graph.{GraphElementTypeUtils,
GraphErrors, ResolvedFlow}
+import org.apache.spark.sql.pipelines.graph.{
+ Flow,
+ GraphElementTypeUtils,
+ GraphErrors,
+ ResolvedFlow
+}
import org.apache.spark.sql.types.{StructField, StructType}
object SchemaInferenceUtils {
+ def resolverFor(caseSensitive: Boolean): Resolver = {
+ if (caseSensitive) {
+ caseSensitiveResolution
+ } else {
+ caseInsensitiveResolution
+ }
+ }
+
+ /**
+ * The effective `spark.sql.caseSensitive` for schema derivation on
`tableIdentifier`, taken from
+ * the flows writing to it rather than from the session.
+ *
+ * A pipeline can set `spark.sql.caseSensitive` for itself, and a `SET` in
pipeline source does
+ * not touch the session:
[[org.apache.spark.sql.pipelines.graph.GraphRegistrationContext]] folds
+ * it into each flow's `sqlConf`, and it is applied when the flow is
analyzed and executed. Schema
+ * derivation therefore has to read it from the same place, or evolution can
disagree with the
+ * flows whose schemas it is deriving from -- e.g. folding an incoming
`Value` onto a persisted
+ * `value` while the flow, resolving case-sensitively, expects `Value` to be
its own column.
+ *
+ * All flows writing to a table must agree: the value decides whether names
differing only in case
+ * identify the same column, so a disagreement would make the resulting
schema depend on the order
+ * the flows are evaluated in. Throws
+ *
[[org.apache.spark.sql.pipelines.graph.GraphErrors.conflictingFlowConfigurationError]]
if they
+ * disagree. Flows that do not set it at all inherit the session's value.
+ */
+ def effectiveCaseSensitivity(
+ tableIdentifier: TableIdentifier,
+ flows: Seq[Flow],
+ sessionCaseSensitive: Boolean): Boolean = {
+ val declaredByFlow = flows.flatMap { flow =>
+ flow.sqlConf.get(SQLConf.CASE_SENSITIVE.key).map(value => value ->
flow.identifier)
+ }
+ if (declaredByFlow.isEmpty) {
+ return sessionCaseSensitive
+ }
+
+ // Compare the parsed booleans, so that e.g. "TRUE" and "true" are not
reported as a conflict,
+ // but report the values as written to keep the error recognizable to the
user.
+ val byParsedValue = declaredByFlow.groupBy { case (value, _) =>
value.trim.toBoolean }
+ // A flow that leaves the conf unset inherits the session value, which
conflicts just as much as
+ // an explicitly opposite value.
+ val flowsWithoutDeclaration = flows.filterNot { flow =>
+ flow.sqlConf.contains(SQLConf.CASE_SENSITIVE.key)
+ }
+ val effectiveValues = byParsedValue.keySet ++
+ Option.when(flowsWithoutDeclaration.nonEmpty)(sessionCaseSensitive)
+ if (effectiveValues.sizeIs > 1) {
+ val valuesByFlow = declaredByFlow
+ .groupBy { case (value, _) => value }
+ .map { case (value, entries) => value -> entries.map { case (_, id) =>
id } } ++
+ Option
+ .when(flowsWithoutDeclaration.nonEmpty)(
+ s"$sessionCaseSensitive (session default)" ->
flowsWithoutDeclaration.map(_.identifier)
+ )
+ .toMap
+ throw GraphErrors.conflictingFlowConfigurationError(
+ tableIdentifier = tableIdentifier,
+ configKey = SQLConf.CASE_SENSITIVE.key,
+ valuesByFlow = valuesByFlow
+ )
+ }
+ effectiveValues.head
+ }
+
/**
* Given a set of flows that write to the same destination and possibly a
user-specified schema,
* we infer the schema of the destination dataset. The logic is as follows:
@@ -39,42 +114,68 @@ object SchemaInferenceUtils {
* The user-specified schema will take precedence over the inferred
schema.
* Returns an error if encountered during schema inference or merging the
inferred schema with
* the user-specified one.
+ *
+ * All merges honor the effective `spark.sql.caseSensitive` of the flows
writing to
+ * `tableIdentifier`, falling back to `sessionCaseSensitive` for flows that
do not set it. Under
+ * case-insensitive analysis, flows emitting column names that differ only
in case contribute a
+ * single column, and a declared column matches a flow column differing only
in case -- consistent
+ * with how the rest of the engine resolves those names.
+ *
+ * When flows differ only in column casing, the surviving spelling is the
one from the flow with
+ * the lowest identifier: `flows` is merged in sorted identifier order, not
in the order given.
+ * Sorting here rather than at the call sites keeps every caller agreeing on
the result, since the
+ * schemas they derive are compared against each other -- the graph's
inferred schema materializes
+ * the table, while
[[org.apache.spark.sql.pipelines.graph.VirtualTableInput]] produces the schema
+ * downstream flows resolve against, and `diffSchemas` keys column identity
on the exact name. Two
+ * callers ordering the same flows differently would spell one column two
ways, leaving a
+ * downstream view disagreeing with its source and turning the next refresh
into a drop-then-add.
*/
def inferSchemaFromFlows(
+ tableIdentifier: TableIdentifier,
flows: Seq[ResolvedFlow],
- userSpecifiedSchema: Option[StructType]): StructType = {
+ userSpecifiedSchema: Option[StructType],
+ sessionCaseSensitive: Boolean): StructType = {
if (flows.isEmpty) {
return userSpecifiedSchema.getOrElse(new StructType())
}
require(
- flows.forall(_.destinationIdentifier ==
flows.head.destinationIdentifier),
+ flows.forall(_.destinationIdentifier == tableIdentifier),
"Expected all flows to have the same destination"
)
- val inferredSchema = flows.map(_.schema).fold(new StructType()) {
(schemaSoFar, schema) =>
- try {
- SchemaMergingUtils.mergeSchemas(schemaSoFar, schema)
- } catch {
- case NonFatal(e) =>
- throw GraphErrors.unableToInferSchemaError(
- flows.head.destinationIdentifier,
- schemaSoFar,
- schema,
- cause = Option(e)
- )
+ val caseSensitive = effectiveCaseSensitivity(
+ tableIdentifier = tableIdentifier,
+ flows = flows,
+ sessionCaseSensitive = sessionCaseSensitive
+ )
+
+ val inferredSchema = flows
+ .sortBy(_.identifier.unquotedString)
+ .map(_.schema)
+ .fold(new StructType()) { (schemaSoFar, schema) =>
+ try {
+ SchemaMergingUtils.mergeSchemas(schemaSoFar, schema, caseSensitive)
+ } catch {
+ case NonFatal(e) =>
+ throw GraphErrors.unableToInferSchemaError(
+ tableIdentifier,
+ schemaSoFar,
+ schema,
+ cause = Option(e)
+ )
+ }
}
- }
- val identifier = flows.head.destinationIdentifier
val datasetType =
GraphElementTypeUtils.getDatasetTypeForMaterializedViewOrStreamingTable(flows)
// We merge the inferred schema with the user-specified schema to pick up
any schema metadata
// that is provided by the user, e.g., comments or column masks.
mergeInferredAndUserSchemasIfNeeded(
- identifier,
+ tableIdentifier,
datasetType,
inferredSchema,
- userSpecifiedSchema
+ userSpecifiedSchema,
+ caseSensitive
)
}
@@ -82,12 +183,13 @@ object SchemaInferenceUtils {
tableIdentifier: TableIdentifier,
datasetType: DatasetType,
inferredSchema: StructType,
- userSpecifiedSchema: Option[StructType]): StructType = {
+ userSpecifiedSchema: Option[StructType],
+ caseSensitive: Boolean): StructType = {
userSpecifiedSchema match {
case Some(userSpecifiedSchema) =>
try {
// Merge the inferred schema with the user-provided schema hint
- SchemaMergingUtils.mergeSchemas(userSpecifiedSchema, inferredSchema)
+ SchemaMergingUtils.mergeSchemas(userSpecifiedSchema, inferredSchema,
caseSensitive)
} catch {
case NonFatal(e) =>
throw GraphErrors.incompatibleUserSpecifiedAndInferredSchemasError(
@@ -110,6 +212,15 @@ object SchemaInferenceUtils {
* 1. New columns that need to be added
* 2. Existing columns that need type updates
*
+ * Column identity is keyed on the exact field name, not on a
case-normalized one. On the
+ * incremental streaming-table path, `targetSchema` is the merge of the
current and desired
+ * schemas, and [[SchemaMergingUtils.mergeSchemas]] has already folded an
incoming
+ * case-only-differing field onto the persisted one. On the non-merging
paths (materialized views
+ * and any full refresh), `targetSchema` is the run's declared schema as-is,
so exact-name keying
+ * keeps a case-only rename visible as an explicit drop-then-add.
+ * Exact keying also avoids silently collapsing two genuinely distinct
declared columns that
+ * differ only in case (`value` and `Value`) into an arbitrary one of the
two.
+ *
* @param currentSchema The current schema of the table
* @param targetSchema The target schema that we want the table to have
* @return A sequence of TableChange objects representing the necessary
changes
diff --git
a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/util/SchemaMergingUtils.scala
b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/util/SchemaMergingUtils.scala
index d15e7ac6425c..aeab5e623431 100644
---
a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/util/SchemaMergingUtils.scala
+++
b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/util/SchemaMergingUtils.scala
@@ -20,7 +20,26 @@ package org.apache.spark.sql.pipelines.util
import org.apache.spark.sql.types.StructType
object SchemaMergingUtils {
- def mergeSchemas(tableSchema: StructType, dataSchema: StructType):
StructType = {
- StructType.merge(tableSchema, dataSchema).asInstanceOf[StructType]
+
+ /**
+ * Additively merges `dataSchema` into `tableSchema`, returning a schema
that is the union of the
+ * two (recursing into nested structs/arrays). On a field present in both,
`tableSchema`'s name
+ * and position win; `dataSchema` only contributes fields absent from
`tableSchema`.
+ *
+ * @param caseSensitive whether two field names that differ only in case are
considered distinct.
+ * When `false` (mirroring a case-insensitive session),
`dataSchema`'s field
+ * is folded onto the matching `tableSchema` field
rather than added as a
+ * separate, case-differing column. Deliberately has no
default: every caller
+ * merges schemas that some pipeline will later resolve
names against, so the
+ * choice belongs to the caller and should be visible
at the call site rather
+ * than silently inherited. Callers should pass the
effective
+ * `spark.sql.caseSensitive` of the flows involved (see
+ * [[SchemaInferenceUtils.effectiveCaseSensitivity]]).
+ */
+ def mergeSchemas(
+ tableSchema: StructType,
+ dataSchema: StructType,
+ caseSensitive: Boolean): StructType = {
+ StructType.merge(tableSchema, dataSchema,
caseSensitive).asInstanceOf[StructType]
}
}
diff --git
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/AutoCdcFlowSuite.scala
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/AutoCdcFlowSuite.scala
index a3d26e6f47e5..4596e31d778a 100644
---
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/AutoCdcFlowSuite.scala
+++
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/AutoCdcFlowSuite.scala
@@ -176,7 +176,10 @@ class AutoCdcFlowSuite extends QueryTest with
SharedSparkSession {
trackHistorySelection = trackHistorySelection
)
)
- new AutoCdcMergeFlow(flow, successfulFuncResult(sourceDf))
+ new AutoCdcMergeFlow(
+ flow,
+ successfulFuncResult(sourceDf),
+ spark.sessionState.conf.caseSensitiveAnalysis)
}
/** A stable 3-column source streaming dataframe used across most schema
tests. */
diff --git
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcConfigDriftSuite.scala
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcConfigDriftSuite.scala
index dda1e5a7455a..c26a5248488b 100644
---
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcConfigDriftSuite.scala
+++
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcConfigDriftSuite.scala
@@ -18,6 +18,7 @@
package org.apache.spark.sql.pipelines.graph
import org.apache.spark.sql.execution.streaming.runtime.MemoryStream
+import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.pipelines.autocdc.{ColumnSelection, ScdType,
UnqualifiedColumnName}
import org.apache.spark.sql.pipelines.utils.{ExecutionTest,
TestGraphRegistrationContext}
import org.apache.spark.sql.test.SharedSparkSession
@@ -53,6 +54,28 @@ class AutoCdcConfigDriftSuite
// Sequencing type drift
//
===========================================================================================
+ test("AutoCDC source validation uses pipeline case sensitivity, not session
default") {
+ withSQLConf(SQLConf.CASE_SENSITIVE.key -> "false") {
+ val stream = MemoryStream[(Int, Long, Long)]
+ stream.addData((1, 1L, 1L))
+
+ val ctx = new TestGraphRegistrationContext(
+ spark,
+ Map(SQLConf.CASE_SENSITIVE.key -> "true")) {
+ registerTable("target", catalog = Some(catalog), database =
Some(namespace))
+ registerFlow(autoCdcFlow(
+ name = "flow",
+ target = "target",
+ query = dfFlowFunc(stream.toDF().toDF("id", "version",
"__start_at")),
+ keys = Seq("id"),
+ sequencing = $"version",
+ scdType = ScdType.Type2))
+ }
+
+ ctx.resolveToDataflowGraph()
+ }
+ }
+
test("an SCD1 flow whose sequencing type differs from the recorded type
triggers " +
"SEQUENCING_TYPE_DRIFT") {
spark.sql(
diff --git
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1AuxiliaryTableSpecSuite.scala
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1AuxiliaryTableSpecSuite.scala
index 226480ea919a..6de8cf52f7b4 100644
---
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1AuxiliaryTableSpecSuite.scala
+++
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1AuxiliaryTableSpecSuite.scala
@@ -75,7 +75,9 @@ class AutoCdcScd1AuxiliaryTableSpecSuite extends PipelineTest
with SharedSparkSe
deleteCondition = None,
storedAsScdType = ScdType.Type1)))
val graph = ctx.resolveToDataflowGraph()
-
graph.auxiliaryTableSpecs(targetIdentifier).asInstanceOf[AutoCdcAuxiliaryTableSpec]
+ val inferredSchemas =
graph.inferSchemas(spark.sessionState.conf.caseSensitiveAnalysis)
+ graph.auxiliaryTableSpecs(inferredSchemas)(targetIdentifier)
+ .asInstanceOf[AutoCdcAuxiliaryTableSpec]
}
test("SCD1 aux schema is exactly the key columns plus the CDC metadata
column") {
diff --git
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1KeyDriftSuite.scala
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1KeyDriftSuite.scala
index 1d27b7d69dab..e180056761f2 100644
---
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1KeyDriftSuite.scala
+++
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1KeyDriftSuite.scala
@@ -307,6 +307,46 @@ class AutoCdcScd1KeyDriftSuite
}
}
+ test("AutoCDC key drift validation uses pipeline case sensitivity, not
session default") {
+ spark.sql(
+ s"CREATE TABLE $catalog.$namespace.target " +
+ s"(id INT NOT NULL, version BIGINT NOT NULL, $scd1MetadataDdl)"
+ )
+
+ val stream1 = MemoryStream[(Int, Long)]
+ stream1.addData((1, 1L))
+ runPipeline(buildPipeline("flow_v1", stream1.toDF().toDF("id", "version"),
Seq("id")))
+
+ withSQLConf(SQLConf.CASE_SENSITIVE.key -> "false") {
+ val stream2 = MemoryStream[(Int, Long)]
+ stream2.addData((1, 2L))
+ val ctx2 = new TestGraphRegistrationContext(
+ spark,
+ Map(SQLConf.CASE_SENSITIVE.key -> "true")) {
+ registerTable("target", catalog = Some(catalog), database =
Some(namespace))
+ registerFlow(autoCdcFlow(
+ name = "flow_v2",
+ target = "target",
+ query = dfFlowFunc(stream2.toDF().toDF("Id", "version")),
+ keys = Seq("Id"),
+ sequencing = $"version"))
+ }
+
+ val ex = intercept[RuntimeException] { runPipeline(ctx2) }
+ checkErrorInPipelineFailure(
+ failure = ex,
+ condition = "AUTOCDC_INVALID_STATE.KEY_SCHEMA_DRIFT",
+ sqlState = Some("42000"),
+ parameters = Map(
+ "tableName" ->
+ fullyQualifiedIdentifier("target", Some(catalog),
Some(namespace)).unquotedString,
+ "expectedKeySchema" -> "Id INT NOT NULL",
+ "recordedKeySchema" -> "id INT NOT NULL"
+ )
+ )
+ }
+ }
+
test("under the default (case-insensitive) resolver, an AutoCDC flow whose
key differs only " +
"in case from the recorded key does NOT trigger drift") {
// Pairs with the case-sensitive test above: same recorded key, but under
the default
@@ -315,11 +355,11 @@ class AutoCdcScd1KeyDriftSuite
// case-sensitive resolver in the validator is caught.
//
// Note that only the *key declaration* (`Seq("Id")`) has different casing
here -- the
- // source DF column name still matches the target's `id` exactly.
Differing the source DF
- // column casing as well would not exercise drift:
[[SchemaMergingUtils.mergeSchemas]] is
- // case-sensitive on column names and would add `Id` as a new column to
the target,
- // producing AMBIGUOUS_REFERENCE during the streaming write rather than
letting drift
- // validation make the call.
+ // source DF column name still matches the target's `id` exactly. This
keeps the test focused
+ // on the drift validator: whether the source DF column were `id` or `Id`,
under the default
+ // (case-insensitive) resolver schema evolution folds it onto the existing
`id` (SPARK-58517),
+ // so the streaming write itself would not fail either way and drift
validation remains the
+ // sole decision-maker.
spark.sql(
s"CREATE TABLE $catalog.$namespace.target " +
s"(id INT NOT NULL, version BIGINT NOT NULL, $scd1MetadataDdl)"
diff --git
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1SchemaEvolutionSuite.scala
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1SchemaEvolutionSuite.scala
index 635e3d93de56..1a626a6ddcdf 100644
---
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1SchemaEvolutionSuite.scala
+++
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1SchemaEvolutionSuite.scala
@@ -34,8 +34,9 @@ import org.apache.spark.sql.test.SharedSparkSession
/**
* Tests covering AutoCDC's interaction with non-key schema evolution across
pipeline runs. The
* suite documents the supported additive cases (new top-level columns, new
nested fields in
- * array-of-struct, broadening / narrowing column selection) and the cases
that fail loudly
- * today (subtractive nested evolution, type-incompatible changes, case-only
renames).
+ * array-of-struct, broadening / narrowing column selection, and -- under
case-insensitive
+ * resolution -- a source column differing from an existing one only in case)
and the cases that
+ * fail loudly today (subtractive nested evolution, type-incompatible changes).
*
* These behaviors are largely inherited from the lower layers
(`SchemaMergingUtils` for
* schema merge, the v2 writer's column-resolution layer for nested-field
handling) rather
@@ -585,18 +586,18 @@ class AutoCdcScd1SchemaEvolutionSuite
)
}
- test("a source DF column whose name differs from the target only by case
fails with " +
- "AMBIGUOUS_REFERENCE under case-insensitive resolution") {
+ test("a source DF column whose name differs from the target only by case is
folded onto the " +
+ "existing column under case-insensitive resolution") {
val session = spark
import session.implicits._
- // `DatasetManager`'s schema-merge compares the existing target schema and
the flow's
- // output schema *case-sensitively*: `SchemaMergingUtils.mergeSchemas`
calls
- // `StructType.merge` without forwarding the session-level
case-sensitivity. When the
- // target has `value` and the source DF emits `Value`, the merged schema
ends up with
- // both as separate columns. Reference resolution downstream is
case-insensitive
- // (Spark's default), so the MERGE plan trips on the duplicate and reports
- // AMBIGUOUS_REFERENCE.
+ // Under case-insensitive resolution (Spark's default), a target `value`
and a source `Value`
+ // are the same column. Schema evolution honors that by threading
case-sensitivity into
+ // `SchemaMergingUtils.mergeSchemas`: the merge maps `Value` onto the
existing `value`, so
+ // `diffSchemas` has no case-only difference left to process. No second
column is added, and
+ // the write succeeds. (Before SPARK-58517 the merge ran case-sensitively
regardless of the
+ // session and added a duplicate `Value` column, after which the
case-insensitive MERGE plan
+ // tripped on the ambiguous reference.)
withSQLConf(SQLConf.CASE_SENSITIVE.key -> "false") {
spark.sql(
s"CREATE TABLE $catalog.$namespace.target " +
@@ -605,8 +606,7 @@ class AutoCdcScd1SchemaEvolutionSuite
val stream = MemoryStream[(Int, Long, String)]
stream.addData((1, 1L, "alice"))
- // Source DF emits `Value` (capital), differing only in case from the
target's
- // `value` column.
+ // Source DF emits `Value` (capital), differing only in case from the
target's `value` column.
val df = stream.toDF().toDF("key", "version", "Value")
val ctx = singleAutoCdcFlowPipeline(
flowName = "auto_cdc_flow",
@@ -615,24 +615,18 @@ class AutoCdcScd1SchemaEvolutionSuite
keys = Seq("key"),
sequencing = functions.col("version"))
- val ex = intercept[RuntimeException] { runPipeline(ctx) }
- // The exact `name` and `referenceNames` parameters depend on internal
merge-plan
- // synthesis; the condition match is the meaningful invariant for this
test.
- checkErrorInPipelineFailure(
- failure = ex,
- condition = "AMBIGUOUS_REFERENCE",
- parameters = Map(
- "name" -> ".*",
- "referenceNames" -> ".*"
- ),
- matchPVals = true,
- queryContext = Array(
- ExpectedContext(
- fragment = s"`$catalog`.`$namespace`.`target`.`Value`",
- start = 0,
- stop = 27
- )
- )
+ runPipeline(ctx)
+
+ // The target schema is unchanged (still a single `value` column,
original case), and the
+ // row lands with the emitted value folded into it.
+ assert(
+ spark.table(s"$catalog.$namespace.target").schema.fieldNames.toSeq ===
+ Seq("key", "version", "value",
AutoCdcReservedNames.cdcMetadataColName),
+ "the target should keep its single `value` column, not gain a `Value`
column"
+ )
+ checkAnswer(
+ spark.table(s"$catalog.$namespace.target"),
+ Seq(Row(1, 1L, "alice", cdcMeta(None, Some(1L))))
)
}
}
diff --git
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2AuxiliaryTableSpecSuite.scala
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2AuxiliaryTableSpecSuite.scala
index a56fed3e5805..d01ec0f54736 100644
---
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2AuxiliaryTableSpecSuite.scala
+++
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2AuxiliaryTableSpecSuite.scala
@@ -76,7 +76,9 @@ class AutoCdcScd2AuxiliaryTableSpecSuite extends PipelineTest
with SharedSparkSe
deleteCondition = None,
storedAsScdType = ScdType.Type2)))
val graph = ctx.resolveToDataflowGraph()
-
graph.auxiliaryTableSpecs(targetIdentifier).asInstanceOf[AutoCdcAuxiliaryTableSpec]
+ val inferredSchemas =
graph.inferSchemas(spark.sessionState.conf.caseSensitiveAnalysis)
+ graph.auxiliaryTableSpecs(inferredSchemas)(targetIdentifier)
+ .asInstanceOf[AutoCdcAuxiliaryTableSpec]
}
/** The SCD2 target (inferred) schema for the default single-flow graph. */
@@ -97,7 +99,8 @@ class AutoCdcScd2AuxiliaryTableSpecSuite extends PipelineTest
with SharedSparkSe
columnSelection = None,
deleteCondition = None,
storedAsScdType = ScdType.Type2)))
- ctx.resolveToDataflowGraph().inferredSchema(targetIdentifier)
+ ctx.resolveToDataflowGraph()
+
.inferSchemas(spark.sessionState.conf.caseSensitiveAnalysis)(targetIdentifier)
}
test("SCD2 aux schema is the full target schema plus the deleted-by-batch-id
marker") {
diff --git
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/ConnectInvalidPipelineSuite.scala
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/ConnectInvalidPipelineSuite.scala
index 62680ebd3835..172d8354932b 100644
---
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/ConnectInvalidPipelineSuite.scala
+++
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/ConnectInvalidPipelineSuite.scala
@@ -31,6 +31,9 @@ import org.apache.spark.sql.types.{IntegerType, StructType}
*/
class ConnectInvalidPipelineSuite extends PipelineTest with SharedSparkSession
{
+ private def validateGraph(graph: DataflowGraph): DataflowGraph =
+ graph.validate(spark.sessionState.conf.caseSensitiveAnalysis)
+
test("Missing source") {
class P extends TestGraphRegistrationContext(spark) {
registerPersistedView("b", query = readFlowFunc("a"))
@@ -39,7 +42,7 @@ class ConnectInvalidPipelineSuite extends PipelineTest with
SharedSparkSession {
val dfg = new P().resolveToDataflowGraph()
assert(!dfg.resolved, "Pipeline should not have resolved properly")
val ex = intercept[UnresolvedPipelineException] {
- dfg.validate()
+ validateGraph(dfg)
}
assert(ex.getMessage.contains("Failed to resolve flows in the pipeline"))
assertAnalysisException(
@@ -64,7 +67,7 @@ class ConnectInvalidPipelineSuite extends PipelineTest with
SharedSparkSession {
val dfg = new P().resolveToDataflowGraph()
assert(!dfg.resolved, "Pipeline should not have resolved properly")
val ex = intercept[UnresolvedPipelineException] {
- dfg.validate()
+ validateGraph(dfg)
}
assert(ex.getMessage.contains("Failed to resolve flows in the pipeline"))
assert(
@@ -141,7 +144,9 @@ class ConnectInvalidPipelineSuite extends PipelineTest with
SharedSparkSession {
registerFlow("a", "a_2", sqlFlowFunc(spark, "SELECT non_existent_col
FROM RANGE(5)"))
registerTable("b", query = Option(readFlowFunc("a")))
}
- val ex = intercept[UnresolvedPipelineException] { new
P().resolveToDataflowGraph().validate() }
+ val ex = intercept[UnresolvedPipelineException] {
+ validateGraph(new P().resolveToDataflowGraph())
+ }
assert(ex.directFailures.keySet == Set(fullyQualifiedIdentifier("a_2")))
assert(ex.downstreamFailures.keySet == Set(fullyQualifiedIdentifier("b")))
@@ -158,7 +163,7 @@ class ConnectInvalidPipelineSuite extends PipelineTest with
SharedSparkSession {
val dfg = new P().resolveToDataflowGraph()
val ex = intercept[UnresolvedPipelineException] {
- dfg.validate()
+ validateGraph(dfg)
}.directFailures(fullyQualifiedIdentifier("b")).getMessage
verifyUnresolveColumnError(ex, "x", Seq("z"))
}
@@ -175,7 +180,7 @@ class ConnectInvalidPipelineSuite extends PipelineTest with
SharedSparkSession {
val dfg = new P().resolveToDataflowGraph()
val ex = intercept[UnresolvedPipelineException] {
- dfg.validate()
+ validateGraph(dfg)
}
assert(
ex.directFailures(fullyQualifiedIdentifier("c"))
@@ -200,7 +205,7 @@ class ConnectInvalidPipelineSuite extends PipelineTest with
SharedSparkSession {
val dfg = new P().resolveToDataflowGraph()
assert(!dfg.resolved)
val ex = intercept[UnresolvedPipelineException] {
- dfg.validate()
+ validateGraph(dfg)
}
assert(
ex.directFailures(fullyQualifiedIdentifier("c"))
@@ -217,7 +222,7 @@ class ConnectInvalidPipelineSuite extends PipelineTest with
SharedSparkSession {
registerPersistedView("a", query = readFlowFunc("a"))
}
val e = intercept[CircularDependencyException] {
- new P().resolveToDataflowGraph().validate()
+ validateGraph(new P().resolveToDataflowGraph())
}
assert(e.upstreamDataset == fullyQualifiedIdentifier("a"))
assert(e.downstreamTable == fullyQualifiedIdentifier("a"))
@@ -229,7 +234,7 @@ class ConnectInvalidPipelineSuite extends PipelineTest with
SharedSparkSession {
registerPersistedView("b", query = readFlowFunc("a"))
}
val e = intercept[CircularDependencyException] {
- new P().resolveToDataflowGraph().validate()
+ validateGraph(new P().resolveToDataflowGraph())
}
val cycle = Set(
fullyQualifiedIdentifier("a"),
@@ -260,7 +265,7 @@ class ConnectInvalidPipelineSuite extends PipelineTest with
SharedSparkSession {
fullyQualifiedIdentifier("d")
)
val e = intercept[CircularDependencyException] {
- new P().resolveToDataflowGraph().validate()
+ validateGraph(new P().resolveToDataflowGraph())
}
assert(e.upstreamDataset != e.downstreamTable)
assert(cycle.contains(e.upstreamDataset))
@@ -287,7 +292,7 @@ class ConnectInvalidPipelineSuite extends PipelineTest with
SharedSparkSession {
fullyQualifiedIdentifier("d")
)
val e = intercept[CircularDependencyException] {
- new P().resolveToDataflowGraph().validate()
+ validateGraph(new P().resolveToDataflowGraph())
}
assert(e.upstreamDataset != e.downstreamTable)
assert(cycle.contains(e.upstreamDataset))
@@ -313,7 +318,7 @@ class ConnectInvalidPipelineSuite extends PipelineTest with
SharedSparkSession {
fullyQualifiedIdentifier("d")
)
val e = intercept[CircularDependencyException] {
- new P().resolveToDataflowGraph().validate()
+ validateGraph(new P().resolveToDataflowGraph())
}
assert(e.upstreamDataset != e.downstreamTable)
assert(cycle.contains(e.upstreamDataset))
@@ -340,7 +345,7 @@ class ConnectInvalidPipelineSuite extends PipelineTest with
SharedSparkSession {
fullyQualifiedIdentifier("d")
)
val e = intercept[CircularDependencyException] {
- new P().resolveToDataflowGraph().validate()
+ validateGraph(new P().resolveToDataflowGraph())
}
assert(e.upstreamDataset != e.downstreamTable)
assert(cycle.contains(e.upstreamDataset))
@@ -408,7 +413,9 @@ class ConnectInvalidPipelineSuite extends PipelineTest with
SharedSparkSession {
registerPersistedView("a", query = dfFlowFunc(Seq(1).toDF()))
registerTable("b", query = Option(readStreamFlowFunc("a")))
}
- val ex = intercept[UnresolvedPipelineException] {
p.resolveToDataflowGraph().validate() }
+ val ex = intercept[UnresolvedPipelineException] {
+ validateGraph(p.resolveToDataflowGraph())
+ }
assert(
ex.directFailures(fullyQualifiedIdentifier("b"))
.getMessage
@@ -429,7 +436,7 @@ class ConnectInvalidPipelineSuite extends PipelineTest with
SharedSparkSession {
registerPersistedView("a", query = dfFlowFunc(mem.toDF()))
registerTable("b", query = Option(readFlowFunc("a")))
}
- val ex = intercept[UnresolvedPipelineException] {
p.resolveToDataflowGraph().validate() }
+ val ex = intercept[UnresolvedPipelineException] {
validateGraph(p.resolveToDataflowGraph()) }
assert(
ex.directFailures(fullyQualifiedIdentifier("b"))
.getMessage
@@ -449,7 +456,7 @@ class ConnectInvalidPipelineSuite extends PipelineTest with
SharedSparkSession {
}.resolveToDataflowGraph()
val ex = intercept[AnalysisException] {
- graph.validate()
+ validateGraph(graph)
}
checkError(
@@ -471,7 +478,7 @@ class ConnectInvalidPipelineSuite extends PipelineTest with
SharedSparkSession {
}.resolveToDataflowGraph()
val ex = intercept[AnalysisException] {
- graph.validate()
+ validateGraph(graph)
}
checkError(
@@ -499,7 +506,7 @@ class ConnectInvalidPipelineSuite extends PipelineTest with
SharedSparkSession {
}.resolveToDataflowGraph()
val ex = intercept[AnalysisException] {
- graph.validate()
+ validateGraph(graph)
}
checkError(
@@ -522,7 +529,7 @@ class ConnectInvalidPipelineSuite extends PipelineTest with
SharedSparkSession {
specifiedSchema = Option(new StructType().add("x", IntegerType))
)
}.resolveToDataflowGraph()
- val ex1 = intercept[AnalysisException] { graph1.validate() }
+ val ex1 = intercept[AnalysisException] { validateGraph(graph1) }
assert(
ex1.getMessage.contains(
s"'${fullyQualifiedIdentifier("a").unquotedString}' " +
@@ -535,7 +542,7 @@ class ConnectInvalidPipelineSuite extends PipelineTest with
SharedSparkSession {
registerTable("a", specifiedSchema = Option(new StructType().add("x",
IntegerType)))
registerFlow("a", "a", query = dfFlowFunc(Seq(true, false).toDF("x")),
once = true)
}.resolveToDataflowGraph()
- val ex2 = intercept[AnalysisException] { graph2.validate() }
+ val ex2 = intercept[AnalysisException] { validateGraph(graph2) }
assert(
ex2.getMessage.contains(
s"'${fullyQualifiedIdentifier("a").unquotedString}' " +
@@ -592,7 +599,7 @@ class ConnectInvalidPipelineSuite extends PipelineTest with
SharedSparkSession {
}.resolveToDataflowGraph()
val ex = intercept[AnalysisException] {
- graph.validate()
+ validateGraph(graph)
}
checkError(
@@ -642,7 +649,7 @@ class ConnectInvalidPipelineSuite extends PipelineTest with
SharedSparkSession {
}.resolveToDataflowGraph()
val ex = intercept[AnalysisException] {
- graph.validate()
+ validateGraph(graph)
}
checkError(
@@ -704,7 +711,7 @@ class ConnectInvalidPipelineSuite extends PipelineTest with
SharedSparkSession {
}.resolveToDataflowGraph()
val ex = intercept[AnalysisException] {
- graph.validate()
+ validateGraph(graph)
}
checkError(
@@ -749,7 +756,7 @@ class ConnectInvalidPipelineSuite extends PipelineTest with
SharedSparkSession {
}.resolveToDataflowGraph()
val ex = intercept[AnalysisException] {
- graph.validate()
+ validateGraph(graph)
}
checkError(
diff --git
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/ConnectValidPipelineSuite.scala
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/ConnectValidPipelineSuite.scala
index ea6d3202ba86..b90ccbbde7ce 100644
---
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/ConnectValidPipelineSuite.scala
+++
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/ConnectValidPipelineSuite.scala
@@ -504,7 +504,7 @@ class ConnectValidPipelineSuite extends PipelineTest with
SharedSparkSession {
registerFlow("sink_a", "sink_flow", query = readStreamFlowFunc("a"))
}
val g = P.resolveToDataflowGraph()
- g.validate()
+ g.validate(spark.sessionState.conf.caseSensitiveAnalysis)
assert(g.resolved)
assert(g.sink(TableIdentifier("sink_a")).isInstanceOf[Sink])
val sink = g.sink(TableIdentifier("sink_a"))
@@ -751,7 +751,7 @@ class ConnectValidPipelineSuite extends PipelineTest with
SharedSparkSession {
}.resolveToDataflowGraph()
assert(!unresolved.resolved, "case-sensitive consumer flow should fail
to resolve")
val ex = intercept[UnresolvedPipelineException] {
- unresolved.validate()
+ unresolved.validate(spark.sessionState.conf.caseSensitiveAnalysis)
}
assertAnalysisException(
ex.directFailures(fullyQualifiedIdentifier("consumer")),
diff --git
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/MaterializeTablesSuite.scala
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/MaterializeTablesSuite.scala
index 46e2d6d9ae63..a4df9867e2ab 100644
---
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/MaterializeTablesSuite.scala
+++
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/MaterializeTablesSuite.scala
@@ -32,6 +32,7 @@ import org.apache.spark.sql.connector.catalog.{
}
import org.apache.spark.sql.connector.expressions.{ClusterByTransform,
Expressions, FieldReference}
import org.apache.spark.sql.execution.streaming.runtime.MemoryStream
+import org.apache.spark.sql.internal.SQLConf
import
org.apache.spark.sql.pipelines.graph.DatasetManager.TableMaterializationException
import org.apache.spark.sql.pipelines.utils.{BaseCoreExecutionTest,
TestGraphRegistrationContext}
import org.apache.spark.sql.test.SharedSparkSession
@@ -565,7 +566,7 @@ abstract class MaterializeTablesSuite extends
BaseCoreExecutionTest {
val graph1 = new TestGraphRegistrationContext(spark) {
registerTable("a", query =
Option(dfFlowFunc(spark.readStream.format("rate").load())))
- }.resolveToDataflowGraph().validate()
+
}.resolveToDataflowGraph().validate(spark.sessionState.conf.caseSensitiveAnalysis)
materializeGraph(graph1, storageRoot = storageRoot)
}
@@ -638,7 +639,7 @@ abstract class MaterializeTablesSuite extends
BaseCoreExecutionTest {
new TestGraphRegistrationContext(spark) {
registerView("a", query = dfFlowFunc(streamInts.toDF()))
registerTable("b", query = Option(sqlFlowFunc(spark, "SELECT value
AS x FROM STREAM a")))
- }.resolveToDataflowGraph().validate()
+
}.resolveToDataflowGraph().validate(spark.sessionState.conf.caseSensitiveAnalysis)
val (refreshSelection, fullRefreshSelection) = if (isFullRefresh) {
(NoTables, AllTables)
@@ -668,7 +669,7 @@ abstract class MaterializeTablesSuite extends
BaseCoreExecutionTest {
new TestGraphRegistrationContext(spark) {
registerView("a", query = dfFlowFunc(streamInts.toDF()))
registerTable("b", query = Option(sqlFlowFunc(spark, "SELECT value
AS y FROM STREAM a")))
- }.resolveToDataflowGraph().validate(),
+
}.resolveToDataflowGraph().validate(spark.sessionState.conf.caseSensitiveAnalysis),
contextOpt = updateContextOpt,
storageRoot = storageRoot
)
@@ -1214,6 +1215,407 @@ abstract class MaterializeTablesSuite extends
BaseCoreExecutionTest {
}
}
+ test("SPARK-58517: re-materializing with a case-only column difference is a
no-op under " +
+ "case-insensitive resolution") {
+ withRecordingCatalog {
+ withSQLConf(SQLConf.CASE_SENSITIVE.key -> "false") {
+ // Create the table with `value`, then re-materialize with the same
column cased as `Value`.
+ // Under case-insensitive resolution these are the same column, so
schema evolution must
+ // fold `Value` onto the existing `value`: no alterTable, and the
persisted column keeps its
+ // original name/case. Before SPARK-58517 the case-sensitive merge
instead added a second
+ // `Value` column, corrupting the table.
+ materializeStreamingTable(
+ "t", new StructType().add("id", IntegerType).add("value",
StringType), Map.empty)
+ assert(recordingCatalog.recordedAlters.isEmpty)
+
+ materializeStreamingTable(
+ "t", new StructType().add("id", IntegerType).add("Value",
StringType), Map.empty)
+ assert(recordingCatalog.recordedAlters.isEmpty,
+ s"expected no alter, got: ${recordingCatalog.recordedAlters}")
+
+ assert(
+ loadTableFromRecordingCatalog("t").columns() sameElements
+ CatalogV2Util.structTypeToV2Columns(
+ new StructType().add("id", IntegerType).add("value", StringType)
+ ),
+ "the persisted schema should keep the original `value` column, not
gain a `Value` column"
+ )
+ }
+ }
+ }
+
+ test("SPARK-58517: multi-flow schema inference folds case-only column
differences under " +
+ "case-insensitive resolution") {
+ withRecordingCatalog {
+ withSQLConf(SQLConf.CASE_SENSITIVE.key -> "false") {
+ // Two append flows write to the same streaming table, one emitting
`value` and the other
+ // `Value`. The target schema is INFERRED by merging the flows'
schemas, which happens
+ // before the evolveTable path runs -- so inference must honor
case-insensitivity too,
+ // otherwise the table is created with both columns and the engine's
own resolver cannot
+ // disambiguate them.
+ val df1 = spark.createDataFrame(
+ spark.sparkContext.parallelize(Seq(Row(1, "a"))),
+ new StructType().add("id", IntegerType).add("value", StringType))
+ val df2 = spark.createDataFrame(
+ spark.sparkContext.parallelize(Seq(Row(2, "b"))),
+ new StructType().add("id", IntegerType).add("Value", StringType))
+
+ val ctx = new TestGraphRegistrationContext(spark) {
+ registerTable(
+ "t",
+ catalog = Option(recordingCatalogName),
+ database = Option(recordingNamespace))
+ registerFlow(
+ "t", "f1", dfFlowFunc(df1),
+ catalog = Option(recordingCatalogName), database =
Option(recordingNamespace))
+ registerFlow(
+ "t", "f2", dfFlowFunc(df2),
+ catalog = Option(recordingCatalogName), database =
Option(recordingNamespace))
+ }
+
+ val graph = ctx.resolveToDataflowGraph()
+ val inferredSchemas =
graph.inferSchemas(spark.sessionState.conf.caseSensitiveAnalysis)
+ val (targetIdentifier, inferred) = inferredSchemas.head
+ val lowestIdentifierFlowValueField =
+ graph.resolvedFlowsTo(targetIdentifier)
+ .sortBy(_.identifier.unquotedString)
+ .head
+ .schema
+ .fieldNames(1)
+ // The two spellings must fold into a single column, and the lowest
flow identifier
+ // supplies the surviving spelling.
+ assert(inferred.fieldNames.toSeq === Seq("id",
lowestIdentifierFlowValueField))
+ }
+ }
+ }
+
+ test("SPARK-58517: a case-only fold picks the same spelling for the
materialized table and for " +
+ "downstream resolution, in either flow declaration order") {
+ // The table's schema is derived twice from the same flows, by two
different callers: the graph
+ // materializes the table from `inferSchemas`, while downstream flows
resolve against the
+ // `VirtualTableInput` schema, whose `availableFlows` is in declaration
order, not sorted. Both
+ // go through `inferSchemaFromFlows`, which merges in sorted flow
identifier order, so the
+ // surviving spelling is the same on both paths and does not depend on
declaration order. Were
+ // the two to disagree, the downstream view would persist a column spelled
differently from the
+ // source column it selects, and -- because `diffSchemas` keys on exact
names -- reordering the
+ // flow definitions would turn the next refresh into a drop-then-add of
that column.
+ withSQLConf(SQLConf.CASE_SENSITIVE.key -> "false") {
+ // `f_eu` sorts before `f_us`, so `value` is the surviving spelling in
both orders.
+ def graphWithFlowsDeclared(usFirst: Boolean): DataflowGraph = {
+ val usDf = spark.createDataFrame(
+ spark.sparkContext.parallelize(Seq(Row(1, "a"))),
+ new StructType().add("id", IntegerType).add("Value", StringType))
+ val euDf = spark.createDataFrame(
+ spark.sparkContext.parallelize(Seq(Row(2, "b"))),
+ new StructType().add("id", IntegerType).add("value", StringType))
+
+ new TestGraphRegistrationContext(spark) {
+ registerTable("events")
+ val registerUs = () => registerFlow("events", "f_us",
dfFlowFunc(usDf))
+ val registerEu = () => registerFlow("events", "f_eu",
dfFlowFunc(euDf))
+ if (usFirst) {
+ registerUs()
+ registerEu()
+ } else {
+ registerEu()
+ registerUs()
+ }
+ // Reads the table, so it resolves against the VirtualTableInput
schema.
+ registerMaterializedView(
+ "events_summary",
+ query = sqlFlowFunc(spark, "SELECT id, value FROM events"))
+ }.resolveToDataflowGraph()
+ }
+
+ Seq(true, false).foreach { usFirst =>
+ val graph = graphWithFlowsDeclared(usFirst)
+ val sessionCaseSensitive =
spark.sessionState.conf.caseSensitiveAnalysis
+ val eventsIdentifier = fullyQualifiedIdentifier("events")
+
+ val materializedSchema =
graph.inferSchemas(sessionCaseSensitive)(eventsIdentifier)
+ assert(
+ materializedSchema.fieldNames.toSeq === Seq("id", "value"),
+ s"materialized schema for usFirst=$usFirst")
+
+ // The downstream view's own schema reflects what it resolved against
upstream: Spark takes
+ // the resolved attribute's name, so a `Value` upstream would surface
here as `Value`.
+ val summarySchema = graph
+
.inferSchemas(sessionCaseSensitive)(fullyQualifiedIdentifier("events_summary"))
+ assert(
+ summarySchema.fieldNames.toSeq === Seq("id", "value"),
+ s"downstream schema for usFirst=$usFirst")
+ }
+ }
+ }
+
+ test("multi-flow schema inference keeps case-only column differences
distinct under " +
+ "case-sensitive resolution") {
+ withRecordingCatalog {
+ withSQLConf(SQLConf.CASE_SENSITIVE.key -> "true") {
+ // The case-sensitive control: `value` and `Value` are distinct
columns, so inference
+ // contributes both.
+ val df1 = spark.createDataFrame(
+ spark.sparkContext.parallelize(Seq(Row(1, "a"))),
+ new StructType().add("id", IntegerType).add("value", StringType))
+ val df2 = spark.createDataFrame(
+ spark.sparkContext.parallelize(Seq(Row(2, "b"))),
+ new StructType().add("id", IntegerType).add("Value", StringType))
+
+ val ctx = new TestGraphRegistrationContext(spark) {
+ registerTable(
+ "t",
+ catalog = Option(recordingCatalogName),
+ database = Option(recordingNamespace))
+ registerFlow(
+ "t", "f1", dfFlowFunc(df1),
+ catalog = Option(recordingCatalogName), database =
Option(recordingNamespace))
+ registerFlow(
+ "t", "f2", dfFlowFunc(df2),
+ catalog = Option(recordingCatalogName), database =
Option(recordingNamespace))
+ }
+
+ val graph = ctx.resolveToDataflowGraph()
+ val inferred = graph.inferSchemas(
+ spark.sessionState.conf.caseSensitiveAnalysis).values.head
+ // Both spellings survive as distinct columns in sorted flow
identifier order.
+ assert(inferred.fieldNames.toSeq === Seq("id", "value", "Value"))
+ }
+ }
+ }
+
+ test("SPARK-58517: a materialized view's case-only column rename is applied
under " +
+ "case-insensitive resolution") {
+ // The non-merging path: for a materialized view `targetSchema` is the
run's declared schema
+ // as-is (no merge with the persisted schema), so a case-only rename must
remain visible to
+ // `diffSchemas` as a drop-then-add. Case-insensitive matching here would
emit no change at all
+ // and freeze the persisted spelling forever -- the table would
permanently disagree with its
+ // definition, with no error pointing at the discrepancy, and a colleague
materializing the same
+ // definition against a fresh table would get the declared casing instead.
+ withSQLConf(SQLConf.CASE_SENSITIVE.key -> "false") {
+ materializeGraph(
+ new TestGraphRegistrationContext(spark) {
+ registerView("src", query = dfFlowFunc(Seq((1, 2L)).toDF("id",
"total")))
+ registerMaterializedView("mv", query = sqlFlowFunc(spark, "SELECT
id, total FROM src"))
+ }.resolveToDataflowGraph(),
+ storageRoot = storageRoot
+ )
+
+ val catalog =
spark.sessionState.catalogManager.currentCatalog.asInstanceOf[TableCatalog]
+ val identifier =
Identifier.of(Array(TestGraphRegistrationContext.DEFAULT_DATABASE), "mv")
+ assert(
+ catalog.loadTable(identifier).columns() sameElements
CatalogV2Util.structTypeToV2Columns(
+ new StructType().add("id", IntegerType).add("total", LongType))
+ )
+
+ // Re-materialize with the column cased as `Total`. The table must
follow the definition.
+ materializeGraph(
+ new TestGraphRegistrationContext(spark) {
+ registerView("src", query = dfFlowFunc(Seq((1, 2L)).toDF("id",
"total")))
+ registerMaterializedView(
+ "mv", query = sqlFlowFunc(spark, "SELECT id, total AS Total FROM
src"))
+ }.resolveToDataflowGraph(),
+ storageRoot = storageRoot
+ )
+ assert(
+ catalog.loadTable(identifier).columns() sameElements
CatalogV2Util.structTypeToV2Columns(
+ new StructType().add("id", IntegerType).add("Total", LongType)),
+ "the materialized view should adopt the declared `Total` casing, not
keep `total`"
+ )
+ }
+ }
+
+ test("SPARK-58517: a full-refreshed streaming table's case-only column
rename is applied " +
+ "under case-insensitive resolution") {
+ // The streaming-table analog of the materialized-view case above: a full
refresh also takes
+ // `targetSchema` as the declared schema without merging, so the same
case-only rename must be
+ // applied rather than silently ignored.
+ withSQLConf(SQLConf.CASE_SENSITIVE.key -> "false") {
+ val graph = materializeGraph(
+ new TestGraphRegistrationContext(spark) {
+ registerView("src", query = dfFlowFunc(Seq((1, 2L)).toDF("id",
"total")))
+ registerTable("st", query = Option(sqlFlowFunc(spark, "SELECT id,
total FROM src")))
+ }.resolveToDataflowGraph(),
+ storageRoot = storageRoot
+ )
+
+ val catalog =
spark.sessionState.catalogManager.currentCatalog.asInstanceOf[TableCatalog]
+ val identifier =
Identifier.of(Array(TestGraphRegistrationContext.DEFAULT_DATABASE), "st")
+ assert(
+ catalog.loadTable(identifier).columns() sameElements
CatalogV2Util.structTypeToV2Columns(
+ new StructType().add("id", IntegerType).add("total", LongType))
+ )
+
+ val renamedGraph =
+ new TestGraphRegistrationContext(spark) {
+ registerView("src", query = dfFlowFunc(Seq((1, 2L)).toDF("id",
"total")))
+ registerTable(
+ "st", query = Option(sqlFlowFunc(spark, "SELECT id, total AS Total
FROM src")))
+ }.resolveToDataflowGraph()
+
+ materializeGraph(
+ renamedGraph,
+ contextOpt = Option(
+ TestPipelineUpdateContext(
+ spark = spark,
+ unresolvedGraph = graph,
+ refreshTables = NoTables,
+ fullRefreshTables = AllTables,
+ storageRoot = storageRoot
+ )
+ ),
+ storageRoot = storageRoot
+ )
+ assert(
+ catalog.loadTable(identifier).columns() sameElements
CatalogV2Util.structTypeToV2Columns(
+ new StructType().add("id", IntegerType).add("Total", LongType)),
+ "a full-refreshed streaming table should adopt the declared `Total`
casing"
+ )
+ }
+ }
+
+ test("re-materializing with a case-only column difference adds a column
under case-sensitive " +
+ "resolution") {
+ withRecordingCatalog {
+ withSQLConf(SQLConf.CASE_SENSITIVE.key -> "true") {
+ // The case-sensitive counterpart: `value` and `Value` are distinct,
so `Value` is added.
+ materializeStreamingTable(
+ "t", new StructType().add("id", IntegerType).add("value",
StringType), Map.empty)
+ assert(recordingCatalog.recordedAlters.isEmpty)
+
+ materializeStreamingTable(
+ "t", new StructType().add("id", IntegerType).add("Value",
StringType), Map.empty)
+ assert(recordingCatalog.recordedAlters.size == 1)
+ val changes = recordingCatalog.recordedAlters.flatten
+ assert(changes.collect { case ac: TableChange.AddColumn =>
ac.fieldNames()(0) } ==
+ Seq("Value"))
+ }
+ }
+ }
+
+ test("SPARK-58517: schema evolution uses the pipeline's case sensitivity,
not the session's") {
+ // A pipeline-level `SET spark.sql.caseSensitive` never reaches the
session, so evolution must
+ // read it from the flows. Here the session default is case-INsensitive
while the pipeline asks
+ // for case-SENSITIVE, so `Value` must become its own column alongside the
persisted `value` --
+ // matching how the flow itself resolves the name.
+ withRecordingCatalog {
+ withSQLConf(SQLConf.CASE_SENSITIVE.key -> "false") {
+ materializeGraph(
+ new TestGraphRegistrationContext(
+ spark, Map(SQLConf.CASE_SENSITIVE.key -> "true")) {
+ registerView("src", query = dfFlowFunc(Seq((1, "a")).toDF("id",
"value")))
+ registerTable(
+ "t",
+ query = Option(sqlFlowFunc(spark, "SELECT id, value FROM src")),
+ catalog = Option(recordingCatalogName),
+ database = Option(recordingNamespace))
+ }.resolveToDataflowGraph(),
+ storageRoot = storageRoot
+ )
+ assert(
+ loadTableFromRecordingCatalog("t").columns() sameElements
+ CatalogV2Util.structTypeToV2Columns(
+ new StructType().add("id", IntegerType).add("value",
StringType)))
+
+ materializeGraph(
+ new TestGraphRegistrationContext(
+ spark, Map(SQLConf.CASE_SENSITIVE.key -> "true")) {
+ registerView("src", query = dfFlowFunc(Seq((1, "a")).toDF("id",
"value")))
+ registerTable(
+ "t",
+ query = Option(sqlFlowFunc(spark, "SELECT id, value AS Value
FROM src")),
+ catalog = Option(recordingCatalogName),
+ database = Option(recordingNamespace))
+ }.resolveToDataflowGraph(),
+ storageRoot = storageRoot
+ )
+ assert(
+ loadTableFromRecordingCatalog("t").columns() sameElements
+ CatalogV2Util.structTypeToV2Columns(
+ new StructType()
+ .add("id", IntegerType)
+ .add("value", StringType)
+ .add("Value", StringType)),
+ "the pipeline asked for case-sensitive resolution, so `Value` must
be its own column")
+ }
+ }
+ }
+
+ test("SPARK-58517: flows writing to one table that disagree on case
sensitivity are rejected") {
+ // The effective value decides whether names differing only in case
identify the same column, so
+ // if the flows disagree the resulting schema would depend on the order
they are evaluated in.
+ // Fail with a clear error instead of picking one arbitrarily.
+ val ctx = new TestGraphRegistrationContext(spark) {
+ registerView("src", query = dfFlowFunc(Seq((1, "a")).toDF("id",
"value")))
+ registerTable("t")
+ registerFlow(
+ "t", "f1", sqlFlowFunc(spark, "SELECT id, value FROM src"),
+ sqlConf = Map(SQLConf.CASE_SENSITIVE.key -> "true"))
+ registerFlow(
+ "t", "f2", sqlFlowFunc(spark, "SELECT id, value AS Value FROM src"),
+ sqlConf = Map(SQLConf.CASE_SENSITIVE.key -> "false"))
+ }
+
+ val ex = intercept[AnalysisException] {
+ ctx.resolveToDataflowGraph().inferSchemas(
+ spark.sessionState.conf.caseSensitiveAnalysis)
+ }
+ checkError(
+ exception = ex,
+ condition = "CONFLICTING_PIPELINE_FLOW_CASE_SENSITIVITY",
+ parameters = Map(
+ "tableName" -> "spark_catalog.test_db.t",
+ "configKey" -> SQLConf.CASE_SENSITIVE.key,
+ "flowConfigurations" ->
+ ("false (spark_catalog.test_db.f2); true (spark_catalog.test_db.f1)")
+ )
+ )
+ }
+
+ test("SPARK-58517: a flow inheriting the session value conflicts with one
that overrides it") {
+ // f2 leaves the conf unset, so it inherits the session's case-INsensitive
default, which
+ // conflicts with f1's explicit case-sensitive request just as much as an
opposite explicit
+ // value would.
+ withSQLConf(SQLConf.CASE_SENSITIVE.key -> "false") {
+ val ctx = new TestGraphRegistrationContext(spark) {
+ registerView("src", query = dfFlowFunc(Seq((1, "a")).toDF("id",
"value")))
+ registerTable("t")
+ registerFlow(
+ "t", "f1", sqlFlowFunc(spark, "SELECT id, value FROM src"),
+ sqlConf = Map(SQLConf.CASE_SENSITIVE.key -> "true"))
+ registerFlow("t", "f2", sqlFlowFunc(spark, "SELECT id, value FROM
src"))
+ }
+
+ val ex = intercept[AnalysisException] {
+ ctx.resolveToDataflowGraph().inferSchemas(
+ spark.sessionState.conf.caseSensitiveAnalysis)
+ }
+ assert(ex.getCondition === "CONFLICTING_PIPELINE_FLOW_CASE_SENSITIVITY")
+ assert(ex.getMessage.contains("session default"))
+ }
+ }
+
+ test("SPARK-58517: flows that agree on case sensitivity are accepted") {
+ // The negative control: identical explicit values are not a conflict, and
neither is a value
+ // that merely differs in spelling from the session's ("TRUE" vs "true").
+ withSQLConf(SQLConf.CASE_SENSITIVE.key -> "false") {
+ val ctx = new TestGraphRegistrationContext(spark) {
+ registerView("src", query = dfFlowFunc(Seq((1, "a")).toDF("id",
"value")))
+ registerTable("t")
+ registerFlow(
+ "t", "f1", sqlFlowFunc(spark, "SELECT id, value FROM src"),
+ sqlConf = Map(SQLConf.CASE_SENSITIVE.key -> "true"))
+ registerFlow(
+ "t", "f2", sqlFlowFunc(spark, "SELECT id, value AS Value FROM src"),
+ sqlConf = Map(SQLConf.CASE_SENSITIVE.key -> "TRUE"))
+ }
+ val inferred = ctx.resolveToDataflowGraph()
+ .inferSchemas(spark.sessionState.conf.caseSensitiveAnalysis)(
+ fullyQualifiedIdentifier("t"))
+ // Case-sensitive, so both spellings survive.
+ assert(inferred.fieldNames.toSeq === Seq("id", "value", "Value"))
+ }
+ }
+
test("re-materializing with a dropped property neither removes it nor issues
an alterTable") {
withRecordingCatalog {
val schema = new StructType().add("id", IntegerType)
diff --git
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/SqlPipelineSuite.scala
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/SqlPipelineSuite.scala
index cec8db6ec528..8ce7971b405f 100644
---
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/SqlPipelineSuite.scala
+++
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/SqlPipelineSuite.scala
@@ -30,6 +30,12 @@ class SqlPipelineSuite extends PipelineTest with
SharedSparkSession {
private val externalTable1Ident = fullyQualifiedIdentifier("external_t1")
private val externalTable2Ident = fullyQualifiedIdentifier("external_t2")
+ private def resolveGraph(graph: DataflowGraph): DataflowGraph =
+ graph.resolve(spark.sessionState.conf.caseSensitiveAnalysis)
+
+ private def validateGraph(graph: DataflowGraph): DataflowGraph =
+ graph.validate(spark.sessionState.conf.caseSensitiveAnalysis)
+
override def beforeEach(): Unit = {
super.beforeEach()
// Create mock external tables that tests can reference, ex. to stream
from.
@@ -53,7 +59,7 @@ class SqlPipelineSuite extends PipelineTest with
SharedSparkSession {
|SELECT * FROM STREAM $externalTable2Ident;
|""".stripMargin
)
- val resolvedDataflowGraph = unresolvedDataflowGraph.resolve()
+ val resolvedDataflowGraph = resolveGraph(unresolvedDataflowGraph)
assert(resolvedDataflowGraph.flows.size == 4)
assert(resolvedDataflowGraph.tables.size == 2)
@@ -127,7 +133,7 @@ class SqlPipelineSuite extends PipelineTest with
SharedSparkSession {
"CREATE MATERIALIZED VIEW a COMMENT 'this is a comment' AS SELECT *
FROM range(1, 4)"
)
- val resolvedDataflowGraph = unresolvedDataflowGraph.resolve()
+ val resolvedDataflowGraph = resolveGraph(unresolvedDataflowGraph)
val flowA =
resolvedDataflowGraph.resolvedFlows
@@ -144,7 +150,7 @@ class SqlPipelineSuite extends PipelineTest with
SharedSparkSession {
|""".stripMargin
)
- val resolvedDataflowGraph = unresolvedDataflowGraph.resolve()
+ val resolvedDataflowGraph = resolveGraph(unresolvedDataflowGraph)
assert(
resolvedDataflowGraph.resolvedFlows
@@ -168,7 +174,7 @@ class SqlPipelineSuite extends PipelineTest with
SharedSparkSession {
|""".stripMargin
)
- val resolvedDataflowGraph = unresolvedDataflowGraph.resolve()
+ val resolvedDataflowGraph = resolveGraph(unresolvedDataflowGraph)
Seq("a", "b", "c", "d").foreach { datasetName =>
val backingFlow = resolvedDataflowGraph.resolvedFlows
@@ -258,7 +264,7 @@ class SqlPipelineSuite extends PipelineTest with
SharedSparkSession {
|PARTITIONED BY (id1, id2)
|AS SELECT id as id1, id as id2 FROM range(1,2)
""".stripMargin
)
- val resolvedDataflowGraph = unresolvedDataflowGraph.resolve()
+ val resolvedDataflowGraph = resolveGraph(unresolvedDataflowGraph)
assert(
resolvedDataflowGraph.tables
@@ -363,7 +369,7 @@ class SqlPipelineSuite extends PipelineTest with
SharedSparkSession {
val unresolvedDataflowGraph = unresolvedDataflowGraphFromSql(
sqlText = "CREATE STREAMING TABLE st TBLPROPERTIES ('prop1'='foo',
'prop2'='bar') AS SELECT 1"
)
- val resolvedDataflowGraph = unresolvedDataflowGraph.resolve()
+ val resolvedDataflowGraph = resolveGraph(unresolvedDataflowGraph)
assert(
resolvedDataflowGraph.tables
.find(_.identifier == fullyQualifiedIdentifier("st"))
@@ -387,7 +393,7 @@ class SqlPipelineSuite extends PipelineTest with
SharedSparkSession {
|""".stripMargin
)
- val resolvedDataflowGraph = unresolvedDataflowGraph.resolve()
+ val resolvedDataflowGraph = resolveGraph(unresolvedDataflowGraph)
assert(
resolvedDataflowGraph.flows
@@ -518,7 +524,7 @@ class SqlPipelineSuite extends PipelineTest with
SharedSparkSession {
|""".stripMargin
)
- val resolvedDataflowGraph = unresolvedDataflowGraph.resolve()
+ val resolvedDataflowGraph = resolveGraph(unresolvedDataflowGraph)
assert(
resolvedDataflowGraph.resolutionFailedFlows
@@ -577,10 +583,10 @@ class SqlPipelineSuite extends PipelineTest with
SharedSparkSession {
|""".stripMargin
)
- val resolvedDataflowGraph = unresolvedDataflowGraph.resolve()
+ val resolvedDataflowGraph = resolveGraph(unresolvedDataflowGraph)
// Let inferred/declared schema mismatch detection execute
- resolvedDataflowGraph.validate()
+ validateGraph(resolvedDataflowGraph)
val expectedSchema = new StructType().add(name = "id", dataType =
LongType, nullable = false)
@@ -651,7 +657,7 @@ class SqlPipelineSuite extends PipelineTest with
SharedSparkSession {
val unresolvedDataflowGraph = unresolvedDataflowGraphFromSql(
sqlText = s"CREATE VIEW b COMMENT 'my persisted comment' AS SELECT *
FROM range(1, 4);"
)
- val graph = unresolvedDataflowGraph.resolve().validate()
+ val graph = validateGraph(resolveGraph(unresolvedDataflowGraph))
val view = graph.views.last
@@ -897,9 +903,7 @@ class SqlPipelineSuite extends PipelineTest with
SharedSparkSession {
checkError(
exception = intercept[AnalysisException] {
- unresolvedDataflowGraph
- .resolve()
- .validate()
+ validateGraph(resolveGraph(unresolvedDataflowGraph))
},
condition = "PIPELINE_DATASET_WITHOUT_FLOW",
sqlState = Option("0A000"),
diff --git
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/TriggeredGraphExecutionSuite.scala
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/TriggeredGraphExecutionSuite.scala
index db8c368ca89d..313f86753222 100644
---
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/TriggeredGraphExecutionSuite.scala
+++
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/TriggeredGraphExecutionSuite.scala
@@ -60,7 +60,7 @@ class TriggeredGraphExecutionSuite extends ExecutionTest with
SharedSparkSession
registerMaterializedView("b", query = readFlowFunc("a"))
}
val unresolvedGraph = pipelineDef.toDataflowGraph
- val resolvedGraph = unresolvedGraph.resolve()
+ val resolvedGraph =
unresolvedGraph.resolve(spark.sessionState.conf.caseSensitiveAnalysis)
assert(resolvedGraph.flows.size == 2)
assert(unresolvedGraph.flows.size == 2)
assert(unresolvedGraph.tables.size == 2)
@@ -109,7 +109,7 @@ class TriggeredGraphExecutionSuite extends ExecutionTest
with SharedSparkSession
}
val unresolvedGraph = pipelineDef.toDataflowGraph
- val resolvedGraph = unresolvedGraph.resolve()
+ val resolvedGraph =
unresolvedGraph.resolve(spark.sessionState.conf.caseSensitiveAnalysis)
assert(resolvedGraph.flows.size == 4)
assert(resolvedGraph.tables.size == 3)
assert(resolvedGraph.views.size == 1)
diff --git
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/UserSpecifiedSchemaValidationSuite.scala
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/UserSpecifiedSchemaValidationSuite.scala
index d216539c93bc..fee3493d01b9 100644
---
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/UserSpecifiedSchemaValidationSuite.scala
+++
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/UserSpecifiedSchemaValidationSuite.scala
@@ -20,10 +20,11 @@ package org.apache.spark.sql.pipelines.graph
import org.apache.spark.sql.AnalysisException
import org.apache.spark.sql.execution.streaming.runtime.MemoryStream
import org.apache.spark.sql.functions
+import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.pipelines.autocdc.{ChangeArgs, ScdType,
UnqualifiedColumnName}
import org.apache.spark.sql.pipelines.utils.{PipelineTest,
TestGraphRegistrationContext}
import org.apache.spark.sql.test.SharedSparkSession
-import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.types.{IntegerType, StringType, StructType}
/**
* Tests for `GraphValidations.validateUserSpecifiedSchemas`, which requires a
table's
@@ -100,10 +101,14 @@ class UserSpecifiedSchemaValidationSuite extends
PipelineTest with SharedSparkSe
/** The full inferred AUTO CDC output schema (data columns plus the reserved
metadata column). */
private def autoCdcInferredSchema(flowName: String): StructType =
- autoCdcGraph(flowName, declaredSchema =
None).inferredSchema(targetIdentifier)
+ autoCdcGraph(flowName, declaredSchema = None)
+
.inferSchemas(spark.sessionState.conf.caseSensitiveAnalysis)(targetIdentifier)
+
+ private def validateGraph(graph: DataflowGraph): DataflowGraph =
+ graph.validate(spark.sessionState.conf.caseSensitiveAnalysis)
private def assertSchemaIncompatible(graph: DataflowGraph): Unit = {
- val ex = intercept[AnalysisException](graph.validate())
+ val ex = intercept[AnalysisException](validateGraph(graph))
assert(ex.getCondition ==
"USER_SPECIFIED_AND_INFERRED_SCHEMA_NOT_COMPATIBLE")
assert(ex.getMessage.contains(targetIdentifier.unquotedString))
}
@@ -111,7 +116,7 @@ class UserSpecifiedSchemaValidationSuite extends
PipelineTest with SharedSparkSe
// Plain flows: the inferred schema is exactly the source's data columns.
test("compatible user-specified schema is accepted for an implicit plain
flow") {
- plainGraph(flowName = "target", declaredSchema =
Some(dataSchema)).validate()
+ validateGraph(plainGraph(flowName = "target", declaredSchema =
Some(dataSchema)))
}
test("incompatible user-specified schema is rejected for an implicit plain
flow") {
@@ -120,7 +125,7 @@ class UserSpecifiedSchemaValidationSuite extends
PipelineTest with SharedSparkSe
}
test("compatible user-specified schema is accepted for a named plain flow") {
- plainGraph(flowName = "plain_flow", declaredSchema =
Some(dataSchema)).validate()
+ validateGraph(plainGraph(flowName = "plain_flow", declaredSchema =
Some(dataSchema)))
}
test("incompatible user-specified schema is rejected for a named plain
flow") {
@@ -128,6 +133,53 @@ class UserSpecifiedSchemaValidationSuite extends
PipelineTest with SharedSparkSe
dataSchemaMissingColumn)))
}
+ test("user-specified schema validation uses pipeline case sensitivity, not
session default") {
+ withSQLConf(SQLConf.CASE_SENSITIVE.key -> "false") {
+ val ctx = new TestGraphRegistrationContext(
+ spark,
+ Map(SQLConf.CASE_SENSITIVE.key -> "true")) {
+ val session = spark
+ import session.implicits._
+
+ registerView("src", query = dfFlowFunc(Seq((1, "alice")).toDF("id",
"value")))
+ registerTable(
+ "target",
+ specifiedSchema = Some(
+ new StructType().add("id", IntegerType).add("value", StringType)))
+ registerFlow(
+ destinationName = "target",
+ name = "case_sensitive_flow",
+ query = sqlFlowFunc(spark, "SELECT id, value AS Value FROM src"))
+ }
+
+ assertSchemaIncompatible(ctx.resolveToDataflowGraph())
+ }
+ }
+
+ test("user-specified schema validation uses case sensitivity inherited from
upstream view") {
+ withSQLConf(SQLConf.CASE_SENSITIVE.key -> "false") {
+ val ctx = new TestGraphRegistrationContext(spark) {
+ val session = spark
+ import session.implicits._
+
+ registerPersistedView(
+ "src",
+ query = dfFlowFunc(Seq((1, "alice")).toDF("id", "value")),
+ sqlConf = Map(SQLConf.CASE_SENSITIVE.key -> "true"))
+ registerTable(
+ "target",
+ specifiedSchema = Some(
+ new StructType().add("id", IntegerType).add("value", StringType)))
+ registerFlow(
+ destinationName = "target",
+ name = "case_sensitive_flow",
+ query = sqlFlowFunc(spark, "SELECT id, value AS Value FROM src"))
+ }
+
+ assertSchemaIncompatible(ctx.resolveToDataflowGraph())
+ }
+ }
+
// AUTO CDC flows: the inferred schema appends a reserved metadata column to
the data columns.
test("data-only user-specified schema is rejected for an implicit AUTO CDC
flow") {
@@ -144,12 +196,14 @@ class UserSpecifiedSchemaValidationSuite extends
PipelineTest with SharedSparkSe
// Schema includes the appended metadata column, matching the inferred
schema exactly.
autoCdcGraph(
flowName = "target",
- declaredSchema = Some(autoCdcInferredSchema("target"))).validate()
+ declaredSchema = Some(autoCdcInferredSchema("target"))).validate(
+ spark.sessionState.conf.caseSensitiveAnalysis)
}
test("full user-specified schema is accepted for a named AUTO CDC flow") {
autoCdcGraph(
flowName = "auto_cdc_flow",
- declaredSchema = Some(autoCdcInferredSchema("auto_cdc_flow"))).validate()
+ declaredSchema = Some(autoCdcInferredSchema("auto_cdc_flow"))).validate(
+ spark.sessionState.conf.caseSensitiveAnalysis)
}
}
diff --git
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/util/SchemaInferenceUtilsSuite.scala
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/util/SchemaInferenceUtilsSuite.scala
index 41d5bbe14a6b..e6aed97f35dd 100644
---
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/util/SchemaInferenceUtilsSuite.scala
+++
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/util/SchemaInferenceUtilsSuite.scala
@@ -17,7 +17,7 @@
package org.apache.spark.sql.pipelines.util
-import org.apache.spark.SparkFunSuite
+import org.apache.spark.{SparkException, SparkFunSuite}
import org.apache.spark.sql.connector.catalog.TableChange
import org.apache.spark.sql.types._
@@ -270,4 +270,132 @@ class SchemaInferenceUtilsSuite extends SparkFunSuite {
assert(addedColumnNames === Set("full_name", "email"))
assert(deletedColumnNames === Set("first_name", "last_name"))
}
+
+ test("determineColumnChanges - a case-only difference is a drop-then-add,
not a match") {
+ // diffSchemas keys column identity on the EXACT field name, with no case
normalization. So a
+ // target `Value` against a persisted `value` is a distinct column:
`value` is dropped and
+ // `Value` added. This is what makes a case-only rename visible on the
non-merging paths
+ // (materialized views, full refresh), where targetSchema is the declared
schema as-is.
+ val currentSchema = new StructType().add("id", IntegerType).add("value",
StringType)
+ val targetSchema = new StructType().add("id", IntegerType).add("Value",
StringType)
+
+ val changes = SchemaInferenceUtils.diffSchemas(currentSchema, targetSchema)
+
+ val addChanges = changes.collect { case ac: TableChange.AddColumn =>
ac.fieldNames()(0) }
+ val deleteChanges = changes.collect { case dc: TableChange.DeleteColumn =>
dc.fieldNames()(0) }
+ assert(addChanges === Seq("Value"))
+ assert(deleteChanges === Seq("value"))
+ }
+
+ test("determineColumnChanges - two declared columns differing only in case
are both kept") {
+ // A declared schema carrying both `value` and `Value` reaches diffSchemas
verbatim (nothing on
+ // the create path rejects duplicate-cased columns). Exact-name keying
must surface BOTH as
+ // additions; normalizing the lookup key would collapse them and silently
keep an arbitrary one
+ // (whichever came last), losing a column the user declared.
+ val currentSchema = new StructType().add("id", IntegerType)
+ val targetSchema = new StructType()
+ .add("id", IntegerType)
+ .add("value", StringType)
+ .add("Value", IntegerType)
+
+ val changes = SchemaInferenceUtils.diffSchemas(currentSchema, targetSchema)
+
+ val added = changes.collect { case ac: TableChange.AddColumn =>
+ ac.fieldNames()(0) -> ac.dataType()
+ }.toMap
+ assert(added === Map("value" -> StringType, "Value" -> IntegerType))
+ }
+
+ test("mergeSchemas - a nested field differing only in case folds onto the
existing field when " +
+ "case-insensitive") {
+ // The nested analog of the top-level case-only fold. `StructType.merge`
propagates the
+ // case-sensitivity flag into nested struct merges (SPARK-58525), so the
incoming `s.Value` is
+ // matched to the existing `s.value` and the struct keeps a single field
with the persisted
+ // (left) spelling -- rather than growing a second, case-differing nested
field.
+ val currentSchema = new StructType()
+ .add("id", IntegerType)
+ .add("s", new StructType().add("value", StringType))
+ val dataSchema = new StructType()
+ .add("id", IntegerType)
+ .add("s", new StructType().add("Value", StringType))
+
+ val merged =
+ SchemaMergingUtils.mergeSchemas(currentSchema, dataSchema, caseSensitive
= false)
+ assert(merged === currentSchema)
+
+ // Because the merge is a no-op, evolution derives no table changes at
all: in particular the
+ // nested struct is NOT rewritten (which would be an UpdateColumnType on
`s`).
+ assert(
+ SchemaInferenceUtils.diffSchemas(currentSchema, merged).isEmpty)
+ }
+
+ test("mergeSchemas - a nested field differing only in case stays distinct
when case-sensitive") {
+ // The case-sensitive control: `s.value` and `s.Value` are different
fields, so the merged
+ // struct carries both.
+ val currentSchema = new StructType().add("s", new
StructType().add("value", StringType))
+ val dataSchema = new StructType().add("s", new StructType().add("Value",
StringType))
+
+ val merged = SchemaMergingUtils.mergeSchemas(currentSchema, dataSchema,
caseSensitive = true)
+ val expectedStruct = new StructType().add("value",
StringType).add("Value", StringType)
+ assert(merged === new StructType().add("s", expectedStruct))
+
+ // Unlike the case-insensitive test above (where the merge is a no-op and
no changes are
+ // derived), evolution here must rewrite the top-level `s` column.
`diffSchemas` compares nested
+ // types wholesale, so the growth of a nested field surfaces as a single
UpdateColumnType on `s`
+ // carrying the full new struct -- not as an add of `s.Value`.
+ val changes = SchemaInferenceUtils.diffSchemas(currentSchema, merged)
+ assert(changes.length === 1)
+ val typeChange = changes.collect { case tc: TableChange.UpdateColumnType
=> tc }
+ assert(typeChange.length === 1)
+ assert(typeChange.head.fieldNames() === Array("s"))
+ assert(typeChange.head.newDataType() === expectedStruct)
+ }
+
+ test("mergeSchemas - a nested case-only field whose type also changes fails
to merge, and " +
+ "diffSchemas reports it as a type change") {
+ // A nested field that differs only in case AND changes type is rejected
rather than silently
+ // resolved. Note this is a *type* incompatibility, not a case one:
`StructType.merge` never
+ // widens numeric types, so `int` -> `long` fails identically for a
same-cased field and at the
+ // top level. The value of pinning it here is that case-insensitive
matching does not turn an
+ // incompatible type change into a silent merge -- the run still fails
loudly, and the user's
+ // remedy is a full refresh.
+ val currentSchema = new StructType().add("s", new
StructType().add("value", IntegerType))
+ val dataSchema = new StructType().add("s", new StructType().add("Value",
LongType))
+
+ val ex = intercept[SparkException] {
+ SchemaMergingUtils.mergeSchemas(currentSchema, dataSchema, caseSensitive
= false)
+ }
+ assert(ex.getCondition === "CANNOT_MERGE_INCOMPATIBLE_DATA_TYPE")
+
+ // Same-cased and top-level widening fail the same way, confirming the
rejection is about the
+ // type change rather than the case difference.
+ intercept[SparkException] {
+ SchemaMergingUtils.mergeSchemas(
+ currentSchema,
+ new StructType().add("s", new StructType().add("value", LongType)),
+ caseSensitive = false)
+ }
+ intercept[SparkException] {
+ SchemaMergingUtils.mergeSchemas(
+ new StructType().add("v", IntegerType),
+ new StructType().add("v", LongType),
+ caseSensitive = false)
+ }
+
+ // Diffing the two schemas directly (rather than diffing against their
merge, which fails
+ // above) reports a TYPE change on the enclosing `s` column -- not a
field-name mismatch, i.e.
+ // not an add of `s.Value` plus a delete of `s.value`. `diffSchemas` keys
column identity only
+ // at the top level and compares nested types wholesale, so the case
difference inside the
+ // struct never surfaces as an add/delete pair.
+ {
+ val changes = SchemaInferenceUtils.diffSchemas(currentSchema, dataSchema)
+ assert(changes.length === 1, s"changes=$changes")
+ val typeChange = changes.collect { case tc: TableChange.UpdateColumnType
=> tc }
+ assert(typeChange.length === 1, s"changes=$changes")
+ assert(typeChange.head.fieldNames() === Array("s"))
+ assert(typeChange.head.newDataType() === new StructType().add("Value",
LongType))
+ assert(!changes.exists(_.isInstanceOf[TableChange.AddColumn]))
+ assert(!changes.exists(_.isInstanceOf[TableChange.DeleteColumn]))
+ }
+ }
}
diff --git
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/utils/TestGraphRegistrationContext.scala
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/utils/TestGraphRegistrationContext.scala
index 068171a46aa1..9d6067cd51cd 100644
---
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/utils/TestGraphRegistrationContext.scala
+++
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/utils/TestGraphRegistrationContext.scala
@@ -305,7 +305,8 @@ class TestGraphRegistrationContext(
query: FlowFunction,
once: Boolean = false,
catalog: Option[String] = None,
- database: Option[String] = None
+ database: Option[String] = None,
+ sqlConf: Map[String, String] = Map.empty
): Unit = {
val rawFlowIdentifier = GraphIdentifierManager.parseTableIdentifier(name,
spark)
val rawDestinationIdentifier =
@@ -345,7 +346,7 @@ class TestGraphRegistrationContext(
currentCatalog = catalog.orElse(Some(defaultCatalog)),
currentDatabase = database.orElse(Some(defaultDatabase))
),
- sqlConf = Map.empty,
+ sqlConf = sqlConf,
once = once,
origin = QueryOrigin(
objectName = Option(flowIdentifier.unquotedString),
@@ -407,7 +408,8 @@ class TestGraphRegistrationContext(
* Generates a dataflow graph from this pipeline definition and resolves it.
* @return
*/
- def resolveToDataflowGraph(): DataflowGraph = toDataflowGraph.resolve()
+ def resolveToDataflowGraph(): DataflowGraph =
+ toDataflowGraph.resolve(spark.sessionState.conf.caseSensitiveAnalysis)
}
object TestGraphRegistrationContext {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
