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 4425a598b2 [core] Add manifest Avro writer (#9194)
4425a598b2 is described below
commit 4425a598b2ec186e84779eec622b9151a1c4cead
Author: YeJunHao <[email protected]>
AuthorDate: Wed Aug 12 21:26:23 2026 +0800
[core] Add manifest Avro writer (#9194)
Extract manifest Avro writing from `ManifestFile` into a dedicated
writer and provide the encoded-record / raw-block writing primitives
required by subsequent manifest compaction optimizations.
This PR only introduces and adopts the writer infrastructure. It does
not include the run-merge sorting algorithm.
---
.../apache/paimon/manifest/ManifestAvroWriter.java | 546 +++++++++++++++++++++
.../org/apache/paimon/manifest/ManifestFile.java | 150 ++----
.../operation/ManifestEntryExternalSort.java | 11 +-
.../paimon/operation/ManifestFileMerger.java | 5 +-
.../apache/paimon/manifest/ManifestFileTest.java | 155 ++++++
.../flink/copy/CopyManifestFileOperator.java | 9 +-
.../apache/paimon/format/avro/AvroBlockWriter.java | 63 +++
.../org/apache/paimon/format/avro/AvroBuilder.java | 34 --
.../apache/paimon/format/avro/AvroBulkWriter.java | 53 --
.../apache/paimon/format/avro/AvroFileFormat.java | 53 +-
.../paimon/format/avro/AvroWriterFactory.java | 48 --
11 files changed, 834 insertions(+), 293 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestAvroWriter.java
b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestAvroWriter.java
new file mode 100644
index 0000000000..6244ed569c
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestAvroWriter.java
@@ -0,0 +1,546 @@
+/*
+ * 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.format.SimpleColStats;
+import org.apache.paimon.format.SimpleStatsCollector;
+import org.apache.paimon.format.avro.AvroBlockWriter;
+import org.apache.paimon.format.avro.AvroFileFormat;
+import org.apache.paimon.format.avro.AvroRawBlock;
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.PositionOutputStream;
+import org.apache.paimon.io.RollingFileWriter;
+import org.apache.paimon.schema.SchemaManager;
+import org.apache.paimon.stats.SimpleStatsConverter;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.IOUtils;
+import org.apache.paimon.utils.ObjectSerializer;
+import org.apache.paimon.utils.PathFactory;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.IdentityHashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Avro writer for manifest entries.
+ *
+ * <p>The writer accepts materialized entries, encoded records and compressed
Avro blocks. It is
+ * intentionally separate from the generic writer abstractions because encoded
Avro data is an
+ * implementation detail of manifest run merging.
+ */
+public final class ManifestAvroWriter implements AutoCloseable {
+
+ private final FileIO fileIO;
+ private final SchemaManager schemaManager;
+ private final RowType partitionType;
+ private final AvroFileFormat avroFileFormat;
+ private final ObjectSerializer<ManifestEntry> serializer;
+ private final String compression;
+ private final PathFactory pathFactory;
+ private final long targetFileSize;
+
+ private final List<ManifestFileMeta> results = new ArrayList<>();
+ private final List<Path> completedPaths = new ArrayList<>();
+ private @Nullable FileWriter currentWriter;
+ private long recordCount;
+ private boolean closed;
+
+ ManifestAvroWriter(
+ FileIO fileIO,
+ SchemaManager schemaManager,
+ RowType partitionType,
+ AvroFileFormat avroFileFormat,
+ ObjectSerializer<ManifestEntry> serializer,
+ String compression,
+ PathFactory pathFactory,
+ long targetFileSize) {
+ this.fileIO = fileIO;
+ this.schemaManager = schemaManager;
+ this.partitionType = partitionType;
+ this.avroFileFormat = avroFileFormat;
+ this.serializer = serializer;
+ this.compression = compression;
+ this.pathFactory = pathFactory;
+ this.targetFileSize = targetFileSize;
+ }
+
+ public void write(ManifestEntry entry) throws IOException {
+ try {
+ currentWriter().write(entry);
+ afterWrite(1, false);
+ } catch (IOException | RuntimeException | Error failure) {
+ abort();
+ throw failure;
+ }
+ }
+
+ public void write(Iterable<? extends ManifestEntry> entries) throws
IOException {
+ for (ManifestEntry entry : entries) {
+ write(entry);
+ }
+ }
+
+ public void writeEncoded(ByteBuffer encodedRecord, EncodedEntry metadata)
throws IOException {
+ try {
+ currentWriter().writeEncoded(encodedRecord, metadata);
+ afterWrite(1, false);
+ } catch (IOException | RuntimeException | Error failure) {
+ abort();
+ throw failure;
+ }
+ }
+
+ public void writeEncodedBlock(AvroRawBlock block, EncodedBlock metadata)
throws IOException {
+ if (metadata.addedFiles < 0 || metadata.deletedFiles < 0) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Manifest block file counts must be non-negative:
added %s, deleted %s.",
+ metadata.addedFiles, metadata.deletedFiles));
+ }
+ long metadataRecordCount = Math.addExact(metadata.addedFiles,
metadata.deletedFiles);
+ if (metadataRecordCount != block.recordCount()) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Manifest block record count mismatch: metadata
%s, block %s.",
+ metadataRecordCount, block.recordCount()));
+ }
+ try {
+ currentWriter().writeEncodedBlock(block, metadata);
+ afterWrite(metadataRecordCount, true);
+ } catch (IOException | RuntimeException | Error failure) {
+ abort();
+ throw failure;
+ }
+ }
+
+ private FileWriter currentWriter() {
+ if (closed) {
+ throw new IllegalStateException("Manifest writer has already
closed.");
+ }
+ if (currentWriter == null) {
+ currentWriter = new FileWriter(pathFactory.newPath());
+ }
+ return currentWriter;
+ }
+
+ private void afterWrite(long addedRecords, boolean forceSizeCheck) throws
IOException {
+ recordCount = Math.addExact(recordCount, addedRecords);
+ if (currentWriter.reachTargetSize(
+ forceSizeCheck || recordCount %
RollingFileWriter.CHECK_ROLLING_RECORD_CNT == 0,
+ targetFileSize)) {
+ closeCurrentWriter();
+ }
+ }
+
+ private void closeCurrentWriter() throws IOException {
+ if (currentWriter == null) {
+ return;
+ }
+ currentWriter.close();
+ ManifestFileMeta result = currentWriter.result();
+ completedPaths.add(currentWriter.path);
+ results.add(result);
+ currentWriter = null;
+ }
+
+ public long recordCount() {
+ return recordCount;
+ }
+
+ public List<ManifestFileMeta> result() {
+ if (!closed) {
+ throw new IllegalStateException(
+ "Cannot access manifest results before closing the
writer.");
+ }
+ return results;
+ }
+
+ public void abort() {
+ if (currentWriter != null) {
+ currentWriter.abort();
+ }
+ for (Path path : completedPaths) {
+ fileIO.deleteQuietly(path);
+ }
+ completedPaths.clear();
+ results.clear();
+ }
+
+ @Override
+ public void close() throws IOException {
+ if (closed) {
+ return;
+ }
+ try {
+ closeCurrentWriter();
+ } catch (IOException | RuntimeException | Error failure) {
+ abort();
+ throw failure;
+ } finally {
+ closed = true;
+ }
+ }
+
+ /** Reusable statistics needed when an encoded manifest record is copied
directly. */
+ public static final class EncodedEntry {
+
+ private byte kind;
+ private BinaryRow partition;
+ private int bucket;
+ private int level;
+ private long schemaId;
+ private long firstRowId;
+ private long rowCount;
+
+ public EncodedEntry replace(
+ byte kind,
+ BinaryRow partition,
+ int bucket,
+ int level,
+ long schemaId,
+ long firstRowId,
+ long rowCount) {
+ this.kind = kind;
+ this.partition = partition;
+ this.bucket = bucket;
+ this.level = level;
+ this.schemaId = schemaId;
+ this.firstRowId = firstRowId;
+ this.rowCount = rowCount;
+ return this;
+ }
+ }
+
+ /** Aggregate statistics for an encoded Avro block copied without
decompression. */
+ public static final class EncodedBlock {
+
+ private final long addedFiles;
+ private final long deletedFiles;
+ private final long schemaId;
+ private final int minBucket;
+ private final int maxBucket;
+ private final int minLevel;
+ private final int maxLevel;
+ private final long minRowId;
+ private final long maxRowId;
+ private final @Nullable BinaryRow nullPartition;
+ private final long nullPartitionCount;
+ private final @Nullable BinaryRow minNonNullPartition;
+ private final @Nullable BinaryRow maxNonNullPartition;
+
+ public EncodedBlock(
+ long addedFiles,
+ long deletedFiles,
+ long schemaId,
+ int minBucket,
+ int maxBucket,
+ int minLevel,
+ int maxLevel,
+ long minRowId,
+ long maxRowId,
+ @Nullable BinaryRow nullPartition,
+ long nullPartitionCount,
+ @Nullable BinaryRow minNonNullPartition,
+ @Nullable BinaryRow maxNonNullPartition) {
+ this.addedFiles = addedFiles;
+ this.deletedFiles = deletedFiles;
+ this.schemaId = schemaId;
+ this.minBucket = minBucket;
+ this.maxBucket = maxBucket;
+ this.minLevel = minLevel;
+ this.maxLevel = maxLevel;
+ this.minRowId = minRowId;
+ this.maxRowId = maxRowId;
+ this.nullPartition = nullPartition;
+ this.nullPartitionCount = nullPartitionCount;
+ this.minNonNullPartition = minNonNullPartition;
+ this.maxNonNullPartition = maxNonNullPartition;
+ }
+ }
+
+ private final class FileWriter {
+
+ private final Path path;
+ private final SimpleStatsCollector partitionStatsCollector;
+ private final SimpleStatsConverter partitionStatsSerializer;
+ private final Map<BinaryRow, long[]> encodedPartitionCounts = new
IdentityHashMap<>();
+ private final long[] repeatedNullCounts = new
long[partitionType.getFieldCount()];
+ private @Nullable PositionOutputStream out;
+ private @Nullable AvroBlockWriter writer;
+ private @Nullable Long outputBytes;
+ private long numAddedFiles;
+ private long numDeletedFiles;
+ private long schemaId = Long.MIN_VALUE;
+ private int minBucket = Integer.MAX_VALUE;
+ private int maxBucket = Integer.MIN_VALUE;
+ private int minLevel = Integer.MAX_VALUE;
+ private int maxLevel = Integer.MIN_VALUE;
+ private @Nullable RowIdStats rowIdStats = new RowIdStats();
+ private boolean closed;
+
+ private FileWriter(Path path) {
+ this.path = path;
+ this.partitionStatsCollector = new
SimpleStatsCollector(partitionType);
+ this.partitionStatsSerializer = new
SimpleStatsConverter(partitionType);
+ boolean outputCreated = false;
+ try {
+ out = fileIO.newOutputStream(path, false);
+ outputCreated = true;
+ writer =
+ avroFileFormat.createBlockWriter(
+ out, ManifestEntry.MANIFEST_ROW_TYPE,
compression);
+ } catch (IOException failure) {
+ IOUtils.closeQuietly(writer);
+ IOUtils.closeQuietly(out);
+ if (outputCreated) {
+ fileIO.deleteQuietly(path);
+ }
+ throw new UncheckedIOException(
+ "Failed to create manifest Avro writer for " + path,
failure);
+ } catch (RuntimeException | Error failure) {
+ IOUtils.closeQuietly(writer);
+ IOUtils.closeQuietly(out);
+ if (outputCreated) {
+ fileIO.deleteQuietly(path);
+ }
+ throw failure;
+ }
+ }
+
+ private void write(ManifestEntry entry) throws IOException {
+ ensureOpen();
+ writer.addElement(
+ entry instanceof ProjectedManifestEntry
+ ? ((ProjectedManifestEntry) entry).fullRow()
+ : serializer.toRow(entry));
+ collectStats(entry);
+ }
+
+ private void writeEncoded(ByteBuffer encodedRecord, EncodedEntry
metadata)
+ throws IOException {
+ ensureOpen();
+ writer.addEncoded(encodedRecord);
+ collectStats(metadata);
+ addEncodedPartition(metadata.partition, 1);
+ }
+
+ private void writeEncodedBlock(AvroRawBlock block, EncodedBlock
metadata)
+ throws IOException {
+ ensureOpen();
+ writer.addEncodedBlock(block);
+ collectStats(metadata);
+ if (metadata.nullPartitionCount > 0) {
+ addEncodedPartition(metadata.nullPartition,
metadata.nullPartitionCount);
+ }
+ if (metadata.minNonNullPartition != null) {
+ addEncodedPartition(metadata.minNonNullPartition, 1);
+ if (metadata.maxNonNullPartition !=
metadata.minNonNullPartition) {
+ addEncodedPartition(metadata.maxNonNullPartition, 1);
+ }
+ }
+ }
+
+ private void collectStats(ManifestEntry entry) {
+ switch (entry.kind()) {
+ case ADD:
+ numAddedFiles++;
+ break;
+ case DELETE:
+ numDeletedFiles++;
+ break;
+ default:
+ throw new UnsupportedOperationException("Unknown entry
kind: " + entry.kind());
+ }
+ schemaId = Math.max(schemaId, entry.file().schemaId());
+ minBucket = Math.min(minBucket, entry.bucket());
+ maxBucket = Math.max(maxBucket, entry.bucket());
+ minLevel = Math.min(minLevel, entry.level());
+ maxLevel = Math.max(maxLevel, entry.level());
+ if (rowIdStats != null) {
+ Long firstRowId = entry.file().firstRowId();
+ if (firstRowId == null) {
+ rowIdStats = null;
+ } else {
+ rowIdStats.collect(firstRowId, entry.file().rowCount());
+ }
+ }
+ partitionStatsCollector.collect(entry.partition());
+ }
+
+ private void collectStats(EncodedEntry entry) {
+ switch (FileKind.fromByteValue(entry.kind)) {
+ case ADD:
+ numAddedFiles++;
+ break;
+ case DELETE:
+ numDeletedFiles++;
+ break;
+ default:
+ throw new UnsupportedOperationException("Unknown entry
kind: " + entry.kind);
+ }
+ schemaId = Math.max(schemaId, entry.schemaId);
+ minBucket = Math.min(minBucket, entry.bucket);
+ maxBucket = Math.max(maxBucket, entry.bucket);
+ minLevel = Math.min(minLevel, entry.level);
+ maxLevel = Math.max(maxLevel, entry.level);
+ if (rowIdStats != null) {
+ rowIdStats.collect(entry.firstRowId, entry.rowCount);
+ }
+ }
+
+ private void collectStats(EncodedBlock block) {
+ numAddedFiles = Math.addExact(numAddedFiles, block.addedFiles);
+ numDeletedFiles = Math.addExact(numDeletedFiles,
block.deletedFiles);
+ schemaId = Math.max(schemaId, block.schemaId);
+ minBucket = Math.min(minBucket, block.minBucket);
+ maxBucket = Math.max(maxBucket, block.maxBucket);
+ minLevel = Math.min(minLevel, block.minLevel);
+ maxLevel = Math.max(maxLevel, block.maxLevel);
+ if (rowIdStats != null) {
+ rowIdStats.collectRange(block.minRowId, block.maxRowId);
+ }
+ }
+
+ private void addEncodedPartition(@Nullable BinaryRow partition, long
count) {
+ if (partition == null || count <= 0) {
+ return;
+ }
+ long[] value =
+ encodedPartitionCounts.computeIfAbsent(partition, ignored
-> new long[1]);
+ value[0] = Math.addExact(value[0], count);
+ }
+
+ private SimpleColStats[] partitionStats() {
+ for (Map.Entry<BinaryRow, long[]> entry :
encodedPartitionCounts.entrySet()) {
+ BinaryRow partition = entry.getKey();
+ partitionStatsCollector.collect(partition);
+ long repeated = entry.getValue()[0] - 1;
+ if (repeated <= 0) {
+ continue;
+ }
+ for (int field = 0; field < partition.getFieldCount();
field++) {
+ if (partition.isNullAt(field)) {
+ repeatedNullCounts[field] =
+ Math.addExact(repeatedNullCounts[field],
repeated);
+ }
+ }
+ }
+ encodedPartitionCounts.clear();
+ SimpleColStats[] stats = partitionStatsCollector.extract();
+ for (int field = 0; field < stats.length; field++) {
+ if (repeatedNullCounts[field] == 0) {
+ continue;
+ }
+ SimpleColStats current = stats[field];
+ stats[field] =
+ new SimpleColStats(
+ current.min(),
+ current.max(),
+ Math.addExact(current.nullCount(),
repeatedNullCounts[field]));
+ }
+ return stats;
+ }
+
+ private boolean reachTargetSize(boolean suggestedCheck, long
targetSize)
+ throws IOException {
+ ensureOpen();
+ return writer.reachTargetSize(suggestedCheck, targetSize);
+ }
+
+ private void ensureOpen() {
+ if (closed || writer == null) {
+ throw new IllegalStateException("Manifest writer has already
closed.");
+ }
+ }
+
+ private void abort() {
+ IOUtils.closeQuietly(writer);
+ writer = null;
+ IOUtils.closeQuietly(out);
+ out = null;
+ fileIO.deleteQuietly(path);
+ closed = true;
+ }
+
+ private void close() throws IOException {
+ if (closed) {
+ return;
+ }
+ try {
+ writer.close();
+ writer = null;
+ out.flush();
+ outputBytes = out.getPos();
+ out.close();
+ out = null;
+ } catch (IOException | RuntimeException | Error failure) {
+ abort();
+ throw failure;
+ } finally {
+ closed = true;
+ }
+ }
+
+ private ManifestFileMeta result() {
+ if (!closed || outputBytes == null) {
+ throw new IllegalStateException(
+ "Cannot access manifest result before closing the
writer.");
+ }
+ return new ManifestFileMeta(
+ path.getName(),
+ outputBytes,
+ numAddedFiles,
+ numDeletedFiles,
+ partitionStatsSerializer.toBinaryAllMode(partitionStats()),
+ numAddedFiles + numDeletedFiles > 0
+ ? schemaId
+ : schemaManager.latest().get().id(),
+ minBucket,
+ maxBucket,
+ minLevel,
+ maxLevel,
+ rowIdStats == null ? null : rowIdStats.minRowId,
+ rowIdStats == null ? null : rowIdStats.maxRowId);
+ }
+ }
+
+ private static class RowIdStats {
+
+ private long minRowId = Long.MAX_VALUE;
+ private long maxRowId = Long.MIN_VALUE;
+
+ private void collect(long firstRowId, long rowCount) {
+ minRowId = Math.min(minRowId, firstRowId);
+ maxRowId = Math.max(maxRowId, firstRowId + rowCount - 1);
+ }
+
+ private void collectRange(long minRowId, long maxRowId) {
+ this.minRowId = Math.min(this.minRowId, minRowId);
+ this.maxRowId = Math.max(this.maxRowId, maxRowId);
+ }
+ }
+}
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 0564ff198e..b9fa876f61 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,19 +21,14 @@ 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.FormatWriterFactory;
-import org.apache.paimon.format.SimpleStatsCollector;
+import org.apache.paimon.format.avro.AvroFileFormat;
import org.apache.paimon.fs.FileIO;
import org.apache.paimon.fs.Path;
import org.apache.paimon.io.DataFileMeta;
-import org.apache.paimon.io.RollingFileWriter;
-import org.apache.paimon.io.RollingFileWriterImpl;
-import org.apache.paimon.io.SingleFileWriter;
import org.apache.paimon.manifest.ProjectedManifestEntry.Projection;
import org.apache.paimon.operation.metrics.CacheMetrics;
import org.apache.paimon.partition.PartitionPredicate;
import org.apache.paimon.schema.SchemaManager;
-import org.apache.paimon.stats.SimpleStatsConverter;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.CloseableIterator;
import org.apache.paimon.utils.FileStorePathFactory;
@@ -62,15 +57,15 @@ public class ManifestFile extends
ObjectsFile<ManifestEntry> {
private final SchemaManager schemaManager;
private final RowType partitionType;
- private final FormatWriterFactory writerFactory;
+ private final AvroFileFormat avroFileFormat;
private final long suggestedFileSize;
private ManifestFile(
FileIO fileIO,
SchemaManager schemaManager,
RowType partitionType,
+ AvroFileFormat avroFileFormat,
ManifestEntrySerializer serializer,
- FormatWriterFactory writerFactory,
String compression,
PathFactory pathFactory,
long suggestedFileSize,
@@ -82,13 +77,13 @@ public class ManifestFile extends
ObjectsFile<ManifestEntry> {
(path, ignoredFileSize) ->
createManifestIterator(
fileIO, path, ManifestEntry.MANIFEST_ROW_TYPE,
null, null),
- writerFactory,
+
avroFileFormat.createWriterFactory(ManifestEntry.MANIFEST_ROW_TYPE),
compression,
pathFactory,
cache);
this.schemaManager = schemaManager;
this.partitionType = partitionType;
- this.writerFactory = writerFactory;
+ this.avroFileFormat = avroFileFormat;
this.suggestedFileSize = suggestedFileSize;
}
@@ -261,7 +256,7 @@ public class ManifestFile extends
ObjectsFile<ManifestEntry> {
* <p>NOTE: This method is atomic.
*/
public List<ManifestFileMeta> write(List<ManifestEntry> entries) {
- RollingFileWriter<ManifestEntry, ManifestFileMeta> writer =
createRollingWriter();
+ ManifestAvroWriter writer = createAvroWriter();
try {
writer.write(entries);
writer.close();
@@ -271,108 +266,52 @@ public class ManifestFile extends
ObjectsFile<ManifestEntry> {
}
}
- public RollingFileWriter<ManifestEntry, ManifestFileMeta>
createRollingWriter() {
- return new RollingFileWriterImpl<>(
- () -> new ManifestEntryWriter(writerFactory,
pathFactory.newPath(), compression),
- suggestedFileSize,
- Long.MAX_VALUE);
+ /** Creates a rolling Avro manifest writer. */
+ public ManifestAvroWriter createAvroWriter() {
+ return new ManifestAvroWriter(
+ fileIO,
+ schemaManager,
+ partitionType,
+ avroFileFormat,
+ serializer,
+ compression,
+ pathFactory,
+ suggestedFileSize);
}
- public ManifestEntryWriter createManifestEntryWriter(Path manifestPath) {
- return new ManifestEntryWriter(writerFactory, manifestPath,
compression);
+ /** Creates an Avro manifest writer for one explicit path. */
+ public ManifestAvroWriter createAvroWriter(Path manifestPath) {
+ return new ManifestAvroWriter(
+ fileIO,
+ schemaManager,
+ partitionType,
+ avroFileFormat,
+ serializer,
+ compression,
+ singlePathFactory(manifestPath),
+ Long.MAX_VALUE);
}
- /** Writer for {@link ManifestEntry}. */
- public class ManifestEntryWriter extends SingleFileWriter<ManifestEntry,
ManifestFileMeta> {
-
- private final SimpleStatsCollector partitionStatsCollector;
- private final SimpleStatsConverter partitionStatsSerializer;
-
- private long numAddedFiles = 0;
- private long numDeletedFiles = 0;
- private long schemaId = Long.MIN_VALUE;
- private int minBucket = Integer.MAX_VALUE;
- private int maxBucket = Integer.MIN_VALUE;
- private int minLevel = Integer.MAX_VALUE;
- private int maxLevel = Integer.MIN_VALUE;
- private @Nullable RowIdStats rowIdStats = new RowIdStats();
-
- ManifestEntryWriter(FormatWriterFactory factory, Path path, String
fileCompression) {
- super(
- ManifestFile.this.fileIO,
- factory,
- path,
- serializer::toRow,
- fileCompression,
- false);
- this.partitionStatsCollector = new
SimpleStatsCollector(partitionType);
- this.partitionStatsSerializer = new
SimpleStatsConverter(partitionType);
- }
+ private PathFactory singlePathFactory(Path manifestPath) {
+ return new PathFactory() {
- @Override
- public void write(ManifestEntry entry) throws IOException {
- if (entry instanceof ProjectedManifestEntry) {
- writeRow(((ProjectedManifestEntry) entry).fullRow());
- } else {
- super.write(entry);
- }
+ private boolean created;
- switch (entry.kind()) {
- case ADD:
- numAddedFiles++;
- break;
- case DELETE:
- numDeletedFiles++;
- break;
- default:
- throw new UnsupportedOperationException("Unknown entry
kind: " + entry.kind());
- }
- schemaId = Math.max(schemaId, entry.file().schemaId());
- minBucket = Math.min(minBucket, entry.bucket());
- maxBucket = Math.max(maxBucket, entry.bucket());
- minLevel = Math.min(minLevel, entry.level());
- maxLevel = Math.max(maxLevel, entry.level());
- if (rowIdStats != null) {
- Long firstRowId = entry.file().firstRowId();
- if (firstRowId == null) {
- rowIdStats = null;
- } else {
- rowIdStats.collect(firstRowId, entry.file().rowCount());
+ @Override
+ public Path newPath() {
+ if (created) {
+ throw new IllegalStateException(
+ "Cannot create more than one fixed-path manifest
file.");
}
+ created = true;
+ return manifestPath;
}
- partitionStatsCollector.collect(entry.partition());
- }
-
- @Override
- public ManifestFileMeta result() throws IOException {
- return new ManifestFileMeta(
- path.getName(),
- outputBytes(),
- numAddedFiles,
- numDeletedFiles,
-
partitionStatsSerializer.toBinaryAllMode(partitionStatsCollector.extract()),
- numAddedFiles + numDeletedFiles > 0
- ? schemaId
- : schemaManager.latest().get().id(),
- minBucket,
- maxBucket,
- minLevel,
- maxLevel,
- rowIdStats == null ? null : rowIdStats.minRowId,
- rowIdStats == null ? null : rowIdStats.maxRowId);
- }
- }
-
- private static class RowIdStats {
-
- private long minRowId = Long.MAX_VALUE;
- private long maxRowId = Long.MIN_VALUE;
-
- private void collect(long firstRowId, long rowCount) {
- minRowId = Math.min(minRowId, firstRowId);
- maxRowId = Math.max(maxRowId, firstRowId + rowCount - 1);
- }
+ @Override
+ public Path toPath(String fileName) {
+ return pathFactory.toPath(fileName);
+ }
+ };
}
/** Creator of {@link ManifestFile}. */
@@ -411,13 +350,12 @@ public class ManifestFile extends
ObjectsFile<ManifestEntry> {
}
public ManifestFile create() {
- RowType entryType = ManifestEntry.MANIFEST_ROW_TYPE;
return new ManifestFile(
fileIO,
schemaManager,
partitionType,
+ (AvroFileFormat) fileFormat,
new ManifestEntrySerializer(),
- fileFormat.createWriterFactory(entryType),
compression,
pathFactory.manifestFileFactory(),
suggestedFileSize,
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/ManifestEntryExternalSort.java
b/paimon-core/src/main/java/org/apache/paimon/operation/ManifestEntryExternalSort.java
index 58866adb82..5c9b83772e 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/ManifestEntryExternalSort.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/ManifestEntryExternalSort.java
@@ -24,9 +24,9 @@ import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.GenericRow;
import org.apache.paimon.data.serializer.InternalRowSerializer;
import org.apache.paimon.disk.IOManager;
-import org.apache.paimon.io.RollingFileWriter;
import org.apache.paimon.manifest.CompactFileIdentifierSet;
import org.apache.paimon.manifest.FileEntry.ReusableIdentifier;
+import org.apache.paimon.manifest.ManifestAvroWriter;
import org.apache.paimon.manifest.ManifestEntry;
import org.apache.paimon.manifest.ManifestFile;
import org.apache.paimon.manifest.ManifestFileMeta;
@@ -233,8 +233,7 @@ public class ManifestEntryExternalSort {
return Collections.emptyList();
}
- RollingFileWriter<ManifestEntry, ManifestFileMeta> writer =
- manifestFile.createRollingWriter();
+ ManifestAvroWriter writer = manifestFile.createAvroWriter();
Exception exception = null;
try {
MutableObjectIterator<BinaryRow> iterator =
sortBuffer.sortedIterator();
@@ -267,10 +266,8 @@ public class ManifestEntryExternalSort {
return Pair.of(Collections.emptyList(),
Collections.emptyList());
}
- RollingFileWriter<ManifestEntry, ManifestFileMeta> addWriter =
- manifestFile.createRollingWriter();
- RollingFileWriter<ManifestEntry, ManifestFileMeta> deleteWriter =
- manifestFile.createRollingWriter();
+ ManifestAvroWriter addWriter = manifestFile.createAvroWriter();
+ ManifestAvroWriter deleteWriter = manifestFile.createAvroWriter();
CompactFileIdentifierSet matchedEntries = new
CompactFileIdentifierSet();
CompactFileIdentifierSet emittedDeletes = new
CompactFileIdentifierSet();
ReusableIdentifier identifier = new ReusableIdentifier();
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileMerger.java
b/paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileMerger.java
index 0313b2b12b..7c5019f1e3 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileMerger.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileMerger.java
@@ -21,8 +21,8 @@ package org.apache.paimon.operation;
import org.apache.paimon.CoreOptions;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.disk.IOManager;
-import org.apache.paimon.io.RollingFileWriter;
import org.apache.paimon.manifest.FileEntry;
+import org.apache.paimon.manifest.ManifestAvroWriter;
import org.apache.paimon.manifest.ManifestEntry;
import org.apache.paimon.manifest.ManifestFile;
import org.apache.paimon.manifest.ManifestFileMeta;
@@ -264,8 +264,7 @@ public class ManifestFileMerger {
return Optional.empty();
}
- RollingFileWriter<ManifestEntry, ManifestFileMeta> writer =
- manifestFile.createRollingWriter();
+ ManifestAvroWriter writer = manifestFile.createAvroWriter();
Function<ManifestFileMeta, List<FullCompactionReadResult>> reader =
file ->
singletonList(
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 0846cb2697..3bd052ab7a 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
@@ -85,6 +85,105 @@ public class ManifestFileTest {
assertThat(actualEntries).isEqualTo(entries);
}
+ @Test
+ void testWriteManifestFileToExplicitPath() throws Exception {
+ List<ManifestEntry> entries = generateData();
+ ManifestFile manifestFile = createManifestFile(tempDir.toString());
+ Path path = new Path(tempDir.toString() +
"/manifest/explicit-manifest");
+
+ ManifestAvroWriter writer = manifestFile.createAvroWriter(path);
+ writer.write(entries);
+ writer.close();
+
+ assertThat(writer.result())
+ .singleElement()
+ .extracting(ManifestFileMeta::fileName)
+ .isEqualTo("explicit-manifest");
+ assertThat(manifestFile.read("explicit-manifest")).isEqualTo(entries);
+ }
+
+ @Test
+ void testAbortedManifestWriterDoesNotExposeResults() throws Exception {
+ ManifestAvroWriter writer =
createManifestFile(tempDir.toString()).createAvroWriter();
+ writer.write(gen.next());
+
+ writer.abort();
+
+ assertThatThrownBy(writer::result)
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessageContaining("before closing");
+ }
+
+ @Test
+ void testWriteEncodedMixedBlockCountsDeletes() throws Exception {
+ assertEncodedBlockCounts(FileKind.ADD, FileKind.DELETE);
+ }
+
+ @Test
+ void testWriteEncodedDeleteOnlyBlockCountsDeletes() throws Exception {
+ assertEncodedBlockCounts(FileKind.DELETE, FileKind.DELETE);
+ }
+
+ @Test
+ void testWriteEncodedRecords() throws Exception {
+ ManifestEntry source = gen.next();
+ List<ManifestEntry> entries =
+ Arrays.asList(
+ ManifestEntry.create(
+ FileKind.ADD,
+ source.partition(),
+ source.bucket(),
+ source.totalBuckets(),
+ source.file().newFirstRowId(10L)),
+ ManifestEntry.create(
+ FileKind.DELETE,
+ source.partition(),
+ source.bucket(),
+ source.totalBuckets(),
+ source.file().newFirstRowId(20L)));
+ ManifestFile manifestFile = createManifestFile(tempDir.toString(),
Long.MAX_VALUE);
+ ManifestFileMeta sourceMeta = writeSingleManifest(manifestFile,
entries);
+ ManifestAvroWriter writer = manifestFile.createAvroWriter();
+ ManifestAvroWriter.EncodedEntry metadata = new
ManifestAvroWriter.EncodedEntry();
+ int position = 0;
+
+ try (ManifestAvroReader reader = openManifestReader(sourceMeta)) {
+ while (reader.hasNext()) {
+ ManifestAvroReader.RowIterator rows =
+ reader.next().toRows(ManifestEntry.MANIFEST_ROW_TYPE);
+ while (rows.hasNext()) {
+ rows.next();
+ ManifestEntry entry = entries.get(position++);
+ writer.writeEncoded(
+ rows.encodedRecord(),
+ metadata.replace(
+ entry.kind().toByteValue(),
+ entry.partition(),
+ entry.bucket(),
+ entry.level(),
+ entry.file().schemaId(),
+ entry.file().firstRowId(),
+ entry.file().rowCount()));
+ }
+ }
+ }
+ writer.close();
+
+ assertThat(position).isEqualTo(entries.size());
+ ManifestFileMeta result = writer.result().get(0);
+
assertThat(result.numAddedFiles()).isEqualTo(sourceMeta.numAddedFiles());
+
assertThat(result.numDeletedFiles()).isEqualTo(sourceMeta.numDeletedFiles());
+
assertThat(result.partitionStats()).isEqualTo(sourceMeta.partitionStats());
+ assertThat(result.schemaId()).isEqualTo(sourceMeta.schemaId());
+ assertThat(result.minBucket()).isEqualTo(sourceMeta.minBucket());
+ assertThat(result.maxBucket()).isEqualTo(sourceMeta.maxBucket());
+ assertThat(result.minLevel()).isEqualTo(sourceMeta.minLevel());
+ assertThat(result.maxLevel()).isEqualTo(sourceMeta.maxLevel());
+ assertThat(result.minRowId()).isEqualTo(sourceMeta.minRowId());
+ assertThat(result.maxRowId()).isEqualTo(sourceMeta.maxRowId());
+
assertThat(manifestFile.read(result.fileName())).containsExactlyElementsOf(entries);
+ }
+
@Test
void testReadMissingManifestFile() {
ManifestFile manifestFile = createManifestFile(tempDir.toString());
@@ -767,6 +866,62 @@ public class ManifestFileTest {
return entries;
}
+ private void assertEncodedBlockCounts(FileKind... kinds) throws Exception {
+ ManifestEntry source = gen.next();
+ List<ManifestEntry> entries = new ArrayList<>();
+ long firstRowId = 10;
+ for (FileKind kind : kinds) {
+ DataFileMeta file = source.file().newFirstRowId(firstRowId);
+ entries.add(
+ ManifestEntry.create(
+ kind,
+ source.partition(),
+ source.bucket(),
+ source.totalBuckets(),
+ file));
+ firstRowId += file.rowCount();
+ }
+
+ ManifestFile manifestFile = createManifestFile(tempDir.toString(),
Long.MAX_VALUE);
+ ManifestFileMeta sourceMeta = writeSingleManifest(manifestFile,
entries);
+ ManifestAvroWriter writer = manifestFile.createAvroWriter();
+ try (ManifestAvroReader reader = openManifestReader(sourceMeta)) {
+ assertThat(reader.hasNext()).isTrue();
+ ManifestAvroReader.RawBlock block = reader.next();
+ assertThat(block.recordCount()).isEqualTo(entries.size());
+ writer.writeEncodedBlock(block.encodedBlock(),
encodedBlock(sourceMeta, entries));
+ assertThat(reader.hasNext()).isFalse();
+ }
+ writer.close();
+
+ ManifestFileMeta result = writer.result().get(0);
+ assertThat(result.numAddedFiles())
+ .isEqualTo(entries.stream().filter(entry -> entry.kind() ==
FileKind.ADD).count());
+ assertThat(result.numDeletedFiles())
+ .isEqualTo(
+ entries.stream().filter(entry -> entry.kind() ==
FileKind.DELETE).count());
+
assertThat(manifestFile.read(result.fileName())).containsExactlyElementsOf(entries);
+ }
+
+ private ManifestAvroWriter.EncodedBlock encodedBlock(
+ ManifestFileMeta meta, List<ManifestEntry> entries) {
+ boolean nullPartition = entries.get(0).partition().isNullAt(0);
+ return new ManifestAvroWriter.EncodedBlock(
+ meta.numAddedFiles(),
+ meta.numDeletedFiles(),
+ meta.schemaId(),
+ meta.minBucket(),
+ meta.maxBucket(),
+ meta.minLevel(),
+ meta.maxLevel(),
+ meta.minRowId(),
+ meta.maxRowId(),
+ nullPartition ? entries.get(0).partition() : null,
+ nullPartition ? entries.size() : 0,
+ nullPartition ? null : entries.get(0).partition(),
+ nullPartition ? null : entries.get(0).partition());
+ }
+
private ManifestAvroReader openManifestReader(ManifestFileMeta manifest)
throws IOException {
FileIO fileIO = LocalFileIO.create();
Path path = new Path(new Path(tempDir.toUri()), "manifest/" +
manifest.fileName());
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/copy/CopyManifestFileOperator.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/copy/CopyManifestFileOperator.java
index 1abcccd0bb..ba6094a44f 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/copy/CopyManifestFileOperator.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/copy/CopyManifestFileOperator.java
@@ -24,9 +24,9 @@ import org.apache.paimon.catalog.Identifier;
import org.apache.paimon.flink.FlinkCatalogFactory;
import org.apache.paimon.fs.FileIO;
import org.apache.paimon.fs.Path;
+import org.apache.paimon.manifest.ManifestAvroWriter;
import org.apache.paimon.manifest.ManifestEntry;
import org.apache.paimon.manifest.ManifestFile;
-import org.apache.paimon.manifest.ManifestFile.ManifestEntryWriter;
import org.apache.paimon.options.Options;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.utils.FileStorePathFactory;
@@ -158,10 +158,9 @@ public class CopyManifestFileOperator extends
AbstractStreamOperator<CopyFileInf
manifestEntry.file().newExternalPath(null));
targetManifestEntries.add(newManifestEntry);
}
- ManifestEntryWriter manifestEntryWriter =
- manifestFile.createManifestEntryWriter(targetPath);
- manifestEntryWriter.write(targetManifestEntries);
- manifestEntryWriter.close();
+ ManifestAvroWriter writer =
manifestFile.createAvroWriter(targetPath);
+ writer.write(targetManifestEntries);
+ writer.close();
} else {
// copy it
IOUtils.copyBytes(
diff --git
a/paimon-format/src/main/java/org/apache/paimon/format/avro/AvroBlockWriter.java
b/paimon-format/src/main/java/org/apache/paimon/format/avro/AvroBlockWriter.java
new file mode 100644
index 0000000000..377a7862d0
--- /dev/null
+++
b/paimon-format/src/main/java/org/apache/paimon/format/avro/AvroBlockWriter.java
@@ -0,0 +1,63 @@
+/*
+ * 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.data.InternalRow;
+import org.apache.paimon.format.FormatWriter;
+import org.apache.paimon.fs.PositionOutputStream;
+
+import org.apache.avro.file.DataFileWriter;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+
+/** Avro writer which accepts normal rows, encoded records and compressed
blocks. */
+public final class AvroBlockWriter implements FormatWriter {
+
+ private final DataFileWriter<InternalRow> writer;
+ private final PositionOutputStream out;
+
+ public AvroBlockWriter(DataFileWriter<InternalRow> writer,
PositionOutputStream out) {
+ this.writer = writer;
+ this.out = out;
+ }
+
+ @Override
+ public void addElement(InternalRow element) throws IOException {
+ writer.append(element);
+ }
+
+ public void addEncoded(ByteBuffer record) throws IOException {
+ writer.appendEncoded(record);
+ }
+
+ public void addEncodedBlock(AvroRawBlock block) throws IOException {
+ writer.appendAllFrom(block.asStream(), false);
+ }
+
+ @Override
+ public boolean reachTargetSize(boolean suggestedCheck, long targetSize)
throws IOException {
+ return suggestedCheck && out.getPos() >= targetSize;
+ }
+
+ @Override
+ public void close() throws IOException {
+ writer.close();
+ }
+}
diff --git
a/paimon-format/src/main/java/org/apache/paimon/format/avro/AvroBuilder.java
b/paimon-format/src/main/java/org/apache/paimon/format/avro/AvroBuilder.java
deleted file mode 100644
index 8c39dc64ae..0000000000
--- a/paimon-format/src/main/java/org/apache/paimon/format/avro/AvroBuilder.java
+++ /dev/null
@@ -1,34 +0,0 @@
-/*
- * 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.avro.file.DataFileWriter;
-
-import java.io.IOException;
-import java.io.OutputStream;
-import java.io.Serializable;
-
-/** A builder to create an {@link AvroBulkWriter} from an {@link
OutputStream}. */
-@FunctionalInterface
-public interface AvroBuilder<T> extends Serializable {
-
- /** Creates and configures an Avro writer to the given output file. */
- DataFileWriter<T> createWriter(OutputStream outputStream, String
compression)
- throws IOException;
-}
diff --git
a/paimon-format/src/main/java/org/apache/paimon/format/avro/AvroBulkWriter.java
b/paimon-format/src/main/java/org/apache/paimon/format/avro/AvroBulkWriter.java
deleted file mode 100644
index 3257638647..0000000000
---
a/paimon-format/src/main/java/org/apache/paimon/format/avro/AvroBulkWriter.java
+++ /dev/null
@@ -1,53 +0,0 @@
-/*
- * 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.avro.file.DataFileWriter;
-
-import java.io.Closeable;
-import java.io.IOException;
-
-/** A simple writer implementation that wraps an Avro {@link DataFileWriter}.
*/
-public class AvroBulkWriter<T> implements Closeable {
-
- /** The underlying Avro writer. */
- private final DataFileWriter<T> dataFileWriter;
-
- /**
- * Create a new AvroBulkWriter wrapping the given Avro {@link
DataFileWriter}.
- *
- * @param dataFileWriter The underlying Avro writer.
- */
- public AvroBulkWriter(DataFileWriter<T> dataFileWriter) {
- this.dataFileWriter = dataFileWriter;
- }
-
- public void addElement(T element) throws IOException {
- dataFileWriter.append(element);
- }
-
- public void flush() throws IOException {
- dataFileWriter.flush();
- }
-
- @Override
- public void close() throws IOException {
- dataFileWriter.close();
- }
-}
diff --git
a/paimon-format/src/main/java/org/apache/paimon/format/avro/AvroFileFormat.java
b/paimon-format/src/main/java/org/apache/paimon/format/avro/AvroFileFormat.java
index 43105fd1cd..73b08eec47 100644
---
a/paimon-format/src/main/java/org/apache/paimon/format/avro/AvroFileFormat.java
+++
b/paimon-format/src/main/java/org/apache/paimon/format/avro/AvroFileFormat.java
@@ -25,6 +25,7 @@ import org.apache.paimon.format.FormatReaderFactory;
import org.apache.paimon.format.FormatWriter;
import org.apache.paimon.format.FormatWriterFactory;
import org.apache.paimon.format.SimpleStatsExtractor;
+import org.apache.paimon.fs.CloseShieldOutputStream;
import org.apache.paimon.fs.PositionOutputStream;
import org.apache.paimon.options.ConfigOption;
import org.apache.paimon.options.ConfigOptions;
@@ -85,6 +86,18 @@ public class AvroFileFormat extends FileFormat {
return new RowAvroWriterFactory(type);
}
+ public AvroBlockWriter createBlockWriter(
+ PositionOutputStream out, RowType rowType, String compression)
throws IOException {
+ Schema schema =
+ AvroSchemaConverter.convertToSchema(rowType,
options.get(AVRO_ROW_NAME_MAPPING));
+ AvroRowDatumWriter datumWriter = new AvroRowDatumWriter(rowType);
+ DataFileWriter<InternalRow> writer = new DataFileWriter<>(datumWriter);
+ writer.setCodec(createCodecFactory(compression));
+ writer.setFlushOnEveryBlock(false);
+ writer.create(schema, new CloseShieldOutputStream(out));
+ return new AvroBlockWriter(writer, out);
+ }
+
@Override
public Optional<SimpleStatsExtractor> createStatsExtractor(
RowType type, SimpleColStatsCollector.Factory[] statsCollectors) {
@@ -113,50 +126,16 @@ public class AvroFileFormat extends FileFormat {
/** A {@link FormatWriterFactory} to write {@link InternalRow}. */
private class RowAvroWriterFactory implements FormatWriterFactory {
- private final AvroWriterFactory<InternalRow> factory;
+ private final RowType rowType;
private RowAvroWriterFactory(RowType rowType) {
- this.factory =
- new AvroWriterFactory<>(
- (out, compression) -> {
- Schema schema =
- AvroSchemaConverter.convertToSchema(
- rowType,
options.get(AVRO_ROW_NAME_MAPPING));
- AvroRowDatumWriter datumWriter = new
AvroRowDatumWriter(rowType);
- DataFileWriter<InternalRow> dataFileWriter =
- new DataFileWriter<>(datumWriter);
-
dataFileWriter.setCodec(createCodecFactory(compression));
- dataFileWriter.setFlushOnEveryBlock(false);
- dataFileWriter.create(schema, out);
- return dataFileWriter;
- });
+ this.rowType = rowType;
}
@Override
public FormatWriter create(PositionOutputStream out, String
compression)
throws IOException {
- AvroBulkWriter<InternalRow> writer = factory.create(out,
compression);
- return new FormatWriter() {
-
- @Override
- public void addElement(InternalRow element) throws IOException
{
- writer.addElement(element);
- }
-
- @Override
- public void close() throws IOException {
- writer.close();
- }
-
- @Override
- public boolean reachTargetSize(boolean suggestedCheck, long
targetSize)
- throws IOException {
- if (out != null) {
- return suggestedCheck && out.getPos() >= targetSize;
- }
- throw new IOException("Failed to get stream length: no
open stream");
- }
- };
+ return createBlockWriter(out, rowType, compression);
}
}
}
diff --git
a/paimon-format/src/main/java/org/apache/paimon/format/avro/AvroWriterFactory.java
b/paimon-format/src/main/java/org/apache/paimon/format/avro/AvroWriterFactory.java
deleted file mode 100644
index 5cbfaca7ee..0000000000
---
a/paimon-format/src/main/java/org/apache/paimon/format/avro/AvroWriterFactory.java
+++ /dev/null
@@ -1,48 +0,0 @@
-/*
- * 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.fs.CloseShieldOutputStream;
-import org.apache.paimon.fs.PositionOutputStream;
-
-import org.apache.avro.file.DataFileWriter;
-
-import java.io.IOException;
-
-/**
- * A factory that creates an {@link AvroBulkWriter}.
- *
- * @param <T> The type of record to write.
- */
-public class AvroWriterFactory<T> {
-
- /** The builder to construct the Avro {@link DataFileWriter}. */
- private final AvroBuilder<T> avroBuilder;
-
- /** Creates a new AvroWriterFactory using the given builder to assemble
the ParquetWriter. */
- public AvroWriterFactory(AvroBuilder<T> avroBuilder) {
- this.avroBuilder = avroBuilder;
- }
-
- public AvroBulkWriter<T> create(PositionOutputStream out, String
compression)
- throws IOException {
- return new AvroBulkWriter<>(
- avroBuilder.createWriter(new CloseShieldOutputStream(out),
compression));
- }
-}
