[HUDI-3838] Implemented drop partition column feature for delta streamer code path

[vingov]

What is the purpose of the pull request

Add support for hoodie.datasource.write.drop.partition.columns config for DeltaStreamer code path.

Brief change log

• Add support for hoodie.datasource.write.drop.partition.columns config for DeltaStreamer code path.

Verify this pull request

(Please pick either of the following options)

This change added tests and can be verified as follows:

• Added method to TestHoodieAvroUtils class to verify the change.
• Manually verified the change by running a job locally using the local docker setup, the BigQuery tables were created properly as expected.

Committer checklist

• Has a corresponding JIRA in PR title & commit

• Commit message is descriptive of the change

• CI is green

• Necessary doc changes done or have another open PR

• For large changes, please consider breaking it into sub-tasks under an umbrella JIRA.

[vingov]

The reason for not invoking rewriteRecord method directly from DeltaStreamer code is because I was getting serialization error org.apache.spark.SparkException: Task not serializable when passing the targetSchema to that method from the map lambda expression.

[codope]

Thanks @vingov for quickly fixing this. Looks good. Left a few minor comments.

CI is failing. I think that's because we're not passing default value. Please check my comments below.

[nsivabalan]

not sure if this is right. we have to move this within map (). for each genRec, we have to first run through key generator and then modify the genRec if need be (drop partition cols).
Also, can we move lines 482 to 494 to separate method and keep this method cleaner/leaner.

[vingov]

To make it cleaner I picked the first record for generating the key, I see your point, updated it.

[codope]

@hudi-bot run azure

[codope]

LGTM. Verified the patch. Below is table config and schema in a commit file:

#Updated at 2022-04-12T10:02:39.874Z
#Tue Apr 12 15:32:39 IST 2022
hoodie.table.precombine.field=tpep_dropoff_datetime
hoodie.datasource.write.drop.partition.columns=false
hoodie.table.partition.fields=date_col
hoodie.table.type=COPY_ON_WRITE
hoodie.archivelog.folder=archived
hoodie.populate.meta.fields=true
hoodie.partition.metafile.use.base.format=false
hoodie.timeline.layout.version=1
hoodie.table.version=4
hoodie.table.metadata.partitions=column_stats,files
hoodie.table.recordkey.fields=tpep_pickup_datetime
hoodie.table.base.file.format=PARQUET
hoodie.table.keygenerator.class=org.apache.hudi.keygen.SimpleKeyGenerator
hoodie.table.name=ny_hudi_tbl
hoodie.table.metadata.partitions.inflight=
hoodie.table.checksum=1182839966

* Schema (note that date_col is the partition column which is not written as part of schema) *

"extraMetadata" : {
    "schema" : "{\"type\":\"record\",\"name\":\"test_struct_name\",\"namespace\":\"test_record_namespace\",\"fields\":[{\"name\":\"VendorID\",\"type\":[\"null\",\"int\"],\"default\":null},{\"name\":\"tpep_pickup_datetime\",\"type\":[\"null\",\"string\"],\"default\":null},{\"name\":\"tpep_dropoff_datetime\",\"type\":[\"null\",\"string\"],\"default\":null},{\"name\":\"passenger_count\",\"type\":[\"null\",\"int\"],\"default\":null},{\"name\":\"trip_distance\",\"type\":[\"null\",\"double\"],\"default\":null},{\"name\":\"RatecodeID\",\"type\":[\"null\",\"int\"],\"default\":null},{\"name\":\"store_and_fwd_flag\",\"type\":[\"null\",\"string\"],\"default\":null},{\"name\":\"PULocationID\",\"type\":[\"null\",\"int\"],\"default\":null},{\"name\":\"DOLocationID\",\"type\":[\"null\",\"int\"],\"default\":null},{\"name\":\"payment_type\",\"type\":[\"null\",\"int\"],\"default\":null},{\"name\":\"fare_amount\",\"type\":[\"null\",\"double\"],\"default\":null},{\"name\":\"extra\",\"type\":[\"null\",\"double\"],\"default\":null},{\"name\":\"mta_tax\",\"type\":[\"null\",\"double\"],\"default\":null},{\"name\":\"tip_amount\",\"type\":[\"null\",\"double\"],\"default\":null},{\"name\":\"tolls_amount\",\"type\":[\"null\",\"double\"],\"default\":null},{\"name\":\"improvement_surcharge\",\"type\":[\"null\",\"double\"],\"default\":null},{\"name\":\"total_amount\",\"type\":[\"null\",\"double\"],\"default\":null},{\"name\":\"congestion_surcharge\",\"type\":[\"null\",\"double\"],\"default\":null}]}",
    "deltastreamer.checkpoint.key" : "1634021149000"
  }

Would be good to add a UT in TestHoodieDeltastreamer as a followup. Tracking in HUDI-3863

[hudi-bot]

CI report:

• e9164c1 Azure: SUCCESS
• afc2f67 UNKNOWN

Bot commands

@hudi-bot supports the following commands:

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

[nsivabalan]

getPartitionColumns(keyGenerator, props) could have been done once in the driver.

[codope]

makes sense.. please land #5302 if it looks good.

[vingov]

I have created the same PR #5303 as well.