ad1happy2go commented on issue #19593: URL: https://github.com/apache/hudi/issues/19593#issuecomment-5596672274
### Reproduction Standalone spark-shell repro on Hudi master (MOR, `metadata.enable=false`, SIMPLE index). It reproduces the duplicate at the query level and isolates the trigger. Full gist: https://gist.github.com/ad1happy2go/3608a9eaa8402ad68286dfd386dcee0b Environment: Hudi master (HEAD), Spark 3.5.6, Scala 2.12, Java 17. Launch with the standard Hudi extensions: ``` spark-shell \ --jars hudi-spark3.5-bundle_2.12-<ver>.jar \ --conf spark.serializer=org.apache.spark.serializer.KryoSerializer \ --conf spark.kryo.registrator=org.apache.spark.HoodieSparkKryoRegistrar \ --conf spark.sql.extensions=org.apache.spark.sql.hudi.HoodieSparkSessionExtension \ --conf spark.sql.catalog.spark_catalog=org.apache.spark.sql.hudi.catalog.HoodieCatalog ``` #### Positive repro — reproduces the duplicate Two copies of the same key reach two independent write tasks in one commit without the intra-batch combine collapsing them. `combine.before.upsert=false` is the deterministic stand-in for a retried/recomputed task or a replayed batch (two copies that never meet in combine). ```scala import org.apache.spark.sql.SaveMode val basePath = "/tmp/hudi_dup_repro" val hudiOptions = Map( "hoodie.table.name" -> "dup_repro_mor", "hoodie.datasource.write.table.type" -> "MERGE_ON_READ", "hoodie.datasource.write.recordkey.field" -> "guid", "hoodie.datasource.write.partitionpath.field" -> "collectionName", "hoodie.datasource.write.precombine.field" -> "operationTime", "hoodie.datasource.write.keygenerator.class" -> "org.apache.hudi.keygen.SimpleKeyGenerator", "hoodie.datasource.write.operation" -> "upsert", "hoodie.metadata.enable" -> "false", "hoodie.index.type" -> "SIMPLE", "hoodie.compact.inline" -> "false", "hoodie.clean.automatic" -> "false", "hoodie.combine.before.upsert" -> "false" ) case class Rec(guid: String, collectionName: String, operationTime: Long, payload: String) // Same guid X, same partition, one commit, two tasks (neither sees the other) Seq(Rec("GUID-X", "partitionP", 100L, "taskA"), Rec("GUID-X", "partitionP", 100L, "taskB")).toDF().repartition(2) .write.format("hudi").options(hudiOptions).mode(SaveMode.Overwrite).save(basePath) // Control: same guid Y in two DIFFERENT partitions -> must stay valid Seq(Rec("GUID-Y", "partitionP", 300L, "vP"), Rec("GUID-Y", "partitionQ", 300L, "vQ")).toDF() .write.format("hudi").options(hudiOptions).mode(SaveMode.Append).save(basePath) spark.read.format("hudi").load(basePath) .groupBy("guid", "collectionName").count().orderBy("guid", "collectionName").show(false) ``` Output: ``` +------+--------------+-----+ |guid |collectionName|count| +------+--------------+-----+ |GUID-X|partitionP |2 | <- BUG: same guid, same partition -> duplicated |GUID-Y|partitionP |1 | <- control: same guid across partitions, |GUID-Y|partitionQ |1 | both kept as valid, NOT flagged +------+--------------+-----+ ``` This satisfies both required edge cases: same-guid-across-partitions stays valid (GUID-Y), same-guid-within-a-partition becomes a real duplicate (GUID-X). #### Negative control — sequential replay is safe With combine at its default (true), a plain sequential replay of the same key does **not** duplicate: ```scala // combine.before.upsert left at default true; same options otherwise case class Rec(guid: String, collectionName: String, operationTime: Long, payload: String) // Batch 1: filler -> partition's only (small) base-file group Seq(Rec("FILLER-F1", "partitionP", 50L, "filler")).toDF() .write.format("hudi").options(hudiOptions).mode(SaveMode.Overwrite).save(basePath) // Batch 2: first write of GUID-X Seq(Rec("GUID-X", "partitionP", 100L, "v1")).toDF() .write.format("hudi").options(hudiOptions).mode(SaveMode.Append).save(basePath) // Batch 3: replay GUID-X Seq(Rec("GUID-X", "partitionP", 200L, "v2")).toDF() .write.format("hudi").options(hudiOptions).mode(SaveMode.Append).save(basePath) ``` Inspecting the partition after each batch shows batch 2 rewrites a **new base file** (same fileId, new instant) rather than leaving X log-only: ``` Batch 1: BASE: 081231ed-...-0_0-13-9_20260907184305322.parquet Batch 2: BASE: 081231ed-...-0_0-28-20_20260907184308216.parquet <- new base, X now in base Batch 3: BASE: 081231ed-...-0_0-13-9_... BASE: 081231ed-...-0_0-28-20_... LOG : 081231ed-...-0_0-43-31_20260907184314682_1.log.parquet count(GUID-X, partitionP) = 1 <- no duplicate ``` Because the re-insert rewrites the base file, the base-file-only index still sees X on the replay, tags it as an UPDATE, and merges it. #### Takeaways 1. The duplicate is real and reproducible on MOR + partition-scoped index; the reader returns two physical rows for one `(guid, collectionName)`. 2. It requires **write-time parallelism**: two same-key copies reaching the write path as separate insert-tagged records that never meet in the combine step. That is what a stage recompute (S4-A) or a replay racing the original commit (S5) looks like. It is not a sequential index-logic defect, as the negative control shows. 3. The partition invariant you need (same guid across partitions = valid; same guid within a partition = duplicate) is not enforced by the write path once two same-key writes bypass combine, a non-global index cannot dedup them, and nothing collapses a cross-file-group duplicate after commit. Scope note: the exact two-distinct-file-group signature (your FG 3250/3251 mirror) needs a genuine Spark stage recompute, which is non-deterministic and was not forced here; the insert partitioner consolidates same-partition inserts into one bucket in a single local run. The query-level duplicate (count=2) is reproduced faithfully. -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
