From f49b86dc47fb1f09a8aaa6642b1b456ef4292391 Mon Sep 17 00:00:00 2001
From: Sam Klock <sklock@akamai.com>
Date: Thu, 19 Apr 2018 16:40:51 -0400
Subject: [PATCH] Optimize bytes skipping when reading SSTables

patch by Samuel Klock; reviewed by  Benedict Elliott Smith, Benjamin Lerer
and Kurt Greaves for CASSANDRA-1441
---
 CHANGES.txt                                   |  1 +
 .../cassandra/io/util/RandomAccessReader.java | 12 +++
 .../io/util/RandomAccessReaderTest.java       | 77 +++++++++++++++++++
 3 files changed, 90 insertions(+)

diff --git a/CHANGES.txt b/CHANGES.txt
index fc98af1604..99e2798dd0 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
 3.11.11
+ * Optimize bytes skipping when reading SSTable files (CASSANDRA-14415)
  * Enable tombstone compactions when unchecked_tombstone_compaction is set in TWCS (CASSANDRA-14496)
  * Read only the required SSTables for single partition queries (CASSANDRA-16737)
  * Fix LeveledCompactionStrategy compacts last level throw an ArrayIndexOutOfBoundsException (CASSANDRA-15669)
diff --git a/src/java/org/apache/cassandra/io/util/RandomAccessReader.java b/src/java/org/apache/cassandra/io/util/RandomAccessReader.java
index 5157eac9ca..b8c1818fe2 100644
--- a/src/java/org/apache/cassandra/io/util/RandomAccessReader.java
+++ b/src/java/org/apache/cassandra/io/util/RandomAccessReader.java
@@ -207,6 +207,18 @@ public class RandomAccessReader extends RebufferingInputStream implements FileDa
         reBufferAt(newPosition);
     }
 
+    @Override
+    public int skipBytes(int n) throws IOException
+    {
+        if (n < 0)
+            return 0;
+        long current = current();
+        long newPosition = Math.min(current + n, length());
+        n = (int)(newPosition - current);
+        seek(newPosition);
+        return n;
+    }
+
     /**
      * Reads a line of text form the current position in this file. A line is
      * represented by zero or more characters followed by {@code '\n'}, {@code
diff --git a/test/unit/org/apache/cassandra/io/util/RandomAccessReaderTest.java b/test/unit/org/apache/cassandra/io/util/RandomAccessReaderTest.java
index 8941d2a6ce..327c656e2b 100644
--- a/test/unit/org/apache/cassandra/io/util/RandomAccessReaderTest.java
+++ b/test/unit/org/apache/cassandra/io/util/RandomAccessReaderTest.java
@@ -498,4 +498,81 @@ public class RandomAccessReaderTest
             }
         }
     }
+
+    @Test
+    public void testSkipBytesLessThanBufferSize() throws IOException
+    {
+        testSkipBytes(new Parameters(8192, 1024), 1);
+    }
+
+    @Test
+    public void testSkipBytesGreaterThanBufferSize() throws IOException
+    {
+        int bufferSize = 16;
+        Parameters params = new Parameters(8192, bufferSize);
+        int numberOfExpectationsInBufferSize = bufferSize / params.expected.length;
+        testSkipBytes(params, numberOfExpectationsInBufferSize + 1);
+    }
+
+    public void testSkipBytesNonPositive() throws IOException
+    {
+        Parameters params = new Parameters(8192, 4096);
+        final File f = writeFile(params);
+        try (FileHandle.Builder builder = new FileHandle.Builder(f.getPath())
+                                                     .bufferType(params.bufferType).bufferSize(params.bufferSize))
+        {
+            builder.mmapped(params.mmappedRegions);
+            try (FileHandle fh = builder.complete();
+                 RandomAccessReader reader = fh.createReader())
+            {
+                assertEquals(0, reader.skipBytes(0));
+                assertEquals(0, reader.skipBytes(-1));
+            }
+        }
+    }
+
+    @Test(expected = NullPointerException.class)
+    public void testSkipBytesClosed() throws IOException
+    {
+        Parameters params = new Parameters(8192, 4096);
+        final File f = writeFile(params);
+        try (FileHandle.Builder builder = new FileHandle.Builder(f.getPath())
+                                                     .bufferType(params.bufferType).bufferSize(params.bufferSize))
+        {
+            try (FileHandle fh = builder.complete();
+                 RandomAccessReader reader = fh.createReader())
+            {
+                reader.close();
+                reader.skipBytes(31415);
+            }
+        }
+    }
+
+    private static void testSkipBytes(Parameters params, int expectationMultiples) throws IOException
+    {
+        final File f = writeFile(params);
+        try (FileHandle.Builder builder = new FileHandle.Builder(f.getPath())
+                                                     .bufferType(params.bufferType).bufferSize(params.bufferSize))
+        {
+            builder.mmapped(params.mmappedRegions);
+            try (FileHandle fh = builder.complete();
+                 RandomAccessReader reader = fh.createReader())
+            {
+                int toSkip = expectationMultiples * params.expected.length;
+                byte[] b = new byte[params.expected.length];
+                long numRead = 0;
+
+                while (numRead < params.fileLength)
+                {
+                    reader.readFully(b);
+                    assertTrue(Arrays.equals(params.expected, b));
+                    numRead += b.length;
+                    int skipped = reader.skipBytes(toSkip);
+                    long expectedSkipped = Math.max(Math.min(toSkip, params.fileLength - numRead), 0);
+                    assertEquals(expectedSkipped, skipped);
+                    numRead += skipped;
+                }
+            }
+        }
+    }
 }