From 4666edc39ceff2d61f9ecfea8fa9e8b83b435215 Mon Sep 17 00:00:00 2001 From: Anish Mahto Date: Mon, 14 Sep 2026 21:45:43 +0000 Subject: [PATCH 1/6] add Scd1ReconciliationStrategy --- .../autocdc/Scd1BatchProcessor.scala | 43 ++----------- .../autocdc/Scd1ReconciliationStrategy.scala | 63 +++++++++++++++++++ 2 files changed, 69 insertions(+), 37 deletions(-) create mode 100644 sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd1ReconciliationStrategy.scala diff --git a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd1BatchProcessor.scala b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd1BatchProcessor.scala index 397775513703..9f9f5b7b89a3 100644 --- a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd1BatchProcessor.scala +++ b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd1BatchProcessor.scala @@ -33,49 +33,18 @@ import org.apache.spark.util.ArrayImplicits._ * @param changeArgs The CDC flow configuration. * @param resolvedSequencingType The post-analysis [[DataType]] of the sequencing column, derived * from the flow's resolved DataFrame at flow setup time. + * @param strategy Strategy used to reconcile the microbatch. */ case class Scd1BatchProcessor( changeArgs: ChangeArgs, - resolvedSequencingType: DataType) { + resolvedSequencingType: DataType, + strategy: Scd1ReconciliationStrategy = Scd1RowLevelReconciliation) { - /** - * Reconcile a CDC microbatch into the canonical form that the auxiliary- and target-table - * merges consume. Composes the per-step transforms in the only order that produces correct - * SCD1 semantics: - * - * 1. [[deduplicateMicrobatch]]: collapse same-key events to the latest by sequence. - * 2. [[extendMicrobatchRowsWithCdcMetadata]]: project the operational `_cdc_metadata` column - * (must run before column selection, which may drop inputs the metadata expressions - * reference). - * 3. [[projectTargetColumnsOntoMicrobatch]]: apply the user-defined column selection while - * preserving the CDC metadata column. - * 4. [[applyTombstonesToMicrobatch]]: filter out late-arriving events superseded by - * tombstones already recorded in the auxiliary table. - * - * The per-step methods are kept package-visible so that focused unit tests can pin each - * transform's behavior independently. This method itself is package-visible so that - * [[Scd1ForeachBatchHandler]] can call it after running [[ScdBatchValidator.validateMicrobatch]] - * - validation is intentionally not folded in here, as it must run before any of these - * transforms touch the data. - * - * @param batchDf The validated incoming CDC microbatch. - * @param auxiliaryTableDf A snapshot of the auxiliary table for tombstone reconciliation. - * Must contain at minimum the key columns + `_cdc_metadata`. - * @return The reconciled microbatch, ready to be merged onto both tables. - */ + /** Reconciles a CDC microbatch into the form consumed by the table merges. */ private[autocdc] def reconcileMicrobatch( batchDf: DataFrame, - auxiliaryTableDf: DataFrame): DataFrame = { - val deduplicated = deduplicateMicrobatch(validatedMicrobatch = batchDf) - val withCdcMetadata = extendMicrobatchRowsWithCdcMetadata(validatedMicrobatch = deduplicated) - val projected = projectTargetColumnsOntoMicrobatch( - microbatchWithCdcMetadataDf = withCdcMetadata - ) - applyTombstonesToMicrobatch( - microbatchDf = projected, - auxiliaryTableDf = auxiliaryTableDf - ) - } + auxiliaryTableDf: DataFrame): DataFrame = + strategy.reconcileMicrobatch(this, batchDf, auxiliaryTableDf) /** * Deduplicate the incoming CDC microbatch by key, keeping the most recent event per key diff --git a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd1ReconciliationStrategy.scala b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd1ReconciliationStrategy.scala new file mode 100644 index 000000000000..a4a4b7bcbb9a --- /dev/null +++ b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd1ReconciliationStrategy.scala @@ -0,0 +1,63 @@ +/* + * 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.classic.DataFrame + +/** Strategy for reconciling an SCD1 microbatch. */ +private[pipelines] trait Scd1ReconciliationStrategy { + + /** + * Resolves the CDC events for each key and removes events superseded by recorded tombstones. + * + * @param batchDf A validated CDC microbatch 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( + processor: Scd1BatchProcessor, + batchDf: DataFrame, + auxiliaryTableDf: DataFrame): DataFrame +} + +/** Row-level SCD1 reconciliation. */ +private[pipelines] object Scd1RowLevelReconciliation extends Scd1ReconciliationStrategy { + + /** + * Keeps the event with the greatest sequencing value for each key, adds its CDC metadata, + * applies the configured column selection, and removes events superseded by auxiliary-table + * tombstones. + */ + override def reconcileMicrobatch( + processor: Scd1BatchProcessor, + batchDf: DataFrame, + auxiliaryTableDf: DataFrame): DataFrame = { + val deduplicated = processor.deduplicateMicrobatch(validatedMicrobatch = batchDf) + val withCdcMetadata = + processor.extendMicrobatchRowsWithCdcMetadata(validatedMicrobatch = deduplicated) + val projected = processor.projectTargetColumnsOntoMicrobatch( + microbatchWithCdcMetadataDf = withCdcMetadata + ) + processor.applyTombstonesToMicrobatch( + microbatchDf = projected, + auxiliaryTableDf = auxiliaryTableDf + ) + } +} From b3840b4d410b11343a22188d0d76818206efc4a1 Mon Sep 17 00:00:00 2001 From: Anish Mahto Date: Wed, 16 Sep 2026 18:02:47 +0000 Subject: [PATCH 2/6] improve scaladoc --- .../spark/sql/pipelines/autocdc/Scd1BatchProcessor.scala | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd1BatchProcessor.scala b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd1BatchProcessor.scala index 9f9f5b7b89a3..e818070ed828 100644 --- a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd1BatchProcessor.scala +++ b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd1BatchProcessor.scala @@ -33,7 +33,11 @@ import org.apache.spark.util.ArrayImplicits._ * @param changeArgs The CDC flow configuration. * @param resolvedSequencingType The post-analysis [[DataType]] of the sequencing column, derived * from the flow's resolved DataFrame at flow setup time. - * @param strategy Strategy used to reconcile the microbatch. + * @param strategy Strategy used to reconcile the microbatch. In the default AutoCDC execution mode + * an event wins wholesale and all columns share its row-level version. Modes such + * as ignore-null however can reconcile leaves independently because different + * events may author them, and therefore require a different reconciliation + * strategy. */ case class Scd1BatchProcessor( changeArgs: ChangeArgs, From b95e0d95241156899ff87551a11d9b6d5afc75a9 Mon Sep 17 00:00:00 2001 From: Anish Mahto Date: Wed, 16 Sep 2026 23:54:34 +0000 Subject: [PATCH 3/6] move more functions --- .../autocdc/Scd1BatchProcessor.scala | 185 +----------------- .../autocdc/Scd1ReconciliationStrategy.scala | 156 ++++++++++++++- .../autocdc/Scd1BatchProcessorSuite.scala | 27 +++ 3 files changed, 181 insertions(+), 187 deletions(-) diff --git a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd1BatchProcessor.scala b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd1BatchProcessor.scala index e818070ed828..8fbde43afd02 100644 --- a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd1BatchProcessor.scala +++ b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd1BatchProcessor.scala @@ -48,180 +48,12 @@ case class Scd1BatchProcessor( private[autocdc] def reconcileMicrobatch( batchDf: DataFrame, auxiliaryTableDf: DataFrame): DataFrame = - strategy.reconcileMicrobatch(this, batchDf, auxiliaryTableDf) - - /** - * Deduplicate the incoming CDC microbatch by key, keeping the most recent event per key - * as ordered by [[ChangeArgs.sequencing]]. - * - * For SCD1 we only care about the most recent (by sequence value) event per key. When - * multiple events share the same key and the same sequence value, the row selected is - * non-deterministic and undefined. - * - * @param validatedMicrobatch A microbatch that has already been validated such that the - * sequencing column should not contain null values, and its data type - * should support ordering. - * - * The schema of the returned dataframe matches the schema of the microbatch exactly. - */ - private[autocdc] def deduplicateMicrobatch(validatedMicrobatch: DataFrame): DataFrame = { - // The `max_by` API can only return a single column, so pack/unpack the entire row into a - // temporary column before and after the `max_by` operation. - val winningRowCol = Scd1BatchProcessor.winningRowColName - - val allMicrobatchColumns = - validatedMicrobatch.columns - .map(colName => F.col(QuotingUtils.quoteIdentifier(colName))) - .toImmutableArraySeq - - validatedMicrobatch - .groupBy(changeArgs.keys.map(k => F.col(k.quoted)): _*) - .agg( - F.max_by(F.struct(allMicrobatchColumns: _*), changeArgs.sequencing) - .as(winningRowCol) - ) - .select(F.col(s"$winningRowCol.*")) - } - - /** - * Project the CDC metadata column onto the microbatch. - * - * This must run before any column selection is applied to the microbatch. The - * [[ChangeArgs.deleteCondition]] and [[ChangeArgs.sequencing]] expressions are evaluated against - * the current microbatch schema, and column selection may drop inputs required by those - * expressions. - * - * Rows are classified as deletes only when [[ChangeArgs.deleteCondition]] evaluates to true. A - * false or null delete condition classifies the row as an upsert. - * - * @param validatedMicrobatch A microbatch that has already been validated such that the - * sequencing column should not contain null values, and its data type - * should support ordering. - * - * The returned dataframe has all of the columns in the input microbatch + the CDC metadata - * column. - */ - private[autocdc] def extendMicrobatchRowsWithCdcMetadata( - validatedMicrobatch: DataFrame): DataFrame = { - val rowDeleteSequence: Column = changeArgs.deleteCondition match { - case Some(deleteCondition) => - F.when(deleteCondition, changeArgs.sequencing).otherwise(F.lit(null)) - case None => - F.lit(null) - } - - val rowUpsertSequence: Column = - // A row that is not a delete must be an upsert, these are mutually exclusive and a complete - // set of CDC event types. - F.when(rowDeleteSequence.isNull, changeArgs.sequencing).otherwise(F.lit(null)) - - validatedMicrobatch.withColumn( - AutoCdcReservedNames.cdcMetadataColName, - Scd1BatchProcessor.constructCdcMetadataCol( - deleteSequence = rowDeleteSequence, - upsertSequence = rowUpsertSequence, - sequencingType = resolvedSequencingType - ) - ) - } - - /** - * Project the user-defined column selection onto the microbatch. By this point the input - * microbatch should already have projected its CDC metadata, because it's possible that the - * user-defined column selection drops columns that are otherwise necessary to compute the - * CDC metadata. - * - * Returned dataframe's schema is: all of the user-selected columns in the input dataframe as per - * [[ChangeArgs.columnSelection]] + the CDC metadata column. - */ - private[autocdc] def projectTargetColumnsOntoMicrobatch( - microbatchWithCdcMetadataDf: DataFrame): DataFrame = { - val resolver = microbatchWithCdcMetadataDf.sparkSession.sessionState.conf.resolver - - // The user schema is the microbatch schema after dropping the system CDC metadata column. - // We project out the system column before applying user selection and project it back in - // afterwards, so that users cannot control whether this [necessary] column shows up in the - // target table. - val userColumnsInMicrobatchSchema = ColumnSelection.applyToSchema( - schemaName = "microbatch", - schema = microbatchWithCdcMetadataDf.schema, - columnSelection = Some( - ColumnSelection.ExcludeColumns( - Seq(UnqualifiedColumnName(AutoCdcReservedNames.cdcMetadataColName)) - ) - ), - resolver = resolver - ) - - val userSelectedColumnsInMicrobatchSchema = - ColumnSelection.applyToSchema( - schemaName = "microbatch", - schema = userColumnsInMicrobatchSchema, - columnSelection = changeArgs.columnSelection, - resolver = resolver - ) - - // In addition to the explicit user-selected columns, re-project the operational CDC metadata - // column as the last column. - val finalColumnsInMicrobatchToSelect = - userSelectedColumnsInMicrobatchSchema.fieldNames.map(colName => { - // Spark drops backticks in the schema, quote all identifiers for safety before executing - // select. Identifiers could have special characters such as '.'. - F.col(QuotingUtils.quoteIdentifier(colName)) - }) :+ F.col( - AutoCdcReservedNames.cdcMetadataColName - ) - - microbatchWithCdcMetadataDf.select( - finalColumnsInMicrobatchToSelect.toImmutableArraySeq: _* + strategy.reconcileMicrobatch( + changeArgs = changeArgs, + resolvedSequencingType = resolvedSequencingType, + batchDf = batchDf, + auxiliaryTableDf = auxiliaryTableDf ) - } - - /** - * Left anti-join the microbatch with the auxiliary table on tombstones that match against and - * effectively delete late-arriving upserts (or stale deletes). - * - * @param microbatchDf The incoming microbatch dataframe with at minimum all of the key - * columns + CDC metadata column. - * @param auxiliaryTableDf Dataframe representing the auxiliary table, with at minimum the key - * columns + CDC metadata column. - * - * The returned filtered dataframe has the same schema as the input microbatch, but with only - * the rows that remain unaffected by any known tombstones. - */ - private[autocdc] def applyTombstonesToMicrobatch( - microbatchDf: DataFrame, - auxiliaryTableDf: DataFrame): DataFrame = { - val aliasedMicrobatchDf = microbatchDf.alias("microbatch") - val aliasedAuxiliaryTableDf = auxiliaryTableDf.alias("auxiliaryTable") - - val cdcMetadata = AutoCdcReservedNames.cdcMetadataColName - - val microbatchCdcMetadata = F.col(s"microbatch.$cdcMetadata") - val effectiveSeq = F.greatest( - Scd1BatchProcessor.deleteSequenceOf(microbatchCdcMetadata), - Scd1BatchProcessor.upsertSequenceOf(microbatchCdcMetadata) - ) - val tombstoneDeleteSeq = - Scd1BatchProcessor.deleteSequenceOf(F.col(s"auxiliaryTable.$cdcMetadata")) - - val keysMatch = changeArgs.keys - .map { k => - F.col(s"microbatch.${k.quoted}") === F.col(s"auxiliaryTable.${k.quoted}") - } - .reduce(_ && _) - - // A microbatch row is considered late-arriving (and therefore deleted by the tombstone) when - // the auxiliary table holds a tombstone for the same key with a strictly larger delete - // sequence. Both late-arriving upserts and deletes are dropped. - val microbatchRowDeletedByTombstone = effectiveSeq < tombstoneDeleteSeq - - aliasedMicrobatchDf.join( - right = aliasedAuxiliaryTableDf, - joinExprs = keysMatch && microbatchRowDeletedByTombstone, - joinType = "left_anti" - ) - } /** * Merge the reconciled (deduplicated per key) microbatch onto the auxiliary table, @@ -381,13 +213,6 @@ case class Scd1BatchProcessor( } object Scd1BatchProcessor { - /** - * Internal columns inserted by AutoCDC reconciliation. Source change-data-feed dataframes must - * not contain any columns starting with [[AutoCdcReservedNames.prefix]]; the invariant is - * enforced at [[org.apache.spark.sql.pipelines.graph.AutoCdcMergeFlow]] construction. - */ - private[autocdc] val winningRowColName: String = s"${AutoCdcReservedNames.prefix}winning_row" - private[pipelines] val cdcDeleteSequenceFieldName: String = "deleteSequence" private[pipelines] val cdcUpsertSequenceFieldName: String = "upsertSequence" diff --git a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd1ReconciliationStrategy.scala b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd1ReconciliationStrategy.scala index a4a4b7bcbb9a..10bb971161ab 100644 --- a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd1ReconciliationStrategy.scala +++ b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd1ReconciliationStrategy.scala @@ -17,7 +17,12 @@ 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 { @@ -32,32 +37,169 @@ private[pipelines] trait Scd1ReconciliationStrategy { * @return A dataframe containing the selected user columns followed by the CDC metadata column. */ def reconcileMicrobatch( - processor: Scd1BatchProcessor, + changeArgs: ChangeArgs, + resolvedSequencingType: DataType, batchDf: 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)) + case None => + F.lit(null) + } + + val rowUpsertSequence: Column = + // A row that is not a delete must be an upsert, these are mutually exclusive and a complete + // set of CDC event types. + F.when(rowDeleteSequence.isNull, changeArgs.sequencing).otherwise(F.lit(null)) + + validatedMicrobatch.withColumn( + AutoCdcReservedNames.cdcMetadataColName, + Scd1BatchProcessor.constructCdcMetadataCol( + deleteSequence = rowDeleteSequence, + upsertSequence = rowUpsertSequence, + sequencingType = resolvedSequencingType + ) + ) + } + + /** + * Applies the user-defined column selection while preserving the CDC metadata column. + * + * Requires CDC metadata to be present because selection may remove columns used to construct it. + */ + protected[autocdc] final def projectTargetColumnsOntoMicrobatch( + changeArgs: ChangeArgs, + microbatchWithCdcMetadataDf: DataFrame): DataFrame = { + val resolver = microbatchWithCdcMetadataDf.sparkSession.sessionState.conf.resolver + val userColumnsInMicrobatchSchema = ColumnSelection.applyToSchema( + schemaName = "microbatch", + schema = microbatchWithCdcMetadataDf.schema, + columnSelection = Some( + ColumnSelection.ExcludeColumns( + Seq(UnqualifiedColumnName(AutoCdcReservedNames.cdcMetadataColName)) + ) + ), + resolver = resolver + ) + val userSelectedColumnsInMicrobatchSchema = ColumnSelection.applyToSchema( + schemaName = "microbatch", + schema = userColumnsInMicrobatchSchema, + columnSelection = changeArgs.columnSelection, + resolver = resolver + ) + val finalColumnsInMicrobatchToSelect = + userSelectedColumnsInMicrobatchSchema.fieldNames.map { columnName => + F.col(QuotingUtils.quoteIdentifier(columnName)) + } :+ F.col(AutoCdcReservedNames.cdcMetadataColName) + + microbatchWithCdcMetadataDf.select( + finalColumnsInMicrobatchToSelect.toImmutableArraySeq: _* + ) + } } /** Row-level SCD1 reconciliation. */ private[pipelines] object Scd1RowLevelReconciliation extends Scd1ReconciliationStrategy { + private val winningRowColName: String = s"${AutoCdcReservedNames.prefix}winning_row" + /** * Keeps the event with the greatest sequencing value for each key, adds its CDC metadata, * applies the configured column selection, and removes events superseded by auxiliary-table * tombstones. */ override def reconcileMicrobatch( - processor: Scd1BatchProcessor, + changeArgs: ChangeArgs, + resolvedSequencingType: DataType, batchDf: DataFrame, auxiliaryTableDf: DataFrame): DataFrame = { - val deduplicated = processor.deduplicateMicrobatch(validatedMicrobatch = batchDf) - val withCdcMetadata = - processor.extendMicrobatchRowsWithCdcMetadata(validatedMicrobatch = deduplicated) - val projected = processor.projectTargetColumnsOntoMicrobatch( + val deduplicated = deduplicateMicrobatch( + changeArgs = changeArgs, + validatedMicrobatch = batchDf + ) + val withCdcMetadata = extendMicrobatchRowsWithCdcMetadata( + changeArgs = changeArgs, + resolvedSequencingType = resolvedSequencingType, + validatedMicrobatch = deduplicated + ) + val projected = projectTargetColumnsOntoMicrobatch( + changeArgs = changeArgs, microbatchWithCdcMetadataDf = withCdcMetadata ) - processor.applyTombstonesToMicrobatch( + applyTombstonesToMicrobatch( + changeArgs = changeArgs, microbatchDf = projected, auxiliaryTableDf = auxiliaryTableDf ) } + + /** + * Deduplicates the microbatch by key, keeping the event with the greatest sequencing value. + * + * Selection between events with equal keys and sequencing values is undefined. + */ + private[autocdc] def deduplicateMicrobatch( + changeArgs: ChangeArgs, + validatedMicrobatch: DataFrame): DataFrame = { + val allMicrobatchColumns = + validatedMicrobatch.columns + .map(colName => F.col(QuotingUtils.quoteIdentifier(colName))) + .toImmutableArraySeq + + validatedMicrobatch + .groupBy(changeArgs.keys.map(k => F.col(k.quoted)): _*) + .agg( + F.max_by(F.struct(allMicrobatchColumns: _*), changeArgs.sequencing) + .as(winningRowColName) + ) + .select(F.col(s"$winningRowColName.*")) + } + + /** + * Left anti-joins the microbatch with matching auxiliary-table tombstones that have greater + * sequencing values. + */ + private[autocdc] def applyTombstonesToMicrobatch( + changeArgs: ChangeArgs, + microbatchDf: DataFrame, + auxiliaryTableDf: DataFrame): DataFrame = { + val aliasedMicrobatchDf = microbatchDf.alias("microbatch") + val aliasedAuxiliaryTableDf = auxiliaryTableDf.alias("auxiliaryTable") + + val cdcMetadata = AutoCdcReservedNames.cdcMetadataColName + val microbatchCdcMetadata = F.col(s"microbatch.$cdcMetadata") + val effectiveSeq = F.greatest( + Scd1BatchProcessor.deleteSequenceOf(microbatchCdcMetadata), + Scd1BatchProcessor.upsertSequenceOf(microbatchCdcMetadata) + ) + val tombstoneDeleteSeq = + Scd1BatchProcessor.deleteSequenceOf(F.col(s"auxiliaryTable.$cdcMetadata")) + + val keysMatch = changeArgs.keys + .map { key => + F.col(s"microbatch.${key.quoted}") === F.col(s"auxiliaryTable.${key.quoted}") + } + .reduce(_ && _) + + val microbatchRowDeletedByTombstone = effectiveSeq < tombstoneDeleteSeq + + aliasedMicrobatchDf.join( + right = aliasedAuxiliaryTableDf, + joinExprs = keysMatch && microbatchRowDeletedByTombstone, + joinType = "left_anti" + ) + } } diff --git a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd1BatchProcessorSuite.scala b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd1BatchProcessorSuite.scala index d2c78442c476..1b69adf7932e 100644 --- a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd1BatchProcessorSuite.scala +++ b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd1BatchProcessorSuite.scala @@ -63,6 +63,33 @@ class Scd1BatchProcessorSuite extends QueryTest with SharedSparkSession { private def columnNamesAndDataTypes(schema: StructType): Seq[(String, DataType)] = schema.fields.map(f => (f.name, f.dataType)).toSeq + private implicit class RowLevelReconciliationOps(processor: Scd1BatchProcessor) { + def deduplicateMicrobatch(batch: DataFrame): DataFrame = + Scd1RowLevelReconciliation.deduplicateMicrobatch(processor.changeArgs, batch) + + def extendMicrobatchRowsWithCdcMetadata(batch: DataFrame): DataFrame = + Scd1RowLevelReconciliation.extendMicrobatchRowsWithCdcMetadata( + changeArgs = processor.changeArgs, + resolvedSequencingType = processor.resolvedSequencingType, + validatedMicrobatch = batch + ) + + def projectTargetColumnsOntoMicrobatch(batch: DataFrame): DataFrame = + Scd1RowLevelReconciliation.projectTargetColumnsOntoMicrobatch( + changeArgs = processor.changeArgs, + microbatchWithCdcMetadataDf = batch + ) + + def applyTombstonesToMicrobatch( + microbatch: DataFrame, + auxiliary: DataFrame): DataFrame = + Scd1RowLevelReconciliation.applyTombstonesToMicrobatch( + changeArgs = processor.changeArgs, + microbatchDf = microbatch, + auxiliaryTableDf = auxiliary + ) + } + // =============== deduplicateMicrobatch tests =============== test("deduplicateMicrobatch keeps only the row with the largest sequence value per key") { From f778a0c299c87b1fb164e0f862bc4885b8249a20 Mon Sep 17 00:00:00 2001 From: Anish Mahto Date: Fri, 18 Sep 2026 17:57:19 +0000 Subject: [PATCH 4/6] rename/refactor test suite --- .../autocdc/Scd1BatchProcessor.scala | 14 +- .../Scd1BatchProcessorMergeSuite.scala | 2 +- ... => Scd1RowLevelReconciliationSuite.scala} | 624 ++++++++---------- 3 files changed, 269 insertions(+), 371 deletions(-) rename sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/{Scd1BatchProcessorSuite.scala => Scd1RowLevelReconciliationSuite.scala} (65%) diff --git a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd1BatchProcessor.scala b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd1BatchProcessor.scala index 8fbde43afd02..d3609135e23b 100644 --- a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd1BatchProcessor.scala +++ b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd1BatchProcessor.scala @@ -33,22 +33,22 @@ import org.apache.spark.util.ArrayImplicits._ * @param changeArgs The CDC flow configuration. * @param resolvedSequencingType The post-analysis [[DataType]] of the sequencing column, derived * from the flow's resolved DataFrame at flow setup time. - * @param strategy Strategy used to reconcile the microbatch. In the default AutoCDC execution mode - * an event wins wholesale and all columns share its row-level version. Modes such - * as ignore-null however can reconcile leaves independently because different - * events may author them, and therefore require a different reconciliation - * strategy. + * @param reconciliationStrategy Strategy used to reconcile the microbatch. In the default AutoCDC + * execution mode an event wins wholesale and all columns share its + * row-level version. Modes such as ignore-null can reconcile leaves + * independently because different events may author them, and + * therefore require a different reconciliation strategy. */ case class Scd1BatchProcessor( changeArgs: ChangeArgs, resolvedSequencingType: DataType, - strategy: Scd1ReconciliationStrategy = Scd1RowLevelReconciliation) { + reconciliationStrategy: Scd1ReconciliationStrategy = Scd1RowLevelReconciliation) { /** Reconciles a CDC microbatch into the form consumed by the table merges. */ private[autocdc] def reconcileMicrobatch( batchDf: DataFrame, auxiliaryTableDf: DataFrame): DataFrame = - strategy.reconcileMicrobatch( + reconciliationStrategy.reconcileMicrobatch( changeArgs = changeArgs, resolvedSequencingType = resolvedSequencingType, batchDf = batchDf, diff --git a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd1BatchProcessorMergeSuite.scala b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd1BatchProcessorMergeSuite.scala index 1aa2cbcd5417..9cfe42267f8b 100644 --- a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd1BatchProcessorMergeSuite.scala +++ b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd1BatchProcessorMergeSuite.scala @@ -30,7 +30,7 @@ import org.apache.spark.sql.types._ * v2 table. These tests require a v2 catalog that supports row-level operations * (set up by [[AutoCdcCatalogExecutionTestBase]]) and run actual writes through Catalyst's * row-level-operations machinery, so they are kept separate from the pure-DataFrame-transform - * tests in [[Scd1BatchProcessorSuite]]. + * tests in [[Scd1RowLevelReconciliationSuite]]. */ class Scd1BatchProcessorMergeSuite extends QueryTest diff --git a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd1BatchProcessorSuite.scala b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd1RowLevelReconciliationSuite.scala similarity index 65% rename from sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd1BatchProcessorSuite.scala rename to sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd1RowLevelReconciliationSuite.scala index 1b69adf7932e..db9275d70efa 100644 --- a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd1BatchProcessorSuite.scala +++ b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd1RowLevelReconciliationSuite.scala @@ -23,7 +23,7 @@ import org.apache.spark.sql.internal.SQLConf import org.apache.spark.sql.test.SharedSparkSession import org.apache.spark.sql.types._ -class Scd1BatchProcessorSuite extends QueryTest with SharedSparkSession { +class Scd1RowLevelReconciliationSuite extends QueryTest with SharedSparkSession { /** * Test Schema for a microbatch that already has the SCD1 CDC metadata column projected. @@ -50,7 +50,6 @@ class Scd1BatchProcessorSuite extends QueryTest with SharedSparkSession { private def cdcMetadataRow(deleteSeq: Option[Long], upsertSeq: Option[Long]): Row = Row(deleteSeq.getOrElse(null), upsertSeq.getOrElse(null)) - /** Build a microbatch [[DataFrame]] from explicit rows and an explicit schema. */ private def microbatchOf(schema: StructType)(rows: Row*): DataFrame = spark.createDataFrame(spark.sparkContext.parallelize(rows), schema) @@ -63,32 +62,37 @@ class Scd1BatchProcessorSuite extends QueryTest with SharedSparkSession { private def columnNamesAndDataTypes(schema: StructType): Seq[(String, DataType)] = schema.fields.map(f => (f.name, f.dataType)).toSeq - private implicit class RowLevelReconciliationOps(processor: Scd1BatchProcessor) { - def deduplicateMicrobatch(batch: DataFrame): DataFrame = - Scd1RowLevelReconciliation.deduplicateMicrobatch(processor.changeArgs, batch) + private def deduplicateMicrobatch( + changeArgs: ChangeArgs, + batch: DataFrame): DataFrame = + Scd1RowLevelReconciliation.deduplicateMicrobatch(changeArgs, batch) - def extendMicrobatchRowsWithCdcMetadata(batch: DataFrame): DataFrame = - Scd1RowLevelReconciliation.extendMicrobatchRowsWithCdcMetadata( - changeArgs = processor.changeArgs, - resolvedSequencingType = processor.resolvedSequencingType, - validatedMicrobatch = batch - ) + private def extendMicrobatchRowsWithCdcMetadata( + changeArgs: ChangeArgs, + batch: DataFrame): DataFrame = + Scd1RowLevelReconciliation.extendMicrobatchRowsWithCdcMetadata( + changeArgs = changeArgs, + resolvedSequencingType = LongType, + validatedMicrobatch = batch + ) - def projectTargetColumnsOntoMicrobatch(batch: DataFrame): DataFrame = - Scd1RowLevelReconciliation.projectTargetColumnsOntoMicrobatch( - changeArgs = processor.changeArgs, - microbatchWithCdcMetadataDf = batch - ) + private def projectTargetColumnsOntoMicrobatch( + changeArgs: ChangeArgs, + batch: DataFrame): DataFrame = + Scd1RowLevelReconciliation.projectTargetColumnsOntoMicrobatch( + changeArgs = changeArgs, + microbatchWithCdcMetadataDf = batch + ) - def applyTombstonesToMicrobatch( - microbatch: DataFrame, - auxiliary: DataFrame): DataFrame = - Scd1RowLevelReconciliation.applyTombstonesToMicrobatch( - changeArgs = processor.changeArgs, - microbatchDf = microbatch, - auxiliaryTableDf = auxiliary - ) - } + private def applyTombstonesToMicrobatch( + changeArgs: ChangeArgs, + microbatch: DataFrame, + auxiliary: DataFrame): DataFrame = + Scd1RowLevelReconciliation.applyTombstonesToMicrobatch( + changeArgs = changeArgs, + microbatchDf = microbatch, + auxiliaryTableDf = auxiliary + ) // =============== deduplicateMicrobatch tests =============== @@ -104,17 +108,14 @@ class Scd1BatchProcessorSuite extends QueryTest with SharedSparkSession { Row(1, 20L, "middle") ) - val processor = Scd1BatchProcessor( - changeArgs = ChangeArgs( - keys = Seq(UnqualifiedColumnName("id")), - sequencing = F.col("seq"), - storedAsScdType = ScdType.Type1 - ), - resolvedSequencingType = LongType + val changeArgs = ChangeArgs( + keys = Seq(UnqualifiedColumnName("id")), + sequencing = F.col("seq"), + storedAsScdType = ScdType.Type1 ) checkAnswer( - df = processor.deduplicateMicrobatch(batch), + df = deduplicateMicrobatch(changeArgs, batch), expectedAnswer = Row(1, 30L, "winner") ) } @@ -129,17 +130,14 @@ class Scd1BatchProcessorSuite extends QueryTest with SharedSparkSession { Row(1, 10L, "only-row") ) - val processor = Scd1BatchProcessor( - changeArgs = ChangeArgs( - keys = Seq(UnqualifiedColumnName("id")), - sequencing = F.col("seq"), - storedAsScdType = ScdType.Type1 - ), - resolvedSequencingType = LongType + val changeArgs = ChangeArgs( + keys = Seq(UnqualifiedColumnName("id")), + sequencing = F.col("seq"), + storedAsScdType = ScdType.Type1 ) checkAnswer( - df = processor.deduplicateMicrobatch(batch), + df = deduplicateMicrobatch(changeArgs, batch), expectedAnswer = Row(1, 10L, "only-row") ) } @@ -155,18 +153,15 @@ class Scd1BatchProcessorSuite extends QueryTest with SharedSparkSession { Row(1, 10L, "second-tied-row") ) - val processor = Scd1BatchProcessor( - changeArgs = ChangeArgs( - keys = Seq(UnqualifiedColumnName("id")), - sequencing = F.col("seq"), - storedAsScdType = ScdType.Type1 - ), - resolvedSequencingType = LongType + val changeArgs = ChangeArgs( + keys = Seq(UnqualifiedColumnName("id")), + sequencing = F.col("seq"), + storedAsScdType = ScdType.Type1 ) // On equal sequence number events for the same key we provide no guarantee on which event will // survive, but the contract is _one_ event will survive - assert that below. - val result = processor.deduplicateMicrobatch(batch).collect() + val result = deduplicateMicrobatch(changeArgs, batch).collect() assert(result.length == 1) assert(result.head.getInt(0) == 1) assert(result.head.getLong(1) == 10L) @@ -187,17 +182,14 @@ class Scd1BatchProcessorSuite extends QueryTest with SharedSparkSession { Row(1, 10L, "non-null-sequence") ) - val processor = Scd1BatchProcessor( - changeArgs = ChangeArgs( - keys = Seq(UnqualifiedColumnName("id")), - sequencing = F.col("seq"), - storedAsScdType = ScdType.Type1 - ), - resolvedSequencingType = LongType + val changeArgs = ChangeArgs( + keys = Seq(UnqualifiedColumnName("id")), + sequencing = F.col("seq"), + storedAsScdType = ScdType.Type1 ) checkAnswer( - df = processor.deduplicateMicrobatch(batch), + df = deduplicateMicrobatch(changeArgs, batch), expectedAnswer = Row(1, 10L, "non-null-sequence") ) } @@ -215,16 +207,13 @@ class Scd1BatchProcessorSuite extends QueryTest with SharedSparkSession { // deduplication if all rows contain a null sequence in the microbatch. Row(1, null, "null-sequence") ) - val processor = Scd1BatchProcessor( - changeArgs = ChangeArgs( - keys = Seq(UnqualifiedColumnName("id")), - sequencing = F.col("seq"), - storedAsScdType = ScdType.Type1 - ), - resolvedSequencingType = LongType + val changeArgs = ChangeArgs( + keys = Seq(UnqualifiedColumnName("id")), + sequencing = F.col("seq"), + storedAsScdType = ScdType.Type1 ) checkAnswer( - df = processor.deduplicateMicrobatch(batch), + df = deduplicateMicrobatch(changeArgs, batch), expectedAnswer = Row(null, null, null) ) } @@ -243,17 +232,14 @@ class Scd1BatchProcessorSuite extends QueryTest with SharedSparkSession { Row(3, 1L, "c1-only") ) - val processor = Scd1BatchProcessor( - changeArgs = ChangeArgs( - keys = Seq(UnqualifiedColumnName("id")), - sequencing = F.col("seq"), - storedAsScdType = ScdType.Type1 - ), - resolvedSequencingType = LongType + val changeArgs = ChangeArgs( + keys = Seq(UnqualifiedColumnName("id")), + sequencing = F.col("seq"), + storedAsScdType = ScdType.Type1 ) checkAnswer( - df = processor.deduplicateMicrobatch(batch), + df = deduplicateMicrobatch(changeArgs, batch), expectedAnswer = Seq( Row(1, 20L, "a2-winner"), Row(2, 50L, "b1-winner"), @@ -274,19 +260,16 @@ class Scd1BatchProcessorSuite extends QueryTest with SharedSparkSession { Row(1, 20L, "winning-name", 200) ) - val processor = Scd1BatchProcessor( - changeArgs = ChangeArgs( - keys = Seq(UnqualifiedColumnName("id")), - sequencing = F.col("seq"), - storedAsScdType = ScdType.Type1 - ), - resolvedSequencingType = LongType + val changeArgs = ChangeArgs( + keys = Seq(UnqualifiedColumnName("id")), + sequencing = F.col("seq"), + storedAsScdType = ScdType.Type1 ) // All non-key columns must come from the row with the largest sequence value, never // a mix of values from multiple rows. checkAnswer( - df = processor.deduplicateMicrobatch(batch), + df = deduplicateMicrobatch(changeArgs, batch), expectedAnswer = Row(1, 20L, "winning-name", 200) ) } @@ -305,17 +288,14 @@ class Scd1BatchProcessorSuite extends QueryTest with SharedSparkSession { Row(1, 20L, Row("new", 200)) ) - val processor = Scd1BatchProcessor( - changeArgs = ChangeArgs( - keys = Seq(UnqualifiedColumnName("id")), - sequencing = F.col("seq"), - storedAsScdType = ScdType.Type1 - ), - resolvedSequencingType = LongType + val changeArgs = ChangeArgs( + keys = Seq(UnqualifiedColumnName("id")), + sequencing = F.col("seq"), + storedAsScdType = ScdType.Type1 ) checkAnswer( - df = processor.deduplicateMicrobatch(batch), + df = deduplicateMicrobatch(changeArgs, batch), expectedAnswer = Row(1, 20L, Row("new", 200)) ) } @@ -336,17 +316,14 @@ class Scd1BatchProcessorSuite extends QueryTest with SharedSparkSession { Row("US", 2, 99L, "us2-only") ) - val processor = Scd1BatchProcessor( - changeArgs = ChangeArgs( - keys = Seq(UnqualifiedColumnName("region"), UnqualifiedColumnName("customer_id")), - sequencing = F.col("seq"), - storedAsScdType = ScdType.Type1 - ), - resolvedSequencingType = LongType + val changeArgs = ChangeArgs( + keys = Seq(UnqualifiedColumnName("region"), UnqualifiedColumnName("customer_id")), + sequencing = F.col("seq"), + storedAsScdType = ScdType.Type1 ) checkAnswer( - df = processor.deduplicateMicrobatch(batch), + df = deduplicateMicrobatch(changeArgs, batch), expectedAnswer = Seq( Row("US", 1, 20L, "us1-new"), Row("EU", 1, 5L, "eu1-only"), @@ -373,17 +350,14 @@ class Scd1BatchProcessorSuite extends QueryTest with SharedSparkSession { Row(1, 15L, 15L, "always-loses") ) - val processor = Scd1BatchProcessor( - changeArgs = ChangeArgs( - keys = Seq(UnqualifiedColumnName("id")), - sequencing = F.greatest(F.col("seq"), F.col("alt_seq")), - storedAsScdType = ScdType.Type1 - ), - resolvedSequencingType = LongType + val changeArgs = ChangeArgs( + keys = Seq(UnqualifiedColumnName("id")), + sequencing = F.greatest(F.col("seq"), F.col("alt_seq")), + storedAsScdType = ScdType.Type1 ) checkAnswer( - df = processor.deduplicateMicrobatch(batch), + df = deduplicateMicrobatch(changeArgs, batch), expectedAnswer = Row(1, 10L, 30L, "winner") ) } @@ -399,17 +373,14 @@ class Scd1BatchProcessorSuite extends QueryTest with SharedSparkSession { Row(1, 20L, "new") ) - val processor = Scd1BatchProcessor( - changeArgs = ChangeArgs( - keys = Seq(UnqualifiedColumnName("`user.id`")), - sequencing = F.col("seq"), - storedAsScdType = ScdType.Type1 - ), - resolvedSequencingType = LongType + val changeArgs = ChangeArgs( + keys = Seq(UnqualifiedColumnName("`user.id`")), + sequencing = F.col("seq"), + storedAsScdType = ScdType.Type1 ) checkAnswer( - df = processor.deduplicateMicrobatch(batch), + df = deduplicateMicrobatch(changeArgs, batch), expectedAnswer = Row(1, 20L, "new") ) } @@ -417,7 +388,7 @@ class Scd1BatchProcessorSuite extends QueryTest with SharedSparkSession { test( "deduplicateMicrobatch fails when a key column collides with the reserved name" ) { - val reservedColName = Scd1BatchProcessor.winningRowColName + val reservedColName = Scd1RowLevelReconciliation.winningRowColName val schema = new StructType() .add(reservedColName, StringType) @@ -429,18 +400,15 @@ class Scd1BatchProcessorSuite extends QueryTest with SharedSparkSession { Row("k1", 20L, "winner") ) - val processor = Scd1BatchProcessor( - changeArgs = ChangeArgs( - keys = Seq(UnqualifiedColumnName(reservedColName)), - sequencing = F.col("seq"), - storedAsScdType = ScdType.Type1 - ), - resolvedSequencingType = LongType + val changeArgs = ChangeArgs( + keys = Seq(UnqualifiedColumnName(reservedColName)), + sequencing = F.col("seq"), + storedAsScdType = ScdType.Type1 ) checkError( exception = intercept[AnalysisException] { - processor.deduplicateMicrobatch(batch).collect() + deduplicateMicrobatch(changeArgs, batch).collect() }, condition = "AMBIGUOUS_REFERENCE", sqlState = "42704", @@ -465,18 +433,15 @@ class Scd1BatchProcessorSuite extends QueryTest with SharedSparkSession { Row("a2", 1, 2.5, 20L, false) ) - val processor = Scd1BatchProcessor( - changeArgs = ChangeArgs( - keys = Seq(UnqualifiedColumnName("id")), - sequencing = F.col("seq"), - storedAsScdType = ScdType.Type1 - ), - resolvedSequencingType = LongType + val changeArgs = ChangeArgs( + keys = Seq(UnqualifiedColumnName("id")), + sequencing = F.col("seq"), + storedAsScdType = ScdType.Type1 ) // Field names and dataTypes must match the input exactly, in the original order. assert( - columnNamesAndDataTypes(processor.deduplicateMicrobatch(batch).schema) == + columnNamesAndDataTypes(deduplicateMicrobatch(changeArgs, batch).schema) == columnNamesAndDataTypes(schema)) } @@ -488,16 +453,13 @@ class Scd1BatchProcessorSuite extends QueryTest with SharedSparkSession { val batch = microbatchOf(schema)() - val processor = Scd1BatchProcessor( - changeArgs = ChangeArgs( - keys = Seq(UnqualifiedColumnName("id")), - sequencing = F.col("seq"), - storedAsScdType = ScdType.Type1 - ), - resolvedSequencingType = LongType + val changeArgs = ChangeArgs( + keys = Seq(UnqualifiedColumnName("id")), + sequencing = F.col("seq"), + storedAsScdType = ScdType.Type1 ) - val result = processor.deduplicateMicrobatch(batch) + val result = deduplicateMicrobatch(changeArgs, batch) assert(result.collect().isEmpty) assert(columnNamesAndDataTypes(result.schema) == columnNamesAndDataTypes(schema)) } @@ -518,21 +480,18 @@ class Scd1BatchProcessorSuite extends QueryTest with SharedSparkSession { Row(4, 40L, true) ) - val processor = Scd1BatchProcessor( - changeArgs = ChangeArgs( - keys = Seq(UnqualifiedColumnName("id")), - sequencing = F.col("seq"), - storedAsScdType = ScdType.Type1, - deleteCondition = Some(F.col("is_delete") === true) - ), - resolvedSequencingType = LongType + val changeArgs = ChangeArgs( + keys = Seq(UnqualifiedColumnName("id")), + sequencing = F.col("seq"), + storedAsScdType = ScdType.Type1, + deleteCondition = Some(F.col("is_delete") === true) ) // Mutual-exclusivity invariant: each row's CDC metadata struct has exactly one of // (deleteSequence, upsertSequence) non-null, and the non-null side carries the row's // sequence value. checkAnswer( - df = processor.extendMicrobatchRowsWithCdcMetadata(batch), + df = extendMicrobatchRowsWithCdcMetadata(changeArgs, batch), expectedAnswer = Seq( Row(1, 10L, false, Row(null, 10L)), Row(2, 20L, true, Row(20L, null)), @@ -552,18 +511,15 @@ class Scd1BatchProcessorSuite extends QueryTest with SharedSparkSession { Row(1, 10L, null) ) - val processor = Scd1BatchProcessor( - changeArgs = ChangeArgs( - keys = Seq(UnqualifiedColumnName("id")), - sequencing = F.col("seq"), - storedAsScdType = ScdType.Type1, - deleteCondition = Some(F.col("is_delete")) - ), - resolvedSequencingType = LongType + val changeArgs = ChangeArgs( + keys = Seq(UnqualifiedColumnName("id")), + sequencing = F.col("seq"), + storedAsScdType = ScdType.Type1, + deleteCondition = Some(F.col("is_delete")) ) checkAnswer( - df = processor.extendMicrobatchRowsWithCdcMetadata(batch), + df = extendMicrobatchRowsWithCdcMetadata(changeArgs, batch), expectedAnswer = Row(1, 10L, null, Row(null, 10L)) ) } @@ -580,18 +536,15 @@ class Scd1BatchProcessorSuite extends QueryTest with SharedSparkSession { Row(2, 20L, "b") ) - val processor = Scd1BatchProcessor( - changeArgs = ChangeArgs( - keys = Seq(UnqualifiedColumnName("id")), - sequencing = F.col("seq"), - storedAsScdType = ScdType.Type1, - deleteCondition = None - ), - resolvedSequencingType = LongType + val changeArgs = ChangeArgs( + keys = Seq(UnqualifiedColumnName("id")), + sequencing = F.col("seq"), + storedAsScdType = ScdType.Type1, + deleteCondition = None ) checkAnswer( - df = processor.extendMicrobatchRowsWithCdcMetadata(batch), + df = extendMicrobatchRowsWithCdcMetadata(changeArgs, batch), expectedAnswer = Seq( Row(1, 10L, "a", Row(null, 10L)), Row(2, 20L, "b", Row(null, 20L)) @@ -609,16 +562,13 @@ class Scd1BatchProcessorSuite extends QueryTest with SharedSparkSession { Row(1, 10L, "a") ) - val processor = Scd1BatchProcessor( - changeArgs = ChangeArgs( - keys = Seq(UnqualifiedColumnName("id")), - sequencing = F.col("seq"), - storedAsScdType = ScdType.Type1 - ), - resolvedSequencingType = LongType + val changeArgs = ChangeArgs( + keys = Seq(UnqualifiedColumnName("id")), + sequencing = F.col("seq"), + storedAsScdType = ScdType.Type1 ) - val result = processor.extendMicrobatchRowsWithCdcMetadata(batch) + val result = extendMicrobatchRowsWithCdcMetadata(changeArgs, batch) // Original columns are preserved in their original order, with CDC metadata appended at // the very end. @@ -639,16 +589,13 @@ class Scd1BatchProcessorSuite extends QueryTest with SharedSparkSession { Row(1, 10, "a") ) - val processor = Scd1BatchProcessor( - changeArgs = ChangeArgs( - keys = Seq(UnqualifiedColumnName("id")), - sequencing = F.col("seq"), - storedAsScdType = ScdType.Type1 - ), - resolvedSequencingType = LongType + val changeArgs = ChangeArgs( + keys = Seq(UnqualifiedColumnName("id")), + sequencing = F.col("seq"), + storedAsScdType = ScdType.Type1 ) - val resultDf = processor.extendMicrobatchRowsWithCdcMetadata(batch) + val resultDf = extendMicrobatchRowsWithCdcMetadata(changeArgs, batch) val cdcMetadataDataType = resultDf.schema(AutoCdcReservedNames.cdcMetadataColName).dataType.asInstanceOf[StructType] @@ -680,18 +627,15 @@ class Scd1BatchProcessorSuite extends QueryTest with SharedSparkSession { Row(1, Row(1L, 0L)) ) - val processor = Scd1BatchProcessor( - changeArgs = ChangeArgs( - keys = Seq(UnqualifiedColumnName("id")), - sequencing = F.col("seq"), - storedAsScdType = ScdType.Type1 - ), - resolvedSequencingType = LongType + val changeArgs = ChangeArgs( + keys = Seq(UnqualifiedColumnName("id")), + sequencing = F.col("seq"), + storedAsScdType = ScdType.Type1 ) val ex = intercept[AnalysisException] { // .schema forces analysis of the underlying logical plan, surfacing the invalid cast. - processor.extendMicrobatchRowsWithCdcMetadata(batch).schema + extendMicrobatchRowsWithCdcMetadata(changeArgs, batch).schema } assert(ex.getCondition == "DATATYPE_MISMATCH.CAST_WITHOUT_SUGGESTION") } @@ -703,17 +647,14 @@ class Scd1BatchProcessorSuite extends QueryTest with SharedSparkSession { Row(2, "bob", 25, Row(20L, null)) ) - val processor = Scd1BatchProcessor( - changeArgs = ChangeArgs( - keys = Seq(UnqualifiedColumnName("id")), - sequencing = F.col("seq"), - storedAsScdType = ScdType.Type1, - columnSelection = None - ), - resolvedSequencingType = LongType + val changeArgs = ChangeArgs( + keys = Seq(UnqualifiedColumnName("id")), + sequencing = F.col("seq"), + storedAsScdType = ScdType.Type1, + columnSelection = None ) - val result = processor.projectTargetColumnsOntoMicrobatch(batch) + val result = projectTargetColumnsOntoMicrobatch(changeArgs, batch) // None selection is no-op on the user columns, and the CDC metadata column is unconditionally // re-projected last, so the output shape exactly matches the input. @@ -733,21 +674,18 @@ class Scd1BatchProcessorSuite extends QueryTest with SharedSparkSession { Row(1, "alice", 30, Row(null, 10L)) ) - val processor = Scd1BatchProcessor( - changeArgs = ChangeArgs( - keys = Seq(UnqualifiedColumnName("id")), - sequencing = F.col("seq"), - storedAsScdType = ScdType.Type1, - columnSelection = Some( - ColumnSelection.IncludeColumns( - Seq(UnqualifiedColumnName("id"), UnqualifiedColumnName("age")) - ) + val changeArgs = ChangeArgs( + keys = Seq(UnqualifiedColumnName("id")), + sequencing = F.col("seq"), + storedAsScdType = ScdType.Type1, + columnSelection = Some( + ColumnSelection.IncludeColumns( + Seq(UnqualifiedColumnName("id"), UnqualifiedColumnName("age")) ) - ), - resolvedSequencingType = LongType + ) ) - val result = processor.projectTargetColumnsOntoMicrobatch(batch) + val result = projectTargetColumnsOntoMicrobatch(changeArgs, batch) assert(result.schema.fieldNames.toSeq == Seq("id", "age", AutoCdcReservedNames.cdcMetadataColName)) @@ -762,21 +700,18 @@ class Scd1BatchProcessorSuite extends QueryTest with SharedSparkSession { Row(1, "alice", 30, Row(null, 10L)) ) - val processor = Scd1BatchProcessor( - changeArgs = ChangeArgs( - keys = Seq(UnqualifiedColumnName("id")), - sequencing = F.col("seq"), - storedAsScdType = ScdType.Type1, - columnSelection = Some( - ColumnSelection.ExcludeColumns( - Seq(UnqualifiedColumnName("age")) - ) + val changeArgs = ChangeArgs( + keys = Seq(UnqualifiedColumnName("id")), + sequencing = F.col("seq"), + storedAsScdType = ScdType.Type1, + columnSelection = Some( + ColumnSelection.ExcludeColumns( + Seq(UnqualifiedColumnName("age")) ) - ), - resolvedSequencingType = LongType + ) ) - val result = processor.projectTargetColumnsOntoMicrobatch(batch) + val result = projectTargetColumnsOntoMicrobatch(changeArgs, batch) assert( result.schema.fieldNames.toSeq == @@ -793,20 +728,17 @@ class Scd1BatchProcessorSuite extends QueryTest with SharedSparkSession { Row(1, "alice", 30, Row(null, 10L)) ) - val processor = Scd1BatchProcessor( - changeArgs = ChangeArgs( - keys = Seq(UnqualifiedColumnName("id")), - sequencing = F.col("seq"), - storedAsScdType = ScdType.Type1, - // User specifies (age, id) -- intentionally different from the schema order (id, age). - columnSelection = Some(ColumnSelection.IncludeColumns( - Seq(UnqualifiedColumnName("age"), UnqualifiedColumnName("id")) - )) - ), - resolvedSequencingType = LongType + val changeArgs = ChangeArgs( + keys = Seq(UnqualifiedColumnName("id")), + sequencing = F.col("seq"), + storedAsScdType = ScdType.Type1, + // User specifies (age, id) -- intentionally different from the schema order (id, age). + columnSelection = Some(ColumnSelection.IncludeColumns( + Seq(UnqualifiedColumnName("age"), UnqualifiedColumnName("id")) + )) ) - val result = processor.projectTargetColumnsOntoMicrobatch(batch) + val result = projectTargetColumnsOntoMicrobatch(changeArgs, batch) // Output column order follows the original microbatch schema (id before age), not the order // in which the user listed columns in IncludeColumns. The CDC metadata column is appended @@ -833,24 +765,21 @@ class Scd1BatchProcessorSuite extends QueryTest with SharedSparkSession { Row(1, "u-100", Row(null, 10L)) ) - val processor = Scd1BatchProcessor( - changeArgs = ChangeArgs( - keys = Seq(UnqualifiedColumnName("id")), - sequencing = F.col("seq"), - storedAsScdType = ScdType.Type1, - columnSelection = Some( - ColumnSelection.IncludeColumns( - Seq( - UnqualifiedColumnName("id"), - UnqualifiedColumnName("`user.id`") - ) + val changeArgs = ChangeArgs( + keys = Seq(UnqualifiedColumnName("id")), + sequencing = F.col("seq"), + storedAsScdType = ScdType.Type1, + columnSelection = Some( + ColumnSelection.IncludeColumns( + Seq( + UnqualifiedColumnName("id"), + UnqualifiedColumnName("`user.id`") ) ) - ), - resolvedSequencingType = LongType + ) ) - val result = processor.projectTargetColumnsOntoMicrobatch(batch) + val result = projectTargetColumnsOntoMicrobatch(changeArgs, batch) assert(result.schema.fieldNames.toSeq == Seq("id", "user.id", AutoCdcReservedNames.cdcMetadataColName)) @@ -867,22 +796,19 @@ class Scd1BatchProcessorSuite extends QueryTest with SharedSparkSession { Row(1, "alice", 30, Row(null, 10L)) ) - val processor = Scd1BatchProcessor( - changeArgs = ChangeArgs( - keys = Seq(UnqualifiedColumnName("id")), - sequencing = F.col("seq"), - storedAsScdType = ScdType.Type1, - // User columns intentionally use a different case than the schema (id, age). - columnSelection = Some( - ColumnSelection.IncludeColumns( - Seq(UnqualifiedColumnName("ID"), UnqualifiedColumnName("AGE")) - ) + val changeArgs = ChangeArgs( + keys = Seq(UnqualifiedColumnName("id")), + sequencing = F.col("seq"), + storedAsScdType = ScdType.Type1, + // User columns intentionally use a different case than the schema (id, age). + columnSelection = Some( + ColumnSelection.IncludeColumns( + Seq(UnqualifiedColumnName("ID"), UnqualifiedColumnName("AGE")) ) - ), - resolvedSequencingType = LongType + ) ) - val result = processor.projectTargetColumnsOntoMicrobatch(batch) + val result = projectTargetColumnsOntoMicrobatch(changeArgs, batch) // Output column names follow the microbatch schema's casing, not the casing in the user's // columnSelection. The CDC metadata column is appended last as always. @@ -898,7 +824,8 @@ class Scd1BatchProcessorSuite extends QueryTest with SharedSparkSession { // =============== applyTombstonesToMicrobatch tests =============== /** - * Schema for the microbatch input to [[Scd1BatchProcessor.applyTombstonesToMicrobatch]] + * Schema for the microbatch input to + * [[Scd1RowLevelReconciliation.applyTombstonesToMicrobatch]] * tests. */ private val applyTombstonesToMicrobatchTestMicrobatchSchema: StructType = new StructType() @@ -910,7 +837,8 @@ class Scd1BatchProcessorSuite extends QueryTest with SharedSparkSession { .add(AutoCdcReservedNames.cdcMetadataColName, cdcMetadataColSchemaType) /** - * Schema for the auxiliary input to [[Scd1BatchProcessor.applyTombstonesToMicrobatch]] tests. + * Schema for the auxiliary input to + * [[Scd1RowLevelReconciliation.applyTombstonesToMicrobatch]] tests. * * In practice for SCD1 the auxiliary table only carries key columns and the CDC metadata * column -- never user data columns -- so we mirror that production-side asymmetry here, @@ -935,18 +863,15 @@ class Scd1BatchProcessorSuite extends QueryTest with SharedSparkSession { Row(1, cdcMetadataRow(deleteSeq = Some(10), upsertSeq = None)) ) - val processor = Scd1BatchProcessor( - changeArgs = ChangeArgs( - keys = Seq(UnqualifiedColumnName("id")), - // Sequencing is irrelevant for applyTombstonesToMicrobatch; it is already encoded - // into the CDC metadata column. - sequencing = F.lit(0L), - storedAsScdType = ScdType.Type1 - ), - resolvedSequencingType = LongType + val changeArgs = ChangeArgs( + keys = Seq(UnqualifiedColumnName("id")), + // Sequencing is irrelevant for applyTombstonesToMicrobatch; it is already encoded + // into the CDC metadata column. + sequencing = F.lit(0L), + storedAsScdType = ScdType.Type1 ) - val result = processor.applyTombstonesToMicrobatch(microbatch, auxiliary) + val result = applyTombstonesToMicrobatch(changeArgs, microbatch, auxiliary) assert(result.collect().isEmpty) } @@ -963,19 +888,16 @@ class Scd1BatchProcessorSuite extends QueryTest with SharedSparkSession { Row(1, cdcMetadataRow(deleteSeq = Some(10), upsertSeq = None)) ) - val processor = Scd1BatchProcessor( - changeArgs = ChangeArgs( - keys = Seq(UnqualifiedColumnName("id")), - // Sequencing is irrelevant for applyTombstonesToMicrobatch; it is already encoded - // into the CDC metadata column. - sequencing = F.lit(0L), - storedAsScdType = ScdType.Type1 - ), - resolvedSequencingType = LongType + val changeArgs = ChangeArgs( + keys = Seq(UnqualifiedColumnName("id")), + // Sequencing is irrelevant for applyTombstonesToMicrobatch; it is already encoded + // into the CDC metadata column. + sequencing = F.lit(0L), + storedAsScdType = ScdType.Type1 ) checkAnswer( - df = processor.applyTombstonesToMicrobatch(microbatch, auxiliary), + df = applyTombstonesToMicrobatch(changeArgs, microbatch, auxiliary), expectedAnswer = Row(1, "tied-upsert", Row(null, 10L)) ) } @@ -990,19 +912,16 @@ class Scd1BatchProcessorSuite extends QueryTest with SharedSparkSession { Row(1, cdcMetadataRow(deleteSeq = Some(10), upsertSeq = None)) ) - val processor = Scd1BatchProcessor( - changeArgs = ChangeArgs( - keys = Seq(UnqualifiedColumnName("id")), - // Sequencing is irrelevant for applyTombstonesToMicrobatch; it is already encoded - // into the CDC metadata column. - sequencing = F.lit(0L), - storedAsScdType = ScdType.Type1 - ), - resolvedSequencingType = LongType + val changeArgs = ChangeArgs( + keys = Seq(UnqualifiedColumnName("id")), + // Sequencing is irrelevant for applyTombstonesToMicrobatch; it is already encoded + // into the CDC metadata column. + sequencing = F.lit(0L), + storedAsScdType = ScdType.Type1 ) checkAnswer( - df = processor.applyTombstonesToMicrobatch(microbatch, auxiliary), + df = applyTombstonesToMicrobatch(changeArgs, microbatch, auxiliary), expectedAnswer = Seq( Row(1, "fresher-upsert", Row(null, 15L)), Row(1, "fresher-delete", Row(20L, null)) @@ -1021,19 +940,16 @@ class Scd1BatchProcessorSuite extends QueryTest with SharedSparkSession { Row(2, cdcMetadataRow(deleteSeq = Some(1000), upsertSeq = None)) ) - val processor = Scd1BatchProcessor( - changeArgs = ChangeArgs( - keys = Seq(UnqualifiedColumnName("id")), - // Sequencing is irrelevant for applyTombstonesToMicrobatch; it is already encoded - // into the CDC metadata column. - sequencing = F.lit(0L), - storedAsScdType = ScdType.Type1 - ), - resolvedSequencingType = LongType + val changeArgs = ChangeArgs( + keys = Seq(UnqualifiedColumnName("id")), + // Sequencing is irrelevant for applyTombstonesToMicrobatch; it is already encoded + // into the CDC metadata column. + sequencing = F.lit(0L), + storedAsScdType = ScdType.Type1 ) checkAnswer( - df = processor.applyTombstonesToMicrobatch(microbatch, auxiliary), + df = applyTombstonesToMicrobatch(changeArgs, microbatch, auxiliary), expectedAnswer = Row(1, "stays", Row(null, 5L)) ) } @@ -1055,19 +971,16 @@ class Scd1BatchProcessorSuite extends QueryTest with SharedSparkSession { Row("US", 99, cdcMetadataRow(deleteSeq = Some(1000), upsertSeq = None)) ) - val processor = Scd1BatchProcessor( - changeArgs = ChangeArgs( - keys = Seq(UnqualifiedColumnName("region"), UnqualifiedColumnName("customer_id")), - // Sequencing is irrelevant for applyTombstonesToMicrobatch; it is already encoded - // into the CDC metadata column. - sequencing = F.lit(0L), - storedAsScdType = ScdType.Type1 - ), - resolvedSequencingType = LongType + val changeArgs = ChangeArgs( + keys = Seq(UnqualifiedColumnName("region"), UnqualifiedColumnName("customer_id")), + // Sequencing is irrelevant for applyTombstonesToMicrobatch; it is already encoded + // into the CDC metadata column. + sequencing = F.lit(0L), + storedAsScdType = ScdType.Type1 ) checkAnswer( - df = processor.applyTombstonesToMicrobatch(microbatch, auxiliary), + df = applyTombstonesToMicrobatch(changeArgs, microbatch, auxiliary), expectedAnswer = Seq( Row("US", 1, Row(null, 5L)), Row("US", 2, Row(null, 5L)) @@ -1087,18 +1000,15 @@ class Scd1BatchProcessorSuite extends QueryTest with SharedSparkSession { Row(1, cdcMetadataRow(deleteSeq = Some(10), upsertSeq = None)) ) - val processor = Scd1BatchProcessor( - changeArgs = ChangeArgs( - keys = Seq(UnqualifiedColumnName("`user.id`")), - // Sequencing is irrelevant for applyTombstonesToMicrobatch; it is already encoded - // into the CDC metadata column. - sequencing = F.lit(0L), - storedAsScdType = ScdType.Type1 - ), - resolvedSequencingType = LongType + val changeArgs = ChangeArgs( + keys = Seq(UnqualifiedColumnName("`user.id`")), + // Sequencing is irrelevant for applyTombstonesToMicrobatch; it is already encoded + // into the CDC metadata column. + sequencing = F.lit(0L), + storedAsScdType = ScdType.Type1 ) - val result = processor.applyTombstonesToMicrobatch(microbatch, auxiliary) + val result = applyTombstonesToMicrobatch(changeArgs, microbatch, auxiliary) assert(result.collect().isEmpty) } @@ -1115,19 +1025,16 @@ class Scd1BatchProcessorSuite extends QueryTest with SharedSparkSession { // against incoming rows in the microbatch. val auxiliary = microbatchOf(applyTombstonesToMicrobatchTestAuxiliarySchema)() - val processor = Scd1BatchProcessor( - changeArgs = ChangeArgs( - keys = Seq(UnqualifiedColumnName("id")), - // Sequencing is irrelevant for applyTombstonesToMicrobatch; it is already encoded - // into the CDC metadata column. - sequencing = F.lit(0L), - storedAsScdType = ScdType.Type1 - ), - resolvedSequencingType = LongType + val changeArgs = ChangeArgs( + keys = Seq(UnqualifiedColumnName("id")), + // Sequencing is irrelevant for applyTombstonesToMicrobatch; it is already encoded + // into the CDC metadata column. + sequencing = F.lit(0L), + storedAsScdType = ScdType.Type1 ) checkAnswer( - df = processor.applyTombstonesToMicrobatch(microbatch, auxiliary), + df = applyTombstonesToMicrobatch(changeArgs, microbatch, auxiliary), expectedAnswer = Seq( Row(1, "kept-upsert", Row(null, 5L)), Row(2, "kept-delete", Row(7L, null)) @@ -1148,19 +1055,16 @@ class Scd1BatchProcessorSuite extends QueryTest with SharedSparkSession { Row(1, cdcMetadataRow(deleteSeq = None, upsertSeq = None)) ) - val processor = Scd1BatchProcessor( - changeArgs = ChangeArgs( - keys = Seq(UnqualifiedColumnName("id")), - // Sequencing is irrelevant for applyTombstonesToMicrobatch; it is already encoded - // into the CDC metadata column. - sequencing = F.lit(0L), - storedAsScdType = ScdType.Type1 - ), - resolvedSequencingType = LongType + val changeArgs = ChangeArgs( + keys = Seq(UnqualifiedColumnName("id")), + // Sequencing is irrelevant for applyTombstonesToMicrobatch; it is already encoded + // into the CDC metadata column. + sequencing = F.lit(0L), + storedAsScdType = ScdType.Type1 ) checkAnswer( - df = processor.applyTombstonesToMicrobatch(microbatch, auxiliary), + df = applyTombstonesToMicrobatch(changeArgs, microbatch, auxiliary), expectedAnswer = Row(1, "kept-upsert", Row(null, 5L)) ) } @@ -1181,18 +1085,15 @@ class Scd1BatchProcessorSuite extends QueryTest with SharedSparkSession { Row(1, cdcMetadataRow(deleteSeq = Some(10), upsertSeq = None)) ) - val processor = Scd1BatchProcessor( - changeArgs = ChangeArgs( - keys = Seq(UnqualifiedColumnName("id")), - // Sequencing is irrelevant for applyTombstonesToMicrobatch; it is already encoded - // into the CDC metadata column. - sequencing = F.lit(0L), - storedAsScdType = ScdType.Type1 - ), - resolvedSequencingType = LongType + val changeArgs = ChangeArgs( + keys = Seq(UnqualifiedColumnName("id")), + // Sequencing is irrelevant for applyTombstonesToMicrobatch; it is already encoded + // into the CDC metadata column. + sequencing = F.lit(0L), + storedAsScdType = ScdType.Type1 ) - val result = processor.applyTombstonesToMicrobatch(microbatch, auxiliary) + val result = applyTombstonesToMicrobatch(changeArgs, microbatch, auxiliary) assert(result.collect().isEmpty) } @@ -1209,18 +1110,15 @@ class Scd1BatchProcessorSuite extends QueryTest with SharedSparkSession { Row(1, cdcMetadataRow(deleteSeq = Some(10), upsertSeq = Some(20))) ) - val processor = Scd1BatchProcessor( - changeArgs = ChangeArgs( - keys = Seq(UnqualifiedColumnName("id")), - // Sequencing is irrelevant for applyTombstonesToMicrobatch; it is already encoded - // into the CDC metadata column. - sequencing = F.lit(0L), - storedAsScdType = ScdType.Type1 - ), - resolvedSequencingType = LongType + val changeArgs = ChangeArgs( + keys = Seq(UnqualifiedColumnName("id")), + // Sequencing is irrelevant for applyTombstonesToMicrobatch; it is already encoded + // into the CDC metadata column. + sequencing = F.lit(0L), + storedAsScdType = ScdType.Type1 ) - val result = processor.applyTombstonesToMicrobatch(microbatch, auxiliary) + val result = applyTombstonesToMicrobatch(changeArgs, microbatch, auxiliary) assert(result.collect().isEmpty) } } From de71434fa70470ae4ddf5ea43352e70c8a18427c Mon Sep 17 00:00:00 2001 From: Anish Mahto Date: Fri, 18 Sep 2026 21:16:11 +0000 Subject: [PATCH 5/6] fix visibility --- .../sql/pipelines/autocdc/Scd1ReconciliationStrategy.scala | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd1ReconciliationStrategy.scala b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd1ReconciliationStrategy.scala index 10bb971161ab..41435d611a9d 100644 --- a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd1ReconciliationStrategy.scala +++ b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd1ReconciliationStrategy.scala @@ -114,7 +114,7 @@ private[pipelines] trait Scd1ReconciliationStrategy { /** Row-level SCD1 reconciliation. */ private[pipelines] object Scd1RowLevelReconciliation extends Scd1ReconciliationStrategy { - private val winningRowColName: String = s"${AutoCdcReservedNames.prefix}winning_row" + private[autocdc] val winningRowColName: String = s"${AutoCdcReservedNames.prefix}winning_row" /** * Keeps the event with the greatest sequencing value for each key, adds its CDC metadata, From 2758de3bcef989f6f0c3204dc46476b1494a9c65 Mon Sep 17 00:00:00 2001 From: Anish Mahto Date: Mon, 21 Sep 2026 16:49:52 +0000 Subject: [PATCH 6/6] improve scaladoc --- .../sql/pipelines/autocdc/Scd1BatchProcessor.scala | 6 +++--- .../autocdc/Scd1ForeachBatchHandler.scala | 2 +- .../autocdc/Scd1ReconciliationStrategy.scala | 14 +++++++++----- 3 files changed, 13 insertions(+), 9 deletions(-) diff --git a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd1BatchProcessor.scala b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd1BatchProcessor.scala index d3609135e23b..4c93e688886e 100644 --- a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd1BatchProcessor.scala +++ b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd1BatchProcessor.scala @@ -44,14 +44,14 @@ case class Scd1BatchProcessor( resolvedSequencingType: DataType, reconciliationStrategy: Scd1ReconciliationStrategy = Scd1RowLevelReconciliation) { - /** Reconciles a CDC microbatch into the form consumed by the table merges. */ + /** Reconciles a validated CDC microbatch into the form consumed by the table merges. */ private[autocdc] def reconcileMicrobatch( - batchDf: DataFrame, + validatedBatchDf: DataFrame, auxiliaryTableDf: DataFrame): DataFrame = reconciliationStrategy.reconcileMicrobatch( changeArgs = changeArgs, resolvedSequencingType = resolvedSequencingType, - batchDf = batchDf, + validatedBatchDf = validatedBatchDf, auxiliaryTableDf = auxiliaryTableDf ) diff --git a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd1ForeachBatchHandler.scala b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd1ForeachBatchHandler.scala index c286f26c8263..61af1054589b 100644 --- a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd1ForeachBatchHandler.scala +++ b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd1ForeachBatchHandler.scala @@ -45,7 +45,7 @@ case class Scd1ForeachBatchHandler( ).validateMicrobatch() val reconciledMicrobatch = batchProcessor.reconcileMicrobatch( - batchDf = batchDf, + validatedBatchDf = batchDf, // Aux holds at most one row per currently-active tombstone (revived keys are GC'd // by mergeMicrobatchOntoAuxiliaryTable), so it generally stays small enough for a broadcast // join. Future optimizations: key-pruned reads, table format-aware clustering and tombstone diff --git a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd1ReconciliationStrategy.scala b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd1ReconciliationStrategy.scala index 41435d611a9d..504f6457c5e1 100644 --- a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd1ReconciliationStrategy.scala +++ b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd1ReconciliationStrategy.scala @@ -30,8 +30,12 @@ private[pipelines] trait Scd1ReconciliationStrategy { /** * Resolves the CDC events for each key and removes events superseded by recorded tombstones. * - * @param batchDf A validated CDC microbatch containing the key columns and every column needed - * to evaluate the sequencing, delete, and column-selection expressions. + * 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. @@ -39,7 +43,7 @@ private[pipelines] trait Scd1ReconciliationStrategy { def reconcileMicrobatch( changeArgs: ChangeArgs, resolvedSequencingType: DataType, - batchDf: DataFrame, + validatedBatchDf: DataFrame, auxiliaryTableDf: DataFrame): DataFrame /** @@ -124,11 +128,11 @@ private[pipelines] object Scd1RowLevelReconciliation extends Scd1ReconciliationS override def reconcileMicrobatch( changeArgs: ChangeArgs, resolvedSequencingType: DataType, - batchDf: DataFrame, + validatedBatchDf: DataFrame, auxiliaryTableDf: DataFrame): DataFrame = { val deduplicated = deduplicateMicrobatch( changeArgs = changeArgs, - validatedMicrobatch = batchDf + validatedMicrobatch = validatedBatchDf ) val withCdcMetadata = extendMicrobatchRowsWithCdcMetadata( changeArgs = changeArgs,