You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by sz...@apache.org on 2026/09/23 21:50:31 UTC
(spark) branch branch-4.x updated: [SPARK-59645][SDP] SCD1 Ignore-null support; Refactor SCD1 microbatch reconciliation into strategies
This is an automated email from the ASF dual-hosted git repository.
szehon-ho pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new aaed0c770872 [SPARK-59645][SDP] SCD1 Ignore-null support; Refactor SCD1 microbatch reconciliation into strategies
aaed0c770872 is described below
commit aaed0c7708723d743a321616dd201f73af7ab982
Author: AnishMahto <an...@gmail.com>
AuthorDate: Wed Sep 23 14:50:07 2026 -0700
[SPARK-59645][SDP] SCD1 Ignore-null support; Refactor SCD1 microbatch reconciliation into strategies
### 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. In future PRs, a leaf-level reconciliation strategy class will be implemented.
Reconciliation behavior shared identically between both row-level and leaf-level reconciliations are kept in the parent `Scd1ReconciliationStrategy` trait.
This is a pure refactor PR, and does not yet introduce the strategy that per-leaf reconciliation (i.e ignore-null) will use. No new behavior is added, and no existing behavior is changed.
### 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.
Closes #58915 from AnishMahto/scd1-leaf-versioning-dispatch.
Authored-by: AnishMahto <an...@gmail.com>
Signed-off-by: Szehon Ho <sz...@gmail.com>
(cherry picked from commit 14e5f6140febbb6d197ab556632a2c4d000c8cbe)
Signed-off-by: Szehon Ho <sz...@gmail.com>
---
.../sql/pipelines/autocdc/Scd1BatchProcessor.scala | 230 +-------
.../autocdc/Scd1ForeachBatchHandler.scala | 2 +-
.../autocdc/Scd1ReconciliationStrategy.scala | 209 ++++++++
.../autocdc/Scd1BatchProcessorMergeSuite.scala | 2 +-
...scala => Scd1RowLevelReconciliationSuite.scala} | 588 +++++++++------------
5 files changed, 473 insertions(+), 558 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 397775513703..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
@@ -33,222 +33,27 @@ 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 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) {
+ resolvedSequencingType: DataType,
+ reconciliationStrategy: 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 validated 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,
+ validatedBatchDf: DataFrame,
+ auxiliaryTableDf: DataFrame): DataFrame =
+ reconciliationStrategy.reconcileMicrobatch(
+ changeArgs = changeArgs,
+ resolvedSequencingType = resolvedSequencingType,
+ validatedBatchDf = validatedBatchDf,
auxiliaryTableDf = 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: _*
- )
- }
-
- /**
- * 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,
@@ -408,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/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
new file mode 100644
index 000000000000..8dbbdb51630e
--- /dev/null
+++ b/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)
+ 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)
+
+ 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[autocdc] 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(
+ changeArgs: ChangeArgs,
+ resolvedSequencingType: DataType,
+ validatedBatchDf: DataFrame,
+ auxiliaryTableDf: DataFrame): DataFrame = {
+ val deduplicated = deduplicateMicrobatch(
+ changeArgs = changeArgs,
+ validatedMicrobatch = validatedBatchDf
+ )
+ val withCdcMetadata = extendMicrobatchRowsWithCdcMetadata(
+ changeArgs = changeArgs,
+ resolvedSequencingType = resolvedSequencingType,
+ validatedMicrobatch = deduplicated
+ )
+ val projected = projectTargetColumnsOntoMicrobatch(
+ changeArgs = changeArgs,
+ microbatchWithCdcMetadataDf = withCdcMetadata
+ )
+ 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/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 67%
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 d2c78442c476..0ae3b2f595e1 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
@@ -17,13 +17,19 @@
package org.apache.spark.sql.pipelines.autocdc
+import Scd1RowLevelReconciliation.{
+ applyTombstonesToMicrobatch,
+ deduplicateMicrobatch,
+ projectTargetColumnsOntoMicrobatch
+}
+
import org.apache.spark.sql.{functions => F, AnalysisException, QueryTest, Row}
import org.apache.spark.sql.classic.DataFrame
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 +56,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,6 +68,15 @@ 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 extendMicrobatchRowsWithCdcMetadata(
+ changeArgs: ChangeArgs,
+ batch: DataFrame): DataFrame =
+ Scd1RowLevelReconciliation.extendMicrobatchRowsWithCdcMetadata(
+ changeArgs = changeArgs,
+ resolvedSequencingType = LongType,
+ validatedMicrobatch = batch
+ )
+
// =============== deduplicateMicrobatch tests ===============
test("deduplicateMicrobatch keeps only the row with the largest sequence value per key") {
@@ -77,17 +91,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")
)
}
@@ -102,17 +113,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")
)
}
@@ -128,18 +136,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)
@@ -160,17 +165,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")
)
}
@@ -188,16 +190,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)
)
}
@@ -216,17 +215,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"),
@@ -247,19 +243,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)
)
}
@@ -278,17 +271,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))
)
}
@@ -309,17 +299,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"),
@@ -346,17 +333,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")
)
}
@@ -372,17 +356,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")
)
}
@@ -390,7 +371,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)
@@ -402,18 +383,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",
@@ -438,18 +416,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))
}
@@ -461,16 +436,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))
}
@@ -491,21 +463,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)),
@@ -525,18 +494,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))
)
}
@@ -553,18 +519,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))
@@ -582,16 +545,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.
@@ -612,16 +572,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]
@@ -653,18 +610,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")
}
@@ -676,17 +630,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.
@@ -706,21 +657,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))
@@ -735,21 +683,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 ==
@@ -766,20 +711,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
@@ -806,24 +748,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))
@@ -840,22 +779,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.
@@ -871,7 +807,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()
@@ -883,7 +820,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,
@@ -908,18 +846,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)
}
@@ -936,19 +871,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))
)
}
@@ -963,19 +895,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))
@@ -994,19 +923,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))
)
}
@@ -1028,19 +954,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))
@@ -1060,18 +983,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)
}
@@ -1088,19 +1008,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))
@@ -1121,19 +1038,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))
)
}
@@ -1154,18 +1068,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)
}
@@ -1182,18 +1093,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)
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org