[HUDI-5193] Enhancing spark-ds write tests for some of the core user flows

[nsivabalan]

Change Logs

We realized we don't have good test coverage for some of the core user flows w/ spark data source writes. So, enhancing the tests to cover the scenarios.

Tests coverage added w/ spark-data source writes:

COW and MOR * (w/ and w/o metadata)
    Partitioned(BLOOM, SIMPLE, GLOBAL_BLOOM), non-partitioned(GLOBAL_BLOOM).
        Immutable data. pure bulk_insert row writing. 
        Immutable w/ file sizing. pure inserts. 
        initial bulk ingest, followed by updates.

Impact

Will help catch any bugs around core user flows upfront.

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

low.

Documentation Update

N/A

Contributor's checklist

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

[jonvex]

def compareUpdateDfWithHudiDf(inputDf: Dataset[Row], hudiDf: Dataset[Row], beforeDf: Dataset[Row], colsToCompare: String ) : Unit = {
    val hudiWithoutMeta = hudiDf.drop(HoodieRecord.RECORD_KEY_METADATA_FIELD, HoodieRecord.PARTITION_PATH_METADATA_FIELD, HoodieRecord.COMMIT_SEQNO_METADATA_FIELD, HoodieRecord.COMMIT_TIME_METADATA_FIELD, HoodieRecord.FILENAME_METADATA_FIELD)
    hudiWithoutMeta.registerTempTable("hudiTbl")
    inputDf.registerTempTable("inputTbl")
    beforeDf.registerTempTable("beforeTbl")
    val hudiDfToCompare = spark.sqlContext.sql("select " + colsToCompare + " from hudiTbl")
    val inputDfToCompare = spark.sqlContext.sql("select " + colsToCompare + " from inputTbl")
    val beforeDfToCompare = spark.sqlContext.sql("select " + colsToCompare + " from beforeTbl")

    assertEquals(hudiDfToCompare.intersect(inputDfToCompare).count, inputDfToCompare.count)
    assertEquals(hudiDfToCompare.except(inputDfToCompare).except(beforeDfToCompare).count, 0)
  }

[jonvex]

In commonOpts you have

DataSourceWriteOptions.RECORDKEY_FIELD.key -> "_row_key",

but then in options you set

(DataSourceWriteOptions.RECORDKEY_FIELD.key() -> recordKeys) +

[jonvex]

I think doMORReadOptimizedQquery should call compareEntireInputDfWithHudiDf instead of compareUpdateDfWithHudiDf but I am less confident about this than my first two suggestions

[jonvex]

read incremental query doesn't have .option(HoodieMetadataConfig.ENABLE.key, isMetadataEnabledOnRead)

[jonvex]

think its supposed to be records4 not records3
var inputDF4 = spark.read.json(spark.sparkContext.parallelize(records3, 2))

[jonvex]

doMORReadOptimizedQquery has two Q's

[jonvex]

Inconsistent use of DF and Df

[jonvex]

Changes look good to me

[hudi-bot]

CI report:

• c92b4ad Azure: SUCCESS

Bot commands

@hudi-bot supports the following commands:

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