jose-torres commented on code in PR #58645:
URL: https://github.com/apache/spark/pull/58645#discussion_r4010358010
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala:
##########
@@ -1445,23 +1508,37 @@ object Scd2BatchProcessor {
private[pipelines] def computeTrackedHistoryColumns(
schema: StructType,
changeArgs: ChangeArgs,
- resolver: Resolver): Seq[String] = {
- val keyColNames = changeArgs.keys.map(_.name)
-
- val eligibleSchema = StructType(schema.fields.filterNot { field =>
- reservedFrameworkColNames.exists(resolver(_, field.name)) ||
- keyColNames.exists(resolver(_, field.name))
- })
-
+ resolver: Resolver): Seq[String] =
ColumnSelection
.applyToSchema(
schemaName = "trackHistorySelection",
- schema = eligibleSchema,
+ schema = computeUserDataSchema(schema, changeArgs, resolver),
columnSelection = changeArgs.trackHistorySelection,
resolver = resolver
)
.fieldNames
.toImmutableArraySeq
+
+ /**
+ * The subset of `schema` that is user data a column selection may act on:
every field that is
+ * neither a framework reserved column nor one of [[ChangeArgs.keys]]. Field
order is preserved.
+ *
+ * Both [[ChangeArgs.trackHistorySelection]] and
[[ChangeArgs.ignoreNullSelection]] resolve
+ * against this, so an exclude-list in either cannot pick up a key or a
framework column, and
+ * an include-list naming one fails as not found.
+ *
+ * `schema` is expected to have already been narrowed by
[[ChangeArgs.columnSelection]] and then
Review Comment:
Always a bit worried about these kind of action at a distance assumptions.
Should there be some abstraction that computes all the various schema subsets
we'll need at different points in execution up front?
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2VersionMap.scala:
##########
@@ -77,15 +83,69 @@ private[pipelines] object Scd2VersionMap {
/**
* Schema of the version map: `Map(String, Boolean)`.
*
- * Keys are dot-delimited paths to *leaf* columns that received a null value
in their
- * corresponding upsert event (e.g. `"address.city"`, `` "`has space`.city"
``). Paths
- * must be formatted by
[[org.apache.spark.sql.catalyst.util.QuotingUtils.quoted]] to
- * ensure segments that need quoting are back-tick escaped.
+ * Keys are compact JSON arrays of the name parts of *leaf* columns that
received a null
+ * value in their upsert event (e.g. `["address","city"]`). Keeping
+ * name parts separate distinguishes a nested path from a column whose name
contains dots
+ * and keeps persisted keys independent of SQL identifier quoting rules.
Name parts use the
+ * persisted target schema's canonical spelling.
*
- * Values indicate authorship. I.e, `true` => authored-null, `false` =>
unauthored-null.
+ * Values indicate authorship: `true` means authored-null, `false` means
unauthored-null.
+ * Null values never appear in the map.
*
* Lack of entry in the map for a null-valued leaf column implies the column
was
* schema-evolved with an unauthored-null.
*/
def mapType: MapType = MapType(StringType, BooleanType, valueContainsNull =
false)
+
+ /** Encodes a leaf path as the compact JSON string persisted as its version
map key. */
+ private[autocdc] def encodePath(path: Seq[String]): String =
+ compact(JArray(path.map(JString(_)).toList))
+
+ /**
+ * Builds the ingest-time version map column for a microbatch. Each row's
map records which
+ * null leaves are authored vs declined, based on the active ignore-null
selection.
+ *
+ * @param schema The schema whose leaves the version map covers.
Null-authorship is tracked
+ * for every leaf column in this schema, as per the version map contract.
+ * @param ignoreNullSelection The ignore-null column selection this schema
is being ingested
+ * under.
+ * @param resolver Case-sensitivity resolver for column name matching.
+ * @return A [[Column]] of [[mapType]] schema.
+ */
+ def buildVersionMap(
+ schema: StructType,
+ ignoreNullSelection: ColumnSelection,
+ resolver: Resolver): Column = {
+ val ignoreNullColumns = ColumnSelection.applyToSchema(
+ schemaName = "ignoreNullSelection",
+ schema = schema,
+ columnSelection = Some(ignoreNullSelection),
+ resolver = resolver
+ )
+ val ignoreNullLeafPaths =
AutoCdcSchemaUtils.extractLeafPaths(ignoreNullColumns).toSet
+
+ // For each leaf, build a nullable struct (key, value). The struct is
non-null only when
+ // the leaf column's runtime value is null (meaning the leaf needs a
version map entry).
+ // The value is a non-nullable BooleanType literal indicating authorship:
true if the null
+ // is authored, false if declined.
Review Comment:
This is described as "declined" a few times, and I was withholding judgment
until I got the implementation part, but even now that I have full context I
don't think that term really makes sense.
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2VersionMap.scala:
##########
@@ -77,15 +83,69 @@ private[pipelines] object Scd2VersionMap {
/**
* Schema of the version map: `Map(String, Boolean)`.
*
- * Keys are dot-delimited paths to *leaf* columns that received a null value
in their
- * corresponding upsert event (e.g. `"address.city"`, `` "`has space`.city"
``). Paths
- * must be formatted by
[[org.apache.spark.sql.catalyst.util.QuotingUtils.quoted]] to
- * ensure segments that need quoting are back-tick escaped.
+ * Keys are compact JSON arrays of the name parts of *leaf* columns that
received a null
+ * value in their upsert event (e.g. `["address","city"]`). Keeping
+ * name parts separate distinguishes a nested path from a column whose name
contains dots
+ * and keeps persisted keys independent of SQL identifier quoting rules.
Name parts use the
+ * persisted target schema's canonical spelling.
*
- * Values indicate authorship. I.e, `true` => authored-null, `false` =>
unauthored-null.
+ * Values indicate authorship: `true` means authored-null, `false` means
unauthored-null.
+ * Null values never appear in the map.
*
* Lack of entry in the map for a null-valued leaf column implies the column
was
* schema-evolved with an unauthored-null.
*/
def mapType: MapType = MapType(StringType, BooleanType, valueContainsNull =
false)
+
+ /** Encodes a leaf path as the compact JSON string persisted as its version
map key. */
Review Comment:
Isn't this a column name? I don't think we should write a custom serializer
for column names they already have canonical forms.
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/AutoCdcSchemaUtils.scala:
##########
@@ -0,0 +1,36 @@
+/*
+ * 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.types.StructType
+
+private[autocdc] object AutoCdcSchemaUtils {
+
+ /**
+ * Enumerates every leaf path in `schema`, in schema order, as its sequence
of name parts.
+ * Structs unfold recursively; every other type (including arrays and maps)
is an opaque leaf.
Review Comment:
In general this transformation is not correct, arrays and maps are also
complex types. I see why we want to exclude them here but the method should be
more explicitly scoped to the context of value history tracking.
--
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]
