(jackrabbit-oak) branch trunk updated: OAK-11238 - De-duplicate entries when writing sorted intermediate batches of node state entries. (#1835)

[nf...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

nfsantos pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/jackrabbit-oak.git


The following commit(s) were added to refs/heads/trunk by this push:
     new c3e7c2055c OAK-11238 - De-duplicate entries when writing sorted intermediate batches of node state entries. (#1835)
c3e7c2055c is described below

commit c3e7c2055c6f876a2791a561ea385c39cf7130e1
Author: Nuno Santos <ns...@adobe.com>
AuthorDate: Wed Oct 30 17:22:03 2024 +0100

    OAK-11238 - De-duplicate entries when writing sorted intermediate batches of node state entries. (#1835)
---
 .../flatfile/pipelined/PipelinedSortBatchTask.java       | 16 +++++++++++++---
 1 file changed, 13 insertions(+), 3 deletions(-)

diff --git a/oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/pipelined/PipelinedSortBatchTask.java b/oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/pipelined/PipelinedSortBatchTask.java
index 1ce54aa09b..a375337ca3 100644
--- a/oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/pipelined/PipelinedSortBatchTask.java
+++ b/oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/pipelined/PipelinedSortBatchTask.java
@@ -36,6 +36,7 @@ import java.nio.charset.StandardCharsets;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Comparator;
 import java.util.Locale;
 import java.util.concurrent.BlockingQueue;
@@ -195,9 +196,16 @@ class PipelinedSortBatchTask implements Callable<PipelinedSortBatchTask.Result>
         Path newtmpfile = Files.createTempFile(sortWorkDir, "sortInBatch", "flatfile");
         long textSize = 0;
         batchesProcessed++;
+        String[] prevPathElements = null;
+        int duplicateEntriesSkippped = 0;
         try (OutputStream os = IndexStoreUtils.createOutputStream(newtmpfile, algorithm)) {
             for (SortKey entry : sortBuffer) {
                 entriesProcessed++;
+                if (Arrays.equals(prevPathElements, entry.getPathElements())) {
+                    duplicateEntriesSkippped++;
+                    continue;
+                }
+                prevPathElements = entry.getPathElements();
                 // Retrieve the entry from the buffer
                 int posInBuffer = entry.getBufferPos();
                 buffer.position(posInBuffer);
@@ -213,11 +221,13 @@ class PipelinedSortBatchTask implements Callable<PipelinedSortBatchTask.Result>
         }
         timeWritingMillis += saveClock.elapsed().toMillis();
         long compressedSize = Files.size(newtmpfile);
-        LOG.info("Wrote batch of size {} (uncompressed {}) with {} entries in {} at {}",
+        int entriesWritten = sortBuffer.size() - duplicateEntriesSkippped;
+        LOG.info("Wrote batch of size {} (uncompressed {}) with {} entries in {} at {}. Duplicate entries skipped: {}",
                 humanReadableByteCountBin(compressedSize),
                 humanReadableByteCountBin(textSize),
-                sortBuffer.size(), saveClock,
-                PipelinedUtils.formatAsTransferSpeedMBs(compressedSize, saveClock.elapsed().toMillis())
+                entriesWritten, saveClock,
+                PipelinedUtils.formatAsTransferSpeedMBs(compressedSize, saveClock.elapsed().toMillis()),
+                duplicateEntriesSkippped
         );
         // Free the memory taken by the entries in the buffer
         sortBuffer.clear();