fix: ensure that InlineFS is seeked to the correct offset upon init (#14178)

voonhous · yihua · commit 4ecd27265730 · 2025-10-30T13:59:31.000-07:00
diff --git a/hudi-hadoop-common/src/main/java/org/apache/hudi/hadoop/fs/inline/InLineFileSystem.java b/hudi-hadoop-common/src/main/java/org/apache/hudi/hadoop/fs/inline/InLineFileSystem.java
@@ -43,6 +43,29 @@
  * - Reading an inlined file at a given offset, length, read it out as if it were an independent file of that length
  * - Inlined path is of the form "inlinefs:///path/to/outer/file/<outer_file_scheme>/?start_offset=<start_offset>&length=<length>
  * <p>
+ * Example:
+ * <pre>
+ * inlinefs://tests_7af7f087-c807-4f5e-a759-65fd9c21063b/hudi_multi_fg_pt_v8_mor/.hoodie/metadata/column_stats/
+ * .col-stats-0001-0_20250429145946675.log.1_1-120-382/local/?start_offset=8036&length=6959
+ * </pre>
+ * <p>
+ * In this example:
+ * <ul>
+ *   <li>The outer file path is: {@code tests_7af7f087-c807-4f5e-a759-65fd9c21063b/hudi_multi_fg_pt_v8_mor/.hoodie/metadata/
+ *       column_stats/.col-stats-0001-0_20250429145946675.log.1_1-120-382}</li>
+ *   <li>The outer file scheme is: {@code local} (representing local file system)</li>
+ *   <li>The inline content starts at byte offset: {@code 8036}</li>
+ *   <li>The inline content length is: {@code 6959} bytes</li>
+ * </ul>
+ * <p>
+ * When this path is opened, the file system will:
+ * <ol>
+ *   <li>Extract the outer file path and scheme from the URL</li>
+ *   <li>Open the outer file using the appropriate file system</li>
+ *   <li>Seek to the start_offset (8036 in the example)</li>
+ *   <li>Present the next 'length' bytes (6959 in the example) as if they were an independent file</li>
+ * </ol>
+ * <p>
  * TODO: The reader/writer may try to use relative paths based on the inlinepath and it may not work. Need to handle
  * this gracefully eg. the parquet summary metadata reading. TODO: If this shows promise, also support directly writing
  * the inlined file to the underneath file without buffer
diff --git a/hudi-io/src/main/java/org/apache/hudi/storage/inline/InLineFSUtils.java b/hudi-io/src/main/java/org/apache/hudi/storage/inline/InLineFSUtils.java
@@ -38,9 +38,18 @@ public class InLineFSUtils {
    * Get the InlineFS Path for a given schema and its Path.
    * <p>
    * Examples:
+   * <pre>
    * Input Path: s3a://file1, origScheme: file, startOffset = 20, length = 40
    * Output: "inlinefs://file1/s3a/?start_offset=20&length=40"
    *
+   * Real-world example:
+   * Input Path: tests_7af7f087-c807-4f5e-a759-65fd9c21063b/hudi_multi_fg_pt_v8_mor/.hoodie/metadata/column_stats/
+   *             .col-stats-0001-0_20250429145946675.log.1_1-120-382
+   * origScheme: local, startOffset = 8036, length = 6959
+   * Output: "inlinefs://tests_7af7f087-c807-4f5e-a759-65fd9c21063b/hudi_multi_fg_pt_v8_mor/.hoodie/metadata/column_stats/
+   *          .col-stats-0001-0_20250429145946675.log.1_1-120-382/local/?start_offset=8036&length=6959"
+   * </pre>
+   *
    * @param outerPath         The outer file path
    * @param origScheme        The file schema
    * @param inLineStartOffset Start offset for the inline file
diff --git a/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/io/InlineSeekableDataInputStream.java b/hudi-trino-plugin/src/main/java/io/trino/plugin/hudi/io/InlineSeekableDataInputStream.java
@@ -17,17 +17,39 @@
 
 import java.io.IOException;
 
+/**
+ * A seekable data input stream that reads inline content from an outer file at a specific offset and length.
+ * <p>
+ * This class is used when reading inline files stored within larger files (e.g., log blocks embedded in Hudi log files).
+ * It provides a view of a segment of the underlying stream as if it were an independent file.
+ * <p>
+ * Example InlineFS URL:
+ * <pre>
+ * inlinefs://tests_7af7f087-c807-4f5e-a759-65fd9c21063b/hudi_multi_fg_pt_v8_mor/.hoodie/metadata/column_stats/
+ * .col-stats-0001-0_20250429145946675.log.1_1-120-382/local/?start_offset=8036&length=6959
+ * </pre>
+ * <p>
+ * Key behaviors:
+ * <ul>
+ *   <li>Upon initialization, the underlying stream is immediately seeked to the start offset to ensure correct positioning</li>
+ *   <li>{@link #getPos()} returns positions relative to the start offset (0-based from the inline content start)</li>
+ *   <li>{@link #seek(long)} accepts positions relative to the start offset and translates them to absolute positions</li>
+ *   <li>Attempting to seek beyond the length throws an {@link IOException}</li>
+ * </ul>
+ */
 public class InlineSeekableDataInputStream
         extends TrinoSeekableDataInputStream
 {
     private final long startOffset;
     private final long length;
 
     public InlineSeekableDataInputStream(TrinoInputStream stream, long startOffset, long length)
+            throws IOException
     {
         super(stream);
         this.startOffset = startOffset;
         this.length = length;
+        stream.seek(startOffset);
     }
 
     @Override
diff --git a/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/io/TestInlineSeekableDataInputStream.java b/hudi-trino-plugin/src/test/java/io/trino/plugin/hudi/io/TestInlineSeekableDataInputStream.java
@@ -0,0 +1,244 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.trino.plugin.hudi.io;
+
+import io.trino.filesystem.TrinoInputStream;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+class TestInlineSeekableDataInputStream
+{
+    public static final String CONST_STR_FOR_BYTES = "0123456789ABCDEFGHIJ";
+
+    @Test
+    void testStreamIsSeekableToStartOffsetUponInitialization()
+            throws IOException
+    {
+        // Create a test stream with data at various positions
+        byte[] data = CONST_STR_FOR_BYTES.getBytes(StandardCharsets.UTF_8);
+        TestTrinoInputStream stream = new TestTrinoInputStream(data);
+
+        long startOffset = 5;
+        long length = 10;
+
+        // Initialize InlineSeekableDataInputStream
+        InlineSeekableDataInputStream inlineStream = new InlineSeekableDataInputStream(stream, startOffset, length);
+
+        // Verify the stream was seeked to the startOffset during initialization
+        assertThat(stream.getPosition()).isEqualTo(startOffset);
+
+        // Verify getPos() returns 0 (relative to startOffset)
+        assertThat(inlineStream.getPos()).isEqualTo(0);
+    }
+
+    @Test
+    void testGetPosReturnsRelativePosition()
+            throws IOException
+    {
+        byte[] data = CONST_STR_FOR_BYTES.getBytes(StandardCharsets.UTF_8);
+        TestTrinoInputStream stream = new TestTrinoInputStream(data);
+
+        long startOffset = 3;
+        long length = 10;
+
+        InlineSeekableDataInputStream inlineStream = new InlineSeekableDataInputStream(stream, startOffset, length);
+
+        // Initially at position 0 (relative)
+        assertThat(inlineStream.getPos()).isEqualTo(0);
+
+        // Seek to position 5 (relative)
+        inlineStream.seek(5);
+        assertThat(inlineStream.getPos()).isEqualTo(5);
+
+        // Verify underlying stream is at startOffset + 5
+        assertThat(stream.getPosition()).isEqualTo(startOffset + 5);
+    }
+
+    @Test
+    void testSeekWithinBounds()
+            throws IOException
+    {
+        byte[] data = CONST_STR_FOR_BYTES.getBytes(StandardCharsets.UTF_8);
+        TestTrinoInputStream stream = new TestTrinoInputStream(data);
+
+        long startOffset = 2;
+        long length = 8;
+
+        InlineSeekableDataInputStream inlineStream = new InlineSeekableDataInputStream(stream, startOffset, length);
+
+        // Seek to the middle
+        inlineStream.seek(4);
+        assertThat(inlineStream.getPos()).isEqualTo(4);
+        assertThat(stream.getPosition()).isEqualTo(startOffset + 4);
+
+        // Seek to the end (length is exclusive, so seeking to length should work)
+        inlineStream.seek(8);
+        assertThat(inlineStream.getPos()).isEqualTo(8);
+        assertThat(stream.getPosition()).isEqualTo(startOffset + 8);
+    }
+
+    @Test
+    void testSeekPastLengthThrowsException()
+            throws IOException
+    {
+        byte[] data = CONST_STR_FOR_BYTES.getBytes(StandardCharsets.UTF_8);
+        TestTrinoInputStream stream = new TestTrinoInputStream(data);
+
+        long startOffset = 5;
+        long length = 10;
+
+        InlineSeekableDataInputStream inlineStream = new InlineSeekableDataInputStream(stream, startOffset, length);
+
+        // Attempting to seek past the length should throw IOException
+        assertThatThrownBy(() -> inlineStream.seek(11))
+                .isInstanceOf(IOException.class)
+                .hasMessageContaining("Attempting to seek past inline content")
+                .hasMessageContaining("position to seek to is 11")
+                .hasMessageContaining("but the length is 10");
+    }
+
+    @Test
+    void testReadDataAtCorrectOffset()
+            throws IOException
+    {
+        byte[] data = CONST_STR_FOR_BYTES.getBytes(StandardCharsets.UTF_8);
+        TestTrinoInputStream stream = new TestTrinoInputStream(data);
+
+        long startOffset = 5; // Start at '5'
+        long length = 5;      // Read 5 bytes: "56789"
+
+        InlineSeekableDataInputStream inlineStream = new InlineSeekableDataInputStream(stream, startOffset, length);
+
+        // Read first byte should be '5' (byte at position startOffset)
+        int firstByte = inlineStream.read();
+        assertThat(firstByte).isEqualTo('5');
+        assertThat(inlineStream.getPos()).isEqualTo(1);
+
+        // Read next byte should be '6'
+        int secondByte = inlineStream.read();
+        assertThat(secondByte).isEqualTo('6');
+        assertThat(inlineStream.getPos()).isEqualTo(2);
+    }
+
+    @Test
+    void testSeekAndRead()
+            throws IOException
+    {
+        byte[] data = CONST_STR_FOR_BYTES.getBytes(StandardCharsets.UTF_8);
+        TestTrinoInputStream stream = new TestTrinoInputStream(data);
+
+        long startOffset = 10; // Start at 'A'
+        long length = 10;      // Length covers "ABCDEFGHIJ"
+
+        InlineSeekableDataInputStream inlineStream = new InlineSeekableDataInputStream(stream, startOffset, length);
+
+        // Seek to position 3 (relative) - should be at 'D' (position 13 absolute)
+        inlineStream.seek(3);
+        assertThat(inlineStream.getPos()).isEqualTo(3);
+
+        // Read should get 'D'
+        int readByte = inlineStream.read();
+        assertThat(readByte).isEqualTo('D');
+        assertThat(inlineStream.getPos()).isEqualTo(4);
+    }
+
+    @Test
+    void testZeroLengthInlineStream()
+            throws IOException
+    {
+        byte[] data = "0123456789".getBytes(StandardCharsets.UTF_8);
+        TestTrinoInputStream stream = new TestTrinoInputStream(data);
+
+        long startOffset = 5;
+        long length = 0;
+
+        InlineSeekableDataInputStream inlineStream = new InlineSeekableDataInputStream(stream, startOffset, length);
+
+        // Position should be 0
+        assertThat(inlineStream.getPos()).isEqualTo(0);
+
+        // Seeking to any position > 0 should fail
+        assertThatThrownBy(() -> inlineStream.seek(1))
+                .isInstanceOf(IOException.class)
+                .hasMessageContaining("Attempting to seek past inline content");
+    }
+
+    /**
+     * Test implementation of TrinoInputStream for unit testing.
+     */
+    private static class TestTrinoInputStream
+            extends TrinoInputStream
+    {
+        private final byte[] data;
+        private int position;
+
+        public TestTrinoInputStream(byte[] data)
+        {
+            this.data = data;
+            this.position = 0;
+        }
+
+        @Override
+        public long getPosition()
+        {
+            return position;
+        }
+
+        @Override
+        public void seek(long position)
+                throws IOException
+        {
+            if (position < 0 || position > data.length) {
+                throw new IOException("Invalid seek position: " + position);
+            }
+            this.position = (int) position;
+        }
+
+        @Override
+        public int read()
+                throws IOException
+        {
+            if (position >= data.length) {
+                return -1;
+            }
+            return data[position++] & 0xFF;
+        }
+
+        @Override
+        public int read(byte[] b, int off, int len)
+                throws IOException
+        {
+            if (position >= data.length) {
+                return -1;
+            }
+            int available = data.length - position;
+            int toRead = Math.min(len, available);
+            System.arraycopy(data, position, b, off, toRead);
+            position += toRead;
+            return toRead;
+        }
+
+        @Override
+        public void close()
+                throws IOException
+        {
+            // No-op for test
+        }
+    }
+}
