Fix uncompressed ORC seek at logical stream end

spark/build.gradle

@@ -24,9 +24,8 @@ if (jdkVersion == '8' && sparkVersions.contains("2.4")) {
   apply from: file("$projectDir/v2.4/build.gradle")
 }
 
-if (sparkVersions.contains("3.1")) {
-  apply from: file("$projectDir/v3.1/build.gradle")
-}
+// Spark 3.1 is always included for this backport branch.
+apply from: file("$projectDir/v3.1/build.gradle")
 
 if (sparkVersions.contains("3.2")) {
   apply from: file("$projectDir/v3.2/build.gradle")

spark/v3.1/build.gradle

@@ -180,6 +180,10 @@ project(":iceberg-spark:iceberg-spark-extensions-3.1_2.12") {
 project(':iceberg-spark:iceberg-spark-runtime-3.1_2.12') {
   apply plugin: 'com.github.johnrengelman.shadow'
 
+  configurations {
+    orcCoreOverlay
+  }
+
   tasks.jar.dependsOn tasks.shadowJar
 
   sourceSets {
@@ -192,8 +196,6 @@ project(':iceberg-spark:iceberg-spark-runtime-3.1_2.12') {
   configurations {
     implementation {
       exclude group: 'org.apache.spark'
-      // included in Spark
-      exclude group: 'org.slf4j'
       exclude group: 'org.apache.commons'
       exclude group: 'commons-pool'
       exclude group: 'commons-codec'
@@ -206,6 +208,10 @@ project(':iceberg-spark:iceberg-spark-runtime-3.1_2.12') {
       exclude group: 'org.abego.treelayout'
       exclude group: 'org.antlr'
     }
+    runtimeClasspath {
+      // included in Spark
+      exclude group: 'org.slf4j'
+    }
   }
 
   dependencies {
@@ -226,6 +232,12 @@ project(':iceberg-spark:iceberg-spark-runtime-3.1_2.12') {
       exclude group: 'net.snowflake' , module: 'snowflake-jdbc'
     }
 
+    compileOnly "org.apache.orc:orc-core::nohive"
+    compileOnly 'org.slf4j:slf4j-api'
+    orcCoreOverlay("org.apache.orc:orc-core::nohive") {
+      transitive = false
+    }
+
     integrationImplementation "org.apache.spark:spark-hive_2.12:${sparkVersion}"
     integrationImplementation 'org.junit.vintage:junit-vintage-engine'
     integrationImplementation 'org.slf4j:slf4j-simple'
@@ -242,6 +254,14 @@ project(':iceberg-spark:iceberg-spark-runtime-3.1_2.12') {
   shadowJar {
     configurations = [project.configurations.runtimeClasspath]
 
+    dependencies {
+      exclude(dependency('org.apache.orc:orc-core'))
+    }
+    from({ project.configurations.orcCoreOverlay.collect { zipTree(it) } }) {
+      // The project overlay is the only source of InStream and its nested classes.
+      exclude 'org/apache/orc/impl/InStream*.class'
+    }
+
     zip64 true
 
     // include the LICENSE and NOTICE files for the shaded Jar
@@ -273,6 +293,12 @@ project(':iceberg-spark:iceberg-spark-runtime-3.1_2.12') {
     archiveClassifier.set(null)
   }
 
+  compileIntegrationJava {
+    classpath += files(shadowJar.archiveFile.get().asFile.path)
+    inputs.file(shadowJar.archiveFile.get().asFile.path)
+  }
+  compileIntegrationJava.dependsOn shadowJar
+
   task integrationTest(type: Test) {
     description = "Test Spark3 Runtime Jar against Spark 3.1"
     group = "verification"
@@ -287,4 +313,3 @@ project(':iceberg-spark:iceberg-spark-runtime-3.1_2.12') {
     enabled = false
   }
 }
-

spark/v3.1/spark-runtime/src/integration/java/org/apache/iceberg/spark/TestOrcUncompressedSeek.java

@@ -0,0 +1,79 @@
+/*
+ * 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.iceberg.spark;
+
+// CHECKSTYLE.OFF: IllegalImport
+import java.io.File;
+import java.nio.ByteBuffer;
+import java.nio.file.Files;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.iceberg.shaded.org.apache.orc.CompressionKind;
+import org.apache.iceberg.shaded.org.apache.orc.OrcFile;
+import org.apache.iceberg.shaded.org.apache.orc.Reader;
+import org.apache.iceberg.shaded.org.apache.orc.TypeDescription;
+import org.apache.iceberg.shaded.org.apache.orc.Writer;
+import org.apache.iceberg.shaded.org.apache.orc.impl.BufferChunk;
+import org.apache.iceberg.shaded.org.apache.orc.impl.InStream;
+import org.apache.iceberg.shaded.org.apache.orc.storage.ql.exec.vector.LongColumnVector;
+import org.apache.iceberg.shaded.org.apache.orc.storage.ql.exec.vector.VectorizedRowBatch;
+import org.junit.Assert;
+import org.junit.Test;
+
+// CHECKSTYLE.ON: IllegalImport
+
+public class TestOrcUncompressedSeek {
+
+  @Test
+  public void readOrcFileThroughShadedRuntime() throws Exception {
+    File orcFile =
+        Files.createTempDirectory("iceberg-orc-uncompressed-seek").resolve("test.orc").toFile();
+    orcFile.deleteOnExit();
+    Configuration conf = new Configuration();
+    Path path = new Path(orcFile.toURI());
+    TypeDescription schema = TypeDescription.fromString("struct<id:bigint>");
+
+    try (Writer writer =
+        OrcFile.createWriter(
+            path, OrcFile.writerOptions(conf).setSchema(schema).compress(CompressionKind.NONE))) {
+      VectorizedRowBatch batch = schema.createRowBatch();
+      ((LongColumnVector) batch.cols[0]).vector[0] = 34;
+      batch.size = 1;
+      writer.addRowBatch(batch);
+    }
+
+    try (Reader reader = OrcFile.createReader(path, OrcFile.readerOptions(conf))) {
+      Assert.assertEquals(1, reader.getNumberOfRows());
+    }
+  }
+
+  @Test
+  public void seekToLogicalEndAtNonFinalRangeBoundary() throws Exception {
+    BufferChunk streamData = new BufferChunk(ByteBuffer.allocate(24), 0);
+    streamData.next = new BufferChunk(ByteBuffer.allocate(8), 32);
+    InStream.UncompressedStream stream =
+        (InStream.UncompressedStream) InStream.create("test", streamData, 0, 24);
+
+    stream.seek(24);
+
+    Assert.assertEquals(0, stream.available());
+    Assert.assertEquals(-1, stream.read());
+    Assert.assertThrows(IllegalArgumentException.class, () -> stream.seek(25));
+  }
+}