[HUDI-5824] Fix: do not combine if write operation is Upsert and COMBINE_BEFORE_UPSERT is false

[kazdy]

Change Logs

Fix shouldCombine, take into account the situation where the write operation is UPSERT but COMBINE_BEFORE_UPSERT is false.
Currently, Hudi always combines records on UPSERT, and option COMBINE_BEFORE_UPSERT is not honored.

Impact

Fixes user-facing option COMBINE_BEFORE_UPSERT

Risk level (write none, low medium or high below)

If medium or high, explain what verification was done to mitigate the risks.

Documentation Update

Describe any necessary documentation update if there is any new feature, config, or user-facing change

• The config description must be updated if new configs are added or the default value of the configs are changed
• Any new feature or user-facing change requires updating the Hudi website. Please create a Jira ticket, attach the
  ticket number here and follow the instruction to make
  changes to the website.

Contributor's checklist

• Read through contributor's guide
• Change Logs and Impact were stated clearly
• Adequate tests were added if applicable
• CI passed

[kazdy]

@hudi-bot run azure

[kazdy]

GH actions tests failed but I don't see why, it passed before.
Azure failed on TestHoodieDeltaStreamerWithMultiWriter, rather unrelated and it also passed it on the first run. Looks like some flaky tests causing issues.

[kazdy]

Found this because of my test failures, reported in HUDI-5839

There seem to be a bug with non-strict insert mode
when using spark datasource it can insert duplicates only in overwrite mode or append mode when data is inserted to the table for the first time, but if I want to insert in append mode for the second time it deduplicates the dataset as if it was working in upsert mode.

opt_insert = {
    'hoodie.table.name': 'huditbl',
    'hoodie.datasource.write.recordkey.field': 'keyid',
    'hoodie.datasource.write.table.name': 'huditbl',
    'hoodie.datasource.write.operation': 'insert',
    'hoodie.sql.insert.mode': 'non-strict',
    'hoodie.upsert.shuffle.parallelism': 2,
    'hoodie.insert.shuffle.parallelism': 2,
    'hoodie.combine.before.upsert': 'false',
    'hoodie.combine.before.insert': 'false',
    'hoodie.datasource.write.insert.drop.duplicates': 'false'
}

df = spark.range(0, 10).toDF("keyid") \
  .withColumn("age", expr("keyid + 1000"))

df.write.format("hudi"). \
options(**opt_insert). \
mode("overwrite"). \
save(path)

spark.read.format("hudi").load(path).count() # returns 10

df = df.union(df) # creates duplicates
df.write.format("hudi"). \
options(**opt_insert). \
mode("append"). \
save(path)

spark.read.format("hudi").load(path).count() # returns 10 but should return 20

[kazdy]

Hi Hudi devs, I would appreciate a review, thanks!

[kazdy]

@hudi-bot run azure

[bvaradar]

@kazdy : One comment about testcase. Otherwise looks good.

[bvaradar]

@kazdy : Can you remove the draft status if you think it is ready.

[hudi-bot]

CI report:

• 27d61f0 UNKNOWN
• b572d73 UNKNOWN
• a44c716 UNKNOWN
• 93db1f0 Azure: SUCCESS

Bot commands

@hudi-bot supports the following commands:

• @hudi-bot run azure re-run the last Azure build

[bvaradar]

LGTM.