jose-torres commented on code in PR #58914:
URL: https://github.com/apache/spark/pull/58914#discussion_r4085487856
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2VersionMap.scala:
##########
@@ -43,49 +48,234 @@ import org.apache.spark.sql.types.{BooleanType, MapType,
StringType}
* unauthored?
*
* Concretely, the contract of the version map is as follows.
- * 1. Every column that received a null in the event but is considered
authored (i.e. not part
- * of ignore-null selection at ingestion), receives an entry of (column
name, true) in the
- * version map.
- * 2. Every column that received a null in the event but is considered
unauthored (i.e. part
- * of the ignore-null selection at ingestion), receives an entry of (column
name, false) in
- * the version map.
- * 3. Every column that was not present in the event, but schema evolved in
later with a null
- * value, will be treated as unauthored BUT does not yet have any entry in
the version map.
- * An entry will be added as per (2).
+ * 1. Every null considered authored receives an entry of (column name, true).
+ * 2. Every null considered unauthored receives an entry of (column name,
false).
+ * 3. Once a non-null map is established, every column added by later schema
evolution has an
+ * unauthored null but initially has no map entry. An entry will be added
as per (2).
*
- * In a single sentence: if a null column in the SCD2 row is either absent
from the version
- * map or has a false value in the version map, the null is considered
unauthored by the
- * upsert event that spawned this row. Otherwise the null value was explicitly
authored by
- * the row.
- *
- * As mentioned above, authorship is dependent on the configured ignore-null
selection, which
- * is free to change between pipeline runs for the same AutoCDC flow. As such,
we choose that
- * the version map strictly reflects authorship as of the ignore-null
selection that was active
- * when the upsert event that produced this row was ingested. This means the
authorship
- * information the version map encoded at creation time is invariant/frozen --
even if the
- * ignore-null selection changes on a future run, the version map is not
rewritten (unless the
- * table is full refreshed).
- *
- * It's worth noting that while contract case (3) materializes new entries in
the version map
- * after creation, it does not change the set of columns whose null values are
considered
- * authored/unauthored. Therefore authorship information encoded by the
mutated version map is
- * still invariant, and independent of a changing ignore-null configuration.
New rows materialized
- * in the map are still compliant with whatever the ignore-null selection was
at ingestion time.
+ * In a single sentence: if a null column in an SCD2 row with a non-null
version map is either
+ * absent from the map or has a false value, the null is considered unauthored
by the upsert event
+ * that spawned this row. Otherwise the row explicitly authored the null.
*/
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 field paths rendered as fully quoted multipart identifiers using
+ * [[QuotingUtils.quoteNameParts]]. Quoting each name part distinguishes a
nested path from a
+ * column whose name contains dots. 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
+ * Lack of entry in a non-null 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)
+
+ /**
+ * Builds a version map from a row's values and an ignore-null selection.
For each null leaf,
+ * the map records whether the row authored that null or left the leaf
unauthored under the
+ * selection. Used both when ingesting a new upsert and when lazily
establishing a map for an
+ * existing row that has none.
+ *
+ * @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 ignoreNullLeafPaths =
+ resolveIgnoreNullLeafPaths(schema, ignoreNullSelection, resolver).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 upsert
+ // event authored the null, false if the event left the leaf unauthored.
+ val candidateEntries =
AutoCdcSchemaUtils.flattenStructFieldPaths(schema).map { path =>
+ val isIgnoreNullLeaf = ignoreNullLeafPaths.contains(path)
+ val leafIsNull = F.col(QuotingUtils.quoteNameParts(path)).isNull
+
+ // If the leaf is not null, this candidate entry will simply resolve to
null and will not be
+ // added to the version map during construction below.
+ F.when(
+ leafIsNull,
+ buildVersionMapEntry(path, authored = !isIgnoreNullLeaf))
+ }
+
+ if (candidateEntries.isEmpty) {
+ F.map().cast(mapType)
+ } else {
+ val nonNullEntries = F.filter(F.array(candidateEntries: _*), (e: Column)
=> e.isNotNull)
+ F.map_from_entries(nonNullEntries)
+ }
+ }
+
+ /**
+ * Resolves an ignore-null selection against the top-level fields of
`schema`, then flattens
+ * selected structs into leaf paths. Arrays and maps remain opaque leaves.
+ *
+ * @param schema Schema against which to resolve and canonicalize the
selection.
+ * @param ignoreNullSelection Top-level columns selected for ignore-null
handling.
+ * @param resolver Case-sensitivity resolver for column name matching.
+ * @return Raw, unquoted leaf-name parts in schema order, with spelling
taken exactly from
+ * `schema`.
+ */
+ private[autocdc] def resolveIgnoreNullLeafPaths(
+ schema: StructType,
+ ignoreNullSelection: ColumnSelection,
+ resolver: Resolver): Seq[Seq[String]] = {
+ val ignoreNullColumns = ColumnSelection.applyToSchema(
+ schemaName = "ignoreNullSelection",
+ schema = schema,
+ columnSelection = Some(ignoreNullSelection),
+ resolver = resolver
+ )
+ AutoCdcSchemaUtils.flattenStructFieldPaths(ignoreNullColumns)
+ }
+
+ /**
+ * Returns the authorship value for `columnPath`, or null if the map or
entry is absent.
+ *
+ * @param versionMap Version map containing keys serialized with
+ * [[QuotingUtils.quoteNameParts]].
+ * @param columnPath Raw, unquoted leaf-name parts. Their spelling must
exactly match the path
Review Comment:
The right invariant here is not "must exactly match" but rather "must have
been normalized to the canonical format". We don't want to let edge cases grow
where normalization doesn't go through a single path.
--
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]
