You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by lz...@apache.org on 2023/06/08 08:14:03 UTC
[incubator-paimon] branch master updated: [core] Extract ObjectsFile.writeWithoutRolling (#1336)
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 5d090a223 [core] Extract ObjectsFile.writeWithoutRolling (#1336)
5d090a223 is described below
commit 5d090a223d76065a414ca077167d8b73cadcd3c2
Author: Jingsong Lee <ji...@gmail.com>
AuthorDate: Thu Jun 8 16:13:59 2023 +0800
[core] Extract ObjectsFile.writeWithoutRolling (#1336)
---
.../org/apache/paimon/manifest/ManifestFile.java | 2 +-
.../org/apache/paimon/manifest/ManifestList.java | 27 +------------
.../java/org/apache/paimon/utils/ObjectsFile.java | 46 ++++++++++++++++++++++
3 files changed, 49 insertions(+), 26 deletions(-)
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 fbaec0dae..989502afa 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
@@ -63,7 +63,7 @@ public class ManifestFile extends ObjectsFile<ManifestEntry> {
PathFactory pathFactory,
long suggestedFileSize,
@Nullable SegmentsCache<String> cache) {
- super(fileIO, serializer, readerFactory, pathFactory, cache);
+ super(fileIO, serializer, readerFactory, writerFactory, pathFactory, cache);
this.schemaManager = schemaManager;
this.partitionType = partitionType;
this.writerFactory = writerFactory;
diff --git a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestList.java b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestList.java
index 34ebe3308..8c06976f0 100644
--- a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestList.java
+++ b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestList.java
@@ -18,14 +18,10 @@
package org.apache.paimon.manifest;
-import org.apache.paimon.CoreOptions;
import org.apache.paimon.format.FileFormat;
import org.apache.paimon.format.FormatReaderFactory;
-import org.apache.paimon.format.FormatWriter;
import org.apache.paimon.format.FormatWriterFactory;
import org.apache.paimon.fs.FileIO;
-import org.apache.paimon.fs.Path;
-import org.apache.paimon.fs.PositionOutputStream;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.FileStorePathFactory;
import org.apache.paimon.utils.ObjectsFile;
@@ -52,7 +48,7 @@ public class ManifestList extends ObjectsFile<ManifestFileMeta> {
FormatWriterFactory writerFactory,
PathFactory pathFactory,
@Nullable SegmentsCache<String> cache) {
- super(fileIO, serializer, readerFactory, pathFactory, cache);
+ super(fileIO, serializer, readerFactory, writerFactory, pathFactory, cache);
this.writerFactory = writerFactory;
}
@@ -62,26 +58,7 @@ public class ManifestList extends ObjectsFile<ManifestFileMeta> {
* <p>NOTE: This method is atomic.
*/
public String write(List<ManifestFileMeta> metas) {
- Path path = pathFactory.newPath();
- try {
- try (PositionOutputStream out = fileIO.newOutputStream(path, false)) {
- FormatWriter writer =
- writerFactory.create(out, CoreOptions.FILE_COMPRESSION.defaultValue());
- try {
- for (ManifestFileMeta record : metas) {
- writer.addElement(serializer.toRow(record));
- }
- } finally {
- writer.flush();
- writer.finish();
- }
- }
- return path.getName();
- } catch (Throwable e) {
- fileIO.deleteQuietly(path);
- throw new RuntimeException(
- "Exception occurs when writing manifest list " + path + ". Clean up.", e);
- }
+ return super.writeWithoutRolling(metas);
}
/** Creator of {@link ManifestList}. */
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 8e9e6bf32..c176d9e1b 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
@@ -18,15 +18,23 @@
package org.apache.paimon.utils;
+import org.apache.paimon.CoreOptions;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.format.FormatReaderFactory;
+import org.apache.paimon.format.FormatWriter;
+import org.apache.paimon.format.FormatWriterFactory;
import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.PositionOutputStream;
import org.apache.paimon.reader.RecordReader;
import javax.annotation.Nullable;
import java.io.IOException;
+import java.io.UncheckedIOException;
import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Iterator;
import java.util.List;
import static org.apache.paimon.utils.FileUtils.createFormatReader;
@@ -37,6 +45,7 @@ public abstract class ObjectsFile<T> {
protected final FileIO fileIO;
protected final ObjectSerializer<T> serializer;
protected final FormatReaderFactory readerFactory;
+ protected final FormatWriterFactory writerFactory;
protected final PathFactory pathFactory;
@Nullable private final ObjectsCache<String, T> cache;
@@ -45,16 +54,26 @@ public abstract class ObjectsFile<T> {
FileIO fileIO,
ObjectSerializer<T> serializer,
FormatReaderFactory readerFactory,
+ FormatWriterFactory writerFactory,
PathFactory pathFactory,
@Nullable SegmentsCache<String> cache) {
this.fileIO = fileIO;
this.serializer = serializer;
this.readerFactory = readerFactory;
+ this.writerFactory = writerFactory;
this.pathFactory = pathFactory;
this.cache =
cache == null ? null : new ObjectsCache<>(cache, serializer, this::createIterator);
}
+ public long fileSize(String fileName) {
+ try {
+ return fileIO.getFileSize(pathFactory.toPath(fileName));
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ }
+
public List<T> read(String fileName) {
return read(fileName, Filter.alwaysTrue(), Filter.alwaysTrue());
}
@@ -79,6 +98,33 @@ public abstract class ObjectsFile<T> {
}
}
+ public String writeWithoutRolling(Collection<T> records) {
+ return writeWithoutRolling(records.iterator());
+ }
+
+ public String writeWithoutRolling(Iterator<T> records) {
+ Path path = pathFactory.newPath();
+ try {
+ try (PositionOutputStream out = fileIO.newOutputStream(path, false)) {
+ FormatWriter writer =
+ writerFactory.create(out, CoreOptions.FILE_COMPRESSION.defaultValue());
+ try {
+ while (records.hasNext()) {
+ writer.addElement(serializer.toRow(records.next()));
+ }
+ } finally {
+ writer.flush();
+ writer.finish();
+ }
+ }
+ return path.getName();
+ } catch (Throwable e) {
+ fileIO.deleteQuietly(path);
+ throw new RuntimeException(
+ "Exception occurs when writing records to " + path + ". Clean up.", e);
+ }
+ }
+
private CloseableIterator<InternalRow> createIterator(String fileName) {
try {
return createFormatReader(fileIO, readerFactory, pathFactory.toPath(fileName))