This is an automated email from the ASF dual-hosted git repository.
leaves12138 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new b898cfc098 [core] Support input streams in Avro block readers (#9842)
b898cfc098 is described below
commit b898cfc0984df476af097a5a77d6468e8a3c293d
Author: YeJunHao <[email protected]>
AuthorDate: Tue Sep 15 17:48:32 2026 +0800
[core] Support input streams in Avro block readers (#9842)
---
.../apache/paimon/manifest/ManifestAvroReader.java | 9 ++-
.../apache/paimon/manifest/ManifestFileTest.java | 9 ++-
.../java/org/apache/avro/file/RawBlockReader.java | 61 +++++++++++----
.../apache/paimon/format/avro/AvroBlockReader.java | 7 +-
.../paimon/format/avro/AvroBlockReaderTest.java | 87 +++++-----------------
5 files changed, 83 insertions(+), 90 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestAvroReader.java
b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestAvroReader.java
index 4e84c690ee..95134f36a0 100644
---
a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestAvroReader.java
+++
b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestAvroReader.java
@@ -26,7 +26,6 @@ import org.apache.paimon.format.avro.AvroRawBlock;
import org.apache.paimon.format.avro.AvroRecordDecoder;
import org.apache.paimon.format.avro.AvroRecordDecoder.FieldDecoder;
import org.apache.paimon.format.avro.AvroRecordDecoder.FieldType;
-import org.apache.paimon.fs.SeekableInputStream;
import org.apache.paimon.partition.PartitionPredicate;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.CloseableIterator;
@@ -35,6 +34,7 @@ import org.apache.paimon.utils.IOUtils;
import javax.annotation.Nullable;
import java.io.IOException;
+import java.io.InputStream;
import java.io.UncheckedIOException;
import java.nio.ByteBuffer;
import java.util.Arrays;
@@ -56,7 +56,7 @@ public final class ManifestAvroReader implements
AutoCloseable {
private long blockOrdinal = -1;
- ManifestAvroReader(SeekableInputStream input) throws IOException {
+ ManifestAvroReader(InputStream input) throws IOException {
AvroBlockReader blockReader = null;
try {
blockReader = new AvroBlockReader(input);
@@ -75,7 +75,10 @@ public final class ManifestAvroReader implements
AutoCloseable {
return blockReader.headerBytes();
}
- /** Returns the physical block offset; read immediately after {@link
#next()}. */
+ /**
+ * Returns the block offset relative to the initial input position; read
immediately after
+ * {@link #next()}.
+ */
public long blockOffset() {
return blockReader.blockOffset();
}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileTest.java
b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileTest.java
index b5ff0fb002..5fb0a7e049 100644
--- a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileTest.java
@@ -55,7 +55,9 @@ import org.junit.jupiter.params.provider.MethodSource;
import javax.annotation.Nullable;
+import java.io.ByteArrayInputStream;
import java.io.IOException;
+import java.io.SequenceInputStream;
import java.io.UncheckedIOException;
import java.nio.ByteBuffer;
import java.nio.file.Files;
@@ -1131,8 +1133,13 @@ public class ManifestFileTest {
int blockCount = 0;
int rowCount = 0;
byte[] bytes =
Files.readAllBytes(tempDir.resolve("manifest").resolve(manifest.fileName()));
+ int split = bytes.length / 2;
- try (ManifestAvroReader reader = openManifestReader(manifest)) {
+ try (ManifestAvroReader reader =
+ new ManifestAvroReader(
+ new SequenceInputStream(
+ new ByteArrayInputStream(bytes, 0, split),
+ new ByteArrayInputStream(bytes, split,
bytes.length - split)))) {
byte[] header = reader.headerBytes();
assertThat(header).isEqualTo(Arrays.copyOf(bytes, header.length));
long nextOffset = header.length;
diff --git
a/paimon-format/src/main/java/org/apache/avro/file/RawBlockReader.java
b/paimon-format/src/main/java/org/apache/avro/file/RawBlockReader.java
index 60ae9b195a..17f39bdebb 100644
--- a/paimon-format/src/main/java/org/apache/avro/file/RawBlockReader.java
+++ b/paimon-format/src/main/java/org/apache/avro/file/RawBlockReader.java
@@ -18,38 +18,35 @@
package org.apache.avro.file;
-import org.apache.paimon.fs.SeekableInputStream;
-import org.apache.paimon.utils.IOUtils;
-
import org.apache.avro.Schema;
import org.apache.avro.io.DatumReader;
import org.apache.avro.io.Decoder;
+import java.io.ByteArrayOutputStream;
+import java.io.FilterInputStream;
import java.io.IOException;
+import java.io.InputStream;
+import java.util.Arrays;
import java.util.NoSuchElementException;
/** Package bridge exposing Avro's compressed blocks without reflection. */
public final class RawBlockReader extends DataFileStream<Void> {
- private final SeekableInputStream input;
+ private final CountingInput input;
private final byte[] headerBytes;
private long blockOffset;
private long blockLength;
private boolean pending;
- public RawBlockReader(SeekableInputStream input) throws IOException {
- this(input, input.getPos());
+ public RawBlockReader(InputStream input) throws IOException {
+ this(new CountingInput(input));
}
- private RawBlockReader(SeekableInputStream input, long headerOffset)
throws IOException {
+ private RawBlockReader(CountingInput input) throws IOException {
super(input, new NoOpDatumReader<Void>());
this.input = input;
- this.headerBytes = new byte[Math.toIntExact(position() -
headerOffset)];
- long resumePosition = input.getPos();
- input.seek(headerOffset);
- IOUtils.readFully(input, headerBytes);
- // Preserve the position past any bytes already buffered by the Avro
decoder.
- input.seek(resumePosition);
+ this.headerBytes = Arrays.copyOf(input.prefix.toByteArray(),
Math.toIntExact(position()));
+ input.prefix = null;
}
/** Returns a copy of the complete OCF header, including schema, codec and
sync marker. */
@@ -58,7 +55,8 @@ public final class RawBlockReader extends
DataFileStream<Void> {
}
/**
- * Returns the physical block offset; read immediately after {@link
#nextRawBlock(RawBlock)}.
+ * Returns the block offset relative to the initial input position; read
immediately after
+ * {@link #nextRawBlock(RawBlock)}.
*/
public long blockOffset() {
return blockOffset;
@@ -71,7 +69,7 @@ public final class RawBlockReader extends
DataFileStream<Void> {
private long position() throws IOException {
// This is the same read-ahead adjustment used by
DataFileReader.blockFinished().
- return input.getPos() - vin.inputStream().available();
+ return input.position - vin.inputStream().available();
}
public boolean hasNextRawBlock() throws IOException {
@@ -94,6 +92,39 @@ public final class RawBlockReader extends
DataFileStream<Void> {
: reuse.replace(raw, resolveCodec(), getSchema());
}
+ private static final class CountingInput extends FilterInputStream {
+ private long position;
+ private ByteArrayOutputStream prefix = new ByteArrayOutputStream();
+
+ private CountingInput(InputStream input) {
+ super(input);
+ }
+
+ @Override
+ public int read() throws IOException {
+ int value = in.read();
+ if (value >= 0) {
+ position++;
+ if (prefix != null) {
+ prefix.write(value);
+ }
+ }
+ return value;
+ }
+
+ @Override
+ public int read(byte[] bytes, int offset, int length) throws
IOException {
+ int n = in.read(bytes, offset, length);
+ if (n > 0) {
+ position += n;
+ if (prefix != null) {
+ prefix.write(bytes, offset, n);
+ }
+ }
+ return n;
+ }
+ }
+
private static final class NoOpDatumReader<D> implements DatumReader<D> {
@Override
diff --git
a/paimon-format/src/main/java/org/apache/paimon/format/avro/AvroBlockReader.java
b/paimon-format/src/main/java/org/apache/paimon/format/avro/AvroBlockReader.java
index 61e387d580..03b70840b6 100644
---
a/paimon-format/src/main/java/org/apache/paimon/format/avro/AvroBlockReader.java
+++
b/paimon-format/src/main/java/org/apache/paimon/format/avro/AvroBlockReader.java
@@ -18,7 +18,6 @@
package org.apache.paimon.format.avro;
-import org.apache.paimon.fs.SeekableInputStream;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.IOUtils;
@@ -31,6 +30,7 @@ import javax.annotation.Nullable;
import java.io.Closeable;
import java.io.IOException;
+import java.io.InputStream;
import java.util.Collections;
/**
@@ -45,7 +45,7 @@ public final class AvroBlockReader implements Closeable {
private @Nullable AvroRawBlock borrowedRawBlock;
- public AvroBlockReader(SeekableInputStream input) throws IOException {
+ public AvroBlockReader(InputStream input) throws IOException {
try {
this.reader = new RawBlockReader(input);
} catch (IOException | RuntimeException | Error e) {
@@ -60,7 +60,8 @@ public final class AvroBlockReader implements Closeable {
}
/**
- * Returns the physical block offset; read immediately after {@link
#nextBorrowedRawBlock()}.
+ * Returns the block offset relative to the initial input position; read
immediately after
+ * {@link #nextBorrowedRawBlock()}.
*/
public long blockOffset() {
return reader.blockOffset();
diff --git
a/paimon-format/src/test/java/org/apache/paimon/format/avro/AvroBlockReaderTest.java
b/paimon-format/src/test/java/org/apache/paimon/format/avro/AvroBlockReaderTest.java
index c0faacd3eb..1aabaa0b12 100644
---
a/paimon-format/src/test/java/org/apache/paimon/format/avro/AvroBlockReaderTest.java
+++
b/paimon-format/src/test/java/org/apache/paimon/format/avro/AvroBlockReaderTest.java
@@ -18,12 +18,6 @@
package org.apache.paimon.format.avro;
-import org.apache.paimon.fs.ByteArraySeekableStream;
-import org.apache.paimon.fs.Path;
-import org.apache.paimon.fs.SeekableInputStream;
-import org.apache.paimon.fs.SeekableInputStreamWrapper;
-import org.apache.paimon.fs.local.LocalFileIO;
-
import org.apache.avro.Schema;
import org.apache.avro.file.CodecFactory;
import org.apache.avro.file.DataFileStream;
@@ -31,17 +25,15 @@ import org.apache.avro.file.DataFileWriter;
import org.apache.avro.generic.GenericDatumReader;
import org.apache.avro.generic.GenericDatumWriter;
import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.io.TempDir;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
+import java.io.FilterInputStream;
import java.io.IOException;
-import java.nio.file.Files;
-import java.util.ArrayList;
+import java.io.InputStream;
import java.util.Arrays;
-import java.util.List;
import java.util.NoSuchElementException;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -53,8 +45,6 @@ class AvroBlockReaderTest {
private static final Schema SCHEMA = Schema.create(Schema.Type.LONG);
- @TempDir private java.nio.file.Path tempDir;
-
@ParameterizedTest
@ValueSource(strings = {"null", "deflate", "snappy", "zstandard"})
void blockMetadataMatchesWriterBoundaries(String codec) throws Exception {
@@ -78,31 +68,19 @@ class AvroBlockReaderTest {
assertThat(boundaries[values.length]).isEqualTo(bytes.length);
for (int maxRead : new int[] {1, 7, Integer.MAX_VALUE}) {
- List<Long> seeks = new ArrayList<>();
- SeekableInputStream input =
- new SeekableInputStreamWrapper(open(bytes)) {
+ InputStream input =
+ new FilterInputStream(new ByteArrayInputStream(bytes)) {
@Override
public int read(byte[] data, int offset, int length)
throws IOException {
return super.read(data, offset, Math.min(length,
maxRead));
}
-
- @Override
- public void seek(long position) throws IOException {
- seeks.add(position);
- super.seek(position);
- }
};
try (AvroBlockReader reader = new AvroBlockReader(input)) {
- long resumePosition = input.getPos();
- assertThat(seeks).containsExactly(0L, resumePosition);
byte[] header = reader.headerBytes();
assertThat(header).isEqualTo(Arrays.copyOf(bytes, (int)
boundaries[0]));
- assertThat(input.getPos()).isEqualTo(resumePosition);
- assertThat(seeks).containsExactly(0L, resumePosition);
byte[] anotherHeader = reader.headerBytes();
anotherHeader[0] = 0;
assertThat(reader.headerBytes()).isEqualTo(header);
- assertThat(seeks).hasSize(2);
AvroRawBlock previous = null;
for (int i = 0; i < values.length; i++) {
// Exercise next() both directly and after repeated
look-ahead calls.
@@ -136,7 +114,7 @@ class AvroBlockReaderTest {
writer.create(SCHEMA, output);
}
byte[] bytes = output.toByteArray();
- try (AvroBlockReader reader = new AvroBlockReader(open(bytes))) {
+ try (AvroBlockReader reader = new AvroBlockReader(new
ByteArrayInputStream(bytes))) {
assertThat(reader.headerBytes()).isEqualTo(bytes);
assertThat(reader.hasNextBlock()).isFalse();
assertThatThrownBy(reader::nextBorrowedRawBlock)
@@ -145,46 +123,28 @@ class AvroBlockReaderTest {
}
@Test
- void failedHeaderReadClosesTheInput() throws IOException {
- ByteArrayOutputStream output = new ByteArrayOutputStream();
- try (DataFileWriter<Long> writer = new DataFileWriter<>(new
GenericDatumWriter<>(SCHEMA))) {
- writer.create(SCHEMA, output);
- writer.append(11L);
- }
- byte[] bytes = output.toByteArray();
+ void failedHeaderReadClosesTheInput() {
AtomicBoolean closed = new AtomicBoolean();
- SeekableInputStream input =
- new SeekableInputStreamWrapper(open(bytes)) {
- private boolean readingHeader;
-
- @Override
- public void seek(long position) throws IOException {
- super.seek(position);
- readingHeader = position == 0;
- }
-
+ InputStream input =
+ new InputStream() {
@Override
- public int read(byte[] data, int offset, int length)
throws IOException {
- if (readingHeader) {
- throw new IOException("header read failed");
- }
- return super.read(data, offset, length);
+ public int read() throws IOException {
+ throw new IOException("header read failed");
}
@Override
- public void close() throws IOException {
- super.close();
+ public void close() {
closed.set(true);
}
};
assertThatThrownBy(() -> new AvroBlockReader(input))
.isInstanceOf(IOException.class)
- .hasMessage("header read failed");
+ .hasRootCauseMessage("header read failed");
assertThat(closed.get()).isTrue();
}
@Test
- void headerFromMemoryCanRestoreEof() throws IOException {
+ void headerExcludesBufferedBlockData() throws IOException {
for (int records : new int[] {0, 1}) {
ByteArrayOutputStream output = new ByteArrayOutputStream();
long headerLength;
@@ -197,12 +157,12 @@ class AvroBlockReaderTest {
}
}
byte[] bytes = output.toByteArray();
- ByteArraySeekableStream input = new ByteArraySeekableStream(bytes);
+ ByteArrayInputStream input = new ByteArrayInputStream(bytes);
try (AvroBlockReader reader = new AvroBlockReader(input)) {
- assertThat(input.getPos()).isEqualTo(bytes.length);
+ assertThat(input.available()).isZero();
byte[] header = reader.headerBytes();
assertThat(header).isEqualTo(Arrays.copyOf(bytes, (int)
headerLength));
- assertThat(input.getPos()).isEqualTo(bytes.length);
+ assertThat(input.available()).isZero();
if (records > 0) {
assertThat(reader.nextBorrowedRawBlock().recordCount()).isEqualTo(records);
assertBlockReadable(
@@ -230,27 +190,18 @@ class AvroBlockReaderTest {
int prefixLength = 13;
byte[] bytes = new byte[prefixLength + avro.length];
System.arraycopy(avro, 0, bytes, prefixLength, avro.length);
- SeekableInputStream input = open(bytes);
- input.seek(prefixLength);
+ InputStream input = new ByteArrayInputStream(bytes, prefixLength,
avro.length);
try (AvroBlockReader reader = new AvroBlockReader(input)) {
- long resumePosition = input.getPos();
byte[] header = reader.headerBytes();
assertThat(header).isEqualTo(Arrays.copyOf(avro, (int)
headerLength));
- assertThat(input.getPos()).isEqualTo(resumePosition);
assertThat(reader.nextBorrowedRawBlock().recordCount()).isEqualTo(1);
- assertThat(reader.blockOffset()).isEqualTo(prefixLength +
headerLength);
+ assertThat(reader.blockOffset()).isEqualTo(headerLength);
assertBlockReadable(
- header, bytes, reader.blockOffset(), reader.blockLength(),
new long[] {17L});
+ header, avro, reader.blockOffset(), reader.blockLength(),
new long[] {17L});
assertThat(reader.hasNextBlock()).isFalse();
}
}
- private SeekableInputStream open(byte[] bytes) throws IOException {
- java.nio.file.Path file = Files.createTempFile(tempDir, "blocks-",
".avro");
- Files.write(file, bytes);
- return LocalFileIO.create().newInputStream(new Path(file.toUri()));
- }
-
private static void assertBlockReadable(
byte[] header, byte[] file, long offset, long length, long[]
expected)
throws IOException {
