Skip to content

Commit 76d775d

Browse files
AnishMahtoAnishMahto
committed
PR feedback
1 parent 2ff07f4 commit 76d775d

2 files changed

Lines changed: 115 additions & 6 deletions

File tree

‎sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd1BatchProcessor.scala‎

Lines changed: 7 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -27,7 +27,6 @@ import org.apache.spark.util.ArrayImplicits._
2727
* configuration.
2828
*/
2929
case class Scd1BatchProcessor(changeArgs: ChangeArgs) {
30-
3130
/**
3231
* Deduplicate the incoming CDC microbatch by key, keeping the most recent event per key
3332
* as ordered by [[ChangeArgs.sequencing]].
@@ -36,19 +35,23 @@ case class Scd1BatchProcessor(changeArgs: ChangeArgs) {
3635
* multiple events share the same key and the same sequence value, the row selected is
3736
* non-deterministic and undefined.
3837
*
38+
* @param validatedMicrobatch A microbatch that has already been validated such that the
39+
* sequencing column should not contain null values, and its data type
40+
* should support ordering.
41+
*
3942
* The schema of the returned dataframe matches the schema of the microbatch exactly.
4043
*/
41-
def deduplicateMicrobatch(microbatchDf: DataFrame): DataFrame = {
44+
def deduplicateMicrobatch(validatedMicrobatch: DataFrame): DataFrame = {
4245
// The `max_by` API can only return a single column, so pack/unpack the entire row into a
4346
// temporary column before and after the `max_by` operation.
4447
val winningRowCol = OutOfOrderCdcMergeUtils.tempColName("__winning_row")
4548

4649
val allMicrobatchColumns =
47-
microbatchDf.columns
50+
validatedMicrobatch.columns
4851
.map(colName => F.col(QuotingUtils.quoteIdentifier(colName)))
4952
.toImmutableArraySeq
5053

51-
microbatchDf
54+
validatedMicrobatch
5255
.groupBy(changeArgs.keys.map(k => F.col(k.quoted)): _*)
5356
.agg(
5457
F.max_by(F.struct(allMicrobatchColumns: _*), changeArgs.sequencing)

‎sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd1BatchProcessorSuite.scala‎

Lines changed: 108 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -17,13 +17,13 @@
1717

1818
package org.apache.spark.sql.pipelines.autocdc
1919

20-
import org.apache.spark.SparkFunSuite
20+
import org.apache.spark.sql.QueryTest
2121
import org.apache.spark.sql.{functions => F, Row}
2222
import org.apache.spark.sql.classic.DataFrame
2323
import org.apache.spark.sql.test.SharedSparkSession
2424
import org.apache.spark.sql.types._
2525

26-
class Scd1BatchProcessorSuite extends SparkFunSuite with SharedSparkSession {
26+
class Scd1BatchProcessorSuite extends QueryTest with SharedSparkSession {
2727

2828
/** Build a microbatch [[DataFrame]] from explicit rows and an explicit schema. */
2929
private def microbatchOf(schema: StructType)(rows: Row*): DataFrame =
@@ -63,6 +63,112 @@ class Scd1BatchProcessorSuite extends SparkFunSuite with SharedSparkSession {
6363
)
6464
}
6565

66+
test("deduplicateMicrobatch is no-op if there's a single event for a key") {
67+
val schema = new StructType()
68+
.add("id", IntegerType)
69+
.add("seq", LongType)
70+
.add("value", StringType)
71+
72+
val batch = microbatchOf(schema)(
73+
Row(1, 10L, "only-row")
74+
)
75+
76+
val processor = Scd1BatchProcessor(
77+
changeArgs = ChangeArgs(
78+
keys = Seq(UnqualifiedColumnName("id")),
79+
sequencing = F.col("seq"),
80+
storedAsScdType = ScdType.Type1
81+
)
82+
)
83+
84+
checkAnswer(
85+
df = processor.deduplicateMicrobatch(batch),
86+
expectedAnswer = Row(1, 10L, "only-row")
87+
)
88+
}
89+
90+
test("deduplicateMicrobatch handles equal sequencing values for the same key") {
91+
val schema = new StructType()
92+
.add("id", IntegerType)
93+
.add("seq", LongType)
94+
.add("value", StringType)
95+
96+
val batch = microbatchOf(schema)(
97+
Row(1, 10L, "first-tied-row"),
98+
Row(1, 10L, "second-tied-row")
99+
)
100+
101+
val processor = Scd1BatchProcessor(
102+
changeArgs = ChangeArgs(
103+
keys = Seq(UnqualifiedColumnName("id")),
104+
sequencing = F.col("seq"),
105+
storedAsScdType = ScdType.Type1
106+
)
107+
)
108+
109+
// On equal sequence number events for the same key we provide no guarantee on which event will
110+
// survive, but the contract is _one_ event will survive - assert that below.
111+
val result = processor.deduplicateMicrobatch(batch).collect()
112+
assert(result.length == 1)
113+
assert(result.head.getInt(0) == 1)
114+
assert(result.head.getLong(1) == 10L)
115+
assert(Set("first-tied-row", "second-tied-row").contains(result.head.getString(2)))
116+
}
117+
118+
test("deduplicateMicrobatch ignores rows with null sequencing when a non-null value exists") {
119+
val schema = new StructType()
120+
.add("id", IntegerType)
121+
.add("seq", LongType)
122+
.add("value", StringType)
123+
124+
val batch = microbatchOf(schema)(
125+
// In production the expectation is the microbatch will have been validated to not contain
126+
// any null sequence values, but demonstrate that null sequence rows are de-prioritized in
127+
// deduplication.
128+
Row(1, null, "null-sequence"),
129+
Row(1, 10L, "non-null-sequence")
130+
)
131+
132+
val processor = Scd1BatchProcessor(
133+
changeArgs = ChangeArgs(
134+
keys = Seq(UnqualifiedColumnName("id")),
135+
sequencing = F.col("seq"),
136+
storedAsScdType = ScdType.Type1
137+
)
138+
)
139+
140+
checkAnswer(
141+
df = processor.deduplicateMicrobatch(batch),
142+
expectedAnswer = Row(1, 10L, "non-null-sequence")
143+
)
144+
}
145+
146+
test(
147+
"deduplicateMicrobatch returns a null row when all sequencing values for a key are null"
148+
) {
149+
val schema = new StructType()
150+
.add("id", IntegerType)
151+
.add("seq", LongType)
152+
.add("value", StringType)
153+
val batch = microbatchOf(schema)(
154+
// In production the expectation is the microbatch will have been validated to not contain
155+
// any null sequence values, but demonstrate that a null row will be returned by
156+
// deduplication if all rows contain a null sequence in the microbatch.
157+
Row(1, null, "null-sequence")
158+
)
159+
val processor = Scd1BatchProcessor(
160+
changeArgs = ChangeArgs(
161+
keys = Seq(UnqualifiedColumnName("id")),
162+
sequencing = F.col("seq"),
163+
storedAsScdType = ScdType.Type1
164+
)
165+
)
166+
checkAnswer(
167+
df = processor.deduplicateMicrobatch(batch),
168+
expectedAnswer = Row(null, null, null)
169+
)
170+
}
171+
66172
test("deduplicateMicrobatch processes multiple keys independently") {
67173
val schema = new StructType()
68174
.add("id", IntegerType)

0 commit comments

Comments
 (0)