File tree Expand file tree Collapse file tree
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc Expand file tree Collapse file tree Original file line number Diff line number Diff line change @@ -99,22 +99,16 @@ case class Scd1BatchProcessor(
9999 }
100100
101101 private def validateCdcMetadataColumnNotPresent (microbatchDf : DataFrame ): Unit = {
102- val ignoreColumnNameCase =
103- ! microbatchDf.sparkSession.sessionState.conf.caseSensitiveAnalysis
102+ val sqlConf = microbatchDf.sparkSession.sessionState.conf
103+ val resolver = sqlConf.resolver
104104
105105 microbatchDf.schema.fieldNames
106- .find { fieldName =>
107- if (ignoreColumnNameCase) {
108- fieldName.equalsIgnoreCase(Scd1BatchProcessor .cdcMetadataColName)
109- } else {
110- fieldName.equals(Scd1BatchProcessor .cdcMetadataColName)
111- }
112- }
106+ .find(resolver(_, Scd1BatchProcessor .cdcMetadataColName))
113107 .foreach { conflictingColumnName =>
114108 throw new AnalysisException (
115109 errorClass = " AUTOCDC_RESERVED_COLUMN_NAME_CONFLICT" ,
116110 messageParameters = Map (
117- " caseSensitivity" -> CaseSensitivityLabels .of(ignoreColumnNameCase ),
111+ " caseSensitivity" -> CaseSensitivityLabels .of(! sqlConf.caseSensitiveAnalysis ),
118112 " columnName" -> conflictingColumnName,
119113 " schemaName" -> " microbatch" ,
120114 " reservedColumnName" -> Scd1BatchProcessor .cdcMetadataColName
You can’t perform that action at this time.
0 commit comments