diff --git a/common/utils/src/main/resources/error/error-conditions.json b/common/utils/src/main/resources/error/error-conditions.json index 2ade38757be3e..8d63c9d4d983d 100644 --- a/common/utils/src/main/resources/error/error-conditions.json +++ b/common/utils/src/main/resources/error/error-conditions.json @@ -1240,6 +1240,14 @@ ], "sqlState" : "KD009" }, + "CONFLICTING_PIPELINE_FLOW_CASE_SENSITIVITY" : { + "message" : [ + "The flows writing to table do not agree on '': .", + "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 '' to the same value for every flow writing to this table." + ], + "sqlState" : "42KD9" + }, "CONNECT" : { "message" : [ "Generic Spark Connect error." 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 a8c3485da31ce..fe28decab357c 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 9251501887a3a..da5b7eb8ad125 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 66f2995ee02d9..f995efbc422a2 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]() @@ -110,7 +110,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 +203,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 c5210976d3f98..512e3634b2901 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. If two flows differ only in column casing, the + * first flow's spelling wins because flows are merged in graph encounter 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 cf5f811d3461e..61d188f302e13 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 856fd6a3923c0..a8be0a4c2ec33 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 c36d824bdf4a6..0663ae385005b 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 @@ -232,7 +232,9 @@ class StreamingTableWrite( override def getOrigin: QueryOrigin = flow.origin def startStream(): StreamingQuery = { - val data = graph.reanalyzeFlow(flow).df + val data = graph.reanalyzeFlow( + flow, + updateContext.spark.sessionState.conf.caseSensitiveAnalysis).df val dataStreamWriter = data .writeStream .queryName(displayName) @@ -260,7 +262,9 @@ 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, + updateContext.spark.sessionState.conf.caseSensitiveAnalysis).df Future { val dataFrameWriter = data.write destination.format.foreach(dataFrameWriter.format) @@ -298,7 +302,9 @@ class SinkWrite( override def getOrigin: QueryOrigin = flow.origin def startStream(): StreamingQuery = { - val data = graph.reanalyzeFlow(flow).df + val data = graph.reanalyzeFlow( + flow, + updateContext.spark.sessionState.conf.caseSensitiveAnalysis).df data.writeStream .queryName(displayName) .option("checkpointLocation", checkpointPath) @@ -328,7 +334,9 @@ class Scd1MergeStreamingWrite( override def getOrigin: QueryOrigin = flow.origin override def startStream(): StreamingQuery = { - val sourceChangeDataFeed = graph.reanalyzeFlow(flow).df + val sourceChangeDataFeed = graph.reanalyzeFlow( + flow, + updateContext.spark.sessionState.conf.caseSensitiveAnalysis).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 +381,9 @@ class Scd2MergeStreamingWrite( override def getOrigin: QueryOrigin = flow.origin override def startStream(): StreamingQuery = { - val sourceChangeDataFeed = graph.reanalyzeFlow(flow).df + val sourceChangeDataFeed = graph.reanalyzeFlow( + flow, + updateContext.spark.sessionState.conf.caseSensitiveAnalysis).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 c835665a0f380..7bc26e6288c14 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 ebe90b677aa21..5b4eaba4e7095 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 AS AUTO CDC INTO `) 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 d35d701d44e57..724ccc56e35c7 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 885755fd78ece..cfaf10a2e657d 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 @@ -186,7 +186,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 = spark.sessionState.conf.caseSensitiveAnalysis) } // 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 4777772342d7d..d0f1d0e00af3e 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,26 +114,41 @@ 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. If two flows differ only in column + * casing, the first flow's spelling wins because flows are merged in the order provided. */ 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 caseSensitive = effectiveCaseSensitivity( + tableIdentifier = tableIdentifier, + flows = flows, + sessionCaseSensitive = sessionCaseSensitive + ) + val inferredSchema = flows.map(_.schema).fold(new StructType()) { (schemaSoFar, schema) => try { - SchemaMergingUtils.mergeSchemas(schemaSoFar, schema) + SchemaMergingUtils.mergeSchemas(schemaSoFar, schema, caseSensitive) } catch { case NonFatal(e) => throw GraphErrors.unableToInferSchemaError( - flows.head.destinationIdentifier, + tableIdentifier, schemaSoFar, schema, cause = Option(e) @@ -66,15 +156,15 @@ object SchemaInferenceUtils { } } - 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 +172,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 +201,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 d15e7ac6425cc..aeab5e623431f 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 a3d26e6f47e55..4596e31d778a5 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 dda1e5a7455a5..c26a5248488b9 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 226480ea919a8..6de8cf52f7b4b 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 1d27b7d69dab2..e180056761f20 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 635e3d93de56d..1a626a6ddcdf9 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 a56fed3e58057..d01ec0f54736c 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 62680ebd3835f..172d8354932b8 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 ea6d3202ba868..b90ccbbde7ce5 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 46e2d6d9ae631..351848bda5a84 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,344 @@ 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 firstFlowValueField = + graph.resolvedFlowsTo(targetIdentifier).head.schema.fieldNames(1) + // The two spellings must fold into a single column, and the first flow in the resolved + // graph's encounter order wins. + assert(inferred.fieldNames.toSeq === Seq("id", firstFlowValueField)) + } + } + } + + 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 merge 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 cec8db6ec5288..8ce7971b405fa 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 db8c368ca89df..313f86753222c 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 d216539c93bc3..fee3493d01b97 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 41d5bbe14a6b1..e6aed97f35ddd 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 068171a46aa16..9d6067cd51cd3 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 {