You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "anew (via GitHub)" <gi...@apache.org> on 2026/07/27 21:59:04 UTC
[PR] [SPARK-58321][SDP] Wire SCD2 AutoCDC streaming write and enable SCD2 end to end [spark]
anew opened a new pull request, #57584:
URL: https://github.com/apache/spark/pull/57584
### What changes were proposed in this pull request?
Adds `Scd2MergeStreamingWrite` (mirroring `Scd1MergeStreamingWrite`): it resolves the auxiliary-table identifier, constructs an `Scd2ForeachBatchHandler` over an `Scd2BatchProcessor`, and drives it via Structured Streaming `foreachBatch`. `FlowPlanner` now routes an SCD2 `AutoCdcMergeFlow` to it, replacing the `AUTOCDC_SCD2_NOT_SUPPORTED` throw.
### Why are the changes needed?
This removes the **last** remaining SCD2 gate. The flow-schema derivation (SPARK-58319), the auxiliary-table spec (SPARK-58320), the reserved-column / track-history validation (SPARK-57251, SPARK-58313), and the per-microbatch reconciliation handler (SPARK-57395) have all merged; this change makes SCD2 AutoCDC flows runnable end to end.
### Does this PR introduce _any_ user-facing change?
Yes: `AUTO CDC ... STORED AS SCD TYPE 2` pipelines are now supported and executable, where previously they failed with `AUTOCDC_SCD2_NOT_SUPPORTED`.
### How was this patch tested?
New `AutoCdcScd2SinglePipelineSuite` runs SCD2 flows end to end through the pipeline and asserts SCD2 target semantics (an upsert opens a current record; an update closes the prior record and opens a new one; a delete closes the current record) plus auxiliary-table materialization. Also refreshed an `AutoCdcFlowSuite` test whose name/comment referenced the removed gate.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Opus 4.8
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58321][SDP] Wire SCD2 AutoCDC streaming write and enable SCD2 end to end [spark]
Posted by "anew (via GitHub)" <gi...@apache.org>.
anew commented on code in PR #57584:
URL: https://github.com/apache/spark/pull/57584#discussion_r3662068538
##########
sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2SinglePipelineSuite.scala:
##########
@@ -0,0 +1,191 @@
+/*
+ * 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.Row
+import org.apache.spark.sql.execution.streaming.runtime.MemoryStream
+import org.apache.spark.sql.functions
+import org.apache.spark.sql.pipelines.autocdc.{
+ AutoCdcReservedNames,
+ ColumnSelection,
+ ScdType,
+ UnqualifiedColumnName
+}
+import org.apache.spark.sql.pipelines.utils.{ExecutionTest, TestGraphRegistrationContext}
+import org.apache.spark.sql.test.SharedSparkSession
+
+/**
+ * End-to-end smoke tests for AutoCDC SCD Type 2 flows running within a single pipeline: one
+ * [[DataflowGraph]] / [[TestPipelineUpdateContext]] executes an SCD2 AutoCDC flow through the
+ * [[Scd2MergeStreamingWrite]] streaming write, and both the target table and the auxiliary
+ * table contents are asserted at the end.
+ *
+ * This exercises the full wiring landed for SCD2: the flow planner routing an SCD2
+ * [[AutoCdcMergeFlow]] to [[Scd2MergeStreamingWrite]], the auxiliary-table materialization, and
+ * the [[org.apache.spark.sql.pipelines.autocdc.Scd2ForeachBatchHandler]] reconciliation.
+ */
+class AutoCdcScd2SinglePipelineSuite
+ extends ExecutionTest
+ with SharedSparkSession
+ with AutoCdcGraphExecutionTestMixin {
+
+ /** The SCD2 target's `_cdc_metadata` struct value for a given recordStartAt. */
+ private def scd2Meta(recordStartAt: Long): Row = Row(recordStartAt)
+
+ /**
+ * DDL for an SCD2 target table with user columns `(id, name, version)` plus the framework
+ * columns `__START_AT` / `__END_AT` (sequencing type BIGINT) and the SCD2 `_cdc_metadata`
+ * struct. `version` is the sequencing column and, unless excluded via a column selection, is
+ * retained as an ordinary user column in the target.
+ */
+ private def createScd2Target(table: String): Unit = {
+ val meta = AutoCdcReservedNames.cdcMetadataColName
+ spark.sql(
+ s"CREATE TABLE $table (" +
+ "id INT NOT NULL, name STRING, version BIGINT NOT NULL, " +
+ "__START_AT BIGINT, __END_AT BIGINT, " +
+ s"$meta STRUCT<__RECORD_START_AT:BIGINT> NOT NULL)"
+ )
+ }
+
+ test("SCD2: an upsert lands an open current record in an empty target table") {
Review Comment:
note: the test coverage here s not exhaustive, but it would be hard to test all possible cases in an end-to-end suite. This suite merely asserts that it works end-to-end. The convergence test, however, should provide decent coverage.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58321][SDP] Wire SCD2 AutoCDC streaming write and enable SCD2 end to end [spark]
Posted by "anew (via GitHub)" <gi...@apache.org>.
anew commented on code in PR #57584:
URL: https://github.com/apache/spark/pull/57584#discussion_r3668553441
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/FlowPlanner.scala:
##########
@@ -96,10 +95,21 @@ class FlowPlanner(
case _ => unsupportedDestinationType(acmf, output)
}
case ScdType.Type2 =>
- throw new AnalysisException(
- errorClass = "AUTOCDC_SCD2_NOT_SUPPORTED",
Review Comment:
good catch, thanks, fixed!
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58321][SDP] Wire SCD2 AutoCDC streaming write and enable SCD2 end to end [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on PR #57584:
URL: https://github.com/apache/spark/pull/57584#issuecomment-5113232451
**Merge Summary:**
- merged into master https://github.com/apache/spark/commit/24ad34b689be2afe543050ab2797918b836d7fe9
- merged into branch-4.x https://github.com/apache/spark/commit/a8f1e432f0629a9c0d80d698bcf33be362f751da
*Posted by `merge_spark_pr.py`*
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58321][SDP] Wire SCD2 AutoCDC streaming write and enable SCD2 end to end [spark]
Posted by "anew (via GitHub)" <gi...@apache.org>.
anew commented on code in PR #57584:
URL: https://github.com/apache/spark/pull/57584#discussion_r3668540496
##########
sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2SinglePipelineSuite.scala:
##########
@@ -0,0 +1,188 @@
+/*
+ * 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.Row
+import org.apache.spark.sql.execution.streaming.runtime.MemoryStream
+import org.apache.spark.sql.functions
+import org.apache.spark.sql.pipelines.autocdc.{
+ AutoCdcReservedNames,
+ ColumnSelection,
+ ScdType,
+ UnqualifiedColumnName
+}
+import org.apache.spark.sql.pipelines.utils.{ExecutionTest, TestGraphRegistrationContext}
+import org.apache.spark.sql.test.SharedSparkSession
+
+/**
+ * End-to-end smoke tests for AutoCDC SCD Type 2 flows running within a single pipeline: one
+ * [[DataflowGraph]] / [[TestPipelineUpdateContext]] executes an SCD2 AutoCDC flow through the
+ * [[Scd2MergeStreamingWrite]] streaming write, and both the target table and the auxiliary
+ * table contents are asserted at the end.
+ *
+ * This exercises the full wiring landed for SCD2: the flow planner routing an SCD2
+ * [[AutoCdcMergeFlow]] to [[Scd2MergeStreamingWrite]], the auxiliary-table materialization, and
+ * the [[org.apache.spark.sql.pipelines.autocdc.Scd2ForeachBatchHandler]] reconciliation.
+ */
+class AutoCdcScd2SinglePipelineSuite
+ extends ExecutionTest
+ with SharedSparkSession
+ with AutoCdcGraphExecutionTestMixin {
+
+ /** The SCD2 target's `_cdc_metadata` struct value for a given recordStartAt. */
+ private def scd2Meta(recordStartAt: Long): Row = Row(recordStartAt)
+
+ /**
+ * DDL for an SCD2 target table with user columns `(id, name, version)` plus the framework
+ * columns `__START_AT` / `__END_AT` (sequencing type BIGINT) and the SCD2 `_cdc_metadata`
+ * struct. `version` is the sequencing column and, unless excluded via a column selection, is
+ * retained as an ordinary user column in the target.
+ */
+ private def createScd2Target(table: String): Unit = {
+ spark.sql(
+ s"CREATE TABLE $table (" +
+ s"id INT NOT NULL, name STRING, version BIGINT NOT NULL, $scd2MetadataDdl)"
+ )
+ }
+
+ test("SCD2: an upsert lands an open current record in an empty target table") {
+ val session = spark
+ import session.implicits._
+ createScd2Target(s"$catalog.$namespace.target")
+
+ val stream = MemoryStream[(Int, String, Long)]
+ stream.addData((1, "alice", 10L))
+
+ val ctx = new TestGraphRegistrationContext(spark) {
+ registerTable("target", catalog = Some(catalog), database = Some(namespace))
+ registerFlow(autoCdcFlow(
+ name = "auto_cdc_flow",
+ target = "target",
+ query = dfFlowFunc(stream.toDF().toDF("id", "name", "version")),
+ keys = Seq("id"),
+ sequencing = functions.col("version"),
+ scdType = ScdType.Type2
+ ))
+ }
+
+ runPipeline(ctx)
+
+ // A single event opens a current record: START_AT = the event sequence, END_AT = null.
+ checkAnswer(
+ spark.table(s"$catalog.$namespace.target"),
+ Seq(Row(1, "alice", 10L, 10L, null, scd2Meta(10L)))
+ )
+ }
+
+ test("SCD2: an update to a key closes the prior record and opens a new one") {
+ val session = spark
+ import session.implicits._
+ createScd2Target(s"$catalog.$namespace.target")
+
+ val stream = MemoryStream[(Int, String, Long)]
+ stream.addData((1, "alice", 10L), (1, "alicia", 20L))
+
+ val ctx = new TestGraphRegistrationContext(spark) {
+ registerTable("target", catalog = Some(catalog), database = Some(namespace))
+ registerFlow(autoCdcFlow(
+ name = "auto_cdc_flow",
+ target = "target",
+ query = dfFlowFunc(stream.toDF().toDF("id", "name", "version")),
+ keys = Seq("id"),
+ sequencing = functions.col("version"),
+ scdType = ScdType.Type2
+ ))
+ }
+
+ runPipeline(ctx)
+
+ // The first value is closed at the second event's sequence; the second value is open.
+ checkAnswer(
+ spark.table(s"$catalog.$namespace.target"),
+ Seq(
+ Row(1, "alice", 10L, 10L, 20L, scd2Meta(10L)),
+ Row(1, "alicia", 20L, 20L, null, scd2Meta(20L))
+ )
+ )
+ }
+
+ test("SCD2: a delete closes the current record with no open record remaining") {
+ val session = spark
+ import session.implicits._
+ // Target omits `is_delete`: the source carries it as a control column driving the delete
+ // condition, and it is excluded from the target projection.
+ createScd2Target(s"$catalog.$namespace.target")
+
+ val stream = MemoryStream[(Int, String, Long, Boolean)]
+ stream.addData((1, "alice", 10L, false), (1, null, 20L, true))
+
+ val ctx = new TestGraphRegistrationContext(spark) {
+ registerTable("target", catalog = Some(catalog), database = Some(namespace))
+ registerFlow(autoCdcFlow(
+ name = "auto_cdc_flow",
+ target = "target",
+ query = dfFlowFunc(stream.toDF().toDF("id", "name", "version", "is_delete")),
+ keys = Seq("id"),
+ sequencing = functions.col("version"),
+ columnSelection = Some(
+ ColumnSelection.ExcludeColumns(Seq(UnqualifiedColumnName("is_delete")))
+ ),
+ deleteCondition = Some(functions.col("is_delete")),
+ scdType = ScdType.Type2
+ ))
+ }
+
+ runPipeline(ctx)
+
+ // The delete closes the open record at the delete's sequence; nothing remains open.
+ checkAnswer(
+ spark.table(s"$catalog.$namespace.target"),
+ Seq(Row(1, "alice", 10L, 10L, 20L, scd2Meta(10L)))
+ )
+ }
+
+ test("SCD2: the auxiliary table is materialized for the target") {
+ val session = spark
+ import session.implicits._
+ createScd2Target(s"$catalog.$namespace.target")
+
+ val stream = MemoryStream[(Int, String, Long)]
+ stream.addData((1, "alice", 10L))
+
+ val ctx = new TestGraphRegistrationContext(spark) {
+ registerTable("target", catalog = Some(catalog), database = Some(namespace))
+ registerFlow(autoCdcFlow(
+ name = "auto_cdc_flow",
+ target = "target",
+ query = dfFlowFunc(stream.toDF().toDF("id", "name", "version")),
+ keys = Seq("id"),
+ sequencing = functions.col("version"),
+ scdType = ScdType.Type2
+ ))
+ }
+
+ runPipeline(ctx)
+
+ // The SCD2 auxiliary table exists and carries the aux-only deleted-by-batch-id marker column
+ // in addition to the full target row schema.
+ val auxColumns = spark.table(auxTableNameFor("target")).schema.fieldNames.toSet
+ assert(auxColumns.contains(AutoCdcReservedNames.cdcMetadataColName))
+ assert(auxColumns.contains("__START_AT"))
+ assert(auxColumns.contains("__END_AT"))
Review Comment:
good suggestion, done.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58321][SDP] Wire SCD2 AutoCDC streaming write and enable SCD2 end to end [spark]
Posted by "jose-torres (via GitHub)" <gi...@apache.org>.
jose-torres commented on code in PR #57584:
URL: https://github.com/apache/spark/pull/57584#discussion_r3663325069
##########
sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcOutOfOrderConvergenceSuite.scala:
##########
@@ -197,11 +219,18 @@ class AutoCdcScd1OutOfOrderConvergenceSuite
runPipeline(outOfOrderCtx)
}
+ // Only the user-visible target must converge. The auxiliary tables legitimately differ by
Review Comment:
This is fine under the assumption, which I think is correct but want to confirm, that any auxiliary row generated at version N is guaranteed to stop affecting results if the ingestion durably advances to some version M not too much higher than N. (It would be a problem, to pick an exaggerated example, if some category of auxiliary row caused different results starting at N + 1000; then just checking consistency betwen the targets would not be enough to confirm that the behavior is meaningfully the same.)
##########
sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2SinglePipelineSuite.scala:
##########
@@ -0,0 +1,191 @@
+/*
+ * 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.Row
+import org.apache.spark.sql.execution.streaming.runtime.MemoryStream
+import org.apache.spark.sql.functions
+import org.apache.spark.sql.pipelines.autocdc.{
+ AutoCdcReservedNames,
+ ColumnSelection,
+ ScdType,
+ UnqualifiedColumnName
+}
+import org.apache.spark.sql.pipelines.utils.{ExecutionTest, TestGraphRegistrationContext}
+import org.apache.spark.sql.test.SharedSparkSession
+
+/**
+ * End-to-end smoke tests for AutoCDC SCD Type 2 flows running within a single pipeline: one
+ * [[DataflowGraph]] / [[TestPipelineUpdateContext]] executes an SCD2 AutoCDC flow through the
+ * [[Scd2MergeStreamingWrite]] streaming write, and both the target table and the auxiliary
+ * table contents are asserted at the end.
+ *
+ * This exercises the full wiring landed for SCD2: the flow planner routing an SCD2
+ * [[AutoCdcMergeFlow]] to [[Scd2MergeStreamingWrite]], the auxiliary-table materialization, and
+ * the [[org.apache.spark.sql.pipelines.autocdc.Scd2ForeachBatchHandler]] reconciliation.
+ */
+class AutoCdcScd2SinglePipelineSuite
+ extends ExecutionTest
+ with SharedSparkSession
+ with AutoCdcGraphExecutionTestMixin {
+
+ /** The SCD2 target's `_cdc_metadata` struct value for a given recordStartAt. */
+ private def scd2Meta(recordStartAt: Long): Row = Row(recordStartAt)
+
+ /**
+ * DDL for an SCD2 target table with user columns `(id, name, version)` plus the framework
+ * columns `__START_AT` / `__END_AT` (sequencing type BIGINT) and the SCD2 `_cdc_metadata`
+ * struct. `version` is the sequencing column and, unless excluded via a column selection, is
+ * retained as an ordinary user column in the target.
+ */
+ private def createScd2Target(table: String): Unit = {
+ val meta = AutoCdcReservedNames.cdcMetadataColName
+ spark.sql(
+ s"CREATE TABLE $table (" +
+ "id INT NOT NULL, name STRING, version BIGINT NOT NULL, " +
+ "__START_AT BIGINT, __END_AT BIGINT, " +
+ s"$meta STRUCT<__RECORD_START_AT:BIGINT> NOT NULL)"
+ )
+ }
+
+ test("SCD2: an upsert lands an open current record in an empty target table") {
Review Comment:
I agree with this analysis. I trust the principle that the results shouldn't be ingestion order dependent more than I trust my ability to read through a big pile of specific rows and intuit the correct result.
##########
sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcOutOfOrderConvergenceSuite.scala:
##########
@@ -140,20 +146,33 @@ class AutoCdcScd1OutOfOrderConvergenceSuite
deleteCondition = Some(functions.col(isDeleteColumn) === true),
columnSelection = Some(ColumnSelection.ExcludeColumns(
Seq(UnqualifiedColumnName(isDeleteColumn))
- ))
+ )),
+ scdType = scdType
))
}
}
- private def createTargetTable(targetTable: String): Unit = {
+ /**
+ * DDL fragment for the SCD-type-specific reserved columns a target table carries after the
+ * user-selected data columns: just the CDC metadata column for SCD1, plus the __START_AT /
+ * __END_AT interval bounds for SCD2. The sequencing type is BIGINT here.
+ */
+ private def reservedColumnsDdl(scdType: ScdType): String = scdType match {
Review Comment:
I ultimately decided I don't care that SCD1 and SCD2 disagree on the encapsulation boundary for the DDL, but I'm leaving this comment as a record of the train of thought in case it jumps out at anyone.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58321][SDP] Wire SCD2 AutoCDC streaming write and enable SCD2 end to end [spark]
Posted by "anew (via GitHub)" <gi...@apache.org>.
anew commented on code in PR #57584:
URL: https://github.com/apache/spark/pull/57584#discussion_r3667668147
##########
sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcOutOfOrderConvergenceSuite.scala:
##########
@@ -197,11 +219,18 @@ class AutoCdcScd1OutOfOrderConvergenceSuite
runPipeline(outOfOrderCtx)
}
+ // Only the user-visible target must converge. The auxiliary tables legitimately differ by
Review Comment:
Confirmed. findAffectedRowsFromAuxiliaryTable bounds aux-row participation per key to recordStartAt >= minSequenceInMicrobatch plus a single "anchor" — the aux row with the largest recordStartAt strictly below the batch's min sequence. There's no unbounded look-back (no "N+1000" case): an aux row from version N only affects microbatches adjacent to N, so once ingestion advances past it (with a newer anchor present) it stops mattering. That's what makes the target-only convergence comparison valid.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58321][SDP] Wire SCD2 AutoCDC streaming write and enable SCD2 end to end [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho closed pull request #57584: [SPARK-58321][SDP] Wire SCD2 AutoCDC streaming write and enable SCD2 end to end
URL: https://github.com/apache/spark/pull/57584
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58321][SDP] Wire SCD2 AutoCDC streaming write and enable SCD2 end to end [spark]
Posted by "anew (via GitHub)" <gi...@apache.org>.
anew commented on code in PR #57584:
URL: https://github.com/apache/spark/pull/57584#discussion_r3662068538
##########
sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2SinglePipelineSuite.scala:
##########
@@ -0,0 +1,191 @@
+/*
+ * 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.Row
+import org.apache.spark.sql.execution.streaming.runtime.MemoryStream
+import org.apache.spark.sql.functions
+import org.apache.spark.sql.pipelines.autocdc.{
+ AutoCdcReservedNames,
+ ColumnSelection,
+ ScdType,
+ UnqualifiedColumnName
+}
+import org.apache.spark.sql.pipelines.utils.{ExecutionTest, TestGraphRegistrationContext}
+import org.apache.spark.sql.test.SharedSparkSession
+
+/**
+ * End-to-end smoke tests for AutoCDC SCD Type 2 flows running within a single pipeline: one
+ * [[DataflowGraph]] / [[TestPipelineUpdateContext]] executes an SCD2 AutoCDC flow through the
+ * [[Scd2MergeStreamingWrite]] streaming write, and both the target table and the auxiliary
+ * table contents are asserted at the end.
+ *
+ * This exercises the full wiring landed for SCD2: the flow planner routing an SCD2
+ * [[AutoCdcMergeFlow]] to [[Scd2MergeStreamingWrite]], the auxiliary-table materialization, and
+ * the [[org.apache.spark.sql.pipelines.autocdc.Scd2ForeachBatchHandler]] reconciliation.
+ */
+class AutoCdcScd2SinglePipelineSuite
+ extends ExecutionTest
+ with SharedSparkSession
+ with AutoCdcGraphExecutionTestMixin {
+
+ /** The SCD2 target's `_cdc_metadata` struct value for a given recordStartAt. */
+ private def scd2Meta(recordStartAt: Long): Row = Row(recordStartAt)
+
+ /**
+ * DDL for an SCD2 target table with user columns `(id, name, version)` plus the framework
+ * columns `__START_AT` / `__END_AT` (sequencing type BIGINT) and the SCD2 `_cdc_metadata`
+ * struct. `version` is the sequencing column and, unless excluded via a column selection, is
+ * retained as an ordinary user column in the target.
+ */
+ private def createScd2Target(table: String): Unit = {
+ val meta = AutoCdcReservedNames.cdcMetadataColName
+ spark.sql(
+ s"CREATE TABLE $table (" +
+ "id INT NOT NULL, name STRING, version BIGINT NOT NULL, " +
+ "__START_AT BIGINT, __END_AT BIGINT, " +
+ s"$meta STRUCT<__RECORD_START_AT:BIGINT> NOT NULL)"
+ )
+ }
+
+ test("SCD2: an upsert lands an open current record in an empty target table") {
Review Comment:
note: the test coverage here s not exhaustive, but it would be hard to test all possible cases in an end-to-end suite. This suite merely asserts that it works end-to-end
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58321][SDP] Wire SCD2 AutoCDC streaming write and enable SCD2 end to end [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #57584:
URL: https://github.com/apache/spark/pull/57584#discussion_r3668142785
##########
sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcOutOfOrderConvergenceSuite.scala:
##########
@@ -173,22 +185,25 @@ class AutoCdcScd1OutOfOrderConvergenceSuite
val shuffledEventStream = rand.shuffle(sortedEventStream)
withClue(
- s"\nseed=$seed (rerun with -D$seedSystemProperty=$seed to reproduce)\n" +
+ s"\nscdType=${scdType.label} seed=$seed " +
+ s"(rerun with -D$seedSystemProperty=$seed to reproduce)\n" +
Review Comment:
The property this message tells the developer to set is still `spark.sql.test.autocdc.scd1OutOfOrderConvergenceSeed` (line 80), so a failing SCD2 run prints an SCD1-named knob. Worth renaming along with the suite.
Separately, the comment on line 193 justifies the scd-type table-name suffix as keeping the SCD1 and SCD2 tests from colliding within a run, but the mixin's `afterEach` calls `SharedTablesInMemoryRowLevelOperationTableCatalog.reset()`, so the tables cannot collide. The suffix is harmless; the stated reason just is not the real one.
##########
sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd2SinglePipelineSuite.scala:
##########
@@ -0,0 +1,188 @@
+/*
+ * 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.Row
+import org.apache.spark.sql.execution.streaming.runtime.MemoryStream
+import org.apache.spark.sql.functions
+import org.apache.spark.sql.pipelines.autocdc.{
+ AutoCdcReservedNames,
+ ColumnSelection,
+ ScdType,
+ UnqualifiedColumnName
+}
+import org.apache.spark.sql.pipelines.utils.{ExecutionTest, TestGraphRegistrationContext}
+import org.apache.spark.sql.test.SharedSparkSession
+
+/**
+ * End-to-end smoke tests for AutoCDC SCD Type 2 flows running within a single pipeline: one
+ * [[DataflowGraph]] / [[TestPipelineUpdateContext]] executes an SCD2 AutoCDC flow through the
+ * [[Scd2MergeStreamingWrite]] streaming write, and both the target table and the auxiliary
+ * table contents are asserted at the end.
+ *
+ * This exercises the full wiring landed for SCD2: the flow planner routing an SCD2
+ * [[AutoCdcMergeFlow]] to [[Scd2MergeStreamingWrite]], the auxiliary-table materialization, and
+ * the [[org.apache.spark.sql.pipelines.autocdc.Scd2ForeachBatchHandler]] reconciliation.
+ */
+class AutoCdcScd2SinglePipelineSuite
+ extends ExecutionTest
+ with SharedSparkSession
+ with AutoCdcGraphExecutionTestMixin {
+
+ /** The SCD2 target's `_cdc_metadata` struct value for a given recordStartAt. */
+ private def scd2Meta(recordStartAt: Long): Row = Row(recordStartAt)
+
+ /**
+ * DDL for an SCD2 target table with user columns `(id, name, version)` plus the framework
+ * columns `__START_AT` / `__END_AT` (sequencing type BIGINT) and the SCD2 `_cdc_metadata`
+ * struct. `version` is the sequencing column and, unless excluded via a column selection, is
+ * retained as an ordinary user column in the target.
+ */
+ private def createScd2Target(table: String): Unit = {
+ spark.sql(
+ s"CREATE TABLE $table (" +
+ s"id INT NOT NULL, name STRING, version BIGINT NOT NULL, $scd2MetadataDdl)"
+ )
+ }
+
+ test("SCD2: an upsert lands an open current record in an empty target table") {
+ val session = spark
+ import session.implicits._
+ createScd2Target(s"$catalog.$namespace.target")
+
+ val stream = MemoryStream[(Int, String, Long)]
+ stream.addData((1, "alice", 10L))
+
+ val ctx = new TestGraphRegistrationContext(spark) {
+ registerTable("target", catalog = Some(catalog), database = Some(namespace))
+ registerFlow(autoCdcFlow(
+ name = "auto_cdc_flow",
+ target = "target",
+ query = dfFlowFunc(stream.toDF().toDF("id", "name", "version")),
+ keys = Seq("id"),
+ sequencing = functions.col("version"),
+ scdType = ScdType.Type2
+ ))
+ }
+
+ runPipeline(ctx)
+
+ // A single event opens a current record: START_AT = the event sequence, END_AT = null.
+ checkAnswer(
+ spark.table(s"$catalog.$namespace.target"),
+ Seq(Row(1, "alice", 10L, 10L, null, scd2Meta(10L)))
+ )
+ }
+
+ test("SCD2: an update to a key closes the prior record and opens a new one") {
+ val session = spark
+ import session.implicits._
+ createScd2Target(s"$catalog.$namespace.target")
+
+ val stream = MemoryStream[(Int, String, Long)]
+ stream.addData((1, "alice", 10L), (1, "alicia", 20L))
+
+ val ctx = new TestGraphRegistrationContext(spark) {
+ registerTable("target", catalog = Some(catalog), database = Some(namespace))
+ registerFlow(autoCdcFlow(
+ name = "auto_cdc_flow",
+ target = "target",
+ query = dfFlowFunc(stream.toDF().toDF("id", "name", "version")),
+ keys = Seq("id"),
+ sequencing = functions.col("version"),
+ scdType = ScdType.Type2
+ ))
+ }
+
+ runPipeline(ctx)
+
+ // The first value is closed at the second event's sequence; the second value is open.
+ checkAnswer(
+ spark.table(s"$catalog.$namespace.target"),
+ Seq(
+ Row(1, "alice", 10L, 10L, 20L, scd2Meta(10L)),
+ Row(1, "alicia", 20L, 20L, null, scd2Meta(20L))
+ )
+ )
+ }
+
+ test("SCD2: a delete closes the current record with no open record remaining") {
+ val session = spark
+ import session.implicits._
+ // Target omits `is_delete`: the source carries it as a control column driving the delete
+ // condition, and it is excluded from the target projection.
+ createScd2Target(s"$catalog.$namespace.target")
+
+ val stream = MemoryStream[(Int, String, Long, Boolean)]
+ stream.addData((1, "alice", 10L, false), (1, null, 20L, true))
+
+ val ctx = new TestGraphRegistrationContext(spark) {
+ registerTable("target", catalog = Some(catalog), database = Some(namespace))
+ registerFlow(autoCdcFlow(
+ name = "auto_cdc_flow",
+ target = "target",
+ query = dfFlowFunc(stream.toDF().toDF("id", "name", "version", "is_delete")),
+ keys = Seq("id"),
+ sequencing = functions.col("version"),
+ columnSelection = Some(
+ ColumnSelection.ExcludeColumns(Seq(UnqualifiedColumnName("is_delete")))
+ ),
+ deleteCondition = Some(functions.col("is_delete")),
+ scdType = ScdType.Type2
+ ))
+ }
+
+ runPipeline(ctx)
+
+ // The delete closes the open record at the delete's sequence; nothing remains open.
+ checkAnswer(
+ spark.table(s"$catalog.$namespace.target"),
+ Seq(Row(1, "alice", 10L, 10L, 20L, scd2Meta(10L)))
+ )
+ }
+
+ test("SCD2: the auxiliary table is materialized for the target") {
+ val session = spark
+ import session.implicits._
+ createScd2Target(s"$catalog.$namespace.target")
+
+ val stream = MemoryStream[(Int, String, Long)]
+ stream.addData((1, "alice", 10L))
+
+ val ctx = new TestGraphRegistrationContext(spark) {
+ registerTable("target", catalog = Some(catalog), database = Some(namespace))
+ registerFlow(autoCdcFlow(
+ name = "auto_cdc_flow",
+ target = "target",
+ query = dfFlowFunc(stream.toDF().toDF("id", "name", "version")),
+ keys = Seq("id"),
+ sequencing = functions.col("version"),
+ scdType = ScdType.Type2
+ ))
+ }
+
+ runPipeline(ctx)
+
+ // The SCD2 auxiliary table exists and carries the aux-only deleted-by-batch-id marker column
+ // in addition to the full target row schema.
+ val auxColumns = spark.table(auxTableNameFor("target")).schema.fieldNames.toSet
+ assert(auxColumns.contains(AutoCdcReservedNames.cdcMetadataColName))
+ assert(auxColumns.contains("__START_AT"))
+ assert(auxColumns.contains("__END_AT"))
Review Comment:
The comment says the aux table "carries the aux-only deleted-by-batch-id marker column", but no assertion checks for it. Relatedly, the class scaladoc says "both the target table and the auxiliary table contents are asserted at the end", whereas what is asserted here is the schema rather than the contents.
`Scd2BatchProcessor.deletedByBatchIdColName` is `private[pipelines]` and `AutoCdcScd2AuxiliaryTableSpecSuite` already uses it from this package, so this could assert the full materialized schema (the target fields plus the marker appended) and then match its own comment. The distinct value of this test over the spec suite is proving the table really is created by a live run, and a full-schema assertion keeps that while making it stronger.
Same thought for the two literals: `Scd2BatchProcessor.startAtColName` / `endAtColName` are reachable here, and those two are precisely the reserved columns that do not carry `AutoCdcReservedNames.prefix`, so a rename would slip past a string literal.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58321][SDP] Wire SCD2 AutoCDC streaming write and enable SCD2 end to end [spark]
Posted by "uros-b (via GitHub)" <gi...@apache.org>.
uros-b commented on code in PR #57584:
URL: https://github.com/apache/spark/pull/57584#discussion_r3667517288
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/FlowPlanner.scala:
##########
@@ -96,10 +95,21 @@ class FlowPlanner(
case _ => unsupportedDestinationType(acmf, output)
}
case ScdType.Type2 =>
- throw new AnalysisException(
- errorClass = "AUTOCDC_SCD2_NOT_SUPPORTED",
Review Comment:
If this error class is no longer used, please remove it completely (for example, from `common/utils/src/main/resources/error/error-conditions.json`) to prevent confusion and to keep the error-class registry up to date
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58321][SDP] Wire SCD2 AutoCDC streaming write and enable SCD2 end to end [spark]
Posted by "anew (via GitHub)" <gi...@apache.org>.
anew commented on code in PR #57584:
URL: https://github.com/apache/spark/pull/57584#discussion_r3668116892
##########
sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcOutOfOrderConvergenceSuite.scala:
##########
@@ -140,20 +146,33 @@ class AutoCdcScd1OutOfOrderConvergenceSuite
deleteCondition = Some(functions.col(isDeleteColumn) === true),
columnSelection = Some(ColumnSelection.ExcludeColumns(
Seq(UnqualifiedColumnName(isDeleteColumn))
- ))
+ )),
+ scdType = scdType
))
}
}
- private def createTargetTable(targetTable: String): Unit = {
+ /**
+ * DDL fragment for the SCD-type-specific reserved columns a target table carries after the
+ * user-selected data columns: just the CDC metadata column for SCD1, plus the __START_AT /
+ * __END_AT interval bounds for SCD2. The sequencing type is BIGINT here.
+ */
+ private def reservedColumnsDdl(scdType: ScdType): String = scdType match {
Review Comment:
I made this symmetric by
- renaming `cdcMetadataDdl` to `scd1MetadataDdl`
- adding a new `scd2MetadataDdl` to the Mixin
- using these constants everywhere.
But digging into this, I realized that there are not many paces where the new `scd2MetadataDdl` are used. The reason is a gap in test coverage: while we have quite a few end to end tests for SCD1, we have very few for SCD2. I will close that test gap in a follow-up PR with Jira: https://issues.apache.org/jira/browse/SPARK-58409
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58321][SDP] Wire SCD2 AutoCDC streaming write and enable SCD2 end to end [spark]
Posted by "anew (via GitHub)" <gi...@apache.org>.
anew commented on code in PR #57584:
URL: https://github.com/apache/spark/pull/57584#discussion_r3668541717
##########
sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcOutOfOrderConvergenceSuite.scala:
##########
@@ -173,22 +185,25 @@ class AutoCdcScd1OutOfOrderConvergenceSuite
val shuffledEventStream = rand.shuffle(sortedEventStream)
withClue(
- s"\nseed=$seed (rerun with -D$seedSystemProperty=$seed to reproduce)\n" +
+ s"\nscdType=${scdType.label} seed=$seed " +
+ s"(rerun with -D$seedSystemProperty=$seed to reproduce)\n" +
Review Comment:
fixed both of them!
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58321][SDP] Wire SCD2 AutoCDC streaming write and enable SCD2 end to end [spark]
Posted by "anew (via GitHub)" <gi...@apache.org>.
anew commented on PR #57584:
URL: https://github.com/apache/spark/pull/57584#issuecomment-5108703808
@szehon-ho thanks for pointng out the test gap. I had just discovered that myself while addressing Jose's comments and I already created a follow-up Jira to close that gap: https://issues.apache.org/jira/browse/SPARK-58409
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org