This is an automated email from the ASF dual-hosted git repository.
JingsongLi 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 79a8d165fa [core] Add block-based manifest Avro reader (#9160)
79a8d165fa is described below
commit 79a8d165fade334c71ee85104082f51a1df37de1
Author: Jingsong Lee <[email protected]>
AuthorDate: Tue Aug 11 13:49:09 2026 +0800
[core] Add block-based manifest Avro reader (#9160)
Speed up manifest reads by decoding Avro object-container blocks
directly and applying partition, bucket, and projection pruning before
materializing nested `DataFileMeta` rows.
---
.../paimon/manifest/BinaryManifestEntry.java | 5 +-
.../apache/paimon/manifest/ManifestAvroReader.java | 317 ++++++++++++++++++++
.../paimon/manifest/ManifestEntrySerializer.java | 2 +-
.../org/apache/paimon/manifest/ManifestFile.java | 74 ++---
.../paimon/manifest/ManifestSchemaUtils.java | 4 +-
.../java/org/apache/paimon/utils/ObjectsFile.java | 30 +-
.../apache/paimon/manifest/ManifestFileTest.java | 320 ++++++++++++++++++---
.../apache/paimon/format/avro/AvroBlockReader.java | 158 ++++++++++
.../paimon/format/avro/AvroRecordDecoder.java | 167 +++++++++++
.../paimon/format/avro/FieldReaderFactory.java | 12 +-
.../paimon/format/avro/AvroFileFormatTest.java | 149 ++++++++++
11 files changed, 1155 insertions(+), 83 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/manifest/BinaryManifestEntry.java
b/paimon-core/src/main/java/org/apache/paimon/manifest/BinaryManifestEntry.java
index 646f19ae44..f1dd962bba 100644
---
a/paimon-core/src/main/java/org/apache/paimon/manifest/BinaryManifestEntry.java
+++
b/paimon-core/src/main/java/org/apache/paimon/manifest/BinaryManifestEntry.java
@@ -39,8 +39,9 @@ import static
org.apache.paimon.utils.SerializationUtils.deserializeBinaryRow;
*
* <p>This class is intended for streaming manifest algorithms which only need
a subset of {@link
* ManifestEntry}. Unlike {@link PojoManifestEntry}, it does not deserialize
the nested {@code
- * _FILE} row into a POJO. The view is mutable and only valid while its
backing {@link InternalRow}
- * is valid; callers must not retain it across reader iterations or batches.
+ * _FILE} row into a POJO. The view is mutable and remains valid while its
backing {@link
+ * InternalRow} remains valid. Producers such as {@link
ManifestFile#scan(String, Long, Projection)}
+ * provide independently backed entries which can be retained.
*/
public final class BinaryManifestEntry implements ManifestEntry {
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
new file mode 100644
index 0000000000..83af2833eb
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestAvroReader.java
@@ -0,0 +1,317 @@
+/*
+ * 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.paimon.manifest;
+
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.format.avro.AvroBlockReader;
+import org.apache.paimon.format.avro.AvroBlockReader.BorrowedBlock;
+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.partition.PartitionPredicate;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.CloseableIterator;
+import org.apache.paimon.utils.IOUtils;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.UncheckedIOException;
+import java.util.NoSuchElementException;
+
+import static org.apache.paimon.utils.SerializationUtils.deserializeBinaryRow;
+
+/**
+ * Schema-aware Avro reader which projects fields and filters before decoding
data file metadata.
+ */
+final class ManifestAvroReader implements CloseableIterator<InternalRow> {
+
+ private final AvroBlockReader blockReader;
+ private final AvroRecordDecoder decoder;
+ private final ManifestRecordDecoder recordDecoder;
+
+ private long recordsRemaining;
+ private @Nullable InternalRow next;
+ private boolean nextReady;
+ private boolean finished;
+
+ ManifestAvroReader(
+ InputStream input,
+ RowType projectedType,
+ @Nullable PartitionPredicate partitionFilter,
+ @Nullable BucketFilter bucketFilter)
+ throws IOException {
+ AvroBlockReader blockReader = new AvroBlockReader(input);
+ try {
+ AvroRecordDecoder decoder = blockReader.createRecordDecoder();
+ this.recordDecoder =
+ new ManifestRecordDecoder(
+ decoder, projectedType, partitionFilter,
bucketFilter);
+ this.decoder = decoder;
+ this.blockReader = blockReader;
+ } catch (RuntimeException | Error e) {
+ IOUtils.closeQuietly(blockReader);
+ throw e;
+ }
+ }
+
+ @Override
+ public boolean hasNext() {
+ if (nextReady) {
+ return true;
+ }
+ if (finished) {
+ return false;
+ }
+
+ try {
+ while (true) {
+ if (recordsRemaining == 0) {
+ if (decoder.isInitialized() && !decoder.isEnd()) {
+ throw new IOException(
+ "Manifest Avro block contains trailing
undecoded bytes.");
+ }
+ if (!blockReader.hasNextBlock()) {
+ finished = true;
+ return false;
+ }
+ BorrowedBlock block = blockReader.nextBorrowedBlock();
+ decoder.reset(block.bytes(), block.offset(),
block.length());
+ recordsRemaining = block.recordCount();
+ }
+
+ InternalRow candidate = recordDecoder.read(decoder);
+ recordsRemaining--;
+ if (candidate != null) {
+ next = candidate;
+ nextReady = true;
+ return true;
+ }
+ }
+ } catch (IOException e) {
+ throw new UncheckedIOException("Failed to decode Manifest Avro
block.", e);
+ }
+ }
+
+ @Override
+ public InternalRow next() {
+ if (!hasNext()) {
+ throw new NoSuchElementException();
+ }
+ InternalRow result = next;
+ next = null;
+ nextReady = false;
+ return result;
+ }
+
+ long decodedDataFiles() {
+ return recordDecoder.decodedDataFiles;
+ }
+
+ long skippedDataFiles() {
+ return recordDecoder.skippedDataFiles;
+ }
+
+ @Override
+ public void close() throws IOException {
+ next = null;
+ nextReady = false;
+ finished = true;
+ blockReader.close();
+ }
+
+ private static class ManifestRecordDecoder {
+
+ private static final String[] TOP_LEVEL_FIELDS = {
+ ManifestSchemaUtils.FORMAT_IDENTIFIER_FIELD,
+ ManifestEntry.KIND,
+ ManifestEntry.PARTITION,
+ ManifestEntry.BUCKET,
+ ManifestEntry.TOTAL_BUCKETS,
+ ManifestEntry.FILE
+ };
+
+ private final int projectedFieldCount;
+ private final int versionPosition;
+ private final int kindPosition;
+ private final int partitionPosition;
+ private final int bucketPosition;
+ private final int totalBucketsPosition;
+ private final int filePosition;
+
+ private final FieldDecoder fileReader;
+
+ private final @Nullable PartitionPredicate partitionFilter;
+ private final @Nullable BucketFilter bucketFilter;
+ private final boolean partitionNeededForFilter;
+ private final boolean bucketNeededForFilter;
+
+ private long decodedDataFiles;
+ private long skippedDataFiles;
+
+ private ManifestRecordDecoder(
+ AvroRecordDecoder decoder,
+ RowType projectedType,
+ @Nullable PartitionPredicate partitionFilter,
+ @Nullable BucketFilter bucketFilter) {
+ this.projectedFieldCount = projectedType.getFieldCount();
+ this.versionPosition =
+
projectedType.getFieldIndex(ManifestSchemaUtils.FORMAT_IDENTIFIER_FIELD);
+ this.kindPosition =
projectedType.getFieldIndex(ManifestEntry.KIND);
+ this.partitionPosition =
projectedType.getFieldIndex(ManifestEntry.PARTITION);
+ this.bucketPosition =
projectedType.getFieldIndex(ManifestEntry.BUCKET);
+ this.totalBucketsPosition =
projectedType.getFieldIndex(ManifestEntry.TOTAL_BUCKETS);
+ this.filePosition =
projectedType.getFieldIndex(ManifestEntry.FILE);
+ this.partitionFilter = partitionFilter;
+ this.bucketFilter = bucketFilter;
+ this.partitionNeededForFilter = partitionFilter != null ||
bucketFilter != null;
+ this.bucketNeededForFilter = bucketFilter != null;
+
+ validateTopLevelFields(decoder);
+ // Manifest v2 has a fixed top-level layout, but DataFileMeta has
gained nullable
+ // fields. Build this reader from the writer schema so legacy
files with fewer nested
+ // fields still decode and expose the missing projected fields as
null.
+ fileReader =
+ decoder.createFieldDecoder(
+ 5, filePosition >= 0 ?
projectedType.getTypeAt(filePosition) : null);
+ }
+
+ private @Nullable InternalRow read(AvroRecordDecoder decoder) throws
IOException {
+ if (!decoder.readRecordStart()) {
+ throw new IOException("Unexpected null or non-record Manifest
Avro value.");
+ }
+
+ int version = decoder.readInt();
+ ManifestEntrySerializer.checkFormatIdentifier(version);
+ int kind = decoder.readInt();
+ byte[] partitionBytes;
+ if (partitionPosition >= 0 || partitionNeededForFilter) {
+ partitionBytes = decoder.readBytes();
+ } else {
+ decoder.skipBytes();
+ partitionBytes = null;
+ }
+
+ BinaryRow partition =
+ partitionNeededForFilter ?
deserializeBinaryRow(partitionBytes) : null;
+ if (partitionFilter != null && !partitionFilter.test(partition)) {
+ skipBucketAndFile(decoder);
+ return null;
+ }
+
+ int bucket;
+ if (bucketPosition >= 0 || bucketNeededForFilter) {
+ bucket = decoder.readInt();
+ } else {
+ decoder.readInt();
+ bucket = 0;
+ }
+
+ int totalBuckets;
+ if (totalBucketsPosition >= 0 || bucketNeededForFilter) {
+ totalBuckets = decoder.readInt();
+ } else {
+ decoder.readInt();
+ totalBuckets = 0;
+ }
+
+ if (bucketFilter != null && !bucketFilter.test(partition, bucket,
totalBuckets)) {
+ skipFile(decoder);
+ return null;
+ }
+
+ Object file;
+ if (filePosition >= 0) {
+ file = fileReader.read(decoder, null);
+ decodedDataFiles++;
+ } else {
+ fileReader.skip(decoder);
+ skippedDataFiles++;
+ file = null;
+ }
+
+ GenericRow row = new GenericRow(projectedFieldCount);
+ setProjected(row, versionPosition, version);
+ setProjected(row, kindPosition, (byte) kind);
+ setProjected(row, partitionPosition, partitionBytes);
+ setProjected(row, bucketPosition, bucket);
+ setProjected(row, totalBucketsPosition, totalBuckets);
+ setProjected(row, filePosition, file);
+ return row;
+ }
+
+ private void skipBucketAndFile(AvroRecordDecoder decoder) throws
IOException {
+ decoder.readInt();
+ decoder.readInt();
+ skipFile(decoder);
+ }
+
+ private void skipFile(AvroRecordDecoder decoder) throws IOException {
+ fileReader.skip(decoder);
+ skippedDataFiles++;
+ }
+
+ private static void setProjected(
+ GenericRow row, int outputPosition, @Nullable Object value) {
+ if (outputPosition >= 0) {
+ row.setField(outputPosition, value);
+ }
+ }
+
+ private static void validateTopLevelFields(AvroRecordDecoder decoder) {
+ if (decoder.fieldCount() != TOP_LEVEL_FIELDS.length) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Manifest Avro schema has %s top-level fields,
expected %s.",
+ decoder.fieldCount(),
TOP_LEVEL_FIELDS.length));
+ }
+ for (int i = 0; i < TOP_LEVEL_FIELDS.length; i++) {
+ String actual = decoder.fieldName(i);
+ String expected = TOP_LEVEL_FIELDS[i];
+ if (!expected.equals(actual)) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Unexpected Manifest Avro field at
position %s: expected %s but found %s.",
+ i, expected, actual));
+ }
+ }
+
+ validateFieldType(decoder, 0, FieldType.INT);
+ validateFieldType(decoder, 1, FieldType.INT);
+ validateFieldType(decoder, 2, FieldType.BYTES);
+ validateFieldType(decoder, 3, FieldType.INT);
+ validateFieldType(decoder, 4, FieldType.INT);
+ validateFieldType(decoder, 5, FieldType.RECORD);
+ }
+
+ private static void validateFieldType(
+ AvroRecordDecoder decoder, int position, FieldType
expectedType) {
+ FieldType actualType = decoder.fieldType(position);
+ if (actualType != expectedType) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Unexpected Manifest Avro type for field %s:
expected %s but found %s.",
+ decoder.fieldName(position), expectedType,
actualType));
+ }
+ }
+ }
+}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestEntrySerializer.java
b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestEntrySerializer.java
index ec52247eba..9dfef81aeb 100644
---
a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestEntrySerializer.java
+++
b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestEntrySerializer.java
@@ -67,7 +67,7 @@ public class ManifestEntrySerializer extends
ObjectSerializer<ManifestEntry> {
return fromDataRow(new OffsetRow(row.getFieldCount() - 1,
1).replace(row));
}
- private void checkFormatIdentifier(int formatIdentifier) {
+ static void checkFormatIdentifier(int formatIdentifier) {
if (formatIdentifier != FORMAT_IDENTIFIER) {
if (formatIdentifier == 1) {
throw new IllegalArgumentException(
diff --git
a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFile.java
b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFile.java
index 4f1aba75ba..45e0a74378 100644
--- a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFile.java
+++ b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFile.java
@@ -21,7 +21,6 @@ package org.apache.paimon.manifest;
import org.apache.paimon.annotation.VisibleForTesting;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.format.FileFormat;
-import org.apache.paimon.format.FormatReaderFactory;
import org.apache.paimon.format.FormatWriterFactory;
import org.apache.paimon.format.SimpleStatsCollector;
import org.apache.paimon.fs.FileIO;
@@ -50,7 +49,6 @@ import java.io.IOException;
import java.io.UncheckedIOException;
import java.util.ArrayList;
import java.util.Arrays;
-import java.util.Collections;
import java.util.List;
import java.util.function.Function;
@@ -64,8 +62,6 @@ public class ManifestFile extends ObjectsFile<ManifestEntry> {
private final SchemaManager schemaManager;
private final RowType partitionType;
- private final FileFormat fileFormat;
- private final RowType manifestType;
private final FormatWriterFactory writerFactory;
private final long suggestedFileSize;
@@ -73,10 +69,7 @@ public class ManifestFile extends ObjectsFile<ManifestEntry>
{
FileIO fileIO,
SchemaManager schemaManager,
RowType partitionType,
- FileFormat fileFormat,
ManifestEntrySerializer serializer,
- RowType manifestType,
- FormatReaderFactory readerFactory,
FormatWriterFactory writerFactory,
String compression,
PathFactory pathFactory,
@@ -85,16 +78,16 @@ public class ManifestFile extends
ObjectsFile<ManifestEntry> {
super(
fileIO,
serializer,
- manifestType,
- readerFactory,
+ ManifestEntry.MANIFEST_ROW_TYPE,
+ (path, ignoredFileSize) ->
+ createManifestIterator(
+ fileIO, path, ManifestEntry.MANIFEST_ROW_TYPE,
null, null),
writerFactory,
compression,
pathFactory,
cache);
this.schemaManager = schemaManager;
this.partitionType = partitionType;
- this.fileFormat = fileFormat;
- this.manifestType = manifestType;
this.writerFactory = writerFactory;
this.suggestedFileSize = suggestedFileSize;
}
@@ -103,7 +96,7 @@ public class ManifestFile extends ObjectsFile<ManifestEntry>
{
protected ManifestEntryCache createCache(
@Nullable SegmentsCache<Path> cache, RowType formatType) {
return new ManifestEntryCache(
- cache, serializer, formatType, super::fileSize,
super::createIterator);
+ cache, serializer, formatType, super::fileSize,
this::createIterator);
}
@Override
@@ -146,8 +139,14 @@ public class ManifestFile extends
ObjectsFile<ManifestEntry> {
return cache.read(path, fileSize, filters, convertor);
}
- return readFromIterator(
- createIterator(path, fileSize), serializer, readFilter,
readTFilter, convertor);
+ CloseableIterator<InternalRow> iterator =
+ createManifestIterator(
+ fileIO,
+ path,
+ ManifestEntry.MANIFEST_ROW_TYPE,
+ partitionFilter,
+ bucketFilter);
+ return readFromIterator(iterator, serializer, readFilter,
readTFilter, convertor);
} catch (IOException e) {
throw new UncheckedIOException(e);
}
@@ -156,46 +155,36 @@ public class ManifestFile extends
ObjectsFile<ManifestEntry> {
/**
* Scans projected manifest entries without materializing {@link
PojoManifestEntry}s.
*
- * <p>The returned iterator reuses the same mutable {@link
BinaryManifestEntry} for all records.
- * An entry is only valid until the next call to {@link
CloseableIterator#hasNext()}, {@link
- * CloseableIterator#next()}, or {@link CloseableIterator#close()}, and
must not be retained.
- * The caller must close the iterator.
+ * <p>Every returned {@link BinaryManifestEntry} has independent backing
data and can be
+ * retained after the iterator advances or closes. The caller must close
the iterator.
*
* <p>This method intentionally bypasses the manifest cache because cached
entries are
* materialized with the complete manifest schema.
*/
public CloseableIterator<BinaryManifestEntry> scan(
String fileName, @Nullable Long fileSize, Projection projection) {
- BinaryManifestEntry entry = projection.createEntry();
try {
CloseableIterator<InternalRow> rows =
- FileUtils.createFormatReader(
- fileIO,
- fileFormat.createReaderFactory(
- manifestType,
- projection.projectedType(),
- Collections.emptyList()),
- pathFactory.toPath(fileName),
- fileSize)
- .toCloseableIterator();
+ createManifestIterator(
+ fileIO,
+ pathFactory.toPath(fileName),
+ projection.projectedType(),
+ null,
+ null);
return new CloseableIterator<BinaryManifestEntry>() {
@Override
public boolean hasNext() {
- entry.clear();
return rows.hasNext();
}
@Override
public BinaryManifestEntry next() {
- entry.clear();
- InternalRow row = rows.next();
- return row == null ? null : entry.replace(row);
+ return projection.createEntry().replace(rows.next());
}
@Override
public void close() throws Exception {
- entry.clear();
rows.close();
}
};
@@ -204,6 +193,22 @@ public class ManifestFile extends
ObjectsFile<ManifestEntry> {
}
}
+ private static CloseableIterator<InternalRow> createManifestIterator(
+ FileIO fileIO,
+ Path path,
+ RowType projectedType,
+ @Nullable PartitionPredicate partitionFilter,
+ @Nullable BucketFilter bucketFilter)
+ throws IOException {
+ try {
+ return new ManifestAvroReader(
+ fileIO.newInputStream(path), projectedType,
partitionFilter, bucketFilter);
+ } catch (IOException e) {
+ FileUtils.checkExists(fileIO, path);
+ throw e;
+ }
+ }
+
@VisibleForTesting
public long suggestedFileSize() {
return suggestedFileSize;
@@ -412,10 +417,7 @@ public class ManifestFile extends
ObjectsFile<ManifestEntry> {
fileIO,
schemaManager,
partitionType,
- fileFormat,
new ManifestEntrySerializer(),
- entryType,
- fileFormat.createReaderFactory(entryType, entryType, new
ArrayList<>()),
fileFormat.createWriterFactory(entryType),
compression,
pathFactory.manifestFileFactory(),
diff --git
a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestSchemaUtils.java
b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestSchemaUtils.java
index e58efc0c13..05842e085d 100644
---
a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestSchemaUtils.java
+++
b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestSchemaUtils.java
@@ -28,13 +28,15 @@ import java.util.List;
/** Utilities for manifest schemas. */
final class ManifestSchemaUtils {
+ static final String FORMAT_IDENTIFIER_FIELD = "_VERSION";
+
private ManifestSchemaUtils() {}
/** Adds the permanent on-disk format identifier field to a manifest row
type. */
static RowType withFormatIdentifier(RowType rowType) {
List<DataField> fields = new ArrayList<>();
// Keep the historical field name for compatibility with existing
manifest files.
- fields.add(new DataField(-1, "_VERSION", new IntType(false)));
+ fields.add(new DataField(-1, FORMAT_IDENTIFIER_FIELD, new
IntType(false)));
fields.addAll(rowType.getFields());
return new RowType(false, fields);
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/utils/ObjectsFile.java
b/paimon-core/src/main/java/org/apache/paimon/utils/ObjectsFile.java
index deb6c91214..7d1a5d483a 100644
--- a/paimon-core/src/main/java/org/apache/paimon/utils/ObjectsFile.java
+++ b/paimon-core/src/main/java/org/apache/paimon/utils/ObjectsFile.java
@@ -46,11 +46,12 @@ public abstract class ObjectsFile<T> implements
SimpleFileReader<T> {
protected final FileIO fileIO;
protected final ObjectSerializer<T> serializer;
- protected final FormatReaderFactory readerFactory;
protected final FormatWriterFactory writerFactory;
protected final String compression;
protected final PathFactory pathFactory;
+ private final BiFunctionWithIOE<Path, Long,
CloseableIterator<InternalRow>> iteratorFactory;
+
@Nullable protected final ObjectsCache<Path, T, ?> cache;
public ObjectsFile(
@@ -62,9 +63,31 @@ public abstract class ObjectsFile<T> implements
SimpleFileReader<T> {
String compression,
PathFactory pathFactory,
@Nullable SegmentsCache<Path> cache) {
+ this(
+ fileIO,
+ serializer,
+ formatType,
+ (file, fileSize) ->
+ FileUtils.createFormatReader(fileIO, readerFactory,
file, fileSize)
+ .toCloseableIterator(),
+ writerFactory,
+ compression,
+ pathFactory,
+ cache);
+ }
+
+ protected ObjectsFile(
+ FileIO fileIO,
+ ObjectSerializer<T> serializer,
+ RowType formatType,
+ BiFunctionWithIOE<Path, Long, CloseableIterator<InternalRow>>
iteratorFactory,
+ FormatWriterFactory writerFactory,
+ String compression,
+ PathFactory pathFactory,
+ @Nullable SegmentsCache<Path> cache) {
this.fileIO = fileIO;
this.serializer = serializer;
- this.readerFactory = readerFactory;
+ this.iteratorFactory = iteratorFactory;
this.writerFactory = writerFactory;
this.compression = compression;
this.pathFactory = pathFactory;
@@ -200,8 +223,7 @@ public abstract class ObjectsFile<T> implements
SimpleFileReader<T> {
public CloseableIterator<InternalRow> createIterator(Path file, @Nullable
Long fileSize)
throws IOException {
- return FileUtils.createFormatReader(fileIO, readerFactory, file,
fileSize)
- .toCloseableIterator();
+ return iteratorFactory.apply(file, fileSize);
}
public long fileSize(Path file) throws IOException {
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 fef9f7b9ac..c1e1898a36 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
@@ -19,14 +19,19 @@
package org.apache.paimon.manifest;
import org.apache.paimon.CoreOptions;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalRow;
import org.apache.paimon.format.FileFormat;
+import org.apache.paimon.format.FormatWriter;
import org.apache.paimon.format.SimpleColStats;
import org.apache.paimon.fs.FileIO;
import org.apache.paimon.fs.FileIOFinder;
import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.PositionOutputStream;
import org.apache.paimon.fs.local.LocalFileIO;
import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.options.Options;
+import org.apache.paimon.partition.PartitionPredicate;
import org.apache.paimon.schema.SchemaManager;
import org.apache.paimon.stats.StatsTestUtils;
import org.apache.paimon.types.DataField;
@@ -42,12 +47,11 @@ import org.junit.jupiter.api.io.TempDir;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collections;
import java.util.List;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.ThreadLocalRandom;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;
import static org.apache.paimon.TestKeyValueGenerator.DEFAULT_PART_TYPE;
@@ -79,6 +83,264 @@ public class ManifestFileTest {
assertThat(actualEntries).isEqualTo(entries);
}
+ @Test
+ void testReadMissingManifestFile() {
+ ManifestFile manifestFile = createManifestFile(tempDir.toString());
+
+ assertThatThrownBy(
+ () ->
+ manifestFile.read(
+ "missing-manifest",
+ null,
+ null,
+ null,
+ row -> true,
+ entry -> true))
+ .hasMessageContaining("not found");
+ }
+
+ @Test
+ void testAvroReaderSkipsDataFileMetaBeforeMaterialization() throws
Exception {
+ List<ManifestEntry> entries = generateData();
+ ManifestEntry selected = entries.get(0);
+ PartitionPredicate partitionFilter =
+ PartitionPredicate.fromMultiple(
+ DEFAULT_PART_TYPE,
Collections.singletonList(selected.partition()));
+ BucketFilter bucketFilter = new BucketFilter(false, selected.bucket(),
null, null);
+ List<ManifestEntry> expected =
+ entries.stream()
+ .filter(
+ entry ->
+
entry.partition().equals(selected.partition())
+ && entry.bucket() ==
selected.bucket())
+ .collect(Collectors.toList());
+
+ ManifestFile manifestFile = createManifestFile(tempDir.toString(),
Long.MAX_VALUE);
+ ManifestFileMeta manifest = writeSingleManifest(manifestFile, entries);
+ FileIO fileIO = LocalFileIO.create();
+ Path path = new Path(new Path(tempDir.toUri()), "manifest/" +
manifest.fileName());
+ ManifestEntrySerializer serializer = new ManifestEntrySerializer();
+ List<ManifestEntry> actual = new ArrayList<>();
+
+ try (ManifestAvroReader reader =
+ new ManifestAvroReader(
+ fileIO.newInputStream(path),
+ ManifestEntry.MANIFEST_ROW_TYPE,
+ partitionFilter,
+ bucketFilter)) {
+ while (reader.hasNext()) {
+ InternalRow row = reader.next();
+ actual.add(serializer.fromRow(row));
+ }
+ assertThat(reader.decodedDataFiles()).isEqualTo(expected.size());
+ assertThat(reader.skippedDataFiles()).isEqualTo(entries.size() -
expected.size());
+ }
+
+ assertThat(actual).containsExactlyElementsOf(expected);
+ assertThat(
+ manifestFile.read(
+ manifest.fileName(),
+ manifest.fileSize(),
+ partitionFilter,
+ bucketFilter,
+ row -> true,
+ entry -> true))
+ .containsExactlyElementsOf(expected);
+ }
+
+ @Test
+ void testAvroReaderSupportsReorderedNestedProjection() throws Exception {
+ List<ManifestEntry> entries = generateData();
+ ManifestFile manifestFile = createManifestFile(tempDir.toString(),
Long.MAX_VALUE);
+ ManifestFileMeta manifest = writeSingleManifest(manifestFile, entries);
+ Path path = new Path(new Path(tempDir.toUri()), "manifest/" +
manifest.fileName());
+ LocalFileIO fileIO = LocalFileIO.create();
+
+ List<DataField> fields = ManifestEntry.MANIFEST_ROW_TYPE.getFields();
+ RowType projectedFileType =
+ DataFileMeta.SCHEMA.project(DataFileMeta.FILE_NAME,
DataFileMeta.ROW_COUNT);
+ RowType projectedType =
+ new RowType(
+ false,
+ Arrays.asList(
+ fields.get(5).newType(projectedFileType),
+ fields.get(2),
+ fields.get(1)));
+ BinaryManifestEntry projectedEntry =
+
BinaryManifestEntry.Projection.create(projectedType).createEntry();
+
+ try (ManifestAvroReader reader =
+ new ManifestAvroReader(fileIO.newInputStream(path),
projectedType, null, null)) {
+ for (ManifestEntry expected : entries) {
+ assertThat(reader.hasNext()).isTrue();
+ InternalRow row = reader.next();
+ assertThat(row.getFieldCount()).isEqualTo(3);
+ assertThat(row.getRow(0,
projectedFileType.getFieldCount()).getFieldCount())
+ .isEqualTo(2);
+
+ projectedEntry.replace(row);
+
assertThat(projectedEntry.fileName()).isEqualTo(expected.fileName());
+
assertThat(projectedEntry.rowCount()).isEqualTo(expected.rowCount());
+
assertThat(projectedEntry.partition()).isEqualTo(expected.partition());
+ assertThat(projectedEntry.kind()).isEqualTo(expected.kind());
+ }
+ assertThat(reader.hasNext()).isFalse();
+ assertThat(reader.decodedDataFiles()).isEqualTo(entries.size());
+ assertThat(reader.skippedDataFiles()).isZero();
+ }
+ }
+
+ @Test
+ void testAvroReaderSkipsUnprojectedDataFile() throws Exception {
+ List<ManifestEntry> entries = generateData();
+ ManifestFile manifestFile = createManifestFile(tempDir.toString(),
Long.MAX_VALUE);
+ ManifestFileMeta manifest = writeSingleManifest(manifestFile, entries);
+ Path path = new Path(new Path(tempDir.toUri()), "manifest/" +
manifest.fileName());
+ LocalFileIO fileIO = LocalFileIO.create();
+
+ List<DataField> fields = ManifestEntry.MANIFEST_ROW_TYPE.getFields();
+ RowType projectedType = new RowType(false,
Arrays.asList(fields.get(2), fields.get(1)));
+ try (ManifestAvroReader reader =
+ new ManifestAvroReader(fileIO.newInputStream(path),
projectedType, null, null)) {
+ for (ManifestEntry expected : entries) {
+ assertThat(reader.hasNext()).isTrue();
+ InternalRow row = reader.next();
+ assertThat(row.getFieldCount()).isEqualTo(2);
+ assertThat(row.getBinary(0))
+ .containsExactly(
+
org.apache.paimon.utils.SerializationUtils.serializeBinaryRow(
+ expected.partition()));
+
assertThat(FileKind.fromByteValue(row.getByte(1))).isEqualTo(expected.kind());
+ }
+ assertThat(reader.hasNext()).isFalse();
+ assertThat(reader.decodedDataFiles()).isZero();
+ assertThat(reader.skippedDataFiles()).isEqualTo(entries.size());
+ }
+ }
+
+ @Test
+ void testProjectedScanRejectsUnsupportedFormatIdentifier() throws
Exception {
+ ManifestEntry entry = gen.next();
+ ManifestFile manifestFile = createManifestFile(tempDir.toString(),
Long.MAX_VALUE);
+ ManifestFileMeta manifest =
+ writeSingleManifest(manifestFile,
Collections.singletonList(entry));
+ Path path = new Path(new Path(tempDir.toUri()), "manifest/" +
manifest.fileName());
+ LocalFileIO fileIO = LocalFileIO.create();
+ ManifestEntrySerializer serializer = new ManifestEntrySerializer();
+ InternalRow valid = serializer.toRow(entry);
+
+ try (PositionOutputStream out = fileIO.newOutputStream(path, true);
+ FormatWriter writer =
+
avro.createWriterFactory(ManifestEntry.MANIFEST_ROW_TYPE)
+ .create(out, "zstd")) {
+ writer.addElement(
+ GenericRow.of(
+ 1,
+ valid.getByte(1),
+ valid.getBinary(2),
+ valid.getInt(3),
+ valid.getInt(4),
+ valid.getRow(5,
DataFileMeta.SCHEMA.getFieldCount())));
+ }
+
+ try (CloseableIterator<BinaryManifestEntry> entries =
+ manifestFile.scan(
+ manifest.fileName(), null,
BinaryManifestEntry.DELETE_ENTRY_PROJECTION)) {
+ assertThatThrownBy(entries::hasNext)
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("not compatible");
+ }
+ }
+
+ @Test
+ void testAvroReaderReadsLegacyDataFileMetaWithFewerFields() throws
Exception {
+ ManifestEntry generated = gen.next();
+ DataFileMeta sourceFile = generated.file().newFirstRowId(42L);
+ ManifestEntry source =
+ ManifestEntry.create(
+ FileKind.ADD,
+ generated.partition(),
+ generated.bucket(),
+ generated.totalBuckets(),
+ sourceFile);
+ RowType legacyFileType =
+ DataFileMeta.SCHEMA.project(
+ new int[] {0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12,
13, 14, 15, 16, 17});
+ List<DataField> legacyManifestFields =
+ ManifestEntry.MANIFEST_ROW_TYPE.getFields().stream()
+ .map(
+ field ->
+ ManifestEntry.FILE.equals(field.name())
+ ? field.newType(legacyFileType)
+ : field)
+ .collect(Collectors.toList());
+ RowType legacyManifestType = new RowType(false, legacyManifestFields);
+ Path path = new Path(new Path(tempDir.toUri()),
"legacy-manifest.avro");
+ LocalFileIO fileIO = LocalFileIO.create();
+ ManifestEntrySerializer serializer = new ManifestEntrySerializer();
+
+ try (PositionOutputStream out = fileIO.newOutputStream(path, false);
+ FormatWriter writer =
+
avro.createWriterFactory(legacyManifestType).create(out, "zstd")) {
+ writer.addElement(serializer.toRow(source));
+ }
+
+ ManifestEntry actual;
+ try (ManifestAvroReader reader =
+ new ManifestAvroReader(
+ fileIO.newInputStream(path),
ManifestEntry.MANIFEST_ROW_TYPE, null, null)) {
+ assertThat(reader.hasNext()).isTrue();
+ actual = serializer.fromRow(reader.next());
+ assertThat(reader.hasNext()).isFalse();
+ }
+
+ assertThat(actual.fileName()).isEqualTo(source.fileName());
+ assertThat(actual.file().firstRowId()).isNull();
+ assertThat(actual.file().writeCols()).isNull();
+ }
+
+ @Test
+ void testAvroReaderRejectsReorderedTopLevelFields() throws Exception {
+ ManifestEntry entry = gen.next();
+ List<DataField> fields = ManifestEntry.MANIFEST_ROW_TYPE.getFields();
+ RowType reorderedType =
+ new RowType(
+ false,
+ Arrays.asList(
+ fields.get(0),
+ fields.get(5),
+ fields.get(1),
+ fields.get(2),
+ fields.get(3),
+ fields.get(4)));
+ Path path = new Path(new Path(tempDir.toUri()),
"reordered-manifest.avro");
+ LocalFileIO fileIO = LocalFileIO.create();
+ ManifestEntrySerializer serializer = new ManifestEntrySerializer();
+
+ try (PositionOutputStream out = fileIO.newOutputStream(path, false);
+ FormatWriter writer =
avro.createWriterFactory(reorderedType).create(out, "zstd")) {
+ InternalRow row = serializer.toRow(entry);
+ writer.addElement(
+ GenericRow.of(
+ row.getInt(0),
+ row.getRow(5, DataFileMeta.SCHEMA.getFieldCount()),
+ row.getByte(1),
+ row.getBinary(2),
+ row.getInt(3),
+ row.getInt(4)));
+ }
+
+ assertThatThrownBy(
+ () ->
+ new ManifestAvroReader(
+ fileIO.newInputStream(path),
+ ManifestEntry.MANIFEST_ROW_TYPE,
+ null,
+ null))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("expected _KIND but found _FILE");
+ }
+
@RepeatedTest(10)
public void testCleanUpForException() throws IOException {
String failingName = UUID.randomUUID().toString();
@@ -227,40 +489,35 @@ public class ManifestFileTest {
}
@Test
- void testScanProjectedManifestEntries() throws Exception {
+ void testScanProjectedManifestEntriesCanBeRetained() throws Exception {
List<ManifestEntry> entries = Arrays.asList(gen.next(), gen.next(),
gen.next());
ManifestFile manifestFile = createManifestFile(tempDir.toString(),
Long.MAX_VALUE);
ManifestFileMeta manifest = writeSingleManifest(manifestFile, entries);
BinaryManifestEntry.Projection projection =
projection(DataFileMeta.FILE_NAME, DataFileMeta.ROW_COUNT);
- List<String> fileNames = new ArrayList<>();
- List<Long> rowCounts = new ArrayList<>();
- AtomicReference<BinaryManifestEntry> retained = new
AtomicReference<>();
+ List<BinaryManifestEntry> actual = new ArrayList<>();
try (CloseableIterator<BinaryManifestEntry> iterator =
manifestFile.scan(manifest.fileName(), manifest.fileSize(),
projection)) {
while (iterator.hasNext()) {
- BinaryManifestEntry entry = iterator.next();
- BinaryManifestEntry previous = retained.getAndSet(entry);
- if (previous != null) {
- assertThat(entry).isSameAs(previous);
- }
- fileNames.add(entry.fileName());
- rowCounts.add(entry.rowCount());
+ actual.add(iterator.next());
}
}
- assertThat(fileNames)
+ assertThat(actual).hasSize(entries.size());
+ for (int i = 1; i < actual.size(); i++) {
+ assertThat(actual.get(i)).isNotSameAs(actual.get(i - 1));
+ }
+
assertThat(actual.stream().map(ManifestEntry::fileName).collect(Collectors.toList()))
.containsExactlyElementsOf(
entries.stream().map(ManifestEntry::fileName).collect(Collectors.toList()));
- assertThat(rowCounts)
+
assertThat(actual.stream().map(ManifestEntry::rowCount).collect(Collectors.toList()))
.containsExactlyElementsOf(
entries.stream().map(ManifestEntry::rowCount).collect(Collectors.toList()));
- assertCleared(retained.get());
}
@Test
- void testScanProjectedManifestInvalidatesEntryWhenAdvancing() throws
Exception {
+ void testScanProjectedManifestKeepsEntryValidWhenAdvancing() throws
Exception {
List<ManifestEntry> entries = Arrays.asList(gen.next(), gen.next());
ManifestFile manifestFile = createManifestFile(tempDir.toString(),
Long.MAX_VALUE);
ManifestFileMeta manifest = writeSingleManifest(manifestFile, entries);
@@ -275,10 +532,10 @@ public class ManifestFileTest {
assertThat(first.fileName()).isEqualTo(entries.get(0).fileName());
assertThat(iterator.hasNext()).isTrue();
- assertCleared(first);
BinaryManifestEntry second = iterator.next();
- assertThat(second).isSameAs(first);
+ assertThat(second).isNotSameAs(first);
assertThat(second.fileName()).isEqualTo(entries.get(1).fileName());
+ assertThat(first.fileName()).isEqualTo(entries.get(0).fileName());
}
}
@@ -287,8 +544,7 @@ public class ManifestFileTest {
List<ManifestEntry> entries = Arrays.asList(gen.next(), gen.next(),
gen.next());
ManifestFile manifestFile = createManifestFile(tempDir.toString(),
Long.MAX_VALUE);
ManifestFileMeta manifest = writeSingleManifest(manifestFile, entries);
- AtomicInteger visited = new AtomicInteger();
- AtomicReference<BinaryManifestEntry> retained = new
AtomicReference<>();
+ List<BinaryManifestEntry> retained = new ArrayList<>();
try (CloseableIterator<BinaryManifestEntry> iterator =
manifestFile.scan(
@@ -297,23 +553,22 @@ public class ManifestFileTest {
projection(DataFileMeta.FILE_NAME))) {
while (iterator.hasNext()) {
BinaryManifestEntry entry = iterator.next();
- retained.set(entry);
- visited.incrementAndGet();
+ retained.add(entry);
break;
}
}
- assertThat(visited).hasValue(1);
- assertCleared(retained.get());
+ assertThat(retained).hasSize(1);
+
assertThat(retained.get(0).fileName()).isEqualTo(entries.get(0).fileName());
}
@Test
- void testScanProjectedManifestClearsEntryWhenProcessingFails() {
+ void testScanProjectedManifestKeepsEntryWhenProcessingFails() {
List<ManifestEntry> entries = Arrays.asList(gen.next(), gen.next());
ManifestFile manifestFile = createManifestFile(tempDir.toString(),
Long.MAX_VALUE);
ManifestFileMeta manifest = writeSingleManifest(manifestFile, entries);
RuntimeException failure = new RuntimeException("Expected processing
failure.");
- AtomicReference<BinaryManifestEntry> retained = new
AtomicReference<>();
+ List<BinaryManifestEntry> retained = new ArrayList<>();
assertThatThrownBy(
() -> {
@@ -324,12 +579,13 @@ public class ManifestFileTest {
projection(DataFileMeta.FILE_NAME))) {
assertThat(iterator.hasNext()).isTrue();
BinaryManifestEntry entry = iterator.next();
- retained.set(entry);
+ retained.add(entry);
throw failure;
}
})
.isSameAs(failure);
- assertCleared(retained.get());
+ assertThat(retained).hasSize(1);
+
assertThat(retained.get(0).fileName()).isEqualTo(entries.get(0).fileName());
}
private List<ManifestEntry> generateData() {
@@ -395,12 +651,6 @@ public class ManifestFileTest {
return BinaryManifestEntry.Projection.create(new RowType(false,
fields));
}
- private static void assertCleared(BinaryManifestEntry entry) {
- assertThatThrownBy(entry::fileName)
- .isInstanceOf(IllegalStateException.class)
- .hasMessageContaining("not backed by a row");
- }
-
private void checkRollingFiles(
ManifestFileMeta expected, List<ManifestFileMeta> actual, long
suggestedFileSize) {
// all but last file should be no smaller than suggestedFileSize
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
new file mode 100644
index 0000000000..ef98b15fa0
--- /dev/null
+++
b/paimon-format/src/main/java/org/apache/paimon/format/avro/AvroBlockReader.java
@@ -0,0 +1,158 @@
+/*
+ * 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.paimon.format.avro;
+
+import org.apache.paimon.utils.IOUtils;
+
+import org.apache.avro.AvroRuntimeException;
+import org.apache.avro.Schema;
+import org.apache.avro.file.DataFileStream;
+import org.apache.avro.generic.GenericDatumReader;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.ByteBuffer;
+
+/**
+ * Reader which exposes decompressed blocks from an Avro object container file.
+ *
+ * <p>This reader owns the input stream and closes it when construction fails
or {@link #close()} is
+ * called.
+ */
+public final class AvroBlockReader implements Closeable {
+
+ private final DataFileStream<Object> reader;
+
+ private long currentBlockRecordCount = -1;
+
+ public AvroBlockReader(InputStream input) throws IOException {
+ try {
+ this.reader = new DataFileStream<>(input, new
GenericDatumReader<>());
+ } catch (IOException | RuntimeException | Error e) {
+ IOUtils.closeQuietly(input);
+ throw e;
+ }
+ }
+
+ Schema schema() {
+ return reader.getSchema();
+ }
+
+ /** Creates a record decoder from the writer schema stored in the Avro
file header. */
+ public AvroRecordDecoder createRecordDecoder() {
+ return new AvroRecordDecoder(reader.getSchema());
+ }
+
+ /** Returns whether another block is available. */
+ public boolean hasNextBlock() throws IOException {
+ return replaceAvroRuntimeException(reader::hasNext);
+ }
+
+ /**
+ * Returns a decompressed copy of the next block.
+ *
+ * <p>The returned array is owned by the caller and remains valid after
this reader advances or
+ * closes.
+ */
+ public byte[] nextBlock() throws IOException {
+ BorrowedBlock block = nextBorrowedBlock();
+ byte[] bytes = new byte[block.length];
+ System.arraycopy(block.bytes, block.offset, bytes, 0, block.length);
+ return bytes;
+ }
+
+ /**
+ * Returns a borrowed view of the next decompressed block.
+ *
+ * <p>The returned bytes are owned by this reader and may be overwritten
by the next call to
+ * {@link #hasNextBlock()}, {@link #nextBlock()}, or this method, or when
this reader is closed.
+ */
+ public BorrowedBlock nextBorrowedBlock() throws IOException {
+ ByteBuffer block = replaceAvroRuntimeException(reader::nextBlock);
+ currentBlockRecordCount = reader.getBlockCount();
+ return new BorrowedBlock(
+ block.array(),
+ block.arrayOffset() + block.position(),
+ block.remaining(),
+ currentBlockRecordCount);
+ }
+
+ /** Returns the record count of the last block returned by a block-reading
method. */
+ public long currentBlockRecordCount() {
+ if (currentBlockRecordCount < 0) {
+ throw new IllegalStateException("No block has been read.");
+ }
+ return currentBlockRecordCount;
+ }
+
+ @Override
+ public void close() throws IOException {
+ reader.close();
+ }
+
+ private static <T> T replaceAvroRuntimeException(IOSupplier<T> supplier)
throws IOException {
+ try {
+ return supplier.get();
+ } catch (AvroRuntimeException e) {
+ if (e.getCause() instanceof IOException) {
+ throw (IOException) e.getCause();
+ }
+ throw e;
+ }
+ }
+
+ /** Borrowed decompressed block contents and its record count. */
+ public static final class BorrowedBlock {
+
+ private final byte[] bytes;
+ private final int offset;
+ private final int length;
+ private final long recordCount;
+
+ private BorrowedBlock(byte[] bytes, int offset, int length, long
recordCount) {
+ this.bytes = bytes;
+ this.offset = offset;
+ this.length = length;
+ this.recordCount = recordCount;
+ }
+
+ public byte[] bytes() {
+ return bytes;
+ }
+
+ public int offset() {
+ return offset;
+ }
+
+ public int length() {
+ return length;
+ }
+
+ public long recordCount() {
+ return recordCount;
+ }
+ }
+
+ @FunctionalInterface
+ private interface IOSupplier<T> {
+
+ T get() throws IOException;
+ }
+}
diff --git
a/paimon-format/src/main/java/org/apache/paimon/format/avro/AvroRecordDecoder.java
b/paimon-format/src/main/java/org/apache/paimon/format/avro/AvroRecordDecoder.java
new file mode 100644
index 0000000000..78b4941897
--- /dev/null
+++
b/paimon-format/src/main/java/org/apache/paimon/format/avro/AvroRecordDecoder.java
@@ -0,0 +1,167 @@
+/*
+ * 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.paimon.format.avro;
+
+import org.apache.paimon.types.DataType;
+
+import org.apache.avro.Schema;
+import org.apache.avro.io.BinaryDecoder;
+import org.apache.avro.io.DecoderFactory;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+
+/**
+ * Decoder for sequentially reading records from Avro blocks without exposing
Avro classes to
+ * callers.
+ */
+public final class AvroRecordDecoder {
+
+ private final Schema recordSchema;
+ private final int recordBranch;
+
+ private @Nullable BinaryDecoder decoder;
+
+ AvroRecordDecoder(Schema writerSchema) {
+ if (writerSchema.getType() == Schema.Type.UNION) {
+ int branchIndex = -1;
+ Schema record = null;
+ for (int i = 0; i < writerSchema.getTypes().size(); i++) {
+ Schema branch = writerSchema.getTypes().get(i);
+ if (branch.getType() == Schema.Type.RECORD) {
+ if (record != null) {
+ throw new IllegalArgumentException(
+ "Avro union contains multiple record
branches.");
+ }
+ record = branch;
+ branchIndex = i;
+ }
+ }
+ if (record == null) {
+ throw new IllegalArgumentException("Avro schema is not a
record or record union.");
+ }
+ this.recordSchema = record;
+ this.recordBranch = branchIndex;
+ } else if (writerSchema.getType() == Schema.Type.RECORD) {
+ this.recordSchema = writerSchema;
+ this.recordBranch = -1;
+ } else {
+ throw new IllegalArgumentException("Avro schema is not a record or
record union.");
+ }
+ }
+
+ /** Returns the number of fields in the writer record. */
+ public int fieldCount() {
+ return recordSchema.getFields().size();
+ }
+
+ /** Returns the writer field name at the given position. */
+ public String fieldName(int position) {
+ return recordSchema.getFields().get(position).name();
+ }
+
+ /** Returns the Avro type of the writer field at the given position. */
+ public FieldType fieldType(int position) {
+ return
FieldType.valueOf(recordSchema.getFields().get(position).schema().getType().name());
+ }
+
+ /** Creates a decoder for one writer field. */
+ public FieldDecoder createFieldDecoder(int position, @Nullable DataType
readType) {
+ FieldReader reader =
+ new FieldReaderFactory()
+
.visit(recordSchema.getFields().get(position).schema(), readType);
+ return new FieldDecoder(reader);
+ }
+
+ /** Reuses this decoder for another block. */
+ public void reset(byte[] bytes, int offset, int length) {
+ decoder = DecoderFactory.get().binaryDecoder(bytes, offset, length,
decoder);
+ }
+
+ /** Returns whether a block has been supplied through {@link
#reset(byte[], int, int)}. */
+ public boolean isInitialized() {
+ return decoder != null;
+ }
+
+ /** Returns whether the current block has been consumed. */
+ public boolean isEnd() throws IOException {
+ return decoder().isEnd();
+ }
+
+ /** Reads the union branch, if present, and returns whether it is the
record branch. */
+ public boolean readRecordStart() throws IOException {
+ return recordBranch < 0 || decoder().readIndex() == recordBranch;
+ }
+
+ public int readInt() throws IOException {
+ return decoder().readInt();
+ }
+
+ public byte[] readBytes() throws IOException {
+ return decoder().readBytes(null).array();
+ }
+
+ public void skipBytes() throws IOException {
+ decoder().skipBytes();
+ }
+
+ private BinaryDecoder decoder() {
+ if (decoder == null) {
+ throw new IllegalStateException("No Avro block has been
supplied.");
+ }
+ return decoder;
+ }
+
+ /** Avro schema types represented without exposing Avro's {@link
Schema.Type}. */
+ public enum FieldType {
+ RECORD,
+ ENUM,
+ ARRAY,
+ MAP,
+ UNION,
+ FIXED,
+ STRING,
+ BYTES,
+ INT,
+ LONG,
+ FLOAT,
+ DOUBLE,
+ BOOLEAN,
+ NULL
+ }
+
+ /** Decoder for one field in the writer record. */
+ public static final class FieldDecoder {
+
+ private final FieldReader reader;
+
+ private FieldDecoder(FieldReader reader) {
+ this.reader = reader;
+ }
+
+ public Object read(AvroRecordDecoder decoder, @Nullable Object reuse)
throws IOException {
+ return reader.read(decoder.decoder(), reuse);
+ }
+
+ public void skip(AvroRecordDecoder decoder) throws IOException {
+ reader.skip(decoder.decoder());
+ }
+ }
+}
diff --git
a/paimon-format/src/main/java/org/apache/paimon/format/avro/FieldReaderFactory.java
b/paimon-format/src/main/java/org/apache/paimon/format/avro/FieldReaderFactory.java
index 4801134507..1369a359eb 100644
---
a/paimon-format/src/main/java/org/apache/paimon/format/avro/FieldReaderFactory.java
+++
b/paimon-format/src/main/java/org/apache/paimon/format/avro/FieldReaderFactory.java
@@ -662,17 +662,21 @@ public class FieldReaderFactory implements
AvroSchemaVisitor<FieldReader> {
row = new GenericRow(mapping.length);
}
- Object[] values = new Object[fieldReaders.length];
for (int i = 0; i < fieldReaders.length; i += 1) {
- if (mappingBack[i] >= 0) {
- values[i] = fieldReaders[i].read(decoder,
row.getField(mappingBack[i]));
+ int outputPosition = mappingBack[i];
+ if (outputPosition >= 0) {
+ row.setField(
+ outputPosition,
+ fieldReaders[i].read(decoder,
row.getField(outputPosition)));
} else {
fieldReaders[i].skip(decoder);
}
}
for (int i = 0; i < mapping.length; i++) {
- row.setField(i, mapping[i] >= 0 ? values[mapping[i]] : null);
+ if (mapping[i] < 0) {
+ row.setField(i, null);
+ }
}
return row;
diff --git
a/paimon-format/src/test/java/org/apache/paimon/format/avro/AvroFileFormatTest.java
b/paimon-format/src/test/java/org/apache/paimon/format/avro/AvroFileFormatTest.java
index cb3d7de27d..b8cb87f1c4 100644
---
a/paimon-format/src/test/java/org/apache/paimon/format/avro/AvroFileFormatTest.java
+++
b/paimon-format/src/test/java/org/apache/paimon/format/avro/AvroFileFormatTest.java
@@ -24,6 +24,7 @@ import org.apache.paimon.format.FileFormat;
import org.apache.paimon.format.FileFormatFactory.FormatContext;
import org.apache.paimon.format.FormatReaderContext;
import org.apache.paimon.format.FormatWriter;
+import org.apache.paimon.format.avro.AvroBlockReader.BorrowedBlock;
import org.apache.paimon.fs.FileIO;
import org.apache.paimon.fs.Path;
import org.apache.paimon.fs.PositionOutputStream;
@@ -35,14 +36,23 @@ import org.apache.paimon.types.DataField;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.RowType;
+import org.apache.avro.Schema;
+import org.apache.avro.SchemaBuilder;
+import org.apache.avro.io.BinaryDecoder;
+import org.apache.avro.io.BinaryEncoder;
+import org.apache.avro.io.DecoderFactory;
+import org.apache.avro.io.EncoderFactory;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
+import java.io.ByteArrayOutputStream;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.NoSuchElementException;
import java.util.UUID;
import java.util.concurrent.ThreadLocalRandom;
@@ -134,6 +144,145 @@ public class AvroFileFormatTest {
}
}
+ @Test
+ void testReadDecompressedBlocks() throws IOException {
+ RowType rowType = DataTypes.ROW(DataTypes.INT().notNull()).notNull();
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path file = new Path(new Path(tempPath.toUri()),
UUID.randomUUID().toString());
+ int numRecords = 100_000;
+
+ try (PositionOutputStream out = fileIO.newOutputStream(file, false)) {
+ FormatWriter writer =
fileFormat.createWriterFactory(rowType).create(out, "zstd");
+ for (int i = 0; i < numRecords; i++) {
+ writer.addElement(GenericRow.of(i));
+ }
+ writer.close();
+ }
+
+ int nextValue = 0;
+ int numBlocks = 0;
+ byte[] firstBlock = null;
+ byte[] firstBlockCopy = null;
+ try (AvroBlockReader reader = new
AvroBlockReader(fileIO.newInputStream(file))) {
+ assertThat(reader.schema()).isNotNull();
+ assertThatThrownBy(reader::currentBlockRecordCount)
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessageContaining("No block");
+
+ AvroRowDatumReader datumReader = new AvroRowDatumReader(rowType);
+ datumReader.setSchema(reader.schema());
+ BinaryDecoder decoder = null;
+ while (reader.hasNextBlock()) {
+ byte[] block = reader.nextBlock();
+ if (firstBlock == null) {
+ firstBlock = block;
+ firstBlockCopy = block.clone();
+ }
+ numBlocks++;
+ decoder = DecoderFactory.get().binaryDecoder(block, decoder);
+ long blockRecordCount = reader.currentBlockRecordCount();
+ assertThat(blockRecordCount).isPositive();
+ for (long i = 0; i < blockRecordCount; i++) {
+ assertThat(datumReader.read(null,
decoder).getInt(0)).isEqualTo(nextValue++);
+ }
+ assertThat(decoder.isEnd()).isTrue();
+ }
+
+ assertThat(reader.hasNextBlock()).isFalse();
+
assertThatThrownBy(reader::nextBlock).isInstanceOf(NoSuchElementException.class);
+ }
+
+ assertThat(numBlocks).isGreaterThan(1);
+ assertThat(nextValue).isEqualTo(numRecords);
+ assertThat(firstBlock).containsExactly(firstBlockCopy);
+ }
+
+ @Test
+ void testReadBlocksFromEmptyFile() throws IOException {
+ RowType rowType = DataTypes.ROW(DataTypes.INT().notNull()).notNull();
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path file = new Path(new Path(tempPath.toUri()),
UUID.randomUUID().toString());
+
+ try (PositionOutputStream out = fileIO.newOutputStream(file, false)) {
+ fileFormat.createWriterFactory(rowType).create(out,
"zstd").close();
+ }
+
+ try (AvroBlockReader reader = new
AvroBlockReader(fileIO.newInputStream(file))) {
+ assertThat(reader.hasNextBlock()).isFalse();
+
assertThatThrownBy(reader::nextBlock).isInstanceOf(NoSuchElementException.class);
+ }
+ }
+
+ @Test
+ void testReadBorrowedBlocks() throws IOException {
+ RowType rowType = DataTypes.ROW(DataTypes.INT().notNull()).notNull();
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path file = new Path(new Path(tempPath.toUri()),
UUID.randomUUID().toString());
+ int numRecords = 100_000;
+
+ try (PositionOutputStream out = fileIO.newOutputStream(file, false)) {
+ FormatWriter writer =
fileFormat.createWriterFactory(rowType).create(out, "zstd");
+ for (int i = 0; i < numRecords; i++) {
+ writer.addElement(GenericRow.of(i));
+ }
+ writer.close();
+ }
+
+ int nextValue = 0;
+ try (AvroBlockReader reader = new
AvroBlockReader(fileIO.newInputStream(file))) {
+ AvroRecordDecoder decoder = reader.createRecordDecoder();
+ AvroRecordDecoder.FieldDecoder fieldDecoder =
+ decoder.createFieldDecoder(0, rowType.getTypeAt(0));
+ while (reader.hasNextBlock()) {
+ BorrowedBlock block = reader.nextBorrowedBlock();
+ assertThat(block.recordCount()).isPositive();
+
assertThat(reader.currentBlockRecordCount()).isEqualTo(block.recordCount());
+ decoder.reset(block.bytes(), block.offset(), block.length());
+ for (long i = 0; i < block.recordCount(); i++) {
+ assertThat(decoder.readRecordStart()).isTrue();
+ assertThat(fieldDecoder.read(decoder,
null)).isEqualTo(nextValue++);
+ }
+ assertThat(decoder.isEnd()).isTrue();
+ }
+ assertThatThrownBy(reader::nextBorrowedBlock)
+ .isInstanceOf(NoSuchElementException.class);
+ }
+
+ assertThat(nextValue).isEqualTo(numRecords);
+ }
+
+ @Test
+ void testRowReaderProjectsIntoReusedRow() throws IOException {
+ Schema writerSchema =
+ SchemaBuilder.record("record")
+ .fields()
+ .requiredInt("first")
+ .requiredInt("second")
+ .endRecord();
+ RowType projectedType =
+ new RowType(
+ false,
+ Arrays.asList(
+ new DataField(0, "second",
DataTypes.INT().notNull()),
+ new DataField(1, "missing", DataTypes.INT())));
+ ByteArrayOutputStream output = new ByteArrayOutputStream();
+ BinaryEncoder encoder = EncoderFactory.get().binaryEncoder(output,
null);
+ encoder.writeInt(10);
+ encoder.writeInt(20);
+ encoder.flush();
+
+ AvroRowDatumReader reader = new AvroRowDatumReader(projectedType);
+ reader.setSchema(writerSchema);
+ BinaryDecoder decoder =
DecoderFactory.get().binaryDecoder(output.toByteArray(), null);
+ GenericRow reuse = GenericRow.of(100, 200);
+ InternalRow result = reader.read(reuse, decoder);
+
+ assertThat(result).isSameAs(reuse);
+ assertThat(result.getInt(0)).isEqualTo(20);
+ assertThat(result.isNullAt(1)).isTrue();
+ assertThat(decoder.isEnd()).isTrue();
+ }
+
@Test
void testGetRealIOException() throws IOException {
RowType rowType = DataTypes.ROW(DataTypes.INT().notNull());
