[
https://issues.apache.org/jira/browse/HUDI-8938?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17922499#comment-17922499
]
Lin Liu commented on HUDI-8938:
-------------------------------
We have identified that there are some marker issues. I paste the timeline and
marker folder structure here. Test is based on
[https://github.com/apache/hudi/pull/12725] .
{code:java}
➜ .hoodie ls -al
total 128
drwxr-xr-x@ 23 linliu staff 736 Jan 30 10:30 .
drwxr-xr-x@ 11 linliu staff 352 Jan 30 10:30 ..
-rw-r--r--@ 1 linliu staff 20 Jan 30 10:29
.20250130182927634.deltacommit.crc
-rw-r--r--@ 1 linliu staff 16 Jan 30 10:29
.20250130182927634.deltacommit.inflight.crc
-rw-r--r--@ 1 linliu staff 8 Jan 30 10:29
.20250130182927634.deltacommit.requested.crc
-rw-r--r--@ 1 linliu staff 44 Jan 30 10:29
.20250130182939372.deltacommit.crc
-rw-r--r--@ 1 linliu staff 24 Jan 30 10:29
.20250130182939372.deltacommit.inflight.crc
-rw-r--r--@ 1 linliu staff 8 Jan 30 10:29
.20250130182939372.deltacommit.requested.crc
-rw-r--r--@ 1 linliu staff 24 Jan 30 10:30
.20250130183014430.deltacommit.inflight.crc
-rw-r--r--@ 1 linliu staff 8 Jan 30 10:30
.20250130183014430.deltacommit.requested.crc
drwxr-xr-x@ 3 linliu staff 96 Jan 30 10:29 .aux
-rw-r--r--@ 1 linliu staff 16 Jan 30 10:29 .hoodie.properties.crc
drwxr-xr-x@ 2 linliu staff 64 Jan 30 10:29 .schema
drwxr-xr-x@ 3 linliu staff 96 Jan 30 10:30 .temp
-rw-r--r--@ 1 linliu staff 1465 Jan 30 10:29 20250130182927634.deltacommit
-rw-r--r--@ 1 linliu staff 883 Jan 30 10:29
20250130182927634.deltacommit.inflight
-rw-r--r--@ 1 linliu staff 0 Jan 30 10:29
20250130182927634.deltacommit.requested
-rw-r--r--@ 1 linliu staff 4383 Jan 30 10:29 20250130182939372.deltacommit
-rw-r--r--@ 1 linliu staff 1690 Jan 30 10:29
20250130182939372.deltacommit.inflight
-rw-r--r--@ 1 linliu staff 0 Jan 30 10:29
20250130182939372.deltacommit.requested
-rw-r--r--@ 1 linliu staff 1698 Jan 30 10:30
20250130183014430.deltacommit.inflight
-rw-r--r--@ 1 linliu staff 0 Jan 30 10:30
20250130183014430.deltacommit.requested
-rw-r--r--@ 1 linliu staff 968 Jan 30 10:29 hoodie.properties
➜ .hoodie cdd ..
zsh: command not found: cdd
➜ .hoodie ls
20250130182927634.deltacommit 20250130182927634.deltacommit.inflight
20250130182927634.deltacommit.requested 20250130182939372.deltacommit
20250130182939372.deltacommit.inflight 20250130182939372.deltacommit.requested
20250130183014430.deltacommit.inflight 20250130183014430.deltacommit.requested
hoodie.properties
➜ .hoodie cd ..
➜ spark-8ff5c320-0599-4d5e-b6f9-15cc63d58ad2 ls -al
total 912
drwxr-xr-x@ 11 linliu staff 352 Jan 30 10:30 .
drwx------@ 3323 linliu staff 106336 Jan 30 10:32 ..
-rw-r--r--@ 1 linliu staff 16 Jan 30 10:29
..a358daa3-9862-4baf-81a3-cdecf2ff5a90-0_20250130182927634.log.1_1-0-1.crc
-rw-r--r--@ 1 linliu staff 16 Jan 30 10:30
..a358daa3-9862-4baf-81a3-cdecf2ff5a90-0_20250130182927634.log.2_1-0-1.crc
-rw-r--r--@ 1 linliu staff 12 Jan 30 10:29
..hoodie_partition_metadata.crc
-rw-r--r--@ 1 linliu staff 3408 Jan 30 10:29
.a358daa3-9862-4baf-81a3-cdecf2ff5a90-0_0-13-28_20250130182927634.parquet.crc
-rw-r--r--@ 1 linliu staff 1003 Jan 30 10:29
.a358daa3-9862-4baf-81a3-cdecf2ff5a90-0_20250130182927634.log.1_1-0-1
-rw-r--r--@ 1 linliu staff 889 Jan 30 10:30
.a358daa3-9862-4baf-81a3-cdecf2ff5a90-0_20250130182927634.log.2_1-0-1
drwxr-xr-x@ 23 linliu staff 736 Jan 30 10:30 .hoodie
-rw-r--r--@ 1 linliu staff 96 Jan 30 10:29 .hoodie_partition_metadata
-rw-r--r--@ 1 linliu staff 435010 Jan 30 10:29
a358daa3-9862-4baf-81a3-cdecf2ff5a90-0_0-13-28_20250130182927634.parquet
➜ spark-8ff5c320-0599-4d5e-b6f9-15cc63d58ad2 cd .hoodie/.temp
➜ .temp ls
20250130183014430
➜ .temp cd 20250130183014430
➜ 20250130183014430 ls
MARKERS.type MARKERS0 MARKERS1
➜ 20250130183014430 ls -al
total 48
drwxr-xr-x@ 8 linliu staff 256 Jan 30 10:30 .
drwxr-xr-x@ 3 linliu staff 96 Jan 30 10:30 ..
-rw-r--r--@ 1 linliu staff 12 Jan 30 10:30 .MARKERS.type.crc
-rw-r--r--@ 1 linliu staff 12 Jan 30 10:30 .MARKERS0.crc
-rw-r--r--@ 1 linliu staff 12 Jan 30 10:30 .MARKERS1.crc
-rw-r--r--@ 1 linliu staff 21 Jan 30 10:30 MARKERS.type
-rw-r--r--@ 1 linliu staff 84 Jan 30 10:30 MARKERS0
-rw-r--r--@ 1 linliu staff 84 Jan 30 10:30 MARKERS1 {code}
The test code.
{code:java}
class TestMergeModeCommitTimeOrdering extends HoodieSparkSqlTestBase {
// TODO(HUDI-8938): add "mor,true,true,6" after the fix
Seq("mor,6,true,true").foreach { args =>
val argList = args.split(',')
val tableType = argList(0)
val tableVersion = argList(1)
val setRecordMergeConfigs = argList(2).toBoolean
val setUpsertOperation = argList(3).toBoolean
val isUpsert = setUpsertOperation || (tableVersion.toInt != 6 &&
setRecordMergeConfigs)
val storage = HoodieTestUtils.getDefaultStorage
val mergeConfigClause = if (setRecordMergeConfigs) {
// with precombine field set, UPSERT operation is used automatically
if (tableVersion.toInt == 6) {
// Table version 6
s", payloadClass = '${classOf[OverwriteWithLatestAvroPayload].getName}'"
} else {
// Current table version (8)
", preCombineField = 'ts',\nhoodie.record.merge.mode =
'COMMIT_TIME_ORDERING'"
}
} else {
// By default, the COMMIT_TIME_ORDERING is used if not specified by the
user
""
}
val writeTableVersionClause = if (tableVersion.toInt == 6) {
s"hoodie.write.table.version = $tableVersion,"
} else {
""
}
val expectedMergeConfigs = if (tableVersion.toInt == 6) {
Map(
HoodieTableConfig.VERSION.key -> "6",
HoodieTableConfig.PAYLOAD_CLASS_NAME.key ->
classOf[OverwriteWithLatestAvroPayload].getName)
} else {
Map(
HoodieTableConfig.VERSION.key -> "8",
HoodieTableConfig.RECORD_MERGE_MODE.key -> COMMIT_TIME_ORDERING.name(),
HoodieTableConfig.PAYLOAD_CLASS_NAME.key ->
classOf[OverwriteWithLatestAvroPayload].getName,
HoodieTableConfig.RECORD_MERGE_STRATEGY_ID.key ->
COMMIT_TIME_BASED_MERGE_STRATEGY_UUID)
}
val nonExistentConfigs = if (tableVersion.toInt == 6) {
Seq(HoodieTableConfig.RECORD_MERGE_MODE.key,
HoodieTableConfig.PRECOMBINE_FIELD.key)
} else {
if (setRecordMergeConfigs) {
Seq()
} else {
Seq(HoodieTableConfig.PRECOMBINE_FIELD.key)
}
}
test(s"Test $tableType table with COMMIT_TIME_ORDERING
(tableVersion=$tableVersion,"
+
s"setRecordMergeConfigs=$setRecordMergeConfigs,setUpsertOperation=$setUpsertOperation)")
{
withSparkSqlSessionConfigWithCondition(
("hoodie.merge.small.file.group.candidates.limit" -> "0", true),
("hoodie.spark.sql.insert.into.operation" -> "upsert",
setUpsertOperation),
// TODO(HUDI-8820): enable MDT after supporting MDT with table version 6
("hoodie.metadata.enable" -> "false", tableVersion.toInt == 6),
("hoodie.file.group.reader.enabled" -> "true", true),
("hoodie.merge.use.record.positions" -> "false", true)
) {
withRecordType()(withTempDir { tmp =>
val tableName = generateTableName
// Create table with COMMIT_TIME_ORDERING
spark.sql(
s"""
| create table $tableName (
| id int,
| name string,
| price double,
| ts long
| ) using hudi
| tblproperties (
| $writeTableVersionClause
| type = '$tableType',
| primaryKey = 'id'
| $mergeConfigClause
| )
| location '${tmp.getCanonicalPath}'
""".stripMargin)
validateTableConfig(
storage, tmp.getCanonicalPath, expectedMergeConfigs,
nonExistentConfigs)
// Insert initial records with ts=100
spark.sql(
s"""
| insert into $tableName
| select 1 as id, 'A' as name, 10.0 as price, 100 as ts
| union all
| select 2, 'B', 20.0, 100
""".stripMargin)
// Verify inserting records with the same ts value are visible
(COMMIT_TIME_ORDERING)
spark.sql(
s"""
| insert into $tableName
| select 1 as id, 'A_equal' as name, 60.0 as price, 100 as ts
| union all
| select 2, 'B_equal', 70.0, 100
""".stripMargin)
validateTableConfig(
storage, tmp.getCanonicalPath, expectedMergeConfigs,
nonExistentConfigs)
checkAnswer(s"select id, name, price, ts from $tableName order by
id")(
(if (isUpsert) {
// With UPSERT operation, there is no duplicate
Seq(
Seq(1, "A_equal", 60.0, 100),
Seq(2, "B_equal", 70.0, 100))
} else {
// With INSERT operation, there are duplicates
Seq(
Seq(1, "A", 10.0, 100),
Seq(1, "A_equal", 60.0, 100),
Seq(2, "B", 20.0, 100),
Seq(2, "B_equal", 70.0, 100))
}): _*)
if (isUpsert) {
// Verify updating records with the same ts value are visible
(COMMIT_TIME_ORDERING)
spark.sql(
s"""
| update $tableName
| set price = 50.0, ts = 100
| where id = 2
""".stripMargin)
validateTableConfig(
storage, tmp.getCanonicalPath, expectedMergeConfigs,
nonExistentConfigs)
checkAnswer(s"select id, name, price, ts from $tableName order by
id")(
Seq(1, "A_equal", 50.0, 100),
Seq(2, "B_equal", 70.0, 100))
}
})
}
}
}
} {code}
> INSERT INTO generates unexpected result on table version 6
> ----------------------------------------------------------
>
> Key: HUDI-8938
> URL: https://issues.apache.org/jira/browse/HUDI-8938
> Project: Apache Hudi
> Issue Type: Sub-task
> Reporter: Y Ethan Guo
> Assignee: Lin Liu
> Priority: Blocker
> Fix For: 1.0.1
>
>
> After fixing HUDI-8850 on existing tests,
> [https://github.com/apache/hudi/pull/12725] introduces more test cases on
> COMMIT_TIME_ORDERING, and one of them failed because after INSERT INTO the
> query result is wrong
> Test dimensions:
> TestMergeModeCommitTimeOrdering
> "mor,6,true,true"
> TestMergeModeEventTimeOrdering
> "mor,6,true", "mor,6,false"
>
> Error:
> {code:java}
> Expected Array([1,A,10.0,100], [1,A_equal,60.0,100], [2,B,20.0,100],
> [2,B_equal,70.0,100]), but got Array([1,A_equal,60.0,100],
> [2,B_equal,70.0,100])
> ScalaTestFailureLocation:
> org.apache.spark.sql.hudi.common.HoodieSparkSqlTestBase at
> (HoodieSparkSqlTestBase.scala:130)
> org.scalatest.exceptions.TestFailedException: Expected Array([1,A,10.0,100],
> [1,A_equal,60.0,100], [2,B,20.0,100], [2,B_equal,70.0,100]), but got
> Array([1,A_equal,60.0,100], [2,B_equal,70.0,100])
> at
> org.scalatest.Assertions.newAssertionFailedException(Assertions.scala:472)
> at
> org.scalatest.Assertions.newAssertionFailedException$(Assertions.scala:471)
> at
> org.scalatest.funsuite.AnyFunSuite.newAssertionFailedException(AnyFunSuite.scala:1562)
> at org.scalatest.Assertions.assertResult(Assertions.scala:867)
> at org.scalatest.Assertions.assertResult$(Assertions.scala:863)
> at org.scalatest.funsuite.AnyFunSuite.assertResult(AnyFunSuite.scala:1562)
> at
> org.apache.spark.sql.hudi.common.HoodieSparkSqlTestBase.checkAnswer(HoodieSparkSqlTestBase.scala:130)
> at
> org.apache.spark.sql.hudi.dml.TestMergeModeCommitTimeOrdering.$anonfun$new$5(TestMergeModeCommitTimeOrdering.scala:144)
> at
> org.apache.spark.sql.hudi.dml.TestMergeModeCommitTimeOrdering.$anonfun$new$5$adapted(TestMergeModeCommitTimeOrdering.scala:93)
> {code}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
