apache / apache/hudi

Partial merging for mor does not work correctly for event time based merging if different fields are updated

Open
#16,995 1 comment 0 reactions 0 assignees View on GitHub
area:sql from-jira priority:high type:bug
Dominant language
Java
Stars
6.2k
Forks
2.5k
Avg merge
2d 8h
Merged PRs (30d)
111

Description

For cow this will fail because  Seq(1, "a1", 12.0, 1023, "a1: updated desc1"),

the 12.0 should never be updated because the coresponding ordering value 999 is lower than the base file. But for mor, this test does not fail. 
It seems like it will merge the two log files and have the partial result

Seq([EMPTY, [EMPTY], 12.0, 999, [EMPTY])  +  Seq([EMPTY], [EMPTY], [EMPTY], 1023, "a1: updated desc1") = Seq([EMPTY], [EMPTY], 12.0, 1023, "a1: updated desc1")

and will not consider the 999 when merging with the base file
{code:java}
spark.sql(
s"""
|create table $tableName (
| id int,
| name string,
| price double,
| _ts int,
| description string
|) using hudi
|tblproperties(
| type ='$tableType',
| primaryKey = 'id',
| preCombineField = '_ts',
| recordMergeMode = '$mergeMode'
|)
|location '$basePath'
""".stripMargin)
val structFields = scala.collection.immutable.List(
StructField("id", IntegerType, nullable = true),
StructField("name", StringType, nullable = true),
StructField("price", DoubleType, nullable = true),
StructField("_ts", IntegerType, nullable = true),
StructField("description", StringType, nullable = true))
spark.sql(s"insert into $tableName values (1, 'a1', 10, 1000, 'a1: desc1')," +
"(2, 'a2', 20, 1200, 'a2: desc2'), (3, 'a3', 30, 1250, 'a3: desc3')")

spark.sql(
s"""
|merge into $tableName t0
|using ( select 1 as id, 'a1' as name, 12.0 as price, 999 as ts
|union select 3 as id, 'a3' as name, 25.0 as price, 1260 as ts) s0
|on t0.id = s0.id
|when matched then update set price = s0.price, _ts = s0.ts
|""".stripMargin)

validateTableSchema(tableName, structFields)
checkAnswer(s"select id, name, price, _ts, description from $tableName")(
Seq(1, "a1", 10.0, 1000, "a1: desc1"),
Seq(2, "a2", 20.0, 1200, "a2: desc2"),
Seq(3, "a3", 25.0, 1260, "a3: desc3")
)

// Partial updates using MERGE INTO statement with changed fields: "description" and "_ts"
spark.sql(
s"""
|merge into $tableName t0
|using ( select 1 as id, 'a1' as name, 'a1: updated desc1' as new_description, 1023 as _ts
|union select 2 as id, 'a2' as name, 'a2: updated desc2' as new_description, 1270 as _ts) s0
|on t0.id = s0.id
|when matched then update set description = s0.new_description, _ts = s0._ts
|""".stripMargin)

validateTableSchema(tableName, structFields)
checkAnswer(s"select id, name, price, _ts, description from $tableName")(
Seq(1, "a1", 12.0, 1023, "a1: updated desc1"),
Seq(2, "a2", 20.0, 1270, "a2: updated desc2"),
Seq(3, "a3", 25.0, 1260, "a3: desc3")
){code}
 

 

## JIRA info

- Link: https://issues.apache.org/jira/browse/HUDI-9375
- Type: Bug

Contributor guide

No contributing guide indexed for this repository

Research direction

Run the supplied Spark SQL reproduction with MOR and compare its final checkAnswer results with the expected values. Trace partial record merging across the two log files, focusing on how the _ts ordering value is considered; done means the lower-timestamp update leaves row 1 at price 10.0 and _ts 1000 while later valid updates remain applied.

Written by the indexing model from the issue text.

Assessment

Tech stack
java, scala, spark
Domain
data-engineering, databases
Issue type
Bug
Difficulty
4/5
Estimated time
3-5 days
Activity status
Active
Clarity
Mostly clear
Newbie friendliness
55/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.