[HUDI-8210] Support writing table version 6 log format from 1.x #12206

vinothchandar · 2024-11-04T23:21:30Z

Change Logs

Annotate log headers, blocks with table version (tv)
Handle both tv=6, and tv=8 for log naming
Eliminate notion of rolloverWriteToken
Simplify/make efficient log writer building across code paths
Handle both tv=6 & 8 for log version, handling
Prevent tv=8 headers from being written with tv=6.
Bring back tv=6 rollback behavior.

Impact

Hudi writers can now write tableVersion 6 compatible log format, in addition to tableVersion 8.

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

low

Documentation Update

No new configurations.

Contributor's checklist

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

vinothchandar · 2024-11-05T01:58:36Z

@hudi-bot run azure

vinothchandar · 2024-11-05T02:05:25Z

@hudi-bot run azure

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieAppendHandle.java

danny0405 · 2024-11-06T04:26:11Z

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieAppendHandle.java

+        prevCommit = instantTime;
+        // Handle log file only case. This is necessary for the concurrent clustering and writer case (e.g., consistent hashing bucket index).
+        // NOTE: flink engine use instantTime to mark operation type, check BaseFlinkCommitActionExecutor::execute
+        if (record.getCurrentLocation() != null && HoodieInstantTimeGenerator.isValidInstantTime(record.getCurrentLocation().getInstantTime())) {


It should be just getFileInstant(record).

do you see an issue with keeping this as-is.. Makes it easy to debug across 0.15 and 1.0.

The logic are actually the same.

@danny0405 can you check if you are ok. renamed and pushed a change

danny0405 · 2024-11-06T04:28:32Z

...di-client-common/src/main/java/org/apache/hudi/table/action/rollback/BaseRollbackHelper.java


          writer = HoodieLogFormat.newWriterBuilder()
              .onParentPath(FSUtils.constructAbsolutePath(metaClient.getBasePath(), rollbackRequest.getPartitionPath()))
              .withFileId(fileId)
-              .withDeltaCommit(instantToRollback.getTimestamp())
+              .withLogWriteToken(CommonClientUtils.generateWriteToken(taskContextSupplier))


The write token is meaningless because the rollback tasks have different task context with the writer task.

I'll check this out.. I was trying to avoid doing FS listings or sth expensive, without this being passed in

at-least for Spark - this makes sense I think. each task is a rollback to a given file group.. So they'd have different write tokens. My intention don't want to compute this from storage .. lmk if you have concerns with proceeding

...di-client-common/src/main/java/org/apache/hudi/table/action/rollback/BaseRollbackHelper.java

...-common/src/main/java/org/apache/hudi/table/action/rollback/MarkerBasedRollbackStrategy.java

hudi-common/src/main/java/org/apache/hudi/common/table/log/HoodieLogFormat.java

danny0405 · 2024-11-06T06:52:01Z

@vinothchandar Thanks for the contribution, is the write version and table version the same stuff, what happens when these two are inconsistent?

vinothchandar · 2024-11-06T11:10:51Z

@danny0405 write fails in that case (post the upgrade flow). I wired in the check CommonClientUtils::validateXXX in the last PR

danny0405 · 2024-11-07T00:55:26Z

@vinothchandar I'm wondering if we can just abstract out the function discrepencies as a new method that can be extensible, then we can have a HoodieAppendHandleFactory for current and legacy impls. Like what we do to HoodieMergeHandle: we introduce the HoodieMergeHandleFactory to handle ramifications of regular write and cdc write. This way we can make the current HoodieAppendHandle much cleaner and more easier to maintain.

Same thoughts to HoodieLogFormat.

vinothchandar · 2024-11-13T03:06:56Z

@danny0405 Filed https://issues.apache.org/jira/browse/HUDI-8509 to track better abstractions. They are pretty localized changes for now. (unlike timeline, which needs a better abstractions due to spread of use)

vinothchandar · 2024-11-13T03:17:14Z

@bvaradar @danny0405 Can you review this again.

@bvaradar Following tests I could not get to given the change of plans to land this first.

When you test this alongside yours, please consider.

Functional test that writes tv=6 CoW/MoR table, can successful compact/rollback (marker based) without needing upgrade to 8.
Spark should be able to read logs with rollbacks (TestHoodieLogFormat already covers non file group reader tests)

hudi-common/src/main/java/org/apache/hudi/common/table/view/AbstractTableFileSystemView.java

vinothchandar · 2024-11-13T18:49:34Z

@danny0405 Can you do a full review please? so I can address in one go - or is that the only comment you had?

- Annotate log headers, blocks with table version (tv) - Handle both tv=6, and tv=8 for log naming - Eliminate notion of rolloverWriteToken - Simplify/make efficient log writer building across code paths - Handle both tv=6 & 8 for log version, handling - Prevent tv=8 headers from being written with tv=6. - Bring back tv=6 rollback behavior.

- Simple adapters in AbstractFileSystemView to handle filtering of uncommitted logs/files in 0.x - Validations for disallowed writer configs around NBCC, log format for table version=6 writing from 1.x - TestHoodieTableFileSystemView tests have all been redone to handle both styles of log naming - New unit tests

danny0405 · 2024-11-14T01:27:57Z

hudi-common/src/main/java/org/apache/hudi/common/table/view/AbstractTableFileSystemView.java

@@ -1039,13 +1109,14 @@ public final Stream<FileSlice> getLatestMergedFileSlicesBeforeOrOn(String partit
  public final Stream<FileSlice> getAllLogsMergedFileSliceBeforeOrOn(String partitionStr, String maxInstantTime) {
    try {
      readLock.lock();
+      HoodieTableVersion version = metaClient.getTableConfig().getTableVersion();


This is not used.

danny0405 · 2024-11-14T01:28:20Z

hudi-common/src/main/java/org/apache/hudi/common/table/view/AbstractTableFileSystemView.java

@@ -987,13 +1051,17 @@ private Option<FileSlice> getLatestFileSliceFilteringUncommittedFiles(Stream<Fil
  public final Map<String, Stream<FileSlice>> getAllLatestFileSlicesBeforeOrOn(String maxCommitTime) {
    try {
      readLock.lock();
+      HoodieTableVersion version = metaClient.getTableConfig().getTableVersion();


This is not used.

danny0405 · 2024-11-14T01:32:26Z

hudi-common/src/main/java/org/apache/hudi/common/table/view/AbstractTableFileSystemView.java

-        return Option.ofNullable(filterUncommittedFiles(fs.get(), true).map(this::addBootstrapBaseFileIfPresent).findFirst().orElse(null));
+        Stream<FileSlice> fileSlices = tableVersion8AndAbove()
+            ? this.filterUncommittedFiles(fs.get(), true)
+            : this.filterBaseFileAfterPendingCompaction(fs.get(), true);


Do we need to also fix the merged log record scanner to bring back the pending instant check for the log blocks because of these unfiltered log files?

vinothchandar · 2024-11-14T06:55:30Z

Pending

Fixing changes/adjusting in AbstractHoodieLogRecordReader.. Context: [HUDI-6952] Skip reading the uncommitted log files for log reader #9879
Corrupt block handling, since there is no way to say if sth is legit or just a rollback block
Look at the FileGroupReader code path as well for handling these types of cases

hudi-bot · 2024-11-14T08:00:50Z

CI report:

408be5e Azure: SUCCESS
c2315d8 Azure: PENDING

Bot commands

@hudi-bot supports the following commands:

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

github-actions bot added the size:L PR with lines of changes in (300, 1000] label Nov 4, 2024

vinothchandar force-pushed the hudi-8210-code-prep branch 2 times, most recently from 94f39b3 to 6b433e7 Compare November 5, 2024 01:42

vinothchandar requested a review from bvaradar November 5, 2024 02:00

apache deleted a comment from hudi-bot Nov 5, 2024

vinothchandar force-pushed the hudi-8210-code-prep branch 2 times, most recently from ac58a5d to 52f31ab Compare November 5, 2024 04:11