szehon-ho commented on code in PR #55836: URL: https://github.com/apache/spark/pull/55836#discussion_r3248831800
########## sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/ChangeArgs.scala: ########## @@ -0,0 +1,179 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.spark.sql.pipelines.autocdc + +import java.util.Locale + +import org.apache.spark.sql.{AnalysisException, Column} +import org.apache.spark.sql.catalyst.parser.CatalystSqlParser +import org.apache.spark.sql.catalyst.util.QuotingUtils +import org.apache.spark.sql.types.StructType + +/** + * A single, unqualified column identifier (no nested path or table/alias qualifier). Backticks + * are consumed: "`a.b`" is stored as "a.b" in [[name]]. Use [[name]] for direct schema-fieldName + * comparison and [[quoted]] for APIs that re-parse identifier strings. + * + * Declared `final class` so the smart constructor is the only path to construction (no synthesized + * `copy` can bypass it). + */ +final class UnqualifiedColumnName private (val name: String) extends Serializable { + + def quoted: String = QuotingUtils.quoteIdentifier(name) Review Comment: is this planned to be used in test only, or actually used? ########## sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/ChangeArgs.scala: ########## @@ -0,0 +1,179 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.spark.sql.pipelines.autocdc + +import java.util.Locale + +import org.apache.spark.sql.{AnalysisException, Column} +import org.apache.spark.sql.catalyst.parser.CatalystSqlParser +import org.apache.spark.sql.catalyst.util.QuotingUtils +import org.apache.spark.sql.types.StructType + +/** + * A single, unqualified column identifier (no nested path or table/alias qualifier). Backticks + * are consumed: "`a.b`" is stored as "a.b" in [[name]]. Use [[name]] for direct schema-fieldName + * comparison and [[quoted]] for APIs that re-parse identifier strings. + * + * Declared `final class` so the smart constructor is the only path to construction (no synthesized Review Comment: just curious, is it that important? there's a lot of case class in spark that's internal. It makes it a bit harder. up to you and @gengliangwang though the comment is also a bit too specific how user may hack around it, maybe just 'this class may not be constructed' , or suggest just remove it ########## sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/ChangeArgs.scala: ########## @@ -0,0 +1,90 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.spark.sql.pipelines.autocdc + +import org.apache.spark.sql.{AnalysisException, Column} +import org.apache.spark.sql.types.StructType + +sealed trait ColumnSelection +object ColumnSelection { + type ColumnList = Seq[String] + case class IncludeColumns(columns: ColumnList) extends ColumnSelection + case class ExcludeColumns(columns: ColumnList) extends ColumnSelection + + /** + * Applies [[ColumnSelection]] to a [[StructType]] and returns the filtered schema. + * Field names are matched exactly. Field order follows the original schema (filtered in place). + */ + def applyToSchema(schema: StructType, columnSelection: Option[ColumnSelection]): StructType = + columnSelection match { + case None => + // A none column selection is interpreted as a no-op. + schema + case Some(IncludeColumns(includeColumns)) => + validateColumnsExistInSchema(columns = includeColumns, schema = schema) Review Comment: yea i meant more , it's already de-duplicated before passing in (in case user passes many duplicates). But i think the code changed since this comment, there's no more includeColumnSet, so its fine ########## common/utils/src/main/resources/error/error-conditions.json: ########## @@ -191,6 +191,18 @@ ], "sqlState" : "0A000" }, + "AUTOCDC_COLUMNS_NOT_FOUND_IN_SCHEMA" : { + "message" : [ + "The following columns are not present in the schema: <missingColumns>. Available columns: <availableColumns>. Matching was <matching>." Review Comment: matching is actually 'case-sensitive' ########## sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/ChangeArgs.scala: ########## @@ -0,0 +1,124 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.spark.sql.pipelines.autocdc + +import org.apache.spark.sql.{AnalysisException, Column} +import org.apache.spark.sql.catalyst.parser.CatalystSqlParser +import org.apache.spark.sql.types.StructType + +/** + * A column reference that must be a single, unqualified identifier (no nested field path and + * no table/alias qualifier). The constructor parses [[name]] with the Spark SQL parser and + * throws an [[AnalysisException]] if it does not resolve to exactly one name part. + */ +case class UnqualifiedColumnName(name: String) { Review Comment: just wondering, can we add a test with UnqualifiedColumnName("`Name`"), or just assert against it somehow -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
