[HUDI-8163] Refactor UnMergedLogHandler with iterators #12608

vamshipasunuru · 2025-01-09T12:56:52Z

Change Logs

Refactor code to deprecate usage of callbacks and replace with iterators

Impact

Describe any public API or user-facing feature change or any performance impact.

Improves performance by using iterator to produce records on-demand.

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

low.

Contributor's checklist

Read through contributor's guide
Change Logs and Impact were stated clearly
Adequate tests were added if applicable, TestHoodieRealtimeRecordReader#testUnMergedReader contains the test the covers newly added iterator
CI passed

vamshipasunuru · 2025-01-09T12:57:34Z

@danny0405

danny0405 · 2025-01-10T02:28:09Z

...t/hudi-spark-client/src/main/java/org/apache/hudi/client/utils/SparkMetadataWriterUtils.java

        .build();
-    scanner.scan(false);
+    Iterator<HoodieRecord<?>> recordIterator = scanner.iterator();


Is it possible we return an iterator too for method getUnmergedLogFileRecords to the list add/get can be simplified.

danny0405 · 2025-01-10T02:30:02Z

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

@@ -617,6 +617,8 @@ && compareTimestamps(logBlock.getLogBlockHeader().get(INSTANT_TIME), GREATER_THA
  /**
   * Iterate over the GenericRecord in the block, read the hoodie key and partition path and call subclass processors to
   * handle it.
+   * TODO:
+   * 1. what is the purpose of this method? should the HoodieRecord be added to a queue and consumed by an iterator?
   */
  private void processDataBlock(HoodieDataBlock dataBlock, Option<KeySpec> keySpecOpt) throws Exception {


This should be the main entry for handling the log blocks, each kind of log block would trigger invocation of specific row-level handling methods.

danny0405 · 2025-01-10T02:33:33Z

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

-    }
-
-    public Builder withRecordDeletionCallback(RecordDeletionCallback recordDeletionCallback) {
-      this.recordDeletionCallback = recordDeletionCallback;


The recordDeletionCallback is only consumer for deleted reords from the delete avro block, we may need another iterator impl specifically for it.

danny0405 · 2025-01-10T02:36:33Z

hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadataUtil.java

+      while (recordIterator.hasNext()) {
+        try {
+          records.add(recordIterator.next());
+        } catch (Exception e) {


The code is exactly the same if we collect all the records in the collection again , maybe we do not remove those two callbacks and just add a new iterator impl for the case that really need it: the Flink streaming reader.

i kept the callbacks and added iterator for flink usage.

This reverts commit 6354f79.

danny0405 · 2025-01-16T01:46:20Z

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

+   * Returns an iterator over the log records.
+   */
+  public Iterator<HoodieRecord<?>> iterator() {
+    List<HoodieRecord<?>> records = new ArrayList<>();


Can we try to eliminate the list and return a nested iterator if possible?

Good idea, replaced with nested ClosableIterator.

danny0405 · 2025-01-16T01:51:43Z

hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/realtime/RealtimeUnmergedRecordReader.java

-                  .build();
-          // Scan all the delta-log files, filling in the queue
-          scanner.scan();
+          HoodieUnMergedLogRecordScanner scanner = scannerBuilder.build();


Nice refactoring, can we also fix the usage of HoodieUnMergedLogRecordScanner for Flink MergeOnReadInputFormat#getUnMergedLogFileIterator

Adjusted the usage within BoundedMemoryRecords in MergeOnReadInputFormat#getUnMergedLogFileIterator

The BoundedMemoryRecords can be eliminated.

danny0405 · 2025-01-18T02:18:42Z

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

+
+      @Override
+      public HoodieRecord<?> next() {
+        if (!hasNext()) {


Maybe we remove this check because hasNext should always be invoked before next.

danny0405 · 2025-01-18T02:19:27Z

hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/FormatUtils.java

+        HoodieUnMergedLogRecordScanner scanner = scannerBuilder.build();
+        try (ClosableIterator<HoodieRecord<?>> logRecordIterator = scanner.iterator()) {
+          while (logRecordIterator.hasNext()) {
+            queue.insertRecord(logRecordIterator.next());


We can remove the whole producers and queue I think.

danny0405 · 2025-01-21T04:06:17Z

hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/FormatUtils.java

@@ -181,9 +172,6 @@ public static HoodieMergedLogRecordScanner logScanner(
   * Utility to read and buffer the records in the unMerged log record scanner.
   */
  public static class BoundedMemoryRecords {
-    // Executor that runs the above producers in parallel


BoundedMemoryRecords can be eliminated too I guess.

vamshipasunuru · 2025-01-21T07:15:38Z

Yes can be removed, theres a post processing that drains the queue, will fix.

…

On Mon, Jan 20, 2025 at 8:06 PM Danny Chan ***@***.***> wrote: ***@***.**** commented on this pull request. ------------------------------ In hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/FormatUtils.java <#12608 (comment)>: > @@ -181,9 +172,6 @@ public static HoodieMergedLogRecordScanner logScanner( * Utility to read and buffer the records in the unMerged log record scanner. */ public static class BoundedMemoryRecords { - // Executor that runs the above producers in parallel BoundedMemoryRecords can be eliminated too I guess. — Reply to this email directly, view it on GitHub <#12608 (review)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/A6NDCQI5UHSGFKFYZWFAKYT2LXBU7AVCNFSM6AAAAABU4DKOA2VHI2DSMVQWIX3LMV43YUDVNRWFEZLROVSXG5CSMV3GSZLXHMZDKNRTGQZTGMRXHE> . You are receiving this because you authored the thread.Message ID: ***@***.***>

danny0405 · 2025-02-04T03:49:23Z

hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/FormatUtils.java

+            flinkConf.getInteger(HoodieMemoryConfig.MAX_DFS_STREAM_BUFFER_SIZE.key(),
+                HoodieMemoryConfig.DEFAULT_MR_MAX_DFS_STREAM_BUFFER_SIZE))
+        .withInstantRange(split.getInstantRange())
+        .withRecordMerger(merger)


Not sure whether the merger is necessary because for HoodieUnMergedLogRecordScanner, the merge never happens.

danny0405 · 2025-02-04T03:49:53Z

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

@@ -372,7 +372,7 @@ private void scanInternalV1(Option<KeySpec> keySpecOpt) {
        }
      }
      // merge the last read block when all the blocks are done reading
-      if (!currentInstantLogBlocks.isEmpty()) {
+      if (!currentInstantLogBlocks.isEmpty() && !skipProcessingBlocks) {


Is this change related with the fix?

yes, without this whatever log instants (currentInstantLogBlocks) that are created during scan() are processed immediately in processQueuedBlocksForInstant.
The v2 scan exposes a interface to skip this processing, but I didn't want to migrate from v1 to v2 as part of this change, hence added this logic to skip in v1.

hudi-bot · 2025-02-04T07:41:08Z

CI report:

ba9a9ef UNKNOWN
f25f5bc UNKNOWN
28c2982 Azure: SUCCESS

Bot commands

@hudi-bot supports the following commands:

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

Replace callbacks with iterator

6354f79

vamshipasunuru marked this pull request as draft January 9, 2025 12:57

github-actions bot added the size:M PR with lines of changes in (100, 300] label Jan 9, 2025

danny0405 reviewed Jan 10, 2025

View reviewed changes

vamshipasunuru and others added 4 commits January 14, 2025 20:25

Revert "Replace callbacks with iterator"

64a9756

This reverts commit 6354f79.

save

6425371

Review comments

8a46112

fix cs

49c86fa

vamshipasunuru marked this pull request as ready for review January 15, 2025 17:17

danny0405 reviewed Jan 16, 2025

View reviewed changes

return nested iterator

930e1e9

vamshipasunuru changed the title ~~[Early feedback] [HUDI-8163] Refactor UnMergedLogHandler with iterators~~ [HUDI-8163] Refactor UnMergedLogHandler with iterators Jan 17, 2025

vamshipasunuru marked this pull request as draft January 17, 2025 08:03

vamshipasunuru added 4 commits January 17, 2025 00:57

Scan before iterating

33c8c4a

fix cs

ba9a9ef

fix cs

5ca19a8

fix cs

cde5b65

danny0405 reviewed Jan 18, 2025

View reviewed changes

vamshipasunuru added 2 commits January 19, 2025 18:03

remove queue

f25f5bc

save cs

d02d371

danny0405 reviewed Jan 21, 2025

View reviewed changes

vamshipasunuru added 2 commits January 25, 2025 13:10

save changes

072c7dd

Replace BoundedMemoryRecords with iterator

3df692e

github-actions bot added size:L PR with lines of changes in (300, 1000] and removed size:M PR with lines of changes in (100, 300] labels Jan 28, 2025

danny0405 reviewed Feb 4, 2025

View reviewed changes

avoid passing merge

28c2982

vamshipasunuru marked this pull request as ready for review February 4, 2025 19:23

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[HUDI-8163] Refactor UnMergedLogHandler with iterators #12608

[HUDI-8163] Refactor UnMergedLogHandler with iterators #12608

vamshipasunuru commented Jan 9, 2025 •

edited

Loading

vamshipasunuru commented Jan 9, 2025

danny0405 Jan 10, 2025

danny0405 Jan 10, 2025

danny0405 Jan 10, 2025

danny0405 Jan 10, 2025

vamshipasunuru Jan 15, 2025

danny0405 Jan 16, 2025

vamshipasunuru Jan 17, 2025 •

edited

Loading

danny0405 Jan 16, 2025

vamshipasunuru Jan 17, 2025

danny0405 Jan 18, 2025

danny0405 Jan 18, 2025

danny0405 Jan 18, 2025

danny0405 Jan 21, 2025

vamshipasunuru commented Jan 21, 2025 via email

danny0405 Feb 4, 2025

danny0405 Feb 4, 2025

vamshipasunuru Feb 4, 2025

hudi-bot commented Feb 4, 2025

[HUDI-8163] Refactor UnMergedLogHandler with iterators #12608

Are you sure you want to change the base?

[HUDI-8163] Refactor UnMergedLogHandler with iterators #12608

Conversation

vamshipasunuru commented Jan 9, 2025 • edited Loading

Change Logs

Impact

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

Contributor's checklist

vamshipasunuru commented Jan 9, 2025

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

vamshipasunuru Jan 17, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

vamshipasunuru commented Jan 21, 2025 via email

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

hudi-bot commented Feb 4, 2025

CI report:

vamshipasunuru commented Jan 9, 2025 •

edited

Loading

vamshipasunuru Jan 17, 2025 •

edited

Loading