You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "AnishMahto (via GitHub)" <gi...@apache.org> on 2026/09/18 18:05:36 UTC
[PR] [SPARK-59645][SDP] SCD1 Ignore-null support; Refactor SCD1 microbatch reconciliation into strategies [spark]
AnishMahto opened a new pull request, #58915:
URL: https://github.com/apache/spark/pull/58915
### What changes were proposed in this pull request?
Default SCD1 reconciliation accepts or rejects an event wholesale, so all columns share one row-level version. Ignore-null permits different events to author different leaves, requiring fundamentally different deduplication and tombstone reconciliation.
Refactor `Scd1BatchProcessor` to delegate to a reconciliation strategy and move existing behavior into the default `Scd1RowLevelReconciliation`. This preserves current behavior while establishing the boundary for future leaf-level reconciliation.
### Why are the changes needed?
Preparing `Scd1BatchProcessor` to support an alternative leaf-level reconciliation strategy, required for ignore-null support.
### Does this PR introduce _any_ user-facing change?
No. This is a pure refactoring change.
### How was this patch tested?
Existing tests. No new tests added as this is a pure refactoring change.
### Was this patch authored or co-authored using generative AI tooling?
Co-authored with GPT 5.6 Sol.
--
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-59645][SDP] SCD1 Ignore-null support; Refactor SCD1 microbatch reconciliation into strategies [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho closed pull request #58915: [SPARK-59645][SDP] SCD1 Ignore-null support; Refactor SCD1 microbatch reconciliation into strategies
URL: https://github.com/apache/spark/pull/58915
--
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-59645][SDP] SCD1 Ignore-null support; Refactor SCD1 microbatch reconciliation into strategies [spark]
Posted by "AnishMahto (via GitHub)" <gi...@apache.org>.
AnishMahto commented on PR #58915:
URL: https://github.com/apache/spark/pull/58915#issuecomment-5781032867
@szehon-ho mind taking a look at this pure refactor PR for AutoCDC SCD1? No new logic or behaviors.
--
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-59645][SDP] SCD1 Ignore-null support; Refactor SCD1 microbatch reconciliation into strategies [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #58915:
URL: https://github.com/apache/spark/pull/58915#discussion_r4077633926
##########
sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd1RowLevelReconciliationSuite.scala:
##########
@@ -63,6 +62,38 @@ class Scd1BatchProcessorSuite extends QueryTest with SharedSparkSession {
private def columnNamesAndDataTypes(schema: StructType): Seq[(String, DataType)] =
schema.fields.map(f => (f.name, f.dataType)).toSeq
+ private def deduplicateMicrobatch(
+ changeArgs: ChangeArgs,
+ batch: DataFrame): DataFrame =
+ Scd1RowLevelReconciliation.deduplicateMicrobatch(changeArgs, batch)
Review Comment:
Optional cleanup: could we import `deduplicateMicrobatch`, `projectTargetColumnsOntoMicrobatch`, and `applyTombstonesToMicrobatch` directly from `Scd1RowLevelReconciliation` and remove the three forwarding helpers? They only pass through their arguments. The `extendMicrobatchRowsWithCdcMetadata` helper can stay since it supplies `LongType`.
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd1ReconciliationStrategy.scala:
##########
@@ -0,0 +1,209 @@
+/*
+ * 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}
+import org.apache.spark.sql.Column
+import org.apache.spark.sql.catalyst.util.QuotingUtils
+import org.apache.spark.sql.classic.DataFrame
+import org.apache.spark.sql.types.DataType
+import org.apache.spark.util.ArrayImplicits._
+
+/** Strategy for reconciling an SCD1 microbatch. */
+private[pipelines] trait Scd1ReconciliationStrategy {
+
+ /**
+ * Resolves the CDC events for each key and removes events superseded by recorded tombstones.
+ *
+ * The sequencing expression must have an orderable data type, and every row must have non-null
+ * sequencing and key values. These invariants are required for per-key ordering and matching.
+ *
+ * @param validatedBatchDf A CDC microbatch satisfying the invariants above and containing the key
+ * columns and every column needed to evaluate the sequencing, delete, and
+ * column-selection expressions.
+ * @param auxiliaryTableDf A snapshot of the auxiliary table containing at least the key columns
+ * and the CDC metadata column.
+ * @return A dataframe containing the selected user columns followed by the CDC metadata column.
+ */
+ def reconcileMicrobatch(
+ changeArgs: ChangeArgs,
+ resolvedSequencingType: DataType,
+ validatedBatchDf: DataFrame,
+ auxiliaryTableDf: DataFrame): DataFrame
+
+ /**
+ * Appends CDC metadata to each microbatch row.
+ *
+ * This must run before column selection because the sequencing and delete expressions may
+ * reference columns that selection removes. A row is a delete only when the delete condition
+ * evaluates to true; a false or null result makes it an upsert.
+ */
+ protected[autocdc] final def extendMicrobatchRowsWithCdcMetadata(
+ changeArgs: ChangeArgs,
+ resolvedSequencingType: DataType,
+ validatedMicrobatch: DataFrame): DataFrame = {
+ val rowDeleteSequence: Column = changeArgs.deleteCondition match {
+ case Some(deleteCondition) =>
+ F.when(deleteCondition, changeArgs.sequencing).otherwise(F.lit(null))
Review Comment:
Optional cleanup: `F.when` already returns null for unmatched conditions, so `.otherwise(F.lit(null))` can be omitted here and in `rowUpsertSequence` below.
--
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-59645][SDP] SCD1 Ignore-null support; Refactor SCD1 microbatch reconciliation into strategies [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on PR #58915:
URL: https://github.com/apache/spark/pull/58915#issuecomment-5803586918
Merged to master, 4.x, thanks!
--
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