HyukjinKwon commented on code in PR #59311:
URL: https://github.com/apache/spark/pull/59311#discussion_r4237343641
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd1BatchProcessor.scala:
##########
@@ -226,29 +227,45 @@ object Scd1BatchProcessor {
/**
* Schema of the CDC metadata struct column for SCD1.
+ *
+ * The CDC metadata of every microbatch, target, and auxiliary-table row
represents either a
+ * delete or an upsert, never both: exactly one of its delete and upsert
sequences is non-null,
+ * and only an upsert may carry a version map.
*/
private[pipelines] def cdcMetadataColSchema(sequencingType: DataType):
StructType =
StructType(
Seq(
// The sequencing of the event if it represents a delete, null
otherwise.
StructField(cdcDeleteSequenceFieldName, sequencingType, nullable =
true),
// The sequencing of the event if it represents an upsert, null
otherwise.
- StructField(cdcUpsertSequenceFieldName, sequencingType, nullable =
true)
+ StructField(cdcUpsertSequenceFieldName, sequencingType, nullable =
true),
+ // On target rows, a null version map indicates row-level
reconciliation; a non-null map
+ // indicates per-leaf reconciliation. Delete rows, including
auxiliary-table tombstones,
+ // always use a null map: a delete's sequence applies to every leaf,
so a map would only
+ // repeat it.
+ StructField(
Review Comment:
**Non-blocking (P2):** Adding `__VERSION_MAP` here changes the SCD1
`__spark_autocdc_metadata` struct that `AutoCdcMergeFlow.schema` publishes as
the flow's output schema. The programming guide (Auto CDC Considerations,
"Declaring a schema explicitly") tells users that a separately declared target,
via `create_streaming_table(schema=...)` or a standalone `CREATE STREAMING
TABLE`, must include the reserved metadata column and match the flow output
exactly. `GraphValidations.validateUserSpecifiedSchemas` enforces that with an
exact comparison. So an existing pipeline that declares
`__spark_autocdc_metadata STRUCT<deleteSequence: BIGINT, upsertSequence:
BIGINT> NOT NULL` will fail graph validation after upgrade, even though the new
field is always null in this PR. No test covers a declared Auto CDC target
schema, so CI won't catch this.
Could you add a Declarative Pipelines entry to
`docs/sql-migration-guide.md`, next to the existing ones, that names the new
field and its type (`MAP<STRING, <sequencing type>>`) and tells users with a
declared target schema to add it? The declared-schema bullet in the programming
guide should be updated to match. Relaxing the exact match for the reserved
column (SPARK-58118) would also solve this, but that is a larger, separate
change.
**Verification:**
- **Inspection:** The migration entry states the exact added field name and
type (value type equals the flow's sequencing type) and the declared-schema
action, consistent with Scd1BatchProcessor.cdcMetadataColSchema.
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd1VersionMap.scala:
##########
@@ -0,0 +1,103 @@
+/*
+ * 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.{functions => F, Column}
+import org.apache.spark.sql.catalyst.analysis.Resolver
+import org.apache.spark.sql.catalyst.util.QuotingUtils
+import org.apache.spark.sql.types.{DataType, MapType, StringType, StructType}
+
+/**
+ * Per-leaf authorship sequences for SCD1 reconciliation.
+ *
+ * An SCD1 target row can combine values authored by different CDC events. The
version map records
+ * which event currently authors each non-key user-data leaf.
+ *
+ * Concretely, the contract of the version map is as follows.
+ * 1. Every user-data leaf present in the row when the map is written receives
an entry.
+ * 2. A non-null entry is the sequencing clock of the event that authored the
leaf's current stored
+ * value.
+ * 3. A null entry means the leaf has no authored value.
+ *
+ * A user-data leaf is a non-framework field obtained by recursively expanding
structs.
Review Comment:
**Nit (P3):** The overview says the map records the author of "each non-key
user-data leaf". Item 1 says every user-data leaf receives an entry, and this
definition doesn't exclude key columns; elsewhere in the module, keys and
framework columns are distinct. `buildVersionMap` writes an entry for every
leaf of whatever schema it's given, so this Scaladoc is the only spec of which
leaves are keyed, and a future writer and reader could disagree about persisted
maps. I'd suggest defining a user-data leaf as a non-key, non-framework field,
so the overview, item 1, the `mapType` doc, and `@param schema` all describe
the same set.
**Verification:**
- **Inspection:** Every Scd1VersionMap statement about which leaves receive
entries uses one definition that excludes keys, matching the overview's
'non-key user-data leaf'.
##########
sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd1VersionMapSuite.scala:
##########
@@ -0,0 +1,278 @@
+/*
+ * 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.{functions => F, QueryTest, Row}
+import org.apache.spark.sql.catalyst.analysis.{caseInsensitiveResolution,
caseSensitiveResolution}
+import org.apache.spark.sql.catalyst.parser.CatalystSqlParser
+import org.apache.spark.sql.classic.DataFrame
+import org.apache.spark.sql.test.SharedSparkSession
+import org.apache.spark.sql.types._
+
+class Scd1VersionMapSuite extends QueryTest with SharedSparkSession {
+
+ private def encodedPath(parts: String*): String =
Scd1VersionMap.serializeKey(parts)
+
+ private def dataFrameOf(schema: StructType)(rows: Row*): DataFrame =
+ spark.createDataFrame(spark.sparkContext.parallelize(rows), schema)
+
+ test("mapType uses the supplied sequencing type and permits null values") {
+ val sequencingType = new StructType()
+ .add("timestamp", LongType)
+ .add("tieBreaker", IntegerType)
+
+ assert(Scd1VersionMap.mapType(sequencingType) ==
+ MapType(StringType, sequencingType, valueContainsNull = true))
+ }
+
+ test("buildVersionMap records every leaf with the expected sequencing
clock") {
+ val userDataSchema = new StructType()
+ .add("selectedNull", StringType)
+ .add("selectedValue", StringType)
+ .add("unselectedNull", StringType)
+ .add("unselectedValue", StringType)
+ val inputSchema = StructType(userDataSchema.fields :+ StructField("seq",
IntegerType))
+ val input = dataFrameOf(inputSchema)(
+ Row(null, "selected", null, "unselected", 10)
+ )
+ val ignoreNullSelection = ColumnSelection.IncludeColumns(Seq(
+ UnqualifiedColumnName("selectedNull"),
+ UnqualifiedColumnName("selectedValue")
+ ))
+
+ val result = input.select(
+ Scd1VersionMap.buildVersionMap(
+ schema = userDataSchema,
+ ignoreNullSelection = ignoreNullSelection,
+ upsertSequence = F.col("seq"),
+ sequencingType = IntegerType,
+ resolver = caseSensitiveResolution
+ ).as("versionMap")
+ )
+
+ assert(result.schema("versionMap").dataType ==
Scd1VersionMap.mapType(IntegerType))
+ checkAnswer(result, Row(Map[String, Any](
+ encodedPath("selectedNull") -> null,
+ encodedPath("selectedValue") -> 10,
+ encodedPath("unselectedNull") -> 10,
+ encodedPath("unselectedValue") -> 10
+ )))
+ }
+
+ test("buildVersionMap supports exclude-list ignore-null selections") {
+ val userDataSchema = new StructType()
+ .add("ignoredNull", StringType)
+ .add("authoredNull", StringType)
+ val inputSchema = StructType(userDataSchema.fields :+ StructField("seq",
LongType))
+ val input = dataFrameOf(inputSchema)(Row(null, null, 10L))
+ val ignoreNullSelection = ColumnSelection.ExcludeColumns(
+ Seq(UnqualifiedColumnName("authoredNull"))
+ )
+
+ val result = input.select(
+ Scd1VersionMap.buildVersionMap(
+ schema = userDataSchema,
+ ignoreNullSelection = ignoreNullSelection,
+ upsertSequence = F.col("seq"),
+ sequencingType = LongType,
+ resolver = caseSensitiveResolution
+ ).as("versionMap")
+ )
+
+ checkAnswer(result, Row(Map[String, Any](
+ encodedPath("ignoredNull") -> null,
+ encodedPath("authoredNull") -> 10L
+ )))
+ }
+
+ test("buildVersionMap recursively expands structs and treats other complex
types as leaves") {
+ val userDataSchema = new StructType()
+ .add("profile", new StructType()
+ .add("city", StringType)
+ .add("contact", new StructType().add("zip", StringType)))
+ .add("profile.city", StringType)
+ .add("tags", ArrayType(StringType))
+ .add("properties", MapType(StringType, StringType))
+ val inputSchema = StructType(userDataSchema.fields :+ StructField("seq",
LongType))
+ val input = dataFrameOf(inputSchema)(
+ Row(Row(null, Row(null)), null, null, null, 10L)
+ )
+ val ignoreNullSelection = ColumnSelection.IncludeColumns(Seq(
+ UnqualifiedColumnName("profile"),
+ UnqualifiedColumnName("tags")
+ ))
+
+ val result = input.select(
+ Scd1VersionMap.buildVersionMap(
+ schema = userDataSchema,
+ ignoreNullSelection = ignoreNullSelection,
+ upsertSequence = F.col("seq"),
+ sequencingType = LongType,
+ resolver = caseSensitiveResolution
+ ).as("versionMap")
+ )
+
+ checkAnswer(result, Row(Map[String, Any](
+ encodedPath("profile", "city") -> null,
+ encodedPath("profile", "contact", "zip") -> null,
+ encodedPath("profile.city") -> 10L,
+ encodedPath("tags") -> null,
+ encodedPath("properties") -> 10L
+ )))
+ }
+
+ test("buildVersionMap respects the supplied resolver for ignore-null
selection") {
+ val userDataSchema = new StructType().add("Value", StringType)
+ val inputSchema = StructType(userDataSchema.fields :+ StructField("seq",
LongType))
+ val input = dataFrameOf(inputSchema)(Row(null, 10L))
+ val ignoreNullSelection = ColumnSelection.IncludeColumns(
+ Seq(UnqualifiedColumnName("value"))
+ )
+
+ val result = input.select(
+ Scd1VersionMap.buildVersionMap(
+ schema = userDataSchema,
+ ignoreNullSelection = ignoreNullSelection,
+ upsertSequence = F.col("seq"),
+ sequencingType = LongType,
+ resolver = caseInsensitiveResolution
+ ).as("versionMap")
+ )
+
+ checkAnswer(result, Row(Map[String, Any](encodedPath("Value") -> null)))
+ }
+
+ test("buildVersionMap produces a typed empty map for an empty schema") {
+ val input = dataFrameOf(new StructType().add("seq", LongType))(Row(10L))
+
+ val result = input.select(
+ Scd1VersionMap.buildVersionMap(
+ schema = new StructType(),
+ ignoreNullSelection = ColumnSelection.ExcludeColumns(Seq.empty),
+ upsertSequence = F.col("seq"),
+ sequencingType = LongType,
+ resolver = caseSensitiveResolution
+ ).as("versionMap")
+ )
+
+ assert(result.schema("versionMap").dataType ==
Scd1VersionMap.mapType(LongType))
+ checkAnswer(result, Row(Map.empty[String, Long]))
+ }
+
+ // Version-map keys are persisted in target tables, so their format must not
change.
+ test("version map keys have a stable persisted serialization") {
+ val leafPathsAndExpectedKeys = Seq(
+ Seq("simple") -> "`simple`",
+ Seq("a.b") -> "`a.b`",
+ Seq("a", "b") -> "`a`.`b`",
+ Seq("wrapper", "leaf") -> "`wrapper`.`leaf`",
+ Seq("wrapper", "a.b") -> "`wrapper`.`a.b`",
+ Seq("wrapper", "has space") -> "`wrapper`.`has space`",
+ Seq("wrapper", "col-one") -> "`wrapper`.`col-one`",
+ Seq("wrapper", "back`tick") -> "`wrapper`.`back``tick`",
+ Seq("wrapper", "`already quoted`") -> "`wrapper`.```already quoted```",
+ Seq("wrapper", "single'quote") -> "`wrapper`.`single'quote`",
+ Seq("wrapper", "double\"quote") -> "`wrapper`.`double\"quote`",
+ Seq("wrapper", "back\\slash") -> "`wrapper`.`back\\slash`",
+ Seq("wrapper", "") -> "`wrapper`.``",
+ Seq("wrapper", "null" + 0.toChar + "byte") -> ("`wrapper`.`null" +
0.toChar + "byte`")
+ )
+ val wrapperLeafNames = leafPathsAndExpectedKeys.collect {
+ case (Seq("wrapper", name), _) => name
+ }
+ val userDataSchema = new StructType()
+ .add("simple", StringType)
+ .add("a.b", StringType)
+ .add("a", new StructType().add("b", StringType))
+ .add("wrapper", StructType(wrapperLeafNames.map(StructField(_,
StringType))))
+ val inputSchema = StructType(userDataSchema.fields :+ StructField("seq",
LongType))
+ val input = dataFrameOf(inputSchema)(
+ Row(null, null, Row(null),
Row.fromSeq(Seq.fill[Any](wrapperLeafNames.size)(null)), 10L))
Review Comment:
**Nit (P3):** Both top-level `a.b` and nested `a`.`b` are selected here, but
every value is null, so if the lookup swapped them the map would be identical.
The comment below says this test resolves each leaf through its key. The
struct-expansion test leaves top-level `profile.city` unselected, and the
round-trip test has no confusable pair, so nothing in the suite can detect a
lookup that reads the wrong confusable column. A case that selects both `a.b`
and `a`.`b` with different nullness, asserting a null entry for one and the
upsert sequence for the other, would give the Scaladoc's quoting claim a real
failure signal.
**Verification:**
- **Regression:** With top-level `a.b` null and nested `a`.`b` non-null (and
the reverse), both selected for ignore-null, the built map has a null entry for
the null leaf and the upsert sequence for the non-null leaf.
--
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]
