You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "anew (via GitHub)" <gi...@apache.org> on 2026/07/20 06:06:38 UTC
[PR] [PSARK-57356][SDP] Implement SCD2 Batch Processor; Cleanup Delete Encoding Rows Post-Reconciliation [spark]
anew opened a new pull request, #57365:
URL: https://github.com/apache/spark/pull/57365
### What changes were proposed in this pull request?
This PR adds two post-reconciliation cleanup stages to the SCD2 batch processor
(`Scd2BatchProcessor`), used by Declarative Pipelines' AUTO CDC / SCD Type 2 processing.
Both operate on the output of `reconcileStartAndEndAt`, which is where an affected key's
rows get their final `startAt` / `endAt` intervals.
- **`dropLeftoverDeletesPostReconciliation`** — drops delete-encoded rows (tombstones and
decomposition tails) that became redundant after reconciliation. A delete-encoded row is
redundant when the immediately preceding row's reconciled `endAt` already equals its own
sequence, i.e. the preceding upsert already encodes the delete boundary. Redundancy is
computed per key using a `lag` over the chronological per-key window (`__previous_end_at` /
`__is_redundant_delete_encoding` temporary columns, dropped before returning).
- e.g. an open upsert `[10, null)` followed by a tombstone at `15` reconciles to a closed
upsert `[10, 15)`, making the tombstone redundant.
- e.g. an existing closed `[10, 20)` bisected by an event at `15` reconciles the event to
`[15, 20)`, making the `[null, 20)` decomposition tail redundant.
- **`promoteDecompositionTailsToTombstones`** — converts decomposition tails that *survive*
the deletion cleanup above into tombstones. A surviving tail is an unmatched delete
boundary; rewriting its `recordStartAt` and `startAt` to the tail's end sequence lets
downstream auxiliary-table handling preserve it as a tombstone. Column metadata and the
input schema are preserved.
Both methods are `private[autocdc]` building blocks with unit coverage; they are not yet
wired into the top-level processing flow (a follow-up composes them into the pipeline).
### Why are the changes needed?
After start/end reconciliation, some delete-encoded rows carry information that the
reconciled upserts already express, and some decomposition tails remain as standalone delete
boundaries. Without this cleanup, redundant tombstones/tails would be routed to the auxiliary
table needlessly, and unmatched tails would not be represented in the form downstream aux
handling expects. These two stages normalize the reconciled output so that each key's history
is encoded once and correctly.
### Does this PR introduce _any_ user-facing change?
No, SCD2 is an unreleased feature.
### How was this patch tested?
Added 13 unit tests to `Scd2BatchProcessorSuite` covering the two new methods (drop-vs-keep
for tombstones and decomposition tails, no-window-predecessor cases, upserts never dropped,
per-key independence, combined redundant drops; and tail-to-tombstone rewriting, pass-through
of tombstones/upserts, user-column and schema/metadata preservation, per-key promotion).
### Was this patch authored or co-authored using generative AI tooling?
Co-authored with Claude Opus 4.7 and 4.8
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57356][SDP] Implement SCD2 Batch Processor; Cleanup Delete Encoding Rows Post-Reconciliation [spark]
Posted by "anew (via GitHub)" <gi...@apache.org>.
anew commented on code in PR #57365:
URL: https://github.com/apache/spark/pull/57365#discussion_r3627441034
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala:
##########
@@ -811,6 +811,93 @@ case class Scd2BatchProcessor(
staged.select(outputColumns.toImmutableArraySeq: _*)
}
+ /**
+ * Drop delete-encoded rows (tombstones and decomposition tails) that became redundant after
+ * reconciliation.
+ *
+ * A tombstone or decomposition tail is redundant when the immediately preceding row's reconciled
+ * [[endAtColName]] equals its own sequence: the preceding upsert already encodes the delete
+ * boundary, so the standalone delete-encoded row should no longer be routed to aux.
+ *
+ * For example, an open upsert `[startAt=10, endAt=null)` followed by a tombstone at `15`
Review Comment:
Good point — the asymmetry was only in the wording, not the logic. Rewrote the doc to state the single rule both cases share: a delete-encoded row is redundant when the immediately preceding reconciled upsert's endAt equals the row's own boundary (its startAt/identity is irrelevant to the check). I also spelled out where the [null, 20) decomposition tail comes from — the open-head + tail split produced when a closed row is bisected by an out-of-order event.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57356][SDP] Implement SCD2 Batch Processor; Cleanup Delete Encoding Rows Post-Reconciliation [spark]
Posted by "anew (via GitHub)" <gi...@apache.org>.
anew commented on code in PR #57365:
URL: https://github.com/apache/spark/pull/57365#discussion_r3627455181
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala:
##########
@@ -811,6 +811,93 @@ case class Scd2BatchProcessor(
staged.select(outputColumns.toImmutableArraySeq: _*)
}
+ /**
+ * Drop delete-encoded rows (tombstones and decomposition tails) that became redundant after
+ * reconciliation.
+ *
+ * A tombstone or decomposition tail is redundant when the immediately preceding row's reconciled
+ * [[endAtColName]] equals its own sequence: the preceding upsert already encodes the delete
+ * boundary, so the standalone delete-encoded row should no longer be routed to aux.
+ *
+ * For example, an open upsert `[startAt=10, endAt=null)` followed by a tombstone at `15`
+ * reconciles into a closed upsert `[10, 15)`, making the tombstone redundant. Likewise, an
+ * existing closed `[10, 20)` bisected by an event at `15` reconciles the event into `[15, 20)`,
+ * making the `[null, 20)` decomposition tail redundant.
+ */
+ private[autocdc] def dropLeftoverDeletesPostReconciliation(
+ reconciledDf: DataFrame): DataFrame = {
+ val recordStartAt =
+ Scd2BatchProcessor.recordStartAtOf(F.col(AutoCdcReservedNames.cdcMetadataColName))
+ val startAt = F.col(Scd2BatchProcessor.startAtColName)
+ val endAt = F.col(Scd2BatchProcessor.endAtColName)
+
+ // Both tombstones and decomposition tails encode a delete boundary in their `endAt`. Either
+ // becomes redundant when the immediately preceding upsert was reconciled to close exactly on
+ // that boundary, since the resulting closed upsert already carries it.
+ val row = Scd2IntervalColumns(recordStartAt, startAt, endAt)
+ val isTombstone = RowClassifier.isTombstone(row)
+ val isDecompositionTail = RowClassifier.isDecompositionTail(row)
+ val isDeleteEncodedRow = isTombstone || isDecompositionTail
+
+ val withWindowCols = reconciledDf
+ .withColumn(
+ Scd2BatchProcessor.previousEndAtColName,
+ F.lag(endAt, 1).over(orderChronologicallyPerKeyWindow)
+ )
+ .withColumn(
+ Scd2BatchProcessor.isRedundantDeleteEncodingColName,
+ isDeleteEncodedRow && (F.col(Scd2BatchProcessor.previousEndAtColName) <=> endAt)
+ )
+
+ withWindowCols
+ .filter(!F.col(Scd2BatchProcessor.isRedundantDeleteEncodingColName))
+ .drop(
+ Scd2BatchProcessor.previousEndAtColName,
+ Scd2BatchProcessor.isRedundantDeleteEncodingColName
+ )
+ }
+
+ /**
+ * Convert surviving decomposition tails into tombstones.
+ *
+ * A decomposition tail that survives deletion cleanup is an unmatched delete boundary; setting
+ * [[recordStartAtFieldName]], and [[startAtColName]] to the tail's end sequence lets downstream
+ * aux handling preserve it as a tombstone.
+ *
+ * For example, if an existing closed upsert `[startAt=10, endAt=20)` is bisected by an event at
+ * `15`, decomposition first produces an open head `[10, null)` plus a tail `[null, 20)`;
+ * reconciliation may close the head as `[10, 15)`, leaving the tail's boundary at `20` to be
+ * promoted into a tombstone.
+ */
+ private[autocdc] def promoteDecompositionTailsToTombstones(
+ reconciledDf: DataFrame): DataFrame = {
+ val recordStartAt =
+ Scd2BatchProcessor.recordStartAtOf(F.col(AutoCdcReservedNames.cdcMetadataColName))
+ val startAt = F.col(Scd2BatchProcessor.startAtColName)
+ val endAt = F.col(Scd2BatchProcessor.endAtColName)
+ val isDecompositionTail =
+ RowClassifier.isDecompositionTail(Scd2IntervalColumns(recordStartAt, startAt, endAt))
+
+ val outputColumns = reconciledDf.columns.map {
+ case c if c == AutoCdcReservedNames.cdcMetadataColName =>
+ val metadata = reconciledDf.schema(c).metadata
+ F.when(
+ isDecompositionTail,
+ Scd2BatchProcessor.constructCdcMetadataCol(
+ recordStartAt = endAt,
+ sequencingType = resolvedSequencingType
+ )
+ ).otherwise(F.col(c)).as(c, metadata)
+ case c if c == Scd2BatchProcessor.startAtColName =>
+ val metadata = reconciledDf.schema(c).metadata
+ F.when(isDecompositionTail, endAt).otherwise(startAt).as(c, metadata)
+ case c =>
+ F.col(QuotingUtils.quoteIdentifier(c))
Review Comment:
Exactly. The other case branches match framework columns. I added a comment to clarify that
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57356][SDP] Implement SCD2 Batch Processor; Cleanup Delete Encoding Rows Post-Reconciliation [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #57365:
URL: https://github.com/apache/spark/pull/57365#discussion_r3632403056
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala:
##########
@@ -811,6 +811,107 @@ case class Scd2BatchProcessor(
staged.select(outputColumns.toImmutableArraySeq: _*)
}
+ /**
+ * Drop delete-encoded rows (tombstones and decomposition tails) that became redundant after
+ * reconciliation.
+ *
+ * The rule is the same for both row kinds: a delete-encoded row encodes a delete boundary in its
+ * [[endAtColName]], and it is redundant once the immediately preceding reconciled upsert closes
+ * (its [[endAtColName]]) exactly on that same boundary. When they coincide, the closed upsert
+ * already carries the boundary, so the standalone delete-encoded row can be dropped rather than
+ * routed to aux. Only the preceding upsert's `endAt` matters here; its `startAt` (identity) is
+ * irrelevant to the comparison.
+ *
+ * Both examples reduce to that single check - the preceding upsert's reconciled `endAt` equals
+ * the delete-encoded row's boundary:
+ * - Tombstone: an open upsert `[startAt=10, endAt=null)` followed by a tombstone at `15`
+ * reconciles into a closed upsert `[10, 15)`. Its `endAt=15` matches the tombstone's
+ * boundary `15`, so the tombstone is redundant.
+ * - Decomposition tail: an existing closed `[10, 20)` is bisected by an out-of-order event at
+ * `15`. Decomposition first splits the original row into an open head `[10, null)` plus a
+ * decomposition tail `[null, 20)` - a synthetic row with a null recordStartAt that carries
+ * the original right boundary `20` so it can be re-closed later. The bisecting event then
+ * reconciles into `[15, 20)`, whose `endAt=20` matches the tail's boundary `20`, so the tail
+ * is redundant. (A tail that does *not* coincide with a preceding upsert survives and is
+ * later turned into a tombstone by [[promoteDecompositionTailsToTombstones]].)
+ */
+ private[autocdc] def dropLeftoverDeletesPostReconciliation(
+ reconciledDf: DataFrame): DataFrame = {
+ val recordStartAt =
+ Scd2BatchProcessor.recordStartAtOf(F.col(AutoCdcReservedNames.cdcMetadataColName))
+ val startAt = F.col(Scd2BatchProcessor.startAtColName)
+ val endAt = F.col(Scd2BatchProcessor.endAtColName)
+
+ // Both tombstones and decomposition tails encode a delete boundary in their `endAt`. Either
+ // becomes redundant when the immediately preceding upsert was reconciled to close exactly on
+ // that boundary, since the resulting closed upsert already carries it.
+ val row = Scd2IntervalColumns(recordStartAt, startAt, endAt)
+ val isTombstone = RowClassifier.isTombstone(row)
+ val isDecompositionTail = RowClassifier.isDecompositionTail(row)
+ val isDeleteEncodedRow = isTombstone || isDecompositionTail
+
+ val withWindowCols = reconciledDf
+ .withColumn(
+ Scd2BatchProcessor.previousEndAtColName,
+ F.lag(endAt, 1).over(orderChronologicallyPerKeyWindow)
+ )
+ .withColumn(
+ Scd2BatchProcessor.isRedundantDeleteEncodingColName,
+ isDeleteEncodedRow && (F.col(Scd2BatchProcessor.previousEndAtColName) <=> endAt)
Review Comment:
Could we defensively verify that the preceding row is an upsert? Adjacent delete-encoded rows with the same boundary should not occur because upstream cleanup prevents them, but checking here would make this method enforce its documented invariant directly. For example:
```scala
val previous = row.lagBy(1, orderChronologicallyPerKeyWindow)
val isRedundantDeleteEncoding =
isDeleteEncodedRow &&
// Defensive: the predecessor should always be an upsert because upstream cleanup prevents
// adjacent delete-encoded rows with the same boundary.
RowClassifier.isUpsertRepresentingRow(previous) &&
(previous.endAt <=> endAt)
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57356][SDP] Implement SCD2 Batch Processor; Cleanup Delete Encoding Rows Post-Reconciliation [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho closed pull request #57365: [SPARK-57356][SDP] Implement SCD2 Batch Processor; Cleanup Delete Encoding Rows Post-Reconciliation
URL: https://github.com/apache/spark/pull/57365
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57356][SDP] Implement SCD2 Batch Processor; Cleanup Delete Encoding Rows Post-Reconciliation [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on PR #57365:
URL: https://github.com/apache/spark/pull/57365#issuecomment-5052131951
**Merge Summary:**
- merged into master https://github.com/apache/spark/commit/e350bb02d90a107c59a4997373008a83585a9007
- merged into branch-4.x https://github.com/apache/spark/commit/cebd378f2e1710b30918e854407b53b0071e13a6
*Posted by `merge_spark_pr.py`*
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57356][SDP] Implement SCD2 Batch Processor; Cleanup Delete Encoding Rows Post-Reconciliation [spark]
Posted by "anew (via GitHub)" <gi...@apache.org>.
anew commented on code in PR #57365:
URL: https://github.com/apache/spark/pull/57365#discussion_r3627452064
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala:
##########
@@ -811,6 +811,93 @@ case class Scd2BatchProcessor(
staged.select(outputColumns.toImmutableArraySeq: _*)
}
+ /**
+ * Drop delete-encoded rows (tombstones and decomposition tails) that became redundant after
+ * reconciliation.
+ *
+ * A tombstone or decomposition tail is redundant when the immediately preceding row's reconciled
+ * [[endAtColName]] equals its own sequence: the preceding upsert already encodes the delete
+ * boundary, so the standalone delete-encoded row should no longer be routed to aux.
+ *
+ * For example, an open upsert `[startAt=10, endAt=null)` followed by a tombstone at `15`
+ * reconciles into a closed upsert `[10, 15)`, making the tombstone redundant. Likewise, an
+ * existing closed `[10, 20)` bisected by an event at `15` reconciles the event into `[15, 20)`,
+ * making the `[null, 20)` decomposition tail redundant.
+ */
+ private[autocdc] def dropLeftoverDeletesPostReconciliation(
+ reconciledDf: DataFrame): DataFrame = {
+ val recordStartAt =
+ Scd2BatchProcessor.recordStartAtOf(F.col(AutoCdcReservedNames.cdcMetadataColName))
+ val startAt = F.col(Scd2BatchProcessor.startAtColName)
+ val endAt = F.col(Scd2BatchProcessor.endAtColName)
+
+ // Both tombstones and decomposition tails encode a delete boundary in their `endAt`. Either
+ // becomes redundant when the immediately preceding upsert was reconciled to close exactly on
+ // that boundary, since the resulting closed upsert already carries it.
+ val row = Scd2IntervalColumns(recordStartAt, startAt, endAt)
+ val isTombstone = RowClassifier.isTombstone(row)
+ val isDecompositionTail = RowClassifier.isDecompositionTail(row)
+ val isDeleteEncodedRow = isTombstone || isDecompositionTail
+
+ val withWindowCols = reconciledDf
+ .withColumn(
+ Scd2BatchProcessor.previousEndAtColName,
+ F.lag(endAt, 1).over(orderChronologicallyPerKeyWindow)
+ )
+ .withColumn(
+ Scd2BatchProcessor.isRedundantDeleteEncodingColName,
+ isDeleteEncodedRow && (F.col(Scd2BatchProcessor.previousEndAtColName) <=> endAt)
+ )
+
+ withWindowCols
+ .filter(!F.col(Scd2BatchProcessor.isRedundantDeleteEncodingColName))
+ .drop(
+ Scd2BatchProcessor.previousEndAtColName,
+ Scd2BatchProcessor.isRedundantDeleteEncodingColName
+ )
+ }
+
+ /**
+ * Convert surviving decomposition tails into tombstones.
Review Comment:
Agreed this is worth a look on the full-algorithm pass. They're handled identically here (both are just delete-boundary encodings), and a decomposition tail is essentially the within-batch analog of a tombstone — the main distinction today is provenance (a tail is synthesized during bisection and carries a null recordStartAt, a tombstone comes from an actual delete event). Whether they can be collapsed into one representation is a good question to settle when we review the algorithm end-to-end; but I think it s outside of this PR's scope.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57356][SDP] Implement SCD2 Batch Processor; Cleanup Delete Encoding Rows Post-Reconciliation [spark]
Posted by "anew (via GitHub)" <gi...@apache.org>.
anew commented on code in PR #57365:
URL: https://github.com/apache/spark/pull/57365#discussion_r3627441034
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala:
##########
@@ -811,6 +811,93 @@ case class Scd2BatchProcessor(
staged.select(outputColumns.toImmutableArraySeq: _*)
}
+ /**
+ * Drop delete-encoded rows (tombstones and decomposition tails) that became redundant after
+ * reconciliation.
+ *
+ * A tombstone or decomposition tail is redundant when the immediately preceding row's reconciled
+ * [[endAtColName]] equals its own sequence: the preceding upsert already encodes the delete
+ * boundary, so the standalone delete-encoded row should no longer be routed to aux.
+ *
+ * For example, an open upsert `[startAt=10, endAt=null)` followed by a tombstone at `15`
Review Comment:
updated the comments to explain the examples in more detail
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57356][SDP] Implement SCD2 Batch Processor; Cleanup Delete Encoding Rows Post-Reconciliation [spark]
Posted by "anew (via GitHub)" <gi...@apache.org>.
anew commented on PR #57365:
URL: https://github.com/apache/spark/pull/57365#issuecomment-5051142322
had to force-push this in order to rebase on latest master (and fix the broken CI caused by the previous master).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57356][SDP] Implement SCD2 Batch Processor; Cleanup Delete Encoding Rows Post-Reconciliation [spark]
Posted by "anew (via GitHub)" <gi...@apache.org>.
anew commented on code in PR #57365:
URL: https://github.com/apache/spark/pull/57365#discussion_r3627447254
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala:
##########
@@ -811,6 +811,93 @@ case class Scd2BatchProcessor(
staged.select(outputColumns.toImmutableArraySeq: _*)
}
+ /**
+ * Drop delete-encoded rows (tombstones and decomposition tails) that became redundant after
+ * reconciliation.
+ *
+ * A tombstone or decomposition tail is redundant when the immediately preceding row's reconciled
+ * [[endAtColName]] equals its own sequence: the preceding upsert already encodes the delete
+ * boundary, so the standalone delete-encoded row should no longer be routed to aux.
+ *
+ * For example, an open upsert `[startAt=10, endAt=null)` followed by a tombstone at `15`
+ * reconciles into a closed upsert `[10, 15)`, making the tombstone redundant. Likewise, an
+ * existing closed `[10, 20)` bisected by an event at `15` reconciles the event into `[15, 20)`,
+ * making the `[null, 20)` decomposition tail redundant.
+ */
+ private[autocdc] def dropLeftoverDeletesPostReconciliation(
+ reconciledDf: DataFrame): DataFrame = {
+ val recordStartAt =
+ Scd2BatchProcessor.recordStartAtOf(F.col(AutoCdcReservedNames.cdcMetadataColName))
+ val startAt = F.col(Scd2BatchProcessor.startAtColName)
+ val endAt = F.col(Scd2BatchProcessor.endAtColName)
+
+ // Both tombstones and decomposition tails encode a delete boundary in their `endAt`. Either
+ // becomes redundant when the immediately preceding upsert was reconciled to close exactly on
+ // that boundary, since the resulting closed upsert already carries it.
+ val row = Scd2IntervalColumns(recordStartAt, startAt, endAt)
+ val isTombstone = RowClassifier.isTombstone(row)
+ val isDecompositionTail = RowClassifier.isDecompositionTail(row)
+ val isDeleteEncodedRow = isTombstone || isDecompositionTail
+
+ val withWindowCols = reconciledDf
+ .withColumn(
+ Scd2BatchProcessor.previousEndAtColName,
+ F.lag(endAt, 1).over(orderChronologicallyPerKeyWindow)
+ )
+ .withColumn(
+ Scd2BatchProcessor.isRedundantDeleteEncodingColName,
+ isDeleteEncodedRow && (F.col(Scd2BatchProcessor.previousEndAtColName) <=> endAt)
+ )
+
+ withWindowCols
Review Comment:
I suppose. this still relates to the previous comment? If so, I hope the updated doc makes it clear
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [WIP][SPARK-57356][SDP] Implement SCD2 Batch Processor; Cleanup Delete Encoding Rows Post-Reconciliation [spark]
Posted by "anew (via GitHub)" <gi...@apache.org>.
anew commented on code in PR #57365:
URL: https://github.com/apache/spark/pull/57365#discussion_r3616934756
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala:
##########
@@ -811,6 +811,93 @@ case class Scd2BatchProcessor(
staged.select(outputColumns.toImmutableArraySeq: _*)
}
+ /**
+ * Drop delete-encoded rows (tombstones and decomposition tails) that became redundant after
+ * reconciliation.
+ *
+ * A tombstone or decomposition tail is redundant when the immediately preceding row's reconciled
+ * [[endAtColName]] equals its own sequence: the preceding upsert already encodes the delete
+ * boundary, so the standalone delete-encoded row should no longer be routed to aux.
+ *
+ * For example, an open upsert `[startAt=10, endAt=null)` followed by a tombstone at `15`
+ * reconciles into a closed upsert `[10, 15)`, making the tombstone redundant. Likewise, an
+ * existing closed `[10, 20)` bisected by an event at `15` reconciles the event into `[15, 20)`,
+ * making the `[null, 20)` decomposition tail redundant.
+ */
+ private[autocdc] def dropLeftoverDeletesPostReconciliation(
+ reconciledDf: DataFrame): DataFrame = {
+ val recordStartAt =
+ Scd2BatchProcessor.recordStartAtOf(F.col(AutoCdcReservedNames.cdcMetadataColName))
+ val startAt = F.col(Scd2BatchProcessor.startAtColName)
+ val endAt = F.col(Scd2BatchProcessor.endAtColName)
+
+ // Both tombstones and decomposition tails encode a delete boundary in their `endAt`. Either
+ // becomes redundant when the immediately preceding upsert was reconciled to close exactly on
+ // that boundary, since the resulting closed upsert already carries it.
+ val row = Scd2IntervalColumns(recordStartAt, startAt, endAt)
+ val isTombstone = RowClassifier.isTombstone(row)
+ val isDecompositionTail = RowClassifier.isDecompositionTail(row)
+ val isDeleteEncodedRow = isTombstone || isDecompositionTail
+
+ val withWindowCols = reconciledDf
+ .withColumn(
+ Scd2BatchProcessor.previousEndAtColName,
+ F.lag(endAt, 1).over(orderChronologicallyPerKeyWindow)
+ )
+ .withColumn(
+ Scd2BatchProcessor.isRedundantDeleteEncodingColName,
+ isDeleteEncodedRow && (F.col(Scd2BatchProcessor.previousEndAtColName) <=> endAt)
+ )
+
+ withWindowCols
+ .filter(!F.col(Scd2BatchProcessor.isRedundantDeleteEncodingColName))
+ .drop(
+ Scd2BatchProcessor.previousEndAtColName,
+ Scd2BatchProcessor.isRedundantDeleteEncodingColName
+ )
+ }
+
+ /**
+ * Convert surviving decomposition tails into tombstones.
+ *
+ * A decomposition tail that survives deletion cleanup is an unmatched delete boundary; setting
+ * [[recordStartAtFieldName]], and [[startAtColName]] to the tail's end sequence lets downstream
+ * aux handling preserve it as a tombstone.
+ *
+ * For example, if an existing closed upsert `[startAt=10, endAt=20)` is bisected by an event at
+ * `15`, decomposition first produces an open head `[10, null)` plus a tail `[null, 20)`;
+ * reconciliation may close the head as `[10, 15)`, leaving the tail's boundary at `20` to be
+ * promoted into a tombstone.
+ */
+ private[autocdc] def promoteDecompositionTailsToTombstones(
+ reconciledDf: DataFrame): DataFrame = {
+ val recordStartAt =
+ Scd2BatchProcessor.recordStartAtOf(F.col(AutoCdcReservedNames.cdcMetadataColName))
+ val startAt = F.col(Scd2BatchProcessor.startAtColName)
+ val endAt = F.col(Scd2BatchProcessor.endAtColName)
+ val isDecompositionTail =
+ RowClassifier.isDecompositionTail(Scd2IntervalColumns(recordStartAt, startAt, endAt))
+
+ val outputColumns = reconciledDf.columns.map {
+ case c if c == AutoCdcReservedNames.cdcMetadataColName =>
+ val metadata = reconciledDf.schema(c).metadata
+ F.when(
+ isDecompositionTail,
+ Scd2BatchProcessor.constructCdcMetadataCol(
+ recordStartAt = endAt,
+ sequencingType = resolvedSequencingType
+ )
+ ).otherwise(F.col(c)).as(c, metadata)
+ case c if c == Scd2BatchProcessor.startAtColName =>
+ val metadata = reconciledDf.schema(c).metadata
+ F.when(isDecompositionTail, endAt).otherwise(startAt).as(c, metadata)
+ case c =>
+ F.col(c)
Review Comment:
fixed and added a test.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [WIP][SPARK-57356][SDP] Implement SCD2 Batch Processor; Cleanup Delete Encoding Rows Post-Reconciliation [spark]
Posted by "anew (via GitHub)" <gi...@apache.org>.
anew commented on code in PR #57365:
URL: https://github.com/apache/spark/pull/57365#discussion_r3616949463
##########
sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessorSuite.scala:
##########
@@ -2661,4 +2661,356 @@ class Scd2BatchProcessorSuite extends QueryTest with SharedSparkSession {
assert(result.collect().isEmpty)
assert(result.columns.toSeq == df.columns.toSeq)
}
+
+ // =============== dropLeftoverDeletesPostReconciliation tests ===============
+
+ test("dropLeftoverDeletesPostReconciliation drops a tombstone whose sequence the preceding " +
+ "upsert was reconciled to close on") {
+ val processor = processorWithKeys(Seq("id"))
+ val userSchema = new StructType().add("id", IntegerType).add("value", StringType)
+
+ // A closed upsert [10, 15) immediately followed by a tombstone at 15. The upsert's reconciled
+ // endAt already encodes the delete boundary, so the standalone tombstone is redundant.
+ val df = targetTableOf(userSchema)(
+ Row(1, "alice", 10L, 15L, Row(10L)),
+ Row(1, "alice", 15L, 15L, Row(15L))
+ )
+
+ val result = processor.dropLeftoverDeletesPostReconciliation(df)
+
+ checkAnswer(
+ df = result,
+ expectedAnswer = Seq(
+ Row(1, "alice", 10L, 15L, Row(10L))
+ )
+ )
+ }
+
+ test("dropLeftoverDeletesPostReconciliation drops a decomposition tail whose boundary the " +
+ "preceding upsert was reconciled to close on") {
+ val processor = processorWithKeys(Seq("id"))
+ val userSchema = new StructType().add("id", IntegerType).add("value", StringType)
+
+ // Reconciled state of an existing closed [10, 20) bisected by an event at 15: the head closes
+ // at 15, the event closes at the tail's boundary 20, leaving the [null, 20) tail redundant
+ // because the [15, 20) upsert already encodes the boundary.
+ val df = targetTableOf(userSchema)(
+ Row(1, "alice", 10L, 15L, Row(10L)),
+ Row(1, "bob", 15L, 20L, Row(15L)),
+ Row(1, "alice", null, 20L, Row(null))
+ )
+
+ val result = processor.dropLeftoverDeletesPostReconciliation(df)
+
+ checkAnswer(
+ df = result,
+ expectedAnswer = Seq(
+ Row(1, "alice", 10L, 15L, Row(10L)),
+ Row(1, "bob", 15L, 20L, Row(15L))
+ )
+ )
+ }
+
+ test("dropLeftoverDeletesPostReconciliation keeps a tombstone whose sequence differs from " +
+ "the preceding upsert's reconciled endAt") {
+ val processor = processorWithKeys(Seq("id"))
+ val userSchema = new StructType().add("id", IntegerType).add("value", StringType)
+
+ // The closed upsert ends at 15 but the tombstone is at 20, so the delete boundary is not
+ // encoded by the preceding row and the tombstone must survive.
+ val df = targetTableOf(userSchema)(
+ Row(1, "alice", 5L, 15L, Row(5L)),
+ Row(1, "alice", 20L, 20L, Row(20L))
+ )
+
+ val result = processor.dropLeftoverDeletesPostReconciliation(df)
+
+ checkAnswer(
+ df = result,
+ expectedAnswer = Seq(
+ Row(1, "alice", 5L, 15L, Row(5L)),
+ Row(1, "alice", 20L, 20L, Row(20L))
+ )
+ )
+ }
+
+ test("dropLeftoverDeletesPostReconciliation keeps a decomposition tail whose boundary the " +
+ "preceding row does not close on") {
+ val processor = processorWithKeys(Seq("id"))
+ val userSchema = new StructType().add("id", IntegerType).add("value", StringType)
+
+ // The preceding closed upsert ends at 12, strictly before the tail's boundary 20, so the
+ // tail is an unmatched delete boundary that must survive for promotion to a tombstone.
+ val df = targetTableOf(userSchema)(
+ Row(1, "alice", 5L, 12L, Row(5L)),
+ Row(1, "alice", null, 20L, Row(null))
+ )
+
+ val result = processor.dropLeftoverDeletesPostReconciliation(df)
+
+ checkAnswer(
+ df = result,
+ expectedAnswer = Seq(
+ Row(1, "alice", 5L, 12L, Row(5L)),
+ Row(1, "alice", null, 20L, Row(null))
+ )
+ )
+ }
+
+ test("dropLeftoverDeletesPostReconciliation keeps a delete-encoded row that has no window " +
+ "predecessor") {
+ val processor = processorWithKeys(Seq("id"))
+ val userSchema = new StructType().add("id", IntegerType).add("value", StringType)
+
+ // With no predecessor, previousEndAt is null and `null <=> endAt` is false, so a leading
+ // tombstone (key 1) and a leading decomposition tail (key 2) both survive.
+ val df = targetTableOf(userSchema)(
+ Row(1, "alice", 15L, 15L, Row(15L)),
+ Row(2, "bob", null, 20L, Row(null))
+ )
+
+ val result = processor.dropLeftoverDeletesPostReconciliation(df)
+
+ checkAnswer(
+ df = result,
+ expectedAnswer = Seq(
+ Row(1, "alice", 15L, 15L, Row(15L)),
+ Row(2, "bob", null, 20L, Row(null))
+ )
+ )
+ }
+
+ test("dropLeftoverDeletesPostReconciliation never drops an upsert even when its endAt " +
+ "equals the preceding row's endAt") {
+ val processor = processorWithKeys(Seq("id"))
+ val userSchema = new StructType().add("id", IntegerType).add("value", StringType)
+
+ // Two closed upserts whose endAts coincide at 20. The `<=>` predicate matches, but neither
+ // row is delete-encoded, so the `isDeleteEncodedRow` guard must keep both. The overlapping
+ // intervals are synthetic here purely to isolate the guard.
+ val df = targetTableOf(userSchema)(
+ Row(1, "alice", 5L, 20L, Row(5L)),
+ Row(1, "bob", 10L, 20L, Row(10L))
+ )
+
+ val result = processor.dropLeftoverDeletesPostReconciliation(df)
+
+ checkAnswer(
+ df = result,
+ expectedAnswer = Seq(
+ Row(1, "alice", 5L, 20L, Row(5L)),
+ Row(1, "bob", 10L, 20L, Row(10L))
+ )
+ )
+ }
+
+ test("dropLeftoverDeletesPostReconciliation evaluates redundancy independently per key") {
+ val processor = processorWithKeys(Seq("id"))
+ val userSchema = new StructType().add("id", IntegerType).add("value", StringType)
+
+ // key 1: a closed upsert [10, 15) followed by a redundant tombstone at 15 (dropped).
+ // key 2: a tombstone at 15 that is the first row in its window (kept) - the key-1 upsert
+ // must not be treated as its predecessor.
+ val df = targetTableOf(userSchema)(
+ Row(1, "alice", 10L, 15L, Row(10L)),
+ Row(1, "alice", 15L, 15L, Row(15L)),
+ Row(2, "bob", 15L, 15L, Row(15L))
+ )
+
+ val result = processor.dropLeftoverDeletesPostReconciliation(df)
+
+ checkAnswer(
+ df = result,
+ expectedAnswer = Seq(
+ Row(1, "alice", 10L, 15L, Row(10L)),
+ Row(2, "bob", 15L, 15L, Row(15L))
+ )
+ )
+ }
+
+ test("dropLeftoverDeletesPostReconciliation drops a redundant tombstone and a redundant " +
+ "decomposition tail in a single pass") {
+ val processor = processorWithKeys(Seq("id"))
+ val userSchema = new StructType().add("id", IntegerType).add("value", StringType)
+
+ // [5, 10) closed upsert then a redundant tombstone at 10, then [15, 20) closed upsert then a
+ // redundant tail at 20. Both delete-encoded rows are encoded by their immediate predecessors
+ // and drop together in one window pass.
+ val df = targetTableOf(userSchema)(
+ Row(1, "alice", 5L, 10L, Row(5L)),
+ Row(1, "alice", 10L, 10L, Row(10L)),
+ Row(1, "bob", 15L, 20L, Row(15L)),
+ Row(1, "alice", null, 20L, Row(null))
+ )
+
+ val result = processor.dropLeftoverDeletesPostReconciliation(df)
+
+ checkAnswer(
+ df = result,
+ expectedAnswer = Seq(
+ Row(1, "alice", 5L, 10L, Row(5L)),
+ Row(1, "bob", 15L, 20L, Row(15L))
+ )
+ )
+ }
+
+ // =============== promoteDecompositionTailsToTombstones tests ===============
+
+ test("promoteDecompositionTailsToTombstones rewrites a decomposition tail into a tombstone") {
+ val processor = processorWithKeys(Seq("id"))
+ val userSchema = new StructType().add("id", IntegerType).add("value", StringType)
+
+ // The [null, 20) tail becomes a tombstone [20, 20] with recordStartAt = 20; the closed
+ // upsert is untouched.
+ val df = targetTableOf(userSchema)(
+ Row(1, "alice", 10L, 20L, Row(10L)),
+ Row(1, "alice", null, 20L, Row(null))
+ )
+
+ val result = processor.promoteDecompositionTailsToTombstones(df)
+
+ checkAnswer(
+ df = result,
+ expectedAnswer = Seq(
+ Row(1, "alice", 10L, 20L, Row(10L)),
+ Row(1, "alice", 20L, 20L, Row(20L))
+ )
+ )
+ }
+
+ test("promoteDecompositionTailsToTombstones leaves tombstones and upserts unchanged") {
+ val processor = processorWithKeys(Seq("id"))
+ val userSchema = new StructType().add("id", IntegerType).add("value", StringType)
+
+ // No decomposition tails present: a tombstone, an open upsert, and a closed upsert must all
+ // pass through identically.
+ val df = targetTableOf(userSchema)(
+ Row(1, "alice", 15L, 15L, Row(15L)),
+ Row(2, "bob", 5L, null, Row(5L)),
+ Row(3, "carol", 5L, 15L, Row(5L))
+ )
+
+ val result = processor.promoteDecompositionTailsToTombstones(df)
+
+ checkAnswer(
+ df = result,
+ expectedAnswer = Seq(
+ Row(1, "alice", 15L, 15L, Row(15L)),
+ Row(2, "bob", 5L, null, Row(5L)),
+ Row(3, "carol", 5L, 15L, Row(5L))
+ )
+ )
+ }
+
+ test("promoteDecompositionTailsToTombstones preserves the tail's inherited user columns") {
+ val processor = processorWithKeys(Seq("id"))
+ val userSchema = new StructType()
+ .add("id", IntegerType)
+ .add("name", StringType)
+ .add("status", StringType)
+
+ // Only the framework columns (startAt and the cdc-metadata recordStartAt) are rewritten to
+ // the tail's boundary; the inherited user columns must survive verbatim.
+ val df = targetTableOf(userSchema)(
+ Row(1, "alice", "active", null, 20L, Row(null))
+ )
+
+ val result = processor.promoteDecompositionTailsToTombstones(df)
+
+ checkAnswer(
+ df = result,
+ expectedAnswer = Seq(
+ Row(1, "alice", "active", 20L, 20L, Row(20L))
+ )
+ )
+ }
+
+ test("promoteDecompositionTailsToTombstones preserves the input schema, column metadata, " +
+ "and row count") {
+ val processor = processorWithKeys(Seq("id"))
+
+ def commentMetadata(comment: String): Metadata =
+ new MetadataBuilder().putString("comment", comment).build()
+
+ val schema = new StructType()
+ .add("id", IntegerType, nullable = false, metadata = commentMetadata("user key"))
+ .add("value", StringType, nullable = true, metadata = commentMetadata("user data"))
+ .add(
+ Scd2BatchProcessor.startAtColName, LongType, nullable = true,
+ metadata = commentMetadata("framework __START_AT"))
+ .add(
+ Scd2BatchProcessor.endAtColName, LongType, nullable = true,
+ metadata = commentMetadata("framework __END_AT"))
+ .add(
+ AutoCdcReservedNames.cdcMetadataColName,
+ Scd2BatchProcessor.cdcMetadataColSchema(LongType), nullable = false,
+ metadata = commentMetadata("framework _cdc_metadata"))
+
+ // A tail (rewritten) plus a non-tail (passed through) so both projection paths are exercised.
+ val df = microbatchOf(schema)(
+ Row(1, "alice", null, 20L, Row(null)),
+ Row(1, "alice", 5L, 20L, Row(5L))
+ )
+
+ val result = processor.promoteDecompositionTailsToTombstones(df)
+
+ schema.fields.zip(result.schema.fields).foreach { case (in, out) =>
+ assert(in.name == out.name)
+ assert(in.dataType == out.dataType)
+ assert(in.nullable == out.nullable)
+ assert(in.metadata == out.metadata)
+ }
+ assert(result.count() == df.count())
+ }
+
+ test("promoteDecompositionTailsToTombstones promotes tails independently across keys") {
+ val processor = processorWithKeys(Seq("id"))
+ val userSchema = new StructType().add("id", IntegerType).add("value", StringType)
+
+ // Each key carries a closed upsert plus a decomposition tail; only the tails are rewritten.
+ val df = targetTableOf(userSchema)(
+ Row(1, "alice", 5L, 20L, Row(5L)),
+ Row(1, "alice", null, 20L, Row(null)),
+ Row(2, "bob", 8L, 30L, Row(8L)),
+ Row(2, "bob", null, 30L, Row(null))
+ )
+
+ val result = processor.promoteDecompositionTailsToTombstones(df)
+
+ checkAnswer(
+ df = result,
+ expectedAnswer = Seq(
+ Row(1, "alice", 5L, 20L, Row(5L)),
+ Row(1, "alice", 20L, 20L, Row(20L)),
+ Row(2, "bob", 8L, 30L, Row(8L)),
+ Row(2, "bob", 30L, 30L, Row(30L))
+ )
+ )
+ }
+
+ test("promoteDecompositionTailsToTombstones quotes pass-through user columns whose names " +
+ "contain a dot") {
+ val processor = processorWithKeys(Seq("id"))
+ val userSchema = new StructType()
+ .add("id", IntegerType)
+ .add("user.name", StringType)
+
+ // The `user.name` user column passes through the default branch unchanged. Without quoting,
+ // `F.col("user.name")` would be parsed as a nested-field access (struct `user`, field
+ // `name`) and fail to resolve, so the select would throw. A decomposition tail (recordStartAt
+ // and startAt null, endAt=20) is promoted to a tombstone at 20 while the dotted user column
+ // survives verbatim.
+ val df = targetTableOf(userSchema)(
Review Comment:
- Without the fix in this last commit, it fails with
```
[UNRESOLVED_COLUMN.WITH_SUGGESTION] A column ... with name `user.name` cannot be resolved
```
- With the fix, the full suite passes
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57356][SDP] Implement SCD2 Batch Processor; Cleanup Delete Encoding Rows Post-Reconciliation [spark]
Posted by "jose-torres (via GitHub)" <gi...@apache.org>.
jose-torres commented on code in PR #57365:
URL: https://github.com/apache/spark/pull/57365#discussion_r3626804067
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala:
##########
@@ -811,6 +811,93 @@ case class Scd2BatchProcessor(
staged.select(outputColumns.toImmutableArraySeq: _*)
}
+ /**
+ * Drop delete-encoded rows (tombstones and decomposition tails) that became redundant after
+ * reconciliation.
+ *
+ * A tombstone or decomposition tail is redundant when the immediately preceding row's reconciled
+ * [[endAtColName]] equals its own sequence: the preceding upsert already encodes the delete
+ * boundary, so the standalone delete-encoded row should no longer be routed to aux.
+ *
+ * For example, an open upsert `[startAt=10, endAt=null)` followed by a tombstone at `15`
+ * reconciles into a closed upsert `[10, 15)`, making the tombstone redundant. Likewise, an
+ * existing closed `[10, 20)` bisected by an event at `15` reconciles the event into `[15, 20)`,
+ * making the `[null, 20)` decomposition tail redundant.
+ */
+ private[autocdc] def dropLeftoverDeletesPostReconciliation(
+ reconciledDf: DataFrame): DataFrame = {
+ val recordStartAt =
+ Scd2BatchProcessor.recordStartAtOf(F.col(AutoCdcReservedNames.cdcMetadataColName))
+ val startAt = F.col(Scd2BatchProcessor.startAtColName)
+ val endAt = F.col(Scd2BatchProcessor.endAtColName)
+
+ // Both tombstones and decomposition tails encode a delete boundary in their `endAt`. Either
+ // becomes redundant when the immediately preceding upsert was reconciled to close exactly on
+ // that boundary, since the resulting closed upsert already carries it.
+ val row = Scd2IntervalColumns(recordStartAt, startAt, endAt)
+ val isTombstone = RowClassifier.isTombstone(row)
+ val isDecompositionTail = RowClassifier.isDecompositionTail(row)
+ val isDeleteEncodedRow = isTombstone || isDecompositionTail
+
+ val withWindowCols = reconciledDf
+ .withColumn(
+ Scd2BatchProcessor.previousEndAtColName,
+ F.lag(endAt, 1).over(orderChronologicallyPerKeyWindow)
+ )
+ .withColumn(
+ Scd2BatchProcessor.isRedundantDeleteEncodingColName,
+ isDeleteEncodedRow && (F.col(Scd2BatchProcessor.previousEndAtColName) <=> endAt)
+ )
+
+ withWindowCols
+ .filter(!F.col(Scd2BatchProcessor.isRedundantDeleteEncodingColName))
+ .drop(
+ Scd2BatchProcessor.previousEndAtColName,
+ Scd2BatchProcessor.isRedundantDeleteEncodingColName
+ )
+ }
+
+ /**
+ * Convert surviving decomposition tails into tombstones.
Review Comment:
Another thing to handle in a post-hoc pass after implementation: we should make sure that these two are actually handled in a different enough way to require two different representations. It kinda sounds to me like a decomposition tail is just the within-batch version of a tombstone.
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala:
##########
@@ -811,6 +811,93 @@ case class Scd2BatchProcessor(
staged.select(outputColumns.toImmutableArraySeq: _*)
}
+ /**
+ * Drop delete-encoded rows (tombstones and decomposition tails) that became redundant after
+ * reconciliation.
+ *
+ * A tombstone or decomposition tail is redundant when the immediately preceding row's reconciled
+ * [[endAtColName]] equals its own sequence: the preceding upsert already encodes the delete
+ * boundary, so the standalone delete-encoded row should no longer be routed to aux.
+ *
+ * For example, an open upsert `[startAt=10, endAt=null)` followed by a tombstone at `15`
Review Comment:
The way this is stated is confusing to me. Why does the history record identity follow the prefix of the range when closing an active row but the suffix when closing an inactive row?
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala:
##########
@@ -811,6 +811,93 @@ case class Scd2BatchProcessor(
staged.select(outputColumns.toImmutableArraySeq: _*)
}
+ /**
+ * Drop delete-encoded rows (tombstones and decomposition tails) that became redundant after
+ * reconciliation.
+ *
+ * A tombstone or decomposition tail is redundant when the immediately preceding row's reconciled
+ * [[endAtColName]] equals its own sequence: the preceding upsert already encodes the delete
+ * boundary, so the standalone delete-encoded row should no longer be routed to aux.
+ *
+ * For example, an open upsert `[startAt=10, endAt=null)` followed by a tombstone at `15`
+ * reconciles into a closed upsert `[10, 15)`, making the tombstone redundant. Likewise, an
+ * existing closed `[10, 20)` bisected by an event at `15` reconciles the event into `[15, 20)`,
+ * making the `[null, 20)` decomposition tail redundant.
+ */
+ private[autocdc] def dropLeftoverDeletesPostReconciliation(
+ reconciledDf: DataFrame): DataFrame = {
+ val recordStartAt =
+ Scd2BatchProcessor.recordStartAtOf(F.col(AutoCdcReservedNames.cdcMetadataColName))
+ val startAt = F.col(Scd2BatchProcessor.startAtColName)
+ val endAt = F.col(Scd2BatchProcessor.endAtColName)
+
+ // Both tombstones and decomposition tails encode a delete boundary in their `endAt`. Either
+ // becomes redundant when the immediately preceding upsert was reconciled to close exactly on
+ // that boundary, since the resulting closed upsert already carries it.
+ val row = Scd2IntervalColumns(recordStartAt, startAt, endAt)
+ val isTombstone = RowClassifier.isTombstone(row)
+ val isDecompositionTail = RowClassifier.isDecompositionTail(row)
+ val isDeleteEncodedRow = isTombstone || isDecompositionTail
+
+ val withWindowCols = reconciledDf
+ .withColumn(
+ Scd2BatchProcessor.previousEndAtColName,
+ F.lag(endAt, 1).over(orderChronologicallyPerKeyWindow)
+ )
+ .withColumn(
+ Scd2BatchProcessor.isRedundantDeleteEncodingColName,
+ isDeleteEncodedRow && (F.col(Scd2BatchProcessor.previousEndAtColName) <=> endAt)
+ )
+
+ withWindowCols
Review Comment:
I suppose the implementation of this method doesn't end up depending on the row identity confusion above, but I still think it's worth clearing up.
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala:
##########
@@ -811,6 +811,93 @@ case class Scd2BatchProcessor(
staged.select(outputColumns.toImmutableArraySeq: _*)
}
+ /**
+ * Drop delete-encoded rows (tombstones and decomposition tails) that became redundant after
+ * reconciliation.
+ *
+ * A tombstone or decomposition tail is redundant when the immediately preceding row's reconciled
+ * [[endAtColName]] equals its own sequence: the preceding upsert already encodes the delete
+ * boundary, so the standalone delete-encoded row should no longer be routed to aux.
+ *
+ * For example, an open upsert `[startAt=10, endAt=null)` followed by a tombstone at `15`
+ * reconciles into a closed upsert `[10, 15)`, making the tombstone redundant. Likewise, an
+ * existing closed `[10, 20)` bisected by an event at `15` reconciles the event into `[15, 20)`,
+ * making the `[null, 20)` decomposition tail redundant.
+ */
+ private[autocdc] def dropLeftoverDeletesPostReconciliation(
+ reconciledDf: DataFrame): DataFrame = {
+ val recordStartAt =
+ Scd2BatchProcessor.recordStartAtOf(F.col(AutoCdcReservedNames.cdcMetadataColName))
+ val startAt = F.col(Scd2BatchProcessor.startAtColName)
+ val endAt = F.col(Scd2BatchProcessor.endAtColName)
+
+ // Both tombstones and decomposition tails encode a delete boundary in their `endAt`. Either
+ // becomes redundant when the immediately preceding upsert was reconciled to close exactly on
+ // that boundary, since the resulting closed upsert already carries it.
+ val row = Scd2IntervalColumns(recordStartAt, startAt, endAt)
+ val isTombstone = RowClassifier.isTombstone(row)
+ val isDecompositionTail = RowClassifier.isDecompositionTail(row)
+ val isDeleteEncodedRow = isTombstone || isDecompositionTail
+
+ val withWindowCols = reconciledDf
+ .withColumn(
+ Scd2BatchProcessor.previousEndAtColName,
+ F.lag(endAt, 1).over(orderChronologicallyPerKeyWindow)
+ )
+ .withColumn(
+ Scd2BatchProcessor.isRedundantDeleteEncodingColName,
+ isDeleteEncodedRow && (F.col(Scd2BatchProcessor.previousEndAtColName) <=> endAt)
+ )
+
+ withWindowCols
+ .filter(!F.col(Scd2BatchProcessor.isRedundantDeleteEncodingColName))
+ .drop(
+ Scd2BatchProcessor.previousEndAtColName,
+ Scd2BatchProcessor.isRedundantDeleteEncodingColName
+ )
+ }
+
+ /**
+ * Convert surviving decomposition tails into tombstones.
+ *
+ * A decomposition tail that survives deletion cleanup is an unmatched delete boundary; setting
+ * [[recordStartAtFieldName]], and [[startAtColName]] to the tail's end sequence lets downstream
+ * aux handling preserve it as a tombstone.
+ *
+ * For example, if an existing closed upsert `[startAt=10, endAt=20)` is bisected by an event at
+ * `15`, decomposition first produces an open head `[10, null)` plus a tail `[null, 20)`;
+ * reconciliation may close the head as `[10, 15)`, leaving the tail's boundary at `20` to be
+ * promoted into a tombstone.
+ */
+ private[autocdc] def promoteDecompositionTailsToTombstones(
+ reconciledDf: DataFrame): DataFrame = {
+ val recordStartAt =
+ Scd2BatchProcessor.recordStartAtOf(F.col(AutoCdcReservedNames.cdcMetadataColName))
+ val startAt = F.col(Scd2BatchProcessor.startAtColName)
+ val endAt = F.col(Scd2BatchProcessor.endAtColName)
+ val isDecompositionTail =
+ RowClassifier.isDecompositionTail(Scd2IntervalColumns(recordStartAt, startAt, endAt))
+
+ val outputColumns = reconciledDf.columns.map {
+ case c if c == AutoCdcReservedNames.cdcMetadataColName =>
+ val metadata = reconciledDf.schema(c).metadata
+ F.when(
+ isDecompositionTail,
+ Scd2BatchProcessor.constructCdcMetadataCol(
+ recordStartAt = endAt,
+ sequencingType = resolvedSequencingType
+ )
+ ).otherwise(F.col(c)).as(c, metadata)
+ case c if c == Scd2BatchProcessor.startAtColName =>
+ val metadata = reconciledDf.schema(c).metadata
+ F.when(isDecompositionTail, endAt).otherwise(startAt).as(c, metadata)
+ case c =>
+ F.col(QuotingUtils.quoteIdentifier(c))
Review Comment:
I guess we only need the quote here because we know the exact names of the other cases and they don't contain special characters?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57356][SDP] Implement SCD2 Batch Processor; Cleanup Delete Encoding Rows Post-Reconciliation [spark]
Posted by "anew (via GitHub)" <gi...@apache.org>.
anew commented on code in PR #57365:
URL: https://github.com/apache/spark/pull/57365#discussion_r3633067122
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala:
##########
@@ -811,6 +811,107 @@ case class Scd2BatchProcessor(
staged.select(outputColumns.toImmutableArraySeq: _*)
}
+ /**
+ * Drop delete-encoded rows (tombstones and decomposition tails) that became redundant after
+ * reconciliation.
+ *
+ * The rule is the same for both row kinds: a delete-encoded row encodes a delete boundary in its
+ * [[endAtColName]], and it is redundant once the immediately preceding reconciled upsert closes
+ * (its [[endAtColName]]) exactly on that same boundary. When they coincide, the closed upsert
+ * already carries the boundary, so the standalone delete-encoded row can be dropped rather than
+ * routed to aux. Only the preceding upsert's `endAt` matters here; its `startAt` (identity) is
+ * irrelevant to the comparison.
+ *
+ * Both examples reduce to that single check - the preceding upsert's reconciled `endAt` equals
+ * the delete-encoded row's boundary:
+ * - Tombstone: an open upsert `[startAt=10, endAt=null)` followed by a tombstone at `15`
+ * reconciles into a closed upsert `[10, 15)`. Its `endAt=15` matches the tombstone's
+ * boundary `15`, so the tombstone is redundant.
+ * - Decomposition tail: an existing closed `[10, 20)` is bisected by an out-of-order event at
+ * `15`. Decomposition first splits the original row into an open head `[10, null)` plus a
+ * decomposition tail `[null, 20)` - a synthetic row with a null recordStartAt that carries
+ * the original right boundary `20` so it can be re-closed later. The bisecting event then
+ * reconciles into `[15, 20)`, whose `endAt=20` matches the tail's boundary `20`, so the tail
+ * is redundant. (A tail that does *not* coincide with a preceding upsert survives and is
+ * later turned into a tombstone by [[promoteDecompositionTailsToTombstones]].)
+ */
+ private[autocdc] def dropLeftoverDeletesPostReconciliation(
+ reconciledDf: DataFrame): DataFrame = {
+ val recordStartAt =
+ Scd2BatchProcessor.recordStartAtOf(F.col(AutoCdcReservedNames.cdcMetadataColName))
+ val startAt = F.col(Scd2BatchProcessor.startAtColName)
+ val endAt = F.col(Scd2BatchProcessor.endAtColName)
+
+ // Both tombstones and decomposition tails encode a delete boundary in their `endAt`. Either
+ // becomes redundant when the immediately preceding upsert was reconciled to close exactly on
+ // that boundary, since the resulting closed upsert already carries it.
+ val row = Scd2IntervalColumns(recordStartAt, startAt, endAt)
+ val isTombstone = RowClassifier.isTombstone(row)
+ val isDecompositionTail = RowClassifier.isDecompositionTail(row)
+ val isDeleteEncodedRow = isTombstone || isDecompositionTail
+
+ val withWindowCols = reconciledDf
+ .withColumn(
+ Scd2BatchProcessor.previousEndAtColName,
+ F.lag(endAt, 1).over(orderChronologicallyPerKeyWindow)
+ )
+ .withColumn(
+ Scd2BatchProcessor.isRedundantDeleteEncodingColName,
+ isDeleteEncodedRow && (F.col(Scd2BatchProcessor.previousEndAtColName) <=> endAt)
Review Comment:
Good idea — done. Adapted it to the existing window setup: I build the lagged interval via `row.lagBy(1, orderChronologicallyPerKeyWindow)` and gate on `RowClassifier.isUpsertRepresentingRow(previous)` before the endAt comparison, which also let me drop the separate previousEndAt temp column. Added a test with two adjacent tombstones on the same boundary — the second now correctly survives (its predecessor isn't an upsert), which the old check would have dropped.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [WIP][SPARK-57356][SDP] Implement SCD2 Batch Processor; Cleanup Delete Encoding Rows Post-Reconciliation [spark]
Posted by "anew (via GitHub)" <gi...@apache.org>.
anew commented on code in PR #57365:
URL: https://github.com/apache/spark/pull/57365#discussion_r3616617784
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala:
##########
@@ -811,6 +811,93 @@ case class Scd2BatchProcessor(
staged.select(outputColumns.toImmutableArraySeq: _*)
}
+ /**
+ * Drop delete-encoded rows (tombstones and decomposition tails) that became redundant after
+ * reconciliation.
+ *
+ * A tombstone or decomposition tail is redundant when the immediately preceding row's reconciled
+ * [[endAtColName]] equals its own sequence: the preceding upsert already encodes the delete
+ * boundary, so the standalone delete-encoded row should no longer be routed to aux.
+ *
+ * For example, an open upsert `[startAt=10, endAt=null)` followed by a tombstone at `15`
+ * reconciles into a closed upsert `[10, 15)`, making the tombstone redundant. Likewise, an
+ * existing closed `[10, 20)` bisected by an event at `15` reconciles the event into `[15, 20)`,
+ * making the `[null, 20)` decomposition tail redundant.
+ */
+ private[autocdc] def dropLeftoverDeletesPostReconciliation(
+ reconciledDf: DataFrame): DataFrame = {
+ val recordStartAt =
+ Scd2BatchProcessor.recordStartAtOf(F.col(AutoCdcReservedNames.cdcMetadataColName))
+ val startAt = F.col(Scd2BatchProcessor.startAtColName)
+ val endAt = F.col(Scd2BatchProcessor.endAtColName)
+
+ // Both tombstones and decomposition tails encode a delete boundary in their `endAt`. Either
+ // becomes redundant when the immediately preceding upsert was reconciled to close exactly on
+ // that boundary, since the resulting closed upsert already carries it.
+ val row = Scd2IntervalColumns(recordStartAt, startAt, endAt)
+ val isTombstone = RowClassifier.isTombstone(row)
+ val isDecompositionTail = RowClassifier.isDecompositionTail(row)
+ val isDeleteEncodedRow = isTombstone || isDecompositionTail
+
+ val withWindowCols = reconciledDf
+ .withColumn(
+ Scd2BatchProcessor.previousEndAtColName,
+ F.lag(endAt, 1).over(orderChronologicallyPerKeyWindow)
+ )
+ .withColumn(
+ Scd2BatchProcessor.isRedundantDeleteEncodingColName,
+ isDeleteEncodedRow && (F.col(Scd2BatchProcessor.previousEndAtColName) <=> endAt)
+ )
+
+ withWindowCols
+ .filter(!F.col(Scd2BatchProcessor.isRedundantDeleteEncodingColName))
+ .drop(
+ Scd2BatchProcessor.previousEndAtColName,
+ Scd2BatchProcessor.isRedundantDeleteEncodingColName
+ )
+ }
+
+ /**
+ * Convert surviving decomposition tails into tombstones.
+ *
+ * A decomposition tail that survives deletion cleanup is an unmatched delete boundary; setting
+ * [[recordStartAtFieldName]], and [[startAtColName]] to the tail's end sequence lets downstream
+ * aux handling preserve it as a tombstone.
+ *
+ * For example, if an existing closed upsert `[startAt=10, endAt=20)` is bisected by an event at
+ * `15`, decomposition first produces an open head `[10, null)` plus a tail `[null, 20)`;
+ * reconciliation may close the head as `[10, 15)`, leaving the tail's boundary at `20` to be
+ * promoted into a tombstone.
+ */
+ private[autocdc] def promoteDecompositionTailsToTombstones(
+ reconciledDf: DataFrame): DataFrame = {
+ val recordStartAt =
+ Scd2BatchProcessor.recordStartAtOf(F.col(AutoCdcReservedNames.cdcMetadataColName))
+ val startAt = F.col(Scd2BatchProcessor.startAtColName)
+ val endAt = F.col(Scd2BatchProcessor.endAtColName)
+ val isDecompositionTail =
+ RowClassifier.isDecompositionTail(Scd2IntervalColumns(recordStartAt, startAt, endAt))
+
+ val outputColumns = reconciledDf.columns.map {
+ case c if c == AutoCdcReservedNames.cdcMetadataColName =>
+ val metadata = reconciledDf.schema(c).metadata
+ F.when(
+ isDecompositionTail,
+ Scd2BatchProcessor.constructCdcMetadataCol(
+ recordStartAt = endAt,
+ sequencingType = resolvedSequencingType
+ )
+ ).otherwise(F.col(c)).as(c, metadata)
+ case c if c == Scd2BatchProcessor.startAtColName =>
+ val metadata = reconciledDf.schema(c).metadata
+ F.when(isDecompositionTail, endAt).otherwise(startAt).as(c, metadata)
+ case c =>
+ F.col(c)
Review Comment:
That will mis-resolve user columns whose literal names contain dots or other identifier-sensitive characters, e.g. user.name, as nested-field paths. The rest of this class already handles these cases with `QuotingUtils.quoteIdentifier`, and there is existing test coverage for dotted tracked-history columns. This should use `F.col(QuotingUtils.quoteIdentifier(c))` for pass-through columns.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org