[HUDI-6579] Adding support for upsert and deletes with spark datasource for pk less table

[nsivabalan]

Change Logs

Adding support for upsert and deletes with spark datasource for pk less table.

Impact

This patch opens up possibility to do updates or delete records with spark datasource writes for a primary key less table.

For eg, if for a primary table, user prefers to delete all records for a given employee.

val df = spark.read.format("hudi").load(basePath).filter("emp_id == '100')
df.write.format("hudi").option("hoodie.datasource.write.operation","delete").mode(Append).save(basePath) 

This is feasible for a pk less table.

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

low.

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

[codope]

For pk-less tables, RECORDKEY_FIELD may not be present and this condition can be true. Do we want to drop the meta fields in such cases? From the comment it seems like for pk-less we don;t want to drop the meta fields.

[nsivabalan]

not sure I get it. we will drop the meta fields down the line while creating the HoodieRecordPayload.
here we are just trying to gauge if we can go w/ prepped writes or regular non-prepped writes. if incoming df has meta fields and if its pk less table, we go with prepped flow.

[codope]

My point was for pk-less this condition would be true and canRemoveMetaFields can return true, isn't it? However, we need the meta fields for pk-less.

!optParams.containsKey(RECORDKEY_FIELD.key()

[nsivabalan]

@codope : addressed all comments.

[codope]

Can you please look into the CI failures?

[codope]

My point was for pk-less this condition would be true and canRemoveMetaFields can return true, isn't it? However, we need the meta fields for pk-less.

!optParams.containsKey(RECORDKEY_FIELD.key()

[nsivabalan]

oops, my bad. not sure how I missed. will fix it

[nsivabalan]

@codope : Updated

[codope]

Will land once the CI succeeds.

[hudi-bot]

CI report:

• edd7d00 Azure: SUCCESS

Bot commands

@hudi-bot supports the following commands:

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

[xushiyan]

dropping meta cols here caused a problem with HoodieStreamingSink: when val sourceDF = spark.readStream.format(hudi).load; sourceDF.writeStream.format(hudi).start(), the source DF is a streaming source and dropping metacols failed the assertion of "Queries with streaming sources must be executed with writeStream.start()" because internally Hudi is writing DF as a batch

we either keep df here the same as sourceDF when using sourceDF.writeStream.format(hudi).start() or use sourceDF.writeStream.foreachBatch{}