diff --git a/build.gradle b/build.gradle index 18f50c23e7..c414f9c64e 100644 --- a/build.gradle +++ b/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') diff --git a/orc/src/main/java/org/apache/orc/impl/InStream$UncompressedStream.java b/orc/src/main/java/org/apache/orc/impl/InStream$UncompressedStream.java new file mode 100644 index 0000000000..5333ebd4c4 --- /dev/null +++ b/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 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()); + } +} diff --git a/orc/src/patch/java/org/apache/orc/impl/InStream.java b/orc/src/patch/java/org/apache/orc/impl/InStream.java new file mode 100644 index 0000000000..936aabc826 --- /dev/null +++ b/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. + * + *

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 modifier); + + public abstract void seek(PositionProvider index) throws IOException; + + static int getRangeNumber(DiskRangeList list, DiskRangeList current) { + return 0; + } +} diff --git a/orc/src/test/java/org/apache/orc/impl/TestInStreamSeek.java b/orc/src/test/java/org/apache/orc/impl/TestInStreamSeek.java new file mode 100644 index 0000000000..8aec12084c --- /dev/null +++ b/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()); + } + } +}