ORC: accept logical end seeks across ranges

build.gradle

@@ -703,6 +703,41 @@ project(':iceberg-hive-metastore') {
 }
 
 project(':iceberg-orc') {
+  def orcInStreamPatchApiSources = fileTree('src/patch/java') {
+    include '**/*.java'
+  }
+  def orcInStreamPatchApiClasses = file("$buildDir/classes/java/orcInStreamPatchApi")
+  def orcInStreamPatchSource =
+      file('src/main/java/org/apache/orc/impl/InStream$UncompressedStream.java')
+  def orcInStreamPatchClasses = file("$buildDir/classes/java/orcInStreamPatch")
+
+  sourceSets.main.java.exclude 'org/apache/orc/impl/InStream$UncompressedStream.java'
+
+  def compileOrcInStreamPatchApi = tasks.register('compileOrcInStreamPatchApi', JavaCompile) {
+    source orcInStreamPatchApiSources
+    classpath = sourceSets.main.compileClasspath
+    destinationDirectory.set(orcInStreamPatchApiClasses)
+  }
+
+  // Error Prone 2.10 crashes on the '$' in the binary-compatible class name.
+  def compileOrcInStreamPatch = tasks.register('compileOrcInStreamPatch', JavaCompile) {
+    source orcInStreamPatchSource
+    classpath = files(orcInStreamPatchApiClasses) + sourceSets.main.compileClasspath
+    destinationDirectory.set(orcInStreamPatchClasses)
+    dependsOn compileOrcInStreamPatchApi
+  }
+
+  pluginManager.withPlugin('com.palantir.baseline-error-prone') {
+    compileOrcInStreamPatchApi.configure {
+      options.errorprone.enabled = false
+    }
+    compileOrcInStreamPatch.configure {
+      options.errorprone.enabled = false
+    }
+  }
+
+  sourceSets.main.output.dir([builtBy: compileOrcInStreamPatch], orcInStreamPatchClasses)
+
   dependencies {
     implementation project(path: ':iceberg-bundled-guava', configuration: 'shadow')
     api project(':iceberg-api')

orc/src/main/java/org/apache/orc/impl/InStream$UncompressedStream.java

@@ -0,0 +1,150 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.orc.impl;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.function.Consumer;
+import org.apache.orc.storage.common.io.DiskRangeList;
+
+public class InStream$UncompressedStream extends InStream {
+  protected ByteBuffer decrypted;
+  protected DiskRangeList currentRange;
+  protected long currentOffset;
+
+  public InStream$UncompressedStream(Object name, long offset, long length) {
+    super(name, offset, length);
+  }
+
+  public InStream$UncompressedStream(Object name, DiskRangeList input, long offset, long length) {
+    super(name, offset, length);
+    reset(input);
+  }
+
+  @Override
+  public int read() {
+    if (decrypted == null || decrypted.remaining() == 0) {
+      if (position == length) {
+        return -1;
+      }
+      setCurrent(currentRange.next, false);
+    }
+    position += 1;
+    return 0xff & decrypted.get();
+  }
+
+  @Override
+  protected void setCurrent(DiskRangeList newRange, boolean isJump) {
+    currentRange = newRange;
+    if (newRange != null) {
+      decrypted = newRange.getData().slice();
+      currentOffset = newRange.getOffset();
+      int start = (int) (position + offset - currentOffset);
+      decrypted.position(start);
+      decrypted.limit(start + (int) Math.min(decrypted.remaining(), length - position));
+    }
+  }
+
+  @Override
+  public int read(byte[] data, int offset, int length) {
+    if (decrypted == null || decrypted.remaining() == 0) {
+      if (position == this.length) {
+        return -1;
+      }
+      setCurrent(currentRange.next, false);
+    }
+    int actualLength = Math.min(length, decrypted.remaining());
+    decrypted.get(data, offset, actualLength);
+    position += actualLength;
+    return actualLength;
+  }
+
+  @Override
+  public int available() {
+    if (decrypted != null && decrypted.remaining() > 0) {
+      return decrypted.remaining();
+    }
+    return (int) (length - position);
+  }
+
+  @Override
+  public void close() {
+    currentRange = null;
+    position = length;
+    decrypted = null;
+    bytes = null;
+  }
+
+  @Override
+  public void changeIv(Consumer<byte[]> modifier) {}
+
+  @Override
+  public void seek(PositionProvider index) throws IOException {
+    seek(index.getNext());
+  }
+
+  public void seek(long desired) throws IOException {
+    if (desired == 0 && bytes == null) {
+      return;
+    }
+
+    long positionFile = desired + offset;
+    if (currentRange != null
+        && positionFile >= currentRange.getOffset()
+        && positionFile < currentRange.getEnd()) {
+      decrypted.position((int) (positionFile - currentOffset));
+      position = desired;
+    } else {
+      for (DiskRangeList currentRange = bytes;
+          currentRange != null;
+          currentRange = currentRange.next) {
+        boolean isLogicalEnd = desired == length && positionFile == currentRange.getEnd();
+        if (currentRange.getOffset() <= positionFile
+            && (isLogicalEnd
+                || (currentRange.next == null
+                    ? positionFile <= currentRange.getEnd()
+                    : positionFile < currentRange.getEnd()))) {
+          position = desired;
+          setCurrent(currentRange, true);
+          return;
+        }
+      }
+      throw new IllegalArgumentException(
+          "Seek in " + name + " to " + desired + " is outside of the data");
+    }
+  }
+
+  @Override
+  public String toString() {
+    return "uncompressed stream "
+        + name
+        + " position: "
+        + position
+        + " length: "
+        + length
+        + " range: "
+        + InStream.getRangeNumber(bytes, currentRange)
+        + " offset: "
+        + currentRange.getOffset()
+        + " position: "
+        + (decrypted == null ? 0 : decrypted.position())
+        + " limit: "
+        + (decrypted == null ? 0 : decrypted.limit());
+  }
+}

orc/src/patch/java/org/apache/orc/impl/InStream.java

@@ -0,0 +1,52 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.orc.impl;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.function.Consumer;
+import org.apache.orc.storage.common.io.DiskRangeList;
+
+/**
+ * Compile API for the binary-compatible ORC 1.8.2 nested-class overlay.
+ *
+ * <p>The patch compiler resolves inherited members through this class. The packaged overlay
+ * resolves them through ORC's {@link InStream} at runtime.
+ */
+public abstract class InStream extends InputStream {
+  protected Object name;
+  protected long offset;
+  protected long length;
+  protected long position;
+  protected DiskRangeList bytes;
+
+  protected InStream(Object name, long offset, long length) {}
+
+  protected void reset(DiskRangeList input) {}
+
+  protected abstract void setCurrent(DiskRangeList newRange, boolean isJump);
+
+  public abstract void changeIv(Consumer<byte[]> modifier);
+
+  public abstract void seek(PositionProvider index) throws IOException;
+
+  static int getRangeNumber(DiskRangeList list, DiskRangeList current) {
+    return 0;
+  }
+}

orc/src/test/java/org/apache/orc/impl/TestInStreamSeek.java

@@ -0,0 +1,43 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.orc.impl;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertSame;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import org.junit.Test;
+
+public class TestInStreamSeek {
+  @Test
+  public void seekToLogicalEndWithLaterDiskRange() throws IOException {
+    BufferChunk streamRange = new BufferChunk(ByteBuffer.allocate(10), 100);
+    streamRange.insertAfter(new BufferChunk(ByteBuffer.allocate(10), 120));
+
+    try (InStream.UncompressedStream stream =
+        new InStream.UncompressedStream("test", streamRange, 100, 10)) {
+      stream.seek(10);
+
+      assertSame(streamRange, stream.currentRange);
+      assertEquals(0, stream.available());
+      assertEquals(-1, stream.read());
+    }
+  }
+}