Don't understand the result

[bithw1]

I am using Hudi 0.15.0, In spark sql, I run following simple query,

When I run select * from hudi_cow_20251229_07, the result is as follows, I wonder why 1,2,3 and 1,3,6 are gone(I am using insert, no duplicates should be dropped)

spark-sql> select * from hudi_cow_20251229_07;
spark-sql> select * from hudi_cow_20251229_07;
_hoodie_commit_time     _hoodie_commit_seqno    _hoodie_record_key      _hoodie_partition_path  _hoodie_file_name       a       b       c
20251229154740370       20251229154740370_0_0   1               6efdbc56-1ebd-4cec-a1d4-6738aed8352b-0_0-15-21_20251229154740370.parquet        1       4       7

set hoodie.spark.sql.insert.into.operation=insert;
set hoodie.datasource.write.insert.drop.duplicates=false;
set hoodie.datasource.write.insert.dup.policy=none;

CREATE TABLE IF NOT EXISTS hudi_cow_20251229_07 (
  a INT,
  b INT,
  c INT
) 

USING hudi

tblproperties(
type='cow',
primaryKey='a',
hoodie.datasource.write.precombine.field='c'
);

insert into  hudi_cow_20251229_07(a,b,c) values(1,2,3),(1,4,7),(1,3,6);

[deepakpanda93]

Hello @bithw1

In this case:

• a is the record key
• c is the precombine field

Hudi guarantees only one record per primary key per commit. When multiple rows share the same primary key within the same insert, Hudi keeps only one row, chosen using the precombine field.

hoodie.table.precombine.field | Field used in preCombining before actual write. By default, when two records have the same key value, the largest value for the precombine field determined by Object.compareTo(..), is picked.

In this case for the recordKey 1, the precombineKey having the largest value is 7. So only (1,4,7) survives.

[bithw1]

Thanks @deepakpanda93

I have specially set the following configurations, so that duplicate records will not be dropped.

Do the following configurations only work for different commits? When primary key is specified, then, duplicates in the same commit will also be dropped even with following configurations?

If this is the case, then it doesn't make much sense to me. If duplicates are allowed, there is no difference whether the duplicate happen in one commit or different commits, the end result data there contains duplicates.

Is it possible to allow duplicates in one commit even with the primary key is specified?

set hoodie.spark.sql.insert.into.operation=insert;
set hoodie.datasource.write.insert.drop.duplicates=false;
set hoodie.datasource.write.insert.dup.policy=none;

[deepakpanda93]

Hello @bithw1 , Tried the same with dataframe approach in spark-shell and it worked !!

scala> import org.apache.spark.sql.functions._
import org.apache.spark.sql.functions._

scala> val df = Seq(
     |   (1, 2, 3),
     |   (1, 4, 7),
     |   (1, 3, 6)
     | ).toDF("a", "b", "c")
25/12/30 06:44:31 WARN MetricsConfig: Cannot locate configuration: tried hadoop-metrics2-s3a-file-system.properties,hadoop-metrics2.properties
df: org.apache.spark.sql.DataFrame = [a: int, b: int ... 1 more field]

scala> df.write
  .format("hudi")
  .option("hoodie.table.name", "hudi_cow_20251229_07")
  .option("hoodie.datasource.write.table.type", "COPY_ON_WRITE")
  .option("hoodie.datasource.write.recordkey.field", "a")
  .option("hoodie.datasource.write.precombine.field", "c")
  .option("hoodie.datasource.write.operation", "insert")
  .option("hoodie.datasource.write.insert.drop.duplicates", "false")
  .option("hoodie.datasource.write.insert.dup.policy", "none")
  .option("hoodie.datasource.write.keygenerator.class", "org.apache.hudi.keygen.NonpartitionedKeyGenerator")
  .mode("append")
  .save("file:///tmp/hudi_cow_20251229_07")

25/12/30 06:46:15 WARN MetricsConfig: Cannot locate configuration: tried hadoop-metrics2-hbase.properties,hadoop-metrics2.properties

scala> val basePath = "file:///tmp/hudi_cow_20251229_07"
basePath: String = file:///tmp/hudi_cow_20251229_07

scala> spark.sql(s"""CREATE TABLE IF NOT EXISTS hudi_cow_20251229_07 USING hudi LOCATION '$basePath'""".stripMargin)
25/12/30 06:49:34 WARN SessionState: METASTORE_FILTER_HOOK will be ignored, since hive.security.authorization.manager is set to instance of HiveAuthorizerFactory.
res19: org.apache.spark.sql.DataFrame = []

scala> spark.sql("select * from hudi_cow_20251229_07").show()
+-------------------+--------------------+------------------+----------------------+--------------------+---+---+---+
|_hoodie_commit_time|_hoodie_commit_seqno|_hoodie_record_key|_hoodie_partition_path|   _hoodie_file_name|  a|  b|  c|
+-------------------+--------------------+------------------+----------------------+--------------------+---+---+---+
|  20251230064613766|20251230064613766...|                 1|                      |aa196a68-02a8-4f0...|  1|  2|  3|
|  20251230064613766|20251230064613766...|                 1|                      |aa196a68-02a8-4f0...|  1|  4|  7|
|  20251230064613766|20251230064613766...|                 1|                      |aa196a68-02a8-4f0...|  1|  3|  6|
+-------------------+--------------------+------------------+----------------------+--------------------+---+---+---+

With spark-sql hoodie.datasource.write.insert.dup.policy is used which is none by default so only no action is being taken in this case.

[bithw1]

Thanks @deepakpanda93

From your experiment, Spark DataFrame will not drop duplicates with the following configurations, while Spark SQL will?
That's too confusing... There must be at least one behavior is wrong.

set hoodie.spark.sql.insert.into.operation=insert;
set hoodie.datasource.write.insert.drop.duplicates=false;
set hoodie.datasource.write.insert.dup.policy=none;

What do you mean by

With spark-sql hoodie.datasource.write.insert.dup.policy is used which is none by default so only no action is being taken in this case.

You mean, if this configuration is set to none, that is, no action is being taken, duplicates will be dropped or will not be dropped?

[deepakpanda93]

@bithw1 hoodie.datasource.write.insert.dup.policy This is only applicable to Spark SQL writing. When it is none, Hudi fall back to use precombineKey field in case of same recordKey in the commit. You can refer the below for more details on this config.

https://hudi.apache.org/docs/0.15.0/configurations#hoodiedatasourceinsertduppolicy

[padhle-yrr-adi]

@bithw1 hyy there, i m new to open source, can you plz help me out? how can i join the dicussion groups?

[deepakpanda93]

Hello @bithw1 After setting this config hoodie.combine.before.insert to false, it allows duplicates within the same batch.

PFB result.

spark-sql (default)> select * from hudi_cow_demo;
20251231085119482       20251231085119482_0_1   1               9dc98f36-f6cf-41aa-b2c4-67b2e7bc891f-0_0-34-48_20251231085119482.parquet        1       2       3
20251231085119482       20251231085119482_0_2   1               9dc98f36-f6cf-41aa-b2c4-67b2e7bc891f-0_0-34-48_20251231085119482.parquet        1       4       7
20251231085119482       20251231085119482_0_3   1               9dc98f36-f6cf-41aa-b2c4-67b2e7bc891f-0_0-34-48_20251231085119482.parquet        1       3       6
Time taken: 0.10 seconds, Fetched 3 row(s)

[bithw1]

Thanks @deepakpanda93

I tried and it works!

I also find this config yesterday, but I didn't try this config because I find the default value is false, not sure why it works if we explicitly set it to be false.

I am using Hudi 0.15.0

  public static final ConfigProperty<String> COMBINE_BEFORE_INSERT = ConfigProperty
      .key("hoodie.combine.before.insert")
      .defaultValue("false")
      .markAdvanced()
      .withDocumentation("When inserted records share same key, controls whether they should be first combined (i.e de-duplicated) before"
          + " writing to storage.");