This is an automated email from the ASF dual-hosted git repository.
jose-torres pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 33cb9b4b45bc [SPARK-58342][SDP] Add unit tests for the SCD1 AutoCDC
auxiliary table spec
33cb9b4b45bc is described below
commit 33cb9b4b45bc4ab6e51669a2429784fbb07d83c4
Author: Andreas Neumann <[email protected]>
AuthorDate: Fri Jul 24 21:03:02 2026 -0700
[SPARK-58342][SDP] Add unit tests for the SCD1 AutoCDC auxiliary table spec
### What changes were proposed in this pull request?
`AutoCdcAuxiliaryTable.buildScd1AuxiliaryTableSpecFor` had no focused unit
test; its behavior was only exercised indirectly via higher-level pipeline
suites. [SPARK-58320](https://issues.apache.org/jira/browse/SPARK-58320) added
`AutoCdcScd2AuxiliaryTableSpecSuite` for the SCD2 spec builder; this adds the
parallel SCD1 coverage.
`AutoCdcScd1AuxiliaryTableSpecSuite` resolves a real SCD1 AutoCDC graph and
asserts the derived auxiliary-table spec:
- the schema is exactly the key columns plus the CDC metadata column --
no user data columns and no SCD2 framework columns, since the SCD1 auxiliary
table stores per-key tombstones rather than full rows;
- the CDC metadata column uses the SCD1 metadata struct schema and is
non-null;
- the spec records `ScdType.Type1`, the key column names (property +
`expectedKeyFields`), and the correct identifiers;
- composite keys are all carried through.
### Why are the changes needed?
Fills a test-coverage gap for the SCD1 auxiliary-table spec derivation,
matching the coverage that now exists for SCD2, so future changes to
`buildScd1AuxiliaryTableSpecFor` (schema shape, recorded drift metadata) are
guarded by a focused unit test.
### Does this PR introduce _any_ user-facing change?
No. Test-only change; no production code is modified.
### How was this patch tested?
- `build/sbt 'pipelines/testOnly *AutoCdcScd1AuxiliaryTableSpecSuite'` --
5/5 pass
- `dev/lint-scala` -- Scalastyle and Scalafmt clean
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Opus 4.8
Closes #57517 from anew/spark-scd1-aux-spec-tests.
Authored-by: Andreas Neumann <[email protected]>
Signed-off-by: Jose Torres <[email protected]>
(cherry picked from commit c784ac6d6c362a937fc4313b1e39aaea22b7dee4)
Signed-off-by: Jose Torres <[email protected]>
---
.../graph/AutoCdcScd1AuxiliaryTableSpecSuite.scala | 126 +++++++++++++++++++++
1 file changed, 126 insertions(+)
diff --git
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1AuxiliaryTableSpecSuite.scala
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1AuxiliaryTableSpecSuite.scala
new file mode 100644
index 000000000000..530cae7cfa58
--- /dev/null
+++
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1AuxiliaryTableSpecSuite.scala
@@ -0,0 +1,126 @@
+/*
+ * 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.graph
+
+import org.apache.spark.sql.execution.streaming.runtime.MemoryStream
+import org.apache.spark.sql.functions
+import org.apache.spark.sql.pipelines.autocdc.{
+ AutoCdcReservedNames,
+ ChangeArgs,
+ ColumnSelection,
+ Scd1BatchProcessor,
+ ScdType,
+ UnqualifiedColumnName
+}
+import org.apache.spark.sql.pipelines.utils.{PipelineTest,
TestGraphRegistrationContext}
+import org.apache.spark.sql.test.SharedSparkSession
+import org.apache.spark.sql.types.LongType
+
+/**
+ * Tests for [[AutoCdcAuxiliaryTable.buildAuxiliaryTableSpecFor]] on the SCD1
branch
+ * (`buildScd1AuxiliaryTableSpecFor`), exercised through the graph the same
way production does:
+ * a resolved SCD1 AutoCDC flow yields an [[AutoCdcAuxiliaryTableSpec]] via
+ * [[DataflowGraph.auxiliaryTableSpecs]].
+ *
+ * Unlike the SCD2 auxiliary table (which stores full hidden rows), the SCD1
auxiliary table
+ * stores only the key columns plus the CDC metadata column: it tracks per-key
tombstones, not
+ * historical rows. Its schema is therefore keys + `_cdc_metadata`, with no
user data columns.
+ */
+class AutoCdcScd1AuxiliaryTableSpecSuite extends PipelineTest with
SharedSparkSession {
+
+ private def targetIdentifier = fullyQualifiedIdentifier("target")
+
+ /** Source change feed with data columns `(id, name, version)`. */
+ private def sourceDf = {
+ val session = spark
+ import session.implicits._
+ val stream = MemoryStream[(Int, String, Long)]
+ stream.addData((1, "alice", 1L))
+ stream.toDF().toDF("id", "name", "version")
+ }
+
+ /** Resolve a graph with a single SCD1 AUTO CDC flow into `target` and
return its aux spec. */
+ private def scd1AuxSpec(
+ keys: Seq[String] = Seq("id"),
+ columnSelection: Option[ColumnSelection] = None):
AutoCdcAuxiliaryTableSpec = {
+ val ctx = new TestGraphRegistrationContext(spark)
+ ctx.registerTable("target")
+ ctx.registerFlow(AutoCdcFlow(
+ identifier = targetIdentifier,
+ destinationIdentifier = targetIdentifier,
+ func = ctx.dfFlowFunc(sourceDf),
+ queryContext = QueryContext(
+ currentCatalog = catalogInPipelineSpec,
+ currentDatabase = databaseInPipelineSpec),
+ origin = QueryOrigin.empty,
+ changeArgs = ChangeArgs(
+ keys = keys.map(k => UnqualifiedColumnName(Seq(k))),
+ sequencing = functions.col("version"),
+ columnSelection = columnSelection,
+ deleteCondition = None,
+ storedAsScdType = ScdType.Type1)))
+ val graph = ctx.resolveToDataflowGraph()
+
graph.auxiliaryTableSpecs(targetIdentifier).asInstanceOf[AutoCdcAuxiliaryTableSpec]
+ }
+
+ test("SCD1 aux schema is exactly the key columns plus the CDC metadata
column") {
+ val spec = scd1AuxSpec()
+ // Only the key (id) and the metadata column -- no user data columns (name
/ version) and no
+ // SCD2 framework columns.
+ assert(
+ spec.schema.fieldNames.toSeq == Seq("id",
AutoCdcReservedNames.cdcMetadataColName))
+ }
+
+ test("SCD1 aux metadata column uses the SCD1 metadata struct schema") {
+ val spec = scd1AuxSpec()
+ val metaField = spec.schema(AutoCdcReservedNames.cdcMetadataColName)
+ // version is a Long, so the SCD1 metadata struct is typed on LongType.
+ assert(metaField.dataType ==
Scd1BatchProcessor.cdcMetadataColSchema(LongType))
+ assert(!metaField.nullable, "the CDC metadata column is non-null")
+ }
+
+ test("SCD1 aux schema omits user data columns") {
+ val fieldNames = scd1AuxSpec().schema.fieldNames.toSeq
+ assert(!fieldNames.contains("name"))
+ assert(!fieldNames.contains("version"))
+ }
+
+ test("SCD1 aux spec records the SCD1 type, key names, and matching
identifier") {
+ val spec = scd1AuxSpec()
+ assert(spec.expectedScdType == ScdType.Type1)
+ assert(spec.properties(AutoCdcAuxiliaryTable.scdTypePropertyKey) ==
ScdType.Type1.label)
+ assert(spec.expectedKeyFields.map(_.name) == Seq("id"))
+ assert(
+ AutoCdcAuxiliaryTable.parseKeyColumnNames(
+
spec.properties(AutoCdcAuxiliaryTable.keyColumnNamesProperty)).contains(Seq("id")))
+ assert(spec.identifier ==
AutoCdcAuxiliaryTable.identifier(targetIdentifier))
+ assert(spec.targetTableIdentifier == targetIdentifier)
+ }
+
+ test("SCD1 aux spec carries all declared key fields for a composite key") {
+ val spec = scd1AuxSpec(keys = Seq("id", "name"))
+ assert(spec.expectedKeyFields.map(_.name) == Seq("id", "name"))
+ assert(
+ AutoCdcAuxiliaryTable.parseKeyColumnNames(
+
spec.properties(AutoCdcAuxiliaryTable.keyColumnNamesProperty)).contains(Seq("id",
"name")))
+ // The aux schema is the two keys followed by the metadata column, and
nothing else.
+ assert(
+ spec.schema.fieldNames.toSeq ==
+ Seq("id", "name", AutoCdcReservedNames.cdcMetadataColName))
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
