[HUDI-1331] Adding support for validating entire dataset and long running tests in test suite framework

[nsivabalan]

What is the purpose of the pull request

• Adding support for validating entire dataset and long running tests in test suite framework. The entire dag declared can be run repeatedly for N no of times. Two config params can be set for repeated test runs, namely --num-rounds and --delay-between-rounds-mins. So, lets say your dag looks like this

insert 1000
update 200
validate

and the configs are set as follows.
--num-rounds = 5
--delay-between-rounds-mins = 10

This will essentially run the (insert, update) for every 10 mins for 5 times.
Since validation can't be statically defined for no of expected records, a new validation node is introduced.
ValidateDatasetNode: will read entire input batch and compare w/ hudi contents. (row equality comparison).
No config needs to be set by default. But there is an optional config called "delete_input_data". If set, after validation is complete, input data will be deleted. This will come in handy when running long running tests.

An example for validation w/ delete_input_data set.

Round1: 
    insert => inputPath/batch1
    upsert -> inputPath/batch2
    Validate with delete_input_data = true
              Validates contents from batch1 and batch2 are in hudi and ensures Row equality
              Since "delete_input_data" is set, deletes contents from batch1 and batch2.
Round2:    
    insert => inputPath/batch3
    upsert -> inputPath/batch4
    Validate with delete_input_data = true
              Validates contents from batch3 and batch4 are in hudi and ensures Row equality
              Since "delete_input_data" is set, deletes contents from batch3 and batch4.
Round3:    
    insert => inputPath/batch5
    upsert -> inputPath/batch6
    Validate with delete_input_data = true
              Validates contents from batch5 and batch6 are in hudi and ensures Row equality
              Since "delete_input_data" is set, deletes contents from batch5 and batch6.   
.
.

Deleting input after reach round will keep the input size in control if we are trying a long running job along w/ validation.

An example for validation w/o any delete flag set.

Round1: 
    insert => inputPath/batch1
    upsert -> inputPath/batch2
    Validate: validates contents from batch1 and batch2 are in hudi and ensures Row equality
Round2:    
    insert => inputPath/batch3
    upsert -> inputPath/batch4
    Validate: validates contents from batch1 to batch4 are in hudi and ensures Row equality
Round3:    
    insert => inputPath/batch5
    upsert -> inputPath/batch6
    Validate: validates contents from batch1 and batch6 are in hudi and ensures Row equality
.
.

You could also have validations in the middle of your dag and not set the "delete_input_data". But set it only in the
last node in the dag so that at the end of every round data is cleaned up.

Round1: 
    insert => inputPath/batch1
    upsert -> inputPath/batch2
    Validate: validates contents from batch1 and batch2 are in hudi and ensures Row equality
    insert => inputPath/batch3
    upsert -> inputPath/batch4
    Validate with delete_input_data = true
             Validates contents from batch1 to batch4 are in hudi and ensures Row equality
             since "delete_input_data" is set to true, this node deletes contents from batch1 and batch4.
Round2: 
    insert => inputPath/batch5
    upsert -> inputPath/batch6
    Validate: validates contents from batch5 and batch6 are in hudi and ensures Row equality
    insert => inputPath/batch7
    upsert -> inputPath/batch8
    Validate: validates contents from batch5 to batch8 are in hudi and ensures Row equality
             since "delete_input_data" is set to true, this node deletes contents from batch5 to batch8.
Round3: 
    insert => inputPath/batch9
    upsert -> inputPath/batch10
    Validate: validates contents from batch9 and batch10 are in hudi and ensures Row equality
    insert => inputPath/batch11
    upsert -> inputPath/batch12
     Validate with delete_input_data = true
             Validates contents from batch9 to batch12 are in hudi and ensures Row equality
             Set "delete_input_data" to true. so this node deletes contents from batch9 to batch12. 
.
.

More info can be found here: https://docs.google.com/document/d/1LrYx_WwaimWyd3gfAg2rrQQsTH3wPkqkbp3_-PpmgEo/edit?usp=sharing

• All existing dags had misspelled num_partitions_insert/upsert config. Have fixed it.
• In all sample dags and config, "timestamp" is used as the partition path. But there was no field for source ordering. Infact same field "timestamp" was set as source ordering field. But for updates, we need a different field as source ordering field. so have introduced a new field called "ts" in source.avsc and have fixed config values appropriately.
• Have added some demo dag yamls for understanding ValidateDatasetNodes (cow-per-round-validate-once.yaml, cow-validate-cumulative-multiple-rounds.yaml, cow-per-round-mixed-validate.yaml)
• Added preCombine field (set to "ts") in configs to be used while generating insert/update records. test suite sets this field's value to batchId. Since this will be used as source ordering field, we leverage batchId for monotonically increasing, so that its easy to reason about updates vs inserts(rather than generating random values).
• Fixed TestSuitePathSelector which was comparing input directors as string rather than integers. And so, after batch9, batch10 wasn't getting returned from path selector.

Brief change log

• Added "ValidateDatasetNode" to compare entire input dataset to hudi content for equality.
• Added two new configs to assist in long running test suite(--num-rounds, --delay-between-rounds-mins)
• All existing dags had misspelled num_partitions_insert/upsert config. Have fixed it.
• In all sample dags and config, "timestamp" is used as the partition path. But there was no field for source ordering. Infact same field "timestamp" was set as source ordering field. But for updates, we need a different field as source ordering field as we can't update the partition path. so have introduced a new field called "ts" in source.avsc and have fixed config values appropriately to be used as source ordering field.
• Have added some demo dag yamls for understanding ValidateDatasetNodes(cow-per-round-validate-once.yaml, cow-validate-cumulative-multiple-rounds.yaml, cow-per-round-mixed-validate.yaml)
• Added preCombine field (set to "ts") in configs to be used while generating insert/update records. test suite sets this field's value to batchId. Since this will be used as source ordering field, we leverage batchId for monotonically increasing, so that its easy to reason about updates vs inserts(rather than generating random values).
• Fixed TestSuitePathSelector which was comparing input directors as string rather than integers. And so, after batch9, batch10 wasn't getting returned from path selector.

Verify this pull request

(Please pick either of the following options)

• Tested locally w/ test inputs in docker setup. These are the test variants verified.
  All tests are done for both COW and MOR(real time). Validation is done by both spark datasource and hive table via spark sql engine.
  a. one round
  b. multiple rounds by validating all data so far.
  c. multiple rounds w/ validation per round

Committer checklist

• [ x] Has a corresponding JIRA in PR title & commit

• Commit message is descriptive of the change

• CI is green

• [ x] Necessary doc changes done or have another open PR

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

[codecov-io]

Codecov Report

Merging #2168 (bc2ff51) into master (5f5c15b) will decrease coverage by 43.12%.
The diff coverage is n/a.

@@              Coverage Diff              @@
##             master    #2168       +/-   ##
=============================================
- Coverage     53.17%   10.05%   -43.13%     
+ Complexity     2711       48     -2663     
=============================================
  Files           346       52      -294     
  Lines         15928     1850    -14078     
  Branches       1636      223     -1413     
=============================================
- Hits           8470      186     -8284     
+ Misses         6769     1651     -5118     
+ Partials        689       13      -676     

Flag	Coverage Δ	Complexity Δ
hudicli	?	?
hudiclient	?	?
hudicommon	?	?
hudihadoopmr	?	?
hudispark	?	?
huditimelineservice	?	?
hudiutilities	10.05% <ø> (-60.05%)	0.00 <ø> (ø)

Flags with carried forward coverage won't be shown. Click here to find out more.

Impacted Files	Coverage Δ	Complexity Δ
...hudi/utilities/schema/FilebasedSchemaProvider.java	0.00% <ø> (-82.36%)	0.00 <0.00> (-5.00)
...va/org/apache/hudi/utilities/IdentitySplitter.java	0.00% <0.00%> (-100.00%)	0.00% <0.00%> (-2.00%)
...va/org/apache/hudi/utilities/schema/SchemaSet.java	0.00% <0.00%> (-100.00%)	0.00% <0.00%> (-3.00%)
...a/org/apache/hudi/utilities/sources/RowSource.java	0.00% <0.00%> (-100.00%)	0.00% <0.00%> (-4.00%)
.../org/apache/hudi/utilities/sources/AvroSource.java	0.00% <0.00%> (-100.00%)	0.00% <0.00%> (-1.00%)
.../org/apache/hudi/utilities/sources/JsonSource.java	0.00% <0.00%> (-100.00%)	0.00% <0.00%> (-1.00%)
...rg/apache/hudi/utilities/sources/CsvDFSSource.java	0.00% <0.00%> (-100.00%)	0.00% <0.00%> (-10.00%)
...g/apache/hudi/utilities/sources/JsonDFSSource.java	0.00% <0.00%> (-100.00%)	0.00% <0.00%> (-4.00%)
...apache/hudi/utilities/sources/JsonKafkaSource.java	0.00% <0.00%> (-100.00%)	0.00% <0.00%> (-6.00%)
...pache/hudi/utilities/sources/ParquetDFSSource.java	0.00% <0.00%> (-100.00%)	0.00% <0.00%> (-5.00%)
... and 326 more

[nsivabalan]

@n3nash : Have fixed the patch to its completion. Please review when you can.

[n3nash]

@nsivabalan why do we need 2 same yaml dags one for once and one for multiple ?

[n3nash]

@nsivabalan
Can we avoid introducing these high level arguments numRound and delayMins to the dagScheduler ? This messes with a simpler design and adds unnecessary overloading to the constructor.
Instead, we could do something like this

Introduce a top level yaml change around this
dag_name:
dag_rounds:
dag_intermittent_delay:
dag_props:
dag_content:
insert_node:
...
upsert_node:
...
You can then introduce a concept called RuntimeComposableDagNode that basically wraps this concept.

Let me know.

[n3nash]

@nsivabalan Left some comments

[nsivabalan]

@n3nash : addressed all feedback man. Please review it when you can.

[nsivabalan]

added some comments for easier review and have asked some clarifying questions.

[nsivabalan]

I haven't fixed these dags yet for long running test-suite. Once we have consensus that things are looking good, I will update all dags. Do check unit test dags under resources dir in hudi-integ-test module for updated version.

[nsivabalan]

as per our discussion, I have added this new field from within the code. So essentially this field will be appended to both source schema and target schema and will be used for source ordering. batch_id is used as values.

[nsivabalan]

high level config param parsing.

[nsivabalan]

as agreed, all node contents go into "dag_content"

[nsivabalan]

this is where the source ordering field is set.

[nsivabalan]

Here is the example of how the dag looks like w/ global params. I have tested this dag for few rounds of validation.

[nsivabalan]

since I am introducing a test schema provider to append source ordering field, I had to make this protected since the test schema provider extends from this.

[nsivabalan]

is this the right way to fetch record key and partition path field ?

[n3nash]

@nsivabalan This will just fetch the KEY names, I'm guessing you need the values for these keys ?

[nsivabalan]

nope. just the record key and partition path field names. we need to group by Hoodiekey in df

inputDf.groupByKey(
        (MapFunction<Row, String>) value -> partitionPathField + "+" + recordKeyField, Encoders.STRING())

[nsivabalan]

@n3nash : the patch is good to be reviewed. would appreciate if you can take some time to close this out.

and a question on DFSHoodieDatasetInputReader. I see when we try to read lesser no of records than total records in one file slice, the reader returns entire contents from one file slice. Is that intentional or is it a bug ?

For eg: if I insert 1500 records
and next node is upsert of 200. this will actually update all 1500 records and not 200(assuming all 1500 is in one file slice).

I ran into this issue while testing out deletes with this patch. I thot there was some issue w/ deletes, since I saw all records were getting deleted ;) and after investigation, found its the reader thats doing this.

if (!numFiles.isPresent() || numFiles.get() == 0) {
      // If num files are not passed, find the number of files to update based on total records to update and records
      // per file
      numFilesToUpdate = (int) Math.ceil((double) numRecordsToUpdate.get() / recordsInSingleFile);
      // recordsInSingleFile is not average so we still need to account for bias is records distribution
      // in the files. Limit to the maximum number of files available.
      int totalExistingFilesCount = partitionToFileIdCountMap.values().stream().reduce((a, b) -> a + b).get();
      numFilesToUpdate = Math.min(numFilesToUpdate, totalExistingFilesCount);
      log.warn("aaa Files to update {}, numRecords toUpdate {}, records in Single file {} ", numFilesToUpdate, numRecordsToUpdate, recordsInSingleFile);
      numRecordsToUpdatePerFile = recordsInSingleFile;
    }

[nsivabalan]

@n3nash : gentle ping.

[n3nash]

@nsivabalan The use-case you described seems to be intentional but the behavior is not correct. If the number of records to update is explicitly asked by the dag, then Option<Long> numRecordsToUpdate should be set. In this case, the code path assumes it's not set. May be there's a bug in setting that variable ?
In general, left 2 comments, after they are addressed, can merge this.

[nsivabalan]

@nsivabalan The use-case you described seems to be intentional but the behavior is not correct. If the number of records to update is explicitly asked by the dag, then Option<Long> numRecordsToUpdate should be set. In this case, the code path assumes it's not set. May be there's a bug in setting that variable ?
In general, left 2 comments, after they are addressed, can merge this.

I don't think so. this code path assume numFiles is not set, not numRecordsToUpdate. Check my comment in code snippet below.

if (!numFiles.isPresent() || numFiles.get() == 0) {
      numFilesToUpdate = (int) Math.ceil((double) numRecordsToUpdate.get() / recordsInSingleFile);
      int totalExistingFilesCount = partitionToFileIdCountMap.values().stream().reduce((a, b) -> a + b).get();
      numFilesToUpdate = Math.min(numFilesToUpdate, totalExistingFilesCount);
      numRecordsToUpdatePerFile = recordsInSingleFile; // this line ignores the numRecordsToUpdate passed in and sets to total records in one single file slice. 
    }

[nsivabalan]

@n3nash : also, if are you ok, I can go ahead and change all dags to this new format.

[n3nash]

@nsivabalan Yes, please go ahead and change the DAG to the new format. If there is any easy way to make sure the DagUtils ensure the new dag and old are backwards compatible (by providing defaults) would be great.

For the following code, see my comment

if (!numFiles.isPresent() || numFiles.get() == 0) {
numFilesToUpdate = (int) Math.ceil((double) numRecordsToUpdate.get() / recordsInSingleFile); // Nishith - This line ensures we are updating only the number of files that we should based on this math. I think the bug will happen when there is only 1 file to update (numRecordsToUpdate < recordsInSingleFile). We should add a check if that's the case and update numRecordsToUpdatePerFile to numRecordsToUpdate in that case.

  int totalExistingFilesCount = partitionToFileIdCountMap.values().stream().reduce((a, b) -> a + b).get();
  numFilesToUpdate = Math.min(numFilesToUpdate, totalExistingFilesCount);
  numRecordsToUpdatePerFile = recordsInSingleFile; // this line ignores the numRecordsToUpdate passed in and sets to total records in one single file slice. 
}

[nsivabalan]

@n3nash : fixed the bug as discussed. you can check it out. Wrt backwards compatability, its just 3 additional configs that one needs to set man. No other changes required. Not sure if we really need to make it backwards compatible. Should be easy to figure it out.

[nsivabalan]

@n3nash : can you please review when you get time.

[n3nash]

LGTM