[SUPPORT] Insert results different than bulk_insert

[parisni]

hudi 0.11 / 0.12 at least

Bulk insert provides the same semantics as insert
https://hudi.apache.org/docs/write_operations#bulk_insert

From my tests this is not true. The below script results in duplicated rows with bulk insert, while insert replaces rows (what we expect from upsert)

I would expect both bulk_insert / insert share same results ?

sc.setLogLevel("WARN")
from pyspark.sql.types import StructType, StructField, StringType

env = "qa"
datalaketable = "nested_table"
tableName = "test_hudi_{datalaketable}".format(datalaketable=datalaketable)
basePath = "/tmp/{tableName}".format(tableName=tableName)
from pyspark.sql.functions import expr

##
# Create a nested table
##
data = [ (("James", None, "Smith"), "OH", "M"),
    (("Anna", "Rose", ""), "NY", "F"),
    (("Julia", "", "Williams"), "OH", "F"),
    (("Maria", "Anne", "Jones"), "NY", "M"),
    (("Jen", "Mary", "Brown"), "NY", "M"),
    (("Mike", "Mary", "Williams"), "OH", "M"),
]

schema = StructType(
    [
        StructField(
            "name",
            StructType(
                [
                    StructField("firstname", StringType(), True),
                    StructField("middlename", StringType(), True),
                    StructField("lastname", StringType(), True),
                ]
            ),
        ),
        StructField("state", StringType(), True),
        StructField("gender", StringType(), True),
    ]
)
df = (
    spark.createDataFrame(data=data, schema=schema)
    .withColumn("event_id", expr("row_number() over(partition by 1 order by 1)"))
    .withColumn("event_date", expr("current_date()"))
    .withColumn("version", expr("current_date()"))
)

df.printSchema()
hudi_options = {
    "hoodie.table.name": tableName,
    "hoodie.datasource.write.recordkey.field": "event_id",
    "hoodie.datasource.write.partitionpath.field": "event_date",
    "hoodie.datasource.write.table.name": tableName,
    "hoodie.datasource.write.operation": "insert", # change with bulk_insert
    "hoodie.datasource.write.precombine.field": "version",
    "hoodie.upsert.shuffle.parallelism": 80,
    "hoodie.insert.shuffle.parallelism": 80,
    "hoodie.delete.shuffle.parallelism": 80,
    "hoodie.datasource.write.keygenerator.class": "org.apache.hudi.keygen.ComplexKeyGenerator",
    "hoodie.datasource.write.hive_style_partitioning": "true",
    "hoodie.datasource.hive_sync.enable": "false",
    "hoodie.datasource.write.keygenerator.class": "org.apache.hudi.keygen.ComplexKeyGenerator",
    "hoodie.metadata.enable": "true",
}
(df.write.format("hudi").options(**hudi_options).mode("overwrite").save(basePath))
spark.read.format("hudi").load(basePath).count()

(df.write.format("hudi").options(**hudi_options).mode("append").save(basePath))

spark.read.format("hudi").load(basePath).count()

// 12 with bulk insert
// 6 with insert

[parisni]

hoodie.merge.allow.duplicate.on.inserts=true fixes the problem.

BTW, I suggest to update the documentation:

INSERT
This operation is very similar to upsert in terms of heuristics/file sizing but completely skips the index lookup step. Thus, it can be a lot faster than upserts for use-cases like log de-duplication (in conjunction with options to filter duplicates mentioned below). Still duplicates are usually merged by default with hoodie.merge.allow.duplicate.on.inserts=false This is also suitable for use-cases where the table can tolerate duplicates, but just need the transactional writes/incremental pull/storage management capabilities of Hudi.

[nsivabalan]

yes, by default bulk_insert will not dedup and thats by design. we just wanted to give user a way to bulk import w/o any index look up.

[nsivabalan]

@bhasudha : can we enhance the docs on this end.

[yihua]

@parisni For bulk insert, you may also dedup within a batch using hoodie.combine.before.insert=true, yet that does not avoid duplicates across batches.

[nsivabalan]

dedup w/ insert could happen by chance if the new batch is routed to the same file group due to small file handling.
so thats just a side effect of small file handling.

[parisni]

Clear explanation. That's a surprising default behavior since insert won't dedup at combine step, I would expect no such collateral effect when choosing insert over upsert

…

On August 31, 2022 11:30:04 PM UTC, Sivabalan Narayanan ***@***.***> wrote: dedup w/ insert could happen by chance if the new batch is routed to the same file group due to small file handling. so thats just a side effect of small file handling. -- Reply to this email directly or view it on GitHub: #6531 (comment) You are receiving this because you were mentioned. Message ID: ***@***.***>

[nsivabalan]

yeah. if not for small file handling, you might as well go w/ bulk_insert. thats why. I am going ahead and closing out the github issue. feel free to open new one if you have any more clarifications or run into issues.