[HUDI-3322][HUDI-3343] Fixing Metadata Table Records Duplication Issues

[alexeykudinkin]

Tips

• Thank you very much for contributing to Apache Hudi.
• Please review https://hudi.apache.org/contribute/how-to-contribute before opening a pull request.

What is the purpose of the pull request

This change is addressing issues in regards to Metadata Table observing ingesting duplicated records leading to it persisting incorrect file-sizes for the files referred to in those records.

There are multiple issues that were leading to that:

1. [HUDI-3322] Incorrect Rollback Plan generation: Rollback Plan generated for MOR tables was overly expansively listing all log-files with the latest base-instant as the ones that have been affected by the rollback, leading to invalid MT records being ingested referring to those.
2. [HUDI-3343] Metadata Table including Uncommitted Log Files during Bootstrap: Since MT is bootstrapped at the end of the commit operation execution (after FS activity, but before committing to the timeline), it was actually incorrectly ingesting some files that were part of the intermediate state of the operation being committed.

This change will unblock Stack of PRs based off #4556

Brief change log

• Cleaned up Rollback request generation flow
• Cleaned up superfluous writtenLogFilesSize payload
• Fixed RollbackUtils to always leverage appropriate instant when fetching metadata
• Propagate HoodieWriteStat w/in ListingBasedRollbackRequest
• Fixed RollbackPlan generation to properly refeference files bearing blocks
• Fixed superfluous RollbackPlan generation
• Fixing MT to be bootstapped before carrying out any FS operations, to make sure it's not ingesting intermediate step upon bootstrapping
• Fix MarkerBasedRollbackStrategy to only ref the latest log-file as the one blocks have been appended to

Verify this pull request

This pull request is already covered by existing tests, such as (please describe tests).

This PR fixing following tests that started to fail after changes in #4556:

TestHoodieSparkMergeOnReadTableRollback#testMORTableRestore
TestHoodieSparkMergeOnReadTableRollback#testRollbackWithDeltaAndCompactionCommit

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.

[alexeykudinkin]

This is addressing HUDI-3322

[alexeykudinkin]

@hudi-bot run azure

[nsivabalan]

Here is my understanding of the fix.
We fix the rollback plan generated in ListingBasedStrategy to just include the log files with full sizes.
For Marker bases strategy, we set file sizes to -1, but fixed the right set of log files to be included in rollback plan.

So, we still have an issue w/ how we reconcile or merge multiple metadata records.
For eg:
Rec1: file1 delta size 100 (commit1)
Rec2: file1 deltasize 200 (commit2)
Rec3: file1 full size 350 (rollback)

when we merge all these 3 records from metadata table, whats the final resolved record look like ?

wrt MDT bootstrap, we ensure to trigger bootstrap before starting any operation.

[nsivabalan]

this may not be backwards compatible while reading rollback metadata written w/ 0.10.0 or previous versions.

[alexeykudinkin]

I've researched this a bit since my intuition was telling that fields removal is BWC change in Avro (since Avro also supports forward compatibility, when you essentially read data with old schema that was produced with a new one, which is essentially the same case)

And seems like deletion is BWC in Avro:
https://docs.confluent.io/platform/current/schema-registry/avro.html

[alexeykudinkin]

This is an excerpt from Avro's docs:
https://avro.apache.org/docs/1.7.7/spec.html#Schema+Resolution

[nsivabalan]

hmmm, interesting. Can you try this out explicitly. write an avro using master branch. and then try to read it using this branch. or you can just try it out using a stand alone java main class too, which ever works.
just wanted to ensure we are good here.

[nsivabalan]

done w/ one pass.

[nsivabalan]

If you don't mind, can we add tests for the fix in this patch. The tests should fail if not the fix and should pass w/ the fix.

[alexeykudinkin]

Here is my understanding of the fix.
We fix the rollback plan generated in ListingBasedStrategy to just include the log files with full sizes.
For Marker bases strategy, we set file sizes to -1, but fixed the right set of log files to be included in rollback plan.

So, we still have an issue w/ how we reconcile or merge multiple metadata records.
For eg:
Rec1: file1 delta size 100 (commit1)
Rec2: file1 deltasize 200 (commit2)
Rec3: file1 full size 350 (rollback)

when we merge all these 3 records from metadata table, whats the final resolved record look like ?

When we do a rollback the value in the plan (carrying the mapping of path to size) is only used to determine whether we should append Rollback Block or delete files. After we actually appended the Rollback Block, we now only modify the record related to the file we've appended the block to (previously we would also update all the log-files from the writtenLogFileSizesMap)

If you don't mind, can we add tests for the fix in this patch. The tests should fail if not the fix and should pass w/ the fix.

There are already tests covering this, which were failing in #4556 (which is how i come to fix this issues).
The reason why these were not failing is simply b/c we're not using MT table when we're reading the data back using Hive's InputFormats.

[hudi-bot]

CI report:

• 62b022e UNKNOWN
• 550d8e3 Azure: SUCCESS

Bot commands

@hudi-bot supports the following commands:

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

[nsivabalan]

just 1 comment on the avro BWC. once verified, we are good to go ahead.

[alexeykudinkin]

@nsivabalan here's the test:

  @Test
  void testAvroBWC() throws IOException {
    Writer w = new Writer("foo", 0xDEAD);
    byte[] buf = HoodieAvroUtils.avroToBytes(w);
    GenericRecord r = HoodieAvroUtils.bytesToAvro(buf, Writer.SCHEMA$, Reader.SCHEMA$);
    assertEquals(w.getKey(), r.get("key"));
  }

For the following schema:

{
  "namespace": "org.apache.hudi.avro.model",
  "type": "record",
  "name": "TestAvroRecord",
  "fields": [
    {
      "name": "writer",
      "type": ["null", {
        "type": "record",
        "name": "Writer",
        "fields": [
          {
            "name": "key",
            "type": ["null", "string"]
          },
          {
            "name": "value",
            "type": ["null", "int"]
          }
        ]
      }]
    },
    {
      "name": "reader",
      "type": ["null",     {
        "name": "Reader",
        "type": "record",
        "fields": [
          {
            "name": "key",
            "type": ["null", "string"]
          }
        ]
      }]
    }
  ]
}

Working as expected

[nsivabalan]

cool. thanks!
btw, can you remind me something. with this patch, whats the size we set in rollback metadata for log files where rollback block command is added?