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 2024/08/19 08:20:38 UTC
(paimon) branch master updated: [core] Add manifestListReader and manifestFileReader to Table (#3991)
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/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new d0524ca3b [core] Add manifestListReader and manifestFileReader to Table (#3991)
d0524ca3b is described below
commit d0524ca3bfeed655f440131ad838bfec73ccc71e
Author: Jingsong Lee <ji...@gmail.com>
AuthorDate: Mon Aug 19 16:20:33 2024 +0800
[core] Add manifestListReader and manifestFileReader to Table (#3991)
---
.../java/org/apache/paimon/catalog/Identifier.java | 13 +-
.../org/apache/paimon/index/IndexFileMeta.java | 40 ++-
.../paimon/index/IndexFileMetaSerializer.java | 2 +-
.../java/org/apache/paimon/io/DataFileMeta.java | 18 +-
.../apache/paimon/io/DataFileMeta08Serializer.java | 5 +-
.../apache/paimon/io/DataFileMetaSerializer.java | 2 +-
.../org/apache/paimon/io/IdentifierSerializer.java | 2 +-
.../java/org/apache/paimon/manifest/FileKind.java | 9 +-
.../apache/paimon/manifest/IndexManifestEntry.java | 46 ++-
.../manifest/IndexManifestEntrySerializer.java | 2 +-
.../apache/paimon/manifest/IndexManifestFile.java | 2 +-
.../org/apache/paimon/manifest/ManifestEntry.java | 97 ++-----
.../paimon/manifest/ManifestEntrySerializer.java | 2 +-
.../org/apache/paimon/manifest/ManifestFile.java | 4 +-
.../apache/paimon/manifest/ManifestFileMeta.java | 316 ++-------------------
.../manifest/ManifestFileMetaSerializer.java | 2 +-
.../org/apache/paimon/manifest/ManifestList.java | 2 +-
.../paimon/manifest/SimpleFileEntrySerializer.java | 2 +-
.../paimon/operation/AbstractFileStoreScan.java | 71 ++++-
.../paimon/operation/FileStoreCommitImpl.java | 2 +-
.../ManifestFileMerger.java} | 132 ++-------
.../paimon/privilege/PrivilegedFileStoreTable.java | 13 +
.../java/org/apache/paimon/stats/SimpleStats.java | 18 ++
.../apache/paimon/stats/SimpleStatsConverter.java | 16 --
.../paimon/table/AbstractFileStoreTable.java | 13 +
.../paimon/table/FallbackReadFileStoreTable.java | 13 +
.../org/apache/paimon/table/ReadonlyTable.java | 19 ++
.../main/java/org/apache/paimon/table/Table.java | 11 +
.../sink/CommitMessageLegacyV2Serializer.java | 5 +-
.../apache/paimon/table/system/AuditLogTable.java | 13 +
.../apache/paimon/table/system/BucketsTable.java | 13 +
.../paimon/table/system/FileMonitorTable.java | 13 +
.../paimon/table/system/ReadOptimizedTable.java | 13 +
.../java/org/apache/paimon/utils/ObjectsFile.java | 3 +-
.../FileKind.java => utils/SimpleFileReader.java} | 36 +--
.../paimon/manifest/ManifestFileMetaTest.java | 26 +-
.../manifest/NoPartitionManifestFileMetaTest.java | 4 +-
37 files changed, 382 insertions(+), 618 deletions(-)
diff --git a/paimon-core/src/main/java/org/apache/paimon/catalog/Identifier.java b/paimon-core/src/main/java/org/apache/paimon/catalog/Identifier.java
index c5473a714..72da69b67 100644
--- a/paimon-core/src/main/java/org/apache/paimon/catalog/Identifier.java
+++ b/paimon-core/src/main/java/org/apache/paimon/catalog/Identifier.java
@@ -19,6 +19,7 @@
package org.apache.paimon.catalog;
import org.apache.paimon.annotation.Public;
+import org.apache.paimon.types.DataField;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.BranchManager;
@@ -28,6 +29,7 @@ import org.apache.paimon.utils.StringUtils;
import javax.annotation.Nullable;
import java.io.Serializable;
+import java.util.Arrays;
import java.util.Objects;
import static org.apache.paimon.utils.Preconditions.checkArgument;
@@ -42,6 +44,13 @@ public class Identifier implements Serializable {
private static final long serialVersionUID = 1L;
+ public static final RowType SCHEMA =
+ new RowType(
+ false,
+ Arrays.asList(
+ new DataField(0, "_DATABASE", DataTypes.STRING()),
+ new DataField(1, "_OBJECT", DataTypes.STRING())));
+
public static final String UNKNOWN_DATABASE = "unknown";
private final String database;
@@ -190,8 +199,4 @@ public class Identifier implements Serializable {
public String toString() {
return String.format("Identifier{database='%s', object='%s'}", database, object);
}
-
- public static RowType schema() {
- return RowType.builder().fields(DataTypes.STRING(), DataTypes.STRING()).build();
- }
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/index/IndexFileMeta.java b/paimon-core/src/main/java/org/apache/paimon/index/IndexFileMeta.java
index b6afd0c44..97bb18255 100644
--- a/paimon-core/src/main/java/org/apache/paimon/index/IndexFileMeta.java
+++ b/paimon-core/src/main/java/org/apache/paimon/index/IndexFileMeta.java
@@ -28,9 +28,8 @@ import org.apache.paimon.utils.Pair;
import javax.annotation.Nullable;
-import java.util.ArrayList;
+import java.util.Arrays;
import java.util.LinkedHashMap;
-import java.util.List;
import java.util.Objects;
import static org.apache.paimon.utils.SerializationUtils.newStringType;
@@ -38,6 +37,24 @@ import static org.apache.paimon.utils.SerializationUtils.newStringType;
/** Metadata of index file. */
public class IndexFileMeta {
+ public static final RowType SCHEMA =
+ new RowType(
+ false,
+ Arrays.asList(
+ new DataField(0, "_INDEX_TYPE", newStringType(false)),
+ new DataField(1, "_FILE_NAME", newStringType(false)),
+ new DataField(2, "_FILE_SIZE", new BigIntType(false)),
+ new DataField(3, "_ROW_COUNT", new BigIntType(false)),
+ new DataField(
+ 4,
+ "_DELETION_VECTORS_RANGES",
+ new ArrayType(
+ true,
+ RowType.of(
+ newStringType(false),
+ new IntType(false),
+ new IntType(false))))));
+
private final String indexType;
private final String fileName;
private final long fileSize;
@@ -123,23 +140,4 @@ public class IndexFileMeta {
+ deletionVectorsRanges
+ '}';
}
-
- public static RowType schema() {
- List<DataField> fields = new ArrayList<>();
- fields.add(new DataField(0, "_INDEX_TYPE", newStringType(false)));
- fields.add(new DataField(1, "_FILE_NAME", newStringType(false)));
- fields.add(new DataField(2, "_FILE_SIZE", new BigIntType(false)));
- fields.add(new DataField(3, "_ROW_COUNT", new BigIntType(false)));
- fields.add(
- new DataField(
- 4,
- "_DELETION_VECTORS_RANGES",
- new ArrayType(
- true,
- RowType.of(
- newStringType(false),
- new IntType(false),
- new IntType(false)))));
- return new RowType(fields);
- }
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/index/IndexFileMetaSerializer.java b/paimon-core/src/main/java/org/apache/paimon/index/IndexFileMetaSerializer.java
index e9d82af46..4b5293262 100644
--- a/paimon-core/src/main/java/org/apache/paimon/index/IndexFileMetaSerializer.java
+++ b/paimon-core/src/main/java/org/apache/paimon/index/IndexFileMetaSerializer.java
@@ -33,7 +33,7 @@ import java.util.LinkedHashMap;
public class IndexFileMetaSerializer extends ObjectSerializer<IndexFileMeta> {
public IndexFileMetaSerializer() {
- super(IndexFileMeta.schema());
+ super(IndexFileMeta.SCHEMA);
}
@Override
diff --git a/paimon-core/src/main/java/org/apache/paimon/io/DataFileMeta.java b/paimon-core/src/main/java/org/apache/paimon/io/DataFileMeta.java
index 9200ad291..364d4d42e 100644
--- a/paimon-core/src/main/java/org/apache/paimon/io/DataFileMeta.java
+++ b/paimon-core/src/main/java/org/apache/paimon/io/DataFileMeta.java
@@ -19,12 +19,12 @@
package org.apache.paimon.io;
import org.apache.paimon.CoreOptions;
+import org.apache.paimon.annotation.Public;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.Timestamp;
import org.apache.paimon.fs.Path;
import org.apache.paimon.manifest.FileSource;
import org.apache.paimon.stats.SimpleStats;
-import org.apache.paimon.stats.SimpleStatsConverter;
import org.apache.paimon.types.ArrayType;
import org.apache.paimon.types.BigIntType;
import org.apache.paimon.types.DataField;
@@ -50,19 +50,25 @@ import static org.apache.paimon.utils.Preconditions.checkArgument;
import static org.apache.paimon.utils.SerializationUtils.newBytesType;
import static org.apache.paimon.utils.SerializationUtils.newStringType;
-/** Metadata of a data file. */
+/**
+ * Metadata of a data file.
+ *
+ * @since 0.9.0
+ */
+@Public
public class DataFileMeta {
public static final RowType SCHEMA =
new RowType(
+ false,
Arrays.asList(
new DataField(0, "_FILE_NAME", newStringType(false)),
new DataField(1, "_FILE_SIZE", new BigIntType(false)),
new DataField(2, "_ROW_COUNT", new BigIntType(false)),
new DataField(3, "_MIN_KEY", newBytesType(false)),
new DataField(4, "_MAX_KEY", newBytesType(false)),
- new DataField(5, "_KEY_STATS", SimpleStatsConverter.schema()),
- new DataField(6, "_VALUE_STATS", SimpleStatsConverter.schema()),
+ new DataField(5, "_KEY_STATS", SimpleStats.SCHEMA),
+ new DataField(6, "_VALUE_STATS", SimpleStats.SCHEMA),
new DataField(7, "_MIN_SEQUENCE_NUMBER", new BigIntType(false)),
new DataField(8, "_MAX_SEQUENCE_NUMBER", new BigIntType(false)),
new DataField(9, "_SCHEMA_ID", new BigIntType(false)),
@@ -490,10 +496,6 @@ public class DataFileMeta {
fileSource);
}
- public static RowType schema() {
- return SCHEMA;
- }
-
public static long getMaxSequenceNumber(List<DataFileMeta> fileMetas) {
return fileMetas.stream()
.map(DataFileMeta::maxSequenceNumber)
diff --git a/paimon-core/src/main/java/org/apache/paimon/io/DataFileMeta08Serializer.java b/paimon-core/src/main/java/org/apache/paimon/io/DataFileMeta08Serializer.java
index 9c12f3834..c65f7e78e 100644
--- a/paimon-core/src/main/java/org/apache/paimon/io/DataFileMeta08Serializer.java
+++ b/paimon-core/src/main/java/org/apache/paimon/io/DataFileMeta08Serializer.java
@@ -24,7 +24,6 @@ import org.apache.paimon.data.safe.SafeBinaryRow;
import org.apache.paimon.data.serializer.InternalRowSerializer;
import org.apache.paimon.data.serializer.InternalSerializers;
import org.apache.paimon.stats.SimpleStats;
-import org.apache.paimon.stats.SimpleStatsConverter;
import org.apache.paimon.types.ArrayType;
import org.apache.paimon.types.BigIntType;
import org.apache.paimon.types.DataField;
@@ -62,8 +61,8 @@ public class DataFileMeta08Serializer implements Serializable {
fields.add(new DataField(2, "_ROW_COUNT", new BigIntType(false)));
fields.add(new DataField(3, "_MIN_KEY", newBytesType(false)));
fields.add(new DataField(4, "_MAX_KEY", newBytesType(false)));
- fields.add(new DataField(5, "_KEY_STATS", SimpleStatsConverter.schema()));
- fields.add(new DataField(6, "_VALUE_STATS", SimpleStatsConverter.schema()));
+ fields.add(new DataField(5, "_KEY_STATS", SimpleStats.SCHEMA));
+ fields.add(new DataField(6, "_VALUE_STATS", SimpleStats.SCHEMA));
fields.add(new DataField(7, "_MIN_SEQUENCE_NUMBER", new BigIntType(false)));
fields.add(new DataField(8, "_MAX_SEQUENCE_NUMBER", new BigIntType(false)));
fields.add(new DataField(9, "_SCHEMA_ID", new BigIntType(false)));
diff --git a/paimon-core/src/main/java/org/apache/paimon/io/DataFileMetaSerializer.java b/paimon-core/src/main/java/org/apache/paimon/io/DataFileMetaSerializer.java
index 78891d064..209aaafd8 100644
--- a/paimon-core/src/main/java/org/apache/paimon/io/DataFileMetaSerializer.java
+++ b/paimon-core/src/main/java/org/apache/paimon/io/DataFileMetaSerializer.java
@@ -36,7 +36,7 @@ public class DataFileMetaSerializer extends ObjectSerializer<DataFileMeta> {
private static final long serialVersionUID = 1L;
public DataFileMetaSerializer() {
- super(DataFileMeta.schema());
+ super(DataFileMeta.SCHEMA);
}
@Override
diff --git a/paimon-core/src/main/java/org/apache/paimon/io/IdentifierSerializer.java b/paimon-core/src/main/java/org/apache/paimon/io/IdentifierSerializer.java
index 8ff39f099..0780726aa 100644
--- a/paimon-core/src/main/java/org/apache/paimon/io/IdentifierSerializer.java
+++ b/paimon-core/src/main/java/org/apache/paimon/io/IdentifierSerializer.java
@@ -28,7 +28,7 @@ import org.apache.paimon.utils.ObjectSerializer;
public class IdentifierSerializer extends ObjectSerializer<Identifier> {
public IdentifierSerializer() {
- super(Identifier.schema());
+ super(Identifier.SCHEMA);
}
@Override
diff --git a/paimon-core/src/main/java/org/apache/paimon/manifest/FileKind.java b/paimon-core/src/main/java/org/apache/paimon/manifest/FileKind.java
index a18d65f15..e7091f583 100644
--- a/paimon-core/src/main/java/org/apache/paimon/manifest/FileKind.java
+++ b/paimon-core/src/main/java/org/apache/paimon/manifest/FileKind.java
@@ -18,7 +18,14 @@
package org.apache.paimon.manifest;
-/** Kind of a file. */
+import org.apache.paimon.annotation.Public;
+
+/**
+ * Kind of a file.
+ *
+ * @since 0.9.0
+ */
+@Public
public enum FileKind {
ADD((byte) 0),
diff --git a/paimon-core/src/main/java/org/apache/paimon/manifest/IndexManifestEntry.java b/paimon-core/src/main/java/org/apache/paimon/manifest/IndexManifestEntry.java
index 6d370a76b..41e94e9d7 100644
--- a/paimon-core/src/main/java/org/apache/paimon/manifest/IndexManifestEntry.java
+++ b/paimon-core/src/main/java/org/apache/paimon/manifest/IndexManifestEntry.java
@@ -27,8 +27,7 @@ import org.apache.paimon.types.IntType;
import org.apache.paimon.types.RowType;
import org.apache.paimon.types.TinyIntType;
-import java.util.ArrayList;
-import java.util.List;
+import java.util.Arrays;
import java.util.Objects;
import static org.apache.paimon.utils.Preconditions.checkArgument;
@@ -38,6 +37,27 @@ import static org.apache.paimon.utils.SerializationUtils.newStringType;
/** Manifest entry for index file. */
public class IndexManifestEntry {
+ public static final RowType SCHEMA =
+ new RowType(
+ false,
+ Arrays.asList(
+ new DataField(0, "_KIND", new TinyIntType(false)),
+ new DataField(1, "_PARTITION", newBytesType(false)),
+ new DataField(2, "_BUCKET", new IntType(false)),
+ new DataField(3, "_INDEX_TYPE", newStringType(false)),
+ new DataField(4, "_FILE_NAME", newStringType(false)),
+ new DataField(5, "_FILE_SIZE", new BigIntType(false)),
+ new DataField(6, "_ROW_COUNT", new BigIntType(false)),
+ new DataField(
+ 7,
+ "_DELETIONS_VECTORS_RANGES",
+ new ArrayType(
+ true,
+ RowType.of(
+ newStringType(false),
+ new IntType(false),
+ new IntType(false))))));
+
private final FileKind kind;
private final BinaryRow partition;
private final int bucket;
@@ -72,28 +92,6 @@ public class IndexManifestEntry {
return indexFile;
}
- public static RowType schema() {
- List<DataField> fields = new ArrayList<>();
- fields.add(new DataField(0, "_KIND", new TinyIntType(false)));
- fields.add(new DataField(1, "_PARTITION", newBytesType(false)));
- fields.add(new DataField(2, "_BUCKET", new IntType(false)));
- fields.add(new DataField(3, "_INDEX_TYPE", newStringType(false)));
- fields.add(new DataField(4, "_FILE_NAME", newStringType(false)));
- fields.add(new DataField(5, "_FILE_SIZE", new BigIntType(false)));
- fields.add(new DataField(6, "_ROW_COUNT", new BigIntType(false)));
- fields.add(
- new DataField(
- 7,
- "_DELETIONS_VECTORS_RANGES",
- new ArrayType(
- true,
- RowType.of(
- newStringType(false),
- new IntType(false),
- new IntType(false)))));
- return new RowType(false, fields);
- }
-
@Override
public boolean equals(Object o) {
if (this == o) {
diff --git a/paimon-core/src/main/java/org/apache/paimon/manifest/IndexManifestEntrySerializer.java b/paimon-core/src/main/java/org/apache/paimon/manifest/IndexManifestEntrySerializer.java
index bdd711442..574e93555 100644
--- a/paimon-core/src/main/java/org/apache/paimon/manifest/IndexManifestEntrySerializer.java
+++ b/paimon-core/src/main/java/org/apache/paimon/manifest/IndexManifestEntrySerializer.java
@@ -33,7 +33,7 @@ import static org.apache.paimon.utils.SerializationUtils.serializeBinaryRow;
public class IndexManifestEntrySerializer extends VersionedObjectSerializer<IndexManifestEntry> {
public IndexManifestEntrySerializer() {
- super(IndexManifestEntry.schema());
+ super(IndexManifestEntry.SCHEMA);
}
@Override
diff --git a/paimon-core/src/main/java/org/apache/paimon/manifest/IndexManifestFile.java b/paimon-core/src/main/java/org/apache/paimon/manifest/IndexManifestFile.java
index 9989f5809..0f30a9fa2 100644
--- a/paimon-core/src/main/java/org/apache/paimon/manifest/IndexManifestFile.java
+++ b/paimon-core/src/main/java/org/apache/paimon/manifest/IndexManifestFile.java
@@ -93,7 +93,7 @@ public class IndexManifestFile extends ObjectsFile<IndexManifestEntry> {
}
public IndexManifestFile create() {
- RowType schema = VersionedObjectSerializer.versionType(IndexManifestEntry.schema());
+ RowType schema = VersionedObjectSerializer.versionType(IndexManifestEntry.SCHEMA);
return new IndexManifestFile(
fileIO,
schema,
diff --git a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestEntry.java b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestEntry.java
index 65cef5517..b6c79295e 100644
--- a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestEntry.java
+++ b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestEntry.java
@@ -18,28 +18,38 @@
package org.apache.paimon.manifest;
+import org.apache.paimon.annotation.Public;
import org.apache.paimon.data.BinaryRow;
-import org.apache.paimon.data.InternalRow;
import org.apache.paimon.io.DataFileMeta;
-import org.apache.paimon.partition.PartitionPredicate;
import org.apache.paimon.types.DataField;
import org.apache.paimon.types.IntType;
import org.apache.paimon.types.RowType;
import org.apache.paimon.types.TinyIntType;
-import org.apache.paimon.utils.Filter;
-import javax.annotation.Nullable;
-
-import java.util.ArrayList;
+import java.util.Arrays;
import java.util.List;
import java.util.Objects;
-import java.util.function.Function;
import static org.apache.paimon.utils.SerializationUtils.newBytesType;
-/** Entry of a manifest file, representing an addition / deletion of a data file. */
+/**
+ * Entry of a manifest file, representing an addition / deletion of a data file.
+ *
+ * @since 0.9.0
+ */
+@Public
public class ManifestEntry implements FileEntry {
+ public static final RowType SCHEMA =
+ new RowType(
+ false,
+ Arrays.asList(
+ new DataField(0, "_KIND", new TinyIntType(false)),
+ new DataField(1, "_PARTITION", newBytesType(false)),
+ new DataField(2, "_BUCKET", new IntType(false)),
+ new DataField(3, "_TOTAL_BUCKETS", new IntType(false)),
+ new DataField(4, "_FILE", DataFileMeta.SCHEMA)));
+
private final FileKind kind;
// for tables without partition this field should be a row with 0 columns (not null)
private final BinaryRow partition;
@@ -104,16 +114,6 @@ public class ManifestEntry implements FileEntry {
return new Identifier(partition, bucket, file.level(), file.fileName());
}
- public static RowType schema() {
- List<DataField> fields = new ArrayList<>();
- fields.add(new DataField(0, "_KIND", new TinyIntType(false)));
- fields.add(new DataField(1, "_PARTITION", newBytesType(false)));
- fields.add(new DataField(2, "_BUCKET", new IntType(false)));
- fields.add(new DataField(3, "_TOTAL_BUCKETS", new IntType(false)));
- fields.add(new DataField(4, "_FILE", DataFileMeta.schema()));
- return new RowType(false, fields);
- }
-
@Override
public boolean equals(Object o) {
if (!(o instanceof ManifestEntry)) {
@@ -137,67 +137,6 @@ public class ManifestEntry implements FileEntry {
return String.format("{%s, %s, %d, %d, %s}", kind, partition, bucket, totalBuckets, file);
}
- /**
- * According to the {@link ManifestCacheFilter}, entry that needs to be cached will be retained,
- * so the entry that will not be accessed in the future will not be cached.
- *
- * <p>Implemented to {@link InternalRow} is for performance (No deserialization).
- */
- public static Filter<InternalRow> createCacheRowFilter(
- @Nullable ManifestCacheFilter manifestCacheFilter, int numOfBuckets) {
- if (manifestCacheFilter == null) {
- return Filter.alwaysTrue();
- }
-
- Function<InternalRow, BinaryRow> partitionGetter =
- ManifestEntrySerializer.partitionGetter();
- Function<InternalRow, Integer> bucketGetter = ManifestEntrySerializer.bucketGetter();
- Function<InternalRow, Integer> totalBucketGetter =
- ManifestEntrySerializer.totalBucketGetter();
- return row -> {
- if (numOfBuckets != totalBucketGetter.apply(row)) {
- return true;
- }
-
- return manifestCacheFilter.test(partitionGetter.apply(row), bucketGetter.apply(row));
- };
- }
-
- /**
- * Read the corresponding entries based on the current required partition and bucket.
- *
- * <p>Implemented to {@link InternalRow} is for performance (No deserialization).
- */
- public static Filter<InternalRow> createEntryRowFilter(
- @Nullable PartitionPredicate partitionFilter,
- @Nullable Filter<Integer> bucketFilter,
- @Nullable Filter<String> fileNameFilter,
- int numOfBuckets) {
- Function<InternalRow, BinaryRow> partitionGetter =
- ManifestEntrySerializer.partitionGetter();
- Function<InternalRow, Integer> bucketGetter = ManifestEntrySerializer.bucketGetter();
- Function<InternalRow, Integer> totalBucketGetter =
- ManifestEntrySerializer.totalBucketGetter();
- Function<InternalRow, String> fileNameGetter = ManifestEntrySerializer.fileNameGetter();
- return row -> {
- if ((partitionFilter != null && !partitionFilter.test(partitionGetter.apply(row)))) {
- return false;
- }
-
- if (bucketFilter != null
- && numOfBuckets == totalBucketGetter.apply(row)
- && !bucketFilter.test(bucketGetter.apply(row))) {
- return false;
- }
-
- if (fileNameFilter != null && !fileNameFilter.test((fileNameGetter.apply(row)))) {
- return false;
- }
-
- return true;
- };
- }
-
public static long recordCount(List<ManifestEntry> manifestEntries) {
return manifestEntries.stream().mapToLong(manifest -> manifest.file().rowCount()).sum();
}
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 83c2be59a..877ef6673 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
@@ -38,7 +38,7 @@ public class ManifestEntrySerializer extends VersionedObjectSerializer<ManifestE
private final DataFileMetaSerializer dataFileMetaSerializer;
public ManifestEntrySerializer() {
- super(ManifestEntry.schema());
+ super(ManifestEntry.SCHEMA);
this.dataFileMetaSerializer = new DataFileMetaSerializer();
}
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 5bd558854..48216b0c4 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
@@ -197,7 +197,7 @@ public class ManifestFile extends ObjectsFile<ManifestEntry> {
}
public ManifestFile create() {
- RowType entryType = VersionedObjectSerializer.versionType(ManifestEntry.schema());
+ RowType entryType = VersionedObjectSerializer.versionType(ManifestEntry.SCHEMA);
return new ManifestFile(
fileIO,
schemaManager,
@@ -213,7 +213,7 @@ public class ManifestFile extends ObjectsFile<ManifestEntry> {
}
public ObjectsFile<SimpleFileEntry> createSimpleFileEntryReader() {
- RowType entryType = VersionedObjectSerializer.versionType(ManifestEntry.schema());
+ RowType entryType = VersionedObjectSerializer.versionType(ManifestEntry.SCHEMA);
return new ObjectsFile<>(
fileIO,
new SimpleFileEntrySerializer(),
diff --git a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFileMeta.java b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFileMeta.java
index 6196dab86..a4f900631 100644
--- a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFileMeta.java
+++ b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFileMeta.java
@@ -18,38 +18,35 @@
package org.apache.paimon.manifest;
-import org.apache.paimon.data.BinaryRow;
-import org.apache.paimon.io.RollingFileWriter;
-import org.apache.paimon.manifest.FileEntry.Identifier;
-import org.apache.paimon.partition.PartitionPredicate;
+import org.apache.paimon.annotation.Public;
import org.apache.paimon.stats.SimpleStats;
-import org.apache.paimon.stats.SimpleStatsConverter;
import org.apache.paimon.types.BigIntType;
import org.apache.paimon.types.DataField;
import org.apache.paimon.types.RowType;
import org.apache.paimon.types.VarCharType;
-import org.apache.paimon.utils.IOUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import javax.annotation.Nullable;
-
-import java.util.ArrayList;
-import java.util.HashSet;
-import java.util.LinkedHashMap;
-import java.util.List;
-import java.util.Map;
+import java.util.Arrays;
import java.util.Objects;
-import java.util.Optional;
-import java.util.Set;
-import static org.apache.paimon.utils.Preconditions.checkArgument;
-
-/** Metadata of a manifest file. */
+/**
+ * Metadata of a manifest file.
+ *
+ * @since 0.9.0
+ */
+@Public
public class ManifestFileMeta {
- private static final Logger LOG = LoggerFactory.getLogger(ManifestFileMeta.class);
+ public static final RowType SCHEMA =
+ new RowType(
+ false,
+ Arrays.asList(
+ new DataField(
+ 0, "_FILE_NAME", new VarCharType(false, Integer.MAX_VALUE)),
+ new DataField(1, "_FILE_SIZE", new BigIntType(false)),
+ new DataField(2, "_NUM_ADDED_FILES", new BigIntType(false)),
+ new DataField(3, "_NUM_DELETED_FILES", new BigIntType(false)),
+ new DataField(4, "_PARTITION_STATS", SimpleStats.SCHEMA),
+ new DataField(5, "_SCHEMA_ID", new BigIntType(false))));
private final String fileName;
private final long fileSize;
@@ -97,17 +94,6 @@ public class ManifestFileMeta {
return schemaId;
}
- public static RowType schema() {
- List<DataField> fields = new ArrayList<>();
- fields.add(new DataField(0, "_FILE_NAME", new VarCharType(false, Integer.MAX_VALUE)));
- fields.add(new DataField(1, "_FILE_SIZE", new BigIntType(false)));
- fields.add(new DataField(2, "_NUM_ADDED_FILES", new BigIntType(false)));
- fields.add(new DataField(3, "_NUM_DELETED_FILES", new BigIntType(false)));
- fields.add(new DataField(4, "_PARTITION_STATS", SimpleStatsConverter.schema()));
- fields.add(new DataField(5, "_SCHEMA_ID", new BigIntType(false)));
- return new RowType(false, fields);
- }
-
@Override
public boolean equals(Object o) {
if (!(o instanceof ManifestFileMeta)) {
@@ -134,268 +120,4 @@ public class ManifestFileMeta {
"{%s, %d, %d, %d, %s, %d}",
fileName, fileSize, numAddedFiles, numDeletedFiles, partitionStats, schemaId);
}
-
- /**
- * Merge several {@link ManifestFileMeta}s. {@link ManifestEntry}s representing first adding and
- * then deleting the same data file will cancel each other.
- *
- * <p>NOTE: This method is atomic.
- */
- public static List<ManifestFileMeta> merge(
- List<ManifestFileMeta> input,
- ManifestFile manifestFile,
- long suggestedMetaSize,
- int suggestedMinMetaCount,
- long manifestFullCompactionSize,
- RowType partitionType,
- @Nullable Integer manifestReadParallelism) {
- // these are the newly created manifest files, clean them up if exception occurs
- List<ManifestFileMeta> newMetas = new ArrayList<>();
-
- try {
- Optional<List<ManifestFileMeta>> fullCompacted =
- tryFullCompaction(
- input,
- newMetas,
- manifestFile,
- suggestedMetaSize,
- manifestFullCompactionSize,
- partitionType,
- manifestReadParallelism);
- return fullCompacted.orElseGet(
- () ->
- tryMinorCompaction(
- input,
- newMetas,
- manifestFile,
- suggestedMetaSize,
- suggestedMinMetaCount,
- manifestReadParallelism));
- } catch (Throwable e) {
- // exception occurs, clean up and rethrow
- for (ManifestFileMeta manifest : newMetas) {
- manifestFile.delete(manifest.fileName);
- }
- throw new RuntimeException(e);
- }
- }
-
- private static List<ManifestFileMeta> tryMinorCompaction(
- List<ManifestFileMeta> input,
- List<ManifestFileMeta> newMetas,
- ManifestFile manifestFile,
- long suggestedMetaSize,
- int suggestedMinMetaCount,
- @Nullable Integer manifestReadParallelism) {
- List<ManifestFileMeta> result = new ArrayList<>();
- List<ManifestFileMeta> candidates = new ArrayList<>();
- long totalSize = 0;
- // merge existing small manifest files
- for (ManifestFileMeta manifest : input) {
- totalSize += manifest.fileSize;
- candidates.add(manifest);
- if (totalSize >= suggestedMetaSize) {
- // reach suggested file size, perform merging and produce new file
- mergeCandidates(
- candidates, manifestFile, result, newMetas, manifestReadParallelism);
- candidates.clear();
- totalSize = 0;
- }
- }
-
- // merge the last bit of manifests if there are too many
- if (candidates.size() >= suggestedMinMetaCount) {
- mergeCandidates(candidates, manifestFile, result, newMetas, manifestReadParallelism);
- } else {
- result.addAll(candidates);
- }
- return result;
- }
-
- private static void mergeCandidates(
- List<ManifestFileMeta> candidates,
- ManifestFile manifestFile,
- List<ManifestFileMeta> result,
- List<ManifestFileMeta> newMetas,
- @Nullable Integer manifestReadParallelism) {
- if (candidates.size() == 1) {
- result.add(candidates.get(0));
- return;
- }
-
- Map<Identifier, ManifestEntry> map = new LinkedHashMap<>();
- FileEntry.mergeEntries(manifestFile, candidates, map, manifestReadParallelism);
- if (!map.isEmpty()) {
- List<ManifestFileMeta> merged = manifestFile.write(new ArrayList<>(map.values()));
- result.addAll(merged);
- newMetas.addAll(merged);
- }
- }
-
- public static Optional<List<ManifestFileMeta>> tryFullCompaction(
- List<ManifestFileMeta> inputs,
- List<ManifestFileMeta> newMetas,
- ManifestFile manifestFile,
- long suggestedMetaSize,
- long sizeTrigger,
- RowType partitionType,
- @Nullable Integer manifestReadParallelism)
- throws Exception {
- // 1. should trigger full compaction
-
- List<ManifestFileMeta> base = new ArrayList<>();
- long totalManifestSize = 0;
- int i = 0;
- for (; i < inputs.size(); i++) {
- ManifestFileMeta file = inputs.get(i);
- if (file.numDeletedFiles == 0 && file.fileSize >= suggestedMetaSize) {
- base.add(file);
- totalManifestSize += file.fileSize;
- } else {
- break;
- }
- }
-
- List<ManifestFileMeta> delta = new ArrayList<>();
- long deltaDeleteFileNum = 0;
- long totalDeltaFileSize = 0;
- for (; i < inputs.size(); i++) {
- ManifestFileMeta file = inputs.get(i);
- delta.add(file);
- totalManifestSize += file.fileSize;
- deltaDeleteFileNum += file.numDeletedFiles();
- totalDeltaFileSize += file.fileSize();
- }
-
- if (totalDeltaFileSize < sizeTrigger) {
- return Optional.empty();
- }
-
- // 2. do full compaction
-
- LOG.info(
- "Start Manifest File Full Compaction, pick the number of delete file: {}, total manifest file size: {}",
- deltaDeleteFileNum,
- totalManifestSize);
-
- // 2.1. try to skip base files by partition filter
-
- Map<Identifier, ManifestEntry> deltaMerged = new LinkedHashMap<>();
- FileEntry.mergeEntries(manifestFile, delta, deltaMerged, manifestReadParallelism);
-
- List<ManifestFileMeta> result = new ArrayList<>();
- int j = 0;
- if (partitionType.getFieldCount() > 0) {
- Set<BinaryRow> deletePartitions = computeDeletePartitions(deltaMerged);
- PartitionPredicate predicate =
- PartitionPredicate.fromMultiple(partitionType, deletePartitions);
- if (predicate != null) {
- for (; j < base.size(); j++) {
- // TODO: optimize this to binary search.
- ManifestFileMeta file = base.get(j);
- if (predicate.test(
- file.numAddedFiles + file.numDeletedFiles,
- file.partitionStats.minValues(),
- file.partitionStats.maxValues(),
- file.partitionStats.nullCounts())) {
- break;
- } else {
- result.add(file);
- }
- }
- } else {
- // There is no DELETE Entry in Delta, Base don't need compaction
- j = base.size();
- result.addAll(base);
- }
- }
-
- // 2.2. try to skip base files by reading entries
-
- Set<Identifier> deleteEntries = new HashSet<>();
- deltaMerged.forEach(
- (k, v) -> {
- if (v.kind() == FileKind.DELETE) {
- deleteEntries.add(k);
- }
- });
-
- List<ManifestEntry> mergedEntries = new ArrayList<>();
- for (; j < base.size(); j++) {
- ManifestFileMeta file = base.get(j);
- boolean contains = false;
- for (ManifestEntry entry : manifestFile.read(file.fileName, file.fileSize)) {
- checkArgument(entry.kind() == FileKind.ADD);
- if (deleteEntries.contains(entry.identifier())) {
- contains = true;
- } else {
- mergedEntries.add(entry);
- }
- }
- if (contains) {
- // already read this file into fullMerged
- j++;
- break;
- } else {
- mergedEntries.clear();
- result.add(file);
- }
- }
-
- // 2.3. merge
-
- RollingFileWriter<ManifestEntry, ManifestFileMeta> writer =
- manifestFile.createRollingWriter();
- Exception exception = null;
- try {
-
- // 2.3.1 merge mergedEntries
- for (ManifestEntry entry : mergedEntries) {
- writer.write(entry);
- }
- mergedEntries.clear();
-
- // 2.3.2 merge base files
- for (ManifestEntry entry :
- FileEntry.readManifestEntries(
- manifestFile, base.subList(j, base.size()), manifestReadParallelism)) {
- checkArgument(entry.kind() == FileKind.ADD);
- if (!deleteEntries.contains(entry.identifier())) {
- writer.write(entry);
- }
- }
-
- // 2.3.3 merge deltaMerged
- for (ManifestEntry entry : deltaMerged.values()) {
- if (entry.kind() == FileKind.ADD) {
- writer.write(entry);
- }
- }
- } catch (Exception e) {
- exception = e;
- } finally {
- if (exception != null) {
- IOUtils.closeQuietly(writer);
- throw exception;
- }
- writer.close();
- }
-
- List<ManifestFileMeta> merged = writer.result();
- result.addAll(merged);
- newMetas.addAll(merged);
- return Optional.of(result);
- }
-
- private static Set<BinaryRow> computeDeletePartitions(
- Map<Identifier, ManifestEntry> deltaMerged) {
- Set<BinaryRow> partitions = new HashSet<>();
- for (ManifestEntry manifestEntry : deltaMerged.values()) {
- if (manifestEntry.kind() == FileKind.DELETE) {
- BinaryRow partition = manifestEntry.partition();
- partitions.add(partition);
- }
- }
- return partitions;
- }
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFileMetaSerializer.java b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFileMetaSerializer.java
index b2888031c..fe791e563 100644
--- a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFileMetaSerializer.java
+++ b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFileMetaSerializer.java
@@ -30,7 +30,7 @@ public class ManifestFileMetaSerializer extends VersionedObjectSerializer<Manife
private static final long serialVersionUID = 1L;
public ManifestFileMetaSerializer() {
- super(ManifestFileMeta.schema());
+ super(ManifestFileMeta.SCHEMA);
}
@Override
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 2980e998b..7768ddee5 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
@@ -140,7 +140,7 @@ public class ManifestList extends ObjectsFile<ManifestFileMeta> {
}
public ManifestList create() {
- RowType metaType = VersionedObjectSerializer.versionType(ManifestFileMeta.schema());
+ RowType metaType = VersionedObjectSerializer.versionType(ManifestFileMeta.SCHEMA);
return new ManifestList(
fileIO,
new ManifestFileMetaSerializer(),
diff --git a/paimon-core/src/main/java/org/apache/paimon/manifest/SimpleFileEntrySerializer.java b/paimon-core/src/main/java/org/apache/paimon/manifest/SimpleFileEntrySerializer.java
index c33abc00c..9c830816a 100644
--- a/paimon-core/src/main/java/org/apache/paimon/manifest/SimpleFileEntrySerializer.java
+++ b/paimon-core/src/main/java/org/apache/paimon/manifest/SimpleFileEntrySerializer.java
@@ -32,7 +32,7 @@ public class SimpleFileEntrySerializer extends VersionedObjectSerializer<SimpleF
private final int version;
public SimpleFileEntrySerializer() {
- super(ManifestEntry.schema());
+ super(ManifestEntry.SCHEMA);
this.version = new ManifestEntrySerializer().getVersion();
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreScan.java b/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreScan.java
index 4bf83e520..485e6a6a6 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreScan.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreScan.java
@@ -21,9 +21,11 @@ package org.apache.paimon.operation;
import org.apache.paimon.CoreOptions;
import org.apache.paimon.Snapshot;
import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.data.InternalRow;
import org.apache.paimon.manifest.FileEntry;
import org.apache.paimon.manifest.ManifestCacheFilter;
import org.apache.paimon.manifest.ManifestEntry;
+import org.apache.paimon.manifest.ManifestEntrySerializer;
import org.apache.paimon.manifest.ManifestFile;
import org.apache.paimon.manifest.ManifestFileMeta;
import org.apache.paimon.manifest.ManifestList;
@@ -481,8 +483,8 @@ public abstract class AbstractFileStoreScan implements FileStoreScan {
.read(
manifest.fileName(),
manifest.fileSize(),
- ManifestEntry.createCacheRowFilter(manifestCacheFilter, numOfBuckets),
- ManifestEntry.createEntryRowFilter(
+ createCacheRowFilter(manifestCacheFilter, numOfBuckets),
+ createEntryRowFilter(
partitionFilter, bucketFilter, fileNameFilter, numOfBuckets));
}
@@ -496,11 +498,72 @@ public abstract class AbstractFileStoreScan implements FileStoreScan {
// use filter for ManifestEntry
// currently, projection is not pushed down to file format
// see SimpleFileEntrySerializer
- ManifestEntry.createCacheRowFilter(manifestCacheFilter, numOfBuckets),
- ManifestEntry.createEntryRowFilter(
+ createCacheRowFilter(manifestCacheFilter, numOfBuckets),
+ createEntryRowFilter(
partitionFilter, bucketFilter, fileNameFilter, numOfBuckets));
}
+ /**
+ * According to the {@link ManifestCacheFilter}, entry that needs to be cached will be retained,
+ * so the entry that will not be accessed in the future will not be cached.
+ *
+ * <p>Implemented to {@link InternalRow} is for performance (No deserialization).
+ */
+ private static Filter<InternalRow> createCacheRowFilter(
+ @Nullable ManifestCacheFilter manifestCacheFilter, int numOfBuckets) {
+ if (manifestCacheFilter == null) {
+ return Filter.alwaysTrue();
+ }
+
+ Function<InternalRow, BinaryRow> partitionGetter =
+ ManifestEntrySerializer.partitionGetter();
+ Function<InternalRow, Integer> bucketGetter = ManifestEntrySerializer.bucketGetter();
+ Function<InternalRow, Integer> totalBucketGetter =
+ ManifestEntrySerializer.totalBucketGetter();
+ return row -> {
+ if (numOfBuckets != totalBucketGetter.apply(row)) {
+ return true;
+ }
+
+ return manifestCacheFilter.test(partitionGetter.apply(row), bucketGetter.apply(row));
+ };
+ }
+
+ /**
+ * Read the corresponding entries based on the current required partition and bucket.
+ *
+ * <p>Implemented to {@link InternalRow} is for performance (No deserialization).
+ */
+ private static Filter<InternalRow> createEntryRowFilter(
+ @Nullable PartitionPredicate partitionFilter,
+ @Nullable Filter<Integer> bucketFilter,
+ @Nullable Filter<String> fileNameFilter,
+ int numOfBuckets) {
+ Function<InternalRow, BinaryRow> partitionGetter =
+ ManifestEntrySerializer.partitionGetter();
+ Function<InternalRow, Integer> bucketGetter = ManifestEntrySerializer.bucketGetter();
+ Function<InternalRow, Integer> totalBucketGetter =
+ ManifestEntrySerializer.totalBucketGetter();
+ Function<InternalRow, String> fileNameGetter = ManifestEntrySerializer.fileNameGetter();
+ return row -> {
+ if ((partitionFilter != null && !partitionFilter.test(partitionGetter.apply(row)))) {
+ return false;
+ }
+
+ if (bucketFilter != null
+ && numOfBuckets == totalBucketGetter.apply(row)
+ && !bucketFilter.test(bucketGetter.apply(row))) {
+ return false;
+ }
+
+ if (fileNameFilter != null && !fileNameFilter.test((fileNameGetter.apply(row)))) {
+ return false;
+ }
+
+ return true;
+ };
+ }
+
// ------------------------------------------------------------------------
// End Thread Safe Methods
// ------------------------------------------------------------------------
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
index 21aeee38d..80ce33ce3 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
@@ -864,7 +864,7 @@ public class FileStoreCommitImpl implements FileStoreCommit {
}
// merge manifest files with changes
newMetas.addAll(
- ManifestFileMeta.merge(
+ ManifestFileMerger.merge(
oldMetas,
manifestFile,
manifestTargetSize.getBytes(),
diff --git a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFileMeta.java b/paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileMerger.java
similarity index 72%
copy from paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFileMeta.java
copy to paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileMerger.java
index 6196dab86..dfaa7af3f 100644
--- a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFileMeta.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileMerger.java
@@ -16,18 +16,17 @@
* limitations under the License.
*/
-package org.apache.paimon.manifest;
+package org.apache.paimon.operation;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.io.RollingFileWriter;
-import org.apache.paimon.manifest.FileEntry.Identifier;
+import org.apache.paimon.manifest.FileEntry;
+import org.apache.paimon.manifest.FileKind;
+import org.apache.paimon.manifest.ManifestEntry;
+import org.apache.paimon.manifest.ManifestFile;
+import org.apache.paimon.manifest.ManifestFileMeta;
import org.apache.paimon.partition.PartitionPredicate;
-import org.apache.paimon.stats.SimpleStats;
-import org.apache.paimon.stats.SimpleStatsConverter;
-import org.apache.paimon.types.BigIntType;
-import org.apache.paimon.types.DataField;
import org.apache.paimon.types.RowType;
-import org.apache.paimon.types.VarCharType;
import org.apache.paimon.utils.IOUtils;
import org.slf4j.Logger;
@@ -40,100 +39,15 @@ import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
-import java.util.Objects;
import java.util.Optional;
import java.util.Set;
import static org.apache.paimon.utils.Preconditions.checkArgument;
-/** Metadata of a manifest file. */
-public class ManifestFileMeta {
-
- private static final Logger LOG = LoggerFactory.getLogger(ManifestFileMeta.class);
-
- private final String fileName;
- private final long fileSize;
- private final long numAddedFiles;
- private final long numDeletedFiles;
- private final SimpleStats partitionStats;
- private final long schemaId;
-
- public ManifestFileMeta(
- String fileName,
- long fileSize,
- long numAddedFiles,
- long numDeletedFiles,
- SimpleStats partitionStats,
- long schemaId) {
- this.fileName = fileName;
- this.fileSize = fileSize;
- this.numAddedFiles = numAddedFiles;
- this.numDeletedFiles = numDeletedFiles;
- this.partitionStats = partitionStats;
- this.schemaId = schemaId;
- }
-
- public String fileName() {
- return fileName;
- }
-
- public long fileSize() {
- return fileSize;
- }
-
- public long numAddedFiles() {
- return numAddedFiles;
- }
-
- public long numDeletedFiles() {
- return numDeletedFiles;
- }
-
- public SimpleStats partitionStats() {
- return partitionStats;
- }
+/** Util for merging manifest files. */
+public class ManifestFileMerger {
- public long schemaId() {
- return schemaId;
- }
-
- public static RowType schema() {
- List<DataField> fields = new ArrayList<>();
- fields.add(new DataField(0, "_FILE_NAME", new VarCharType(false, Integer.MAX_VALUE)));
- fields.add(new DataField(1, "_FILE_SIZE", new BigIntType(false)));
- fields.add(new DataField(2, "_NUM_ADDED_FILES", new BigIntType(false)));
- fields.add(new DataField(3, "_NUM_DELETED_FILES", new BigIntType(false)));
- fields.add(new DataField(4, "_PARTITION_STATS", SimpleStatsConverter.schema()));
- fields.add(new DataField(5, "_SCHEMA_ID", new BigIntType(false)));
- return new RowType(false, fields);
- }
-
- @Override
- public boolean equals(Object o) {
- if (!(o instanceof ManifestFileMeta)) {
- return false;
- }
- ManifestFileMeta that = (ManifestFileMeta) o;
- return Objects.equals(fileName, that.fileName)
- && fileSize == that.fileSize
- && numAddedFiles == that.numAddedFiles
- && numDeletedFiles == that.numDeletedFiles
- && Objects.equals(partitionStats, that.partitionStats)
- && schemaId == that.schemaId;
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(
- fileName, fileSize, numAddedFiles, numDeletedFiles, partitionStats, schemaId);
- }
-
- @Override
- public String toString() {
- return String.format(
- "{%s, %d, %d, %d, %s, %d}",
- fileName, fileSize, numAddedFiles, numDeletedFiles, partitionStats, schemaId);
- }
+ private static final Logger LOG = LoggerFactory.getLogger(ManifestFileMerger.class);
/**
* Merge several {@link ManifestFileMeta}s. {@link ManifestEntry}s representing first adding and
@@ -174,7 +88,7 @@ public class ManifestFileMeta {
} catch (Throwable e) {
// exception occurs, clean up and rethrow
for (ManifestFileMeta manifest : newMetas) {
- manifestFile.delete(manifest.fileName);
+ manifestFile.delete(manifest.fileName());
}
throw new RuntimeException(e);
}
@@ -192,7 +106,7 @@ public class ManifestFileMeta {
long totalSize = 0;
// merge existing small manifest files
for (ManifestFileMeta manifest : input) {
- totalSize += manifest.fileSize;
+ totalSize += manifest.fileSize();
candidates.add(manifest);
if (totalSize >= suggestedMetaSize) {
// reach suggested file size, perform merging and produce new file
@@ -223,7 +137,7 @@ public class ManifestFileMeta {
return;
}
- Map<Identifier, ManifestEntry> map = new LinkedHashMap<>();
+ Map<FileEntry.Identifier, ManifestEntry> map = new LinkedHashMap<>();
FileEntry.mergeEntries(manifestFile, candidates, map, manifestReadParallelism);
if (!map.isEmpty()) {
List<ManifestFileMeta> merged = manifestFile.write(new ArrayList<>(map.values()));
@@ -248,9 +162,9 @@ public class ManifestFileMeta {
int i = 0;
for (; i < inputs.size(); i++) {
ManifestFileMeta file = inputs.get(i);
- if (file.numDeletedFiles == 0 && file.fileSize >= suggestedMetaSize) {
+ if (file.numDeletedFiles() == 0 && file.fileSize() >= suggestedMetaSize) {
base.add(file);
- totalManifestSize += file.fileSize;
+ totalManifestSize += file.fileSize();
} else {
break;
}
@@ -262,7 +176,7 @@ public class ManifestFileMeta {
for (; i < inputs.size(); i++) {
ManifestFileMeta file = inputs.get(i);
delta.add(file);
- totalManifestSize += file.fileSize;
+ totalManifestSize += file.fileSize();
deltaDeleteFileNum += file.numDeletedFiles();
totalDeltaFileSize += file.fileSize();
}
@@ -280,7 +194,7 @@ public class ManifestFileMeta {
// 2.1. try to skip base files by partition filter
- Map<Identifier, ManifestEntry> deltaMerged = new LinkedHashMap<>();
+ Map<FileEntry.Identifier, ManifestEntry> deltaMerged = new LinkedHashMap<>();
FileEntry.mergeEntries(manifestFile, delta, deltaMerged, manifestReadParallelism);
List<ManifestFileMeta> result = new ArrayList<>();
@@ -294,10 +208,10 @@ public class ManifestFileMeta {
// TODO: optimize this to binary search.
ManifestFileMeta file = base.get(j);
if (predicate.test(
- file.numAddedFiles + file.numDeletedFiles,
- file.partitionStats.minValues(),
- file.partitionStats.maxValues(),
- file.partitionStats.nullCounts())) {
+ file.numAddedFiles() + file.numDeletedFiles(),
+ file.partitionStats().minValues(),
+ file.partitionStats().maxValues(),
+ file.partitionStats().nullCounts())) {
break;
} else {
result.add(file);
@@ -312,7 +226,7 @@ public class ManifestFileMeta {
// 2.2. try to skip base files by reading entries
- Set<Identifier> deleteEntries = new HashSet<>();
+ Set<FileEntry.Identifier> deleteEntries = new HashSet<>();
deltaMerged.forEach(
(k, v) -> {
if (v.kind() == FileKind.DELETE) {
@@ -324,7 +238,7 @@ public class ManifestFileMeta {
for (; j < base.size(); j++) {
ManifestFileMeta file = base.get(j);
boolean contains = false;
- for (ManifestEntry entry : manifestFile.read(file.fileName, file.fileSize)) {
+ for (ManifestEntry entry : manifestFile.read(file.fileName(), file.fileSize())) {
checkArgument(entry.kind() == FileKind.ADD);
if (deleteEntries.contains(entry.identifier())) {
contains = true;
@@ -388,7 +302,7 @@ public class ManifestFileMeta {
}
private static Set<BinaryRow> computeDeletePartitions(
- Map<Identifier, ManifestEntry> deltaMerged) {
+ Map<FileEntry.Identifier, ManifestEntry> deltaMerged) {
Set<BinaryRow> partitions = new HashSet<>();
for (ManifestEntry manifestEntry : deltaMerged.values()) {
if (manifestEntry.kind() == FileKind.DELETE) {
diff --git a/paimon-core/src/main/java/org/apache/paimon/privilege/PrivilegedFileStoreTable.java b/paimon-core/src/main/java/org/apache/paimon/privilege/PrivilegedFileStoreTable.java
index 6842f0142..af48c1337 100644
--- a/paimon-core/src/main/java/org/apache/paimon/privilege/PrivilegedFileStoreTable.java
+++ b/paimon-core/src/main/java/org/apache/paimon/privilege/PrivilegedFileStoreTable.java
@@ -21,6 +21,8 @@ package org.apache.paimon.privilege;
import org.apache.paimon.FileStore;
import org.apache.paimon.catalog.Identifier;
import org.apache.paimon.manifest.ManifestCacheFilter;
+import org.apache.paimon.manifest.ManifestEntry;
+import org.apache.paimon.manifest.ManifestFileMeta;
import org.apache.paimon.schema.TableSchema;
import org.apache.paimon.stats.Statistics;
import org.apache.paimon.table.DelegatedFileStoreTable;
@@ -35,6 +37,7 @@ import org.apache.paimon.table.source.InnerTableRead;
import org.apache.paimon.table.source.StreamDataTableScan;
import org.apache.paimon.table.source.snapshot.SnapshotReader;
import org.apache.paimon.utils.BranchManager;
+import org.apache.paimon.utils.SimpleFileReader;
import org.apache.paimon.utils.TagManager;
import java.time.Duration;
@@ -91,6 +94,16 @@ public class PrivilegedFileStoreTable extends DelegatedFileStoreTable {
wrapped.copy(dynamicOptions), privilegeChecker, identifier);
}
+ @Override
+ public SimpleFileReader<ManifestFileMeta> manifestListReader() {
+ return wrapped.manifestListReader();
+ }
+
+ @Override
+ public SimpleFileReader<ManifestEntry> manifestFileReader() {
+ return wrapped.manifestFileReader();
+ }
+
@Override
public FileStoreTable copy(TableSchema newTableSchema) {
return new PrivilegedFileStoreTable(
diff --git a/paimon-core/src/main/java/org/apache/paimon/stats/SimpleStats.java b/paimon-core/src/main/java/org/apache/paimon/stats/SimpleStats.java
index cf00c71f4..62ad8577a 100644
--- a/paimon-core/src/main/java/org/apache/paimon/stats/SimpleStats.java
+++ b/paimon-core/src/main/java/org/apache/paimon/stats/SimpleStats.java
@@ -18,15 +18,22 @@
package org.apache.paimon.stats;
+import org.apache.paimon.annotation.Public;
import org.apache.paimon.data.BinaryArray;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.GenericRow;
import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.types.ArrayType;
+import org.apache.paimon.types.BigIntType;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.RowType;
+import java.util.Arrays;
import java.util.Objects;
import static org.apache.paimon.data.BinaryRow.EMPTY_ROW;
import static org.apache.paimon.utils.SerializationUtils.deserializeBinaryRow;
+import static org.apache.paimon.utils.SerializationUtils.newBytesType;
import static org.apache.paimon.utils.SerializationUtils.serializeBinaryRow;
/**
@@ -40,9 +47,20 @@ import static org.apache.paimon.utils.SerializationUtils.serializeBinaryRow;
*
* <p>All statistics are stored in the form of a Binary, which can significantly reduce its memory
* consumption, but the cost is that the column type needs to be known when getting.
+ *
+ * @since 0.9.0
*/
+@Public
public class SimpleStats {
+ public static final RowType SCHEMA =
+ new RowType(
+ false,
+ Arrays.asList(
+ new DataField(0, "_MIN_VALUES", newBytesType(false)),
+ new DataField(1, "_MAX_VALUES", newBytesType(false)),
+ new DataField(2, "_NULL_COUNTS", new ArrayType(new BigIntType(true)))));
+
/** Empty stats for 0 column number. */
public static final SimpleStats EMPTY_STATS =
new SimpleStats(EMPTY_ROW, EMPTY_ROW, BinaryArray.fromLongArray(new Long[0]));
diff --git a/paimon-core/src/main/java/org/apache/paimon/stats/SimpleStatsConverter.java b/paimon-core/src/main/java/org/apache/paimon/stats/SimpleStatsConverter.java
index 3351a8783..b76dc72b8 100644
--- a/paimon-core/src/main/java/org/apache/paimon/stats/SimpleStatsConverter.java
+++ b/paimon-core/src/main/java/org/apache/paimon/stats/SimpleStatsConverter.java
@@ -31,20 +31,12 @@ import org.apache.paimon.data.InternalRow;
import org.apache.paimon.data.Timestamp;
import org.apache.paimon.data.serializer.InternalRowSerializer;
import org.apache.paimon.format.SimpleColStats;
-import org.apache.paimon.types.ArrayType;
-import org.apache.paimon.types.BigIntType;
-import org.apache.paimon.types.DataField;
import org.apache.paimon.types.DataType;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.ProjectedRow;
import javax.annotation.Nullable;
-import java.util.ArrayList;
-import java.util.List;
-
-import static org.apache.paimon.utils.SerializationUtils.newBytesType;
-
/** Converter for array of {@link SimpleColStats}. */
public class SimpleStatsConverter {
@@ -108,14 +100,6 @@ public class SimpleStatsConverter {
return new NullCountsEvoArray(indexMapping, nullCounts, rowCount);
}
- public static RowType schema() {
- List<DataField> fields = new ArrayList<>();
- fields.add(new DataField(0, "_MIN_VALUES", newBytesType(false)));
- fields.add(new DataField(1, "_MAX_VALUES", newBytesType(false)));
- fields.add(new DataField(2, "_NULL_COUNTS", new ArrayType(new BigIntType(true))));
- return new RowType(fields);
- }
-
private static RowType toAllFieldsNullableRowType(RowType rowType) {
// as stated in RollingFile.Writer#finish, col stats are not collected currently so
// min/max values are all nulls
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java b/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java
index 3b61d3edb..a2609cee4 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java
@@ -25,6 +25,8 @@ import org.apache.paimon.consumer.ConsumerManager;
import org.apache.paimon.fs.FileIO;
import org.apache.paimon.fs.Path;
import org.apache.paimon.iceberg.IcebergCommitCallback;
+import org.apache.paimon.manifest.ManifestEntry;
+import org.apache.paimon.manifest.ManifestFileMeta;
import org.apache.paimon.metastore.AddPartitionCommitCallback;
import org.apache.paimon.metastore.AddPartitionTagCallback;
import org.apache.paimon.metastore.MetastoreClient;
@@ -60,6 +62,7 @@ import org.apache.paimon.tag.TagPreview;
import org.apache.paimon.utils.BranchManager;
import org.apache.paimon.utils.Preconditions;
import org.apache.paimon.utils.SegmentsCache;
+import org.apache.paimon.utils.SimpleFileReader;
import org.apache.paimon.utils.SnapshotManager;
import org.apache.paimon.utils.SnapshotNotExistException;
import org.apache.paimon.utils.TagManager;
@@ -126,6 +129,16 @@ abstract class AbstractFileStoreTable implements FileStoreTable {
return store().snapshotManager().snapshot(snapshotId);
}
+ @Override
+ public SimpleFileReader<ManifestFileMeta> manifestListReader() {
+ return store().manifestListFactory().create();
+ }
+
+ @Override
+ public SimpleFileReader<ManifestEntry> manifestFileReader() {
+ return store().manifestFileFactory().create();
+ }
+
@Override
public String name() {
return identifier().getObjectName();
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/FallbackReadFileStoreTable.java b/paimon-core/src/main/java/org/apache/paimon/table/FallbackReadFileStoreTable.java
index ead4452b0..4bc102058 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/FallbackReadFileStoreTable.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/FallbackReadFileStoreTable.java
@@ -23,6 +23,8 @@ import org.apache.paimon.Snapshot;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.disk.IOManager;
+import org.apache.paimon.manifest.ManifestEntry;
+import org.apache.paimon.manifest.ManifestFileMeta;
import org.apache.paimon.metrics.MetricRegistry;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.reader.RecordReader;
@@ -38,6 +40,7 @@ import org.apache.paimon.types.DataType;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.Filter;
import org.apache.paimon.utils.Preconditions;
+import org.apache.paimon.utils.SimpleFileReader;
import java.io.IOException;
import java.util.ArrayList;
@@ -72,6 +75,16 @@ public class FallbackReadFileStoreTable extends DelegatedFileStoreTable {
fallback.copy(rewriteFallbackOptions(dynamicOptions)));
}
+ @Override
+ public SimpleFileReader<ManifestFileMeta> manifestListReader() {
+ return wrapped.manifestListReader();
+ }
+
+ @Override
+ public SimpleFileReader<ManifestEntry> manifestFileReader() {
+ return wrapped.manifestFileReader();
+ }
+
@Override
public FileStoreTable copy(TableSchema newTableSchema) {
return new FallbackReadFileStoreTable(
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/ReadonlyTable.java b/paimon-core/src/main/java/org/apache/paimon/table/ReadonlyTable.java
index 6373e9944..027f26e44 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/ReadonlyTable.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/ReadonlyTable.java
@@ -19,6 +19,8 @@
package org.apache.paimon.table;
import org.apache.paimon.Snapshot;
+import org.apache.paimon.manifest.ManifestEntry;
+import org.apache.paimon.manifest.ManifestFileMeta;
import org.apache.paimon.stats.Statistics;
import org.apache.paimon.table.sink.BatchWriteBuilder;
import org.apache.paimon.table.sink.InnerTableCommit;
@@ -26,6 +28,7 @@ import org.apache.paimon.table.sink.InnerTableWrite;
import org.apache.paimon.table.sink.StreamWriteBuilder;
import org.apache.paimon.table.sink.WriteSelector;
import org.apache.paimon.table.source.StreamDataTableScan;
+import org.apache.paimon.utils.SimpleFileReader;
import java.time.Duration;
import java.util.Collections;
@@ -121,6 +124,22 @@ public interface ReadonlyTable extends InnerTable {
this.getClass().getSimpleName()));
}
+ @Override
+ default SimpleFileReader<ManifestFileMeta> manifestListReader() {
+ throw new UnsupportedOperationException(
+ String.format(
+ "Readonly Table %s does not support manifestListReader.",
+ this.getClass().getSimpleName()));
+ }
+
+ @Override
+ default SimpleFileReader<ManifestEntry> manifestFileReader() {
+ throw new UnsupportedOperationException(
+ String.format(
+ "Readonly Table %s does not support manifestFileReader.",
+ this.getClass().getSimpleName()));
+ }
+
@Override
default void rollbackTo(long snapshotId) {
throw new UnsupportedOperationException(
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/Table.java b/paimon-core/src/main/java/org/apache/paimon/table/Table.java
index c588a5f0a..ef2304040 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/Table.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/Table.java
@@ -21,11 +21,14 @@ package org.apache.paimon.table;
import org.apache.paimon.Snapshot;
import org.apache.paimon.annotation.Experimental;
import org.apache.paimon.annotation.Public;
+import org.apache.paimon.manifest.ManifestEntry;
+import org.apache.paimon.manifest.ManifestFileMeta;
import org.apache.paimon.stats.Statistics;
import org.apache.paimon.table.sink.BatchWriteBuilder;
import org.apache.paimon.table.sink.StreamWriteBuilder;
import org.apache.paimon.table.source.ReadBuilder;
import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.SimpleFileReader;
import java.io.Serializable;
import java.time.Duration;
@@ -83,6 +86,14 @@ public interface Table extends Serializable {
@Experimental
Snapshot snapshot(long snapshotId);
+ /** Reader to read manifest file meta from manifest list file. */
+ @Experimental
+ SimpleFileReader<ManifestFileMeta> manifestListReader();
+
+ /** Reader to read manifest entry from manifest file. */
+ @Experimental
+ SimpleFileReader<ManifestEntry> manifestFileReader();
+
/** Rollback table's state to a specific snapshot. */
@Experimental
void rollbackTo(long snapshotId);
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/sink/CommitMessageLegacyV2Serializer.java b/paimon-core/src/main/java/org/apache/paimon/table/sink/CommitMessageLegacyV2Serializer.java
index 1c6564b2e..989f32f50 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/sink/CommitMessageLegacyV2Serializer.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/sink/CommitMessageLegacyV2Serializer.java
@@ -28,7 +28,6 @@ import org.apache.paimon.io.DataIncrement;
import org.apache.paimon.io.DataInputView;
import org.apache.paimon.io.IndexIncrement;
import org.apache.paimon.stats.SimpleStats;
-import org.apache.paimon.stats.SimpleStatsConverter;
import org.apache.paimon.types.ArrayType;
import org.apache.paimon.types.BigIntType;
import org.apache.paimon.types.DataField;
@@ -90,8 +89,8 @@ public class CommitMessageLegacyV2Serializer {
fields.add(new DataField(2, "_ROW_COUNT", new BigIntType(false)));
fields.add(new DataField(3, "_MIN_KEY", newBytesType(false)));
fields.add(new DataField(4, "_MAX_KEY", newBytesType(false)));
- fields.add(new DataField(5, "_KEY_STATS", SimpleStatsConverter.schema()));
- fields.add(new DataField(6, "_VALUE_STATS", SimpleStatsConverter.schema()));
+ fields.add(new DataField(5, "_KEY_STATS", SimpleStats.SCHEMA));
+ fields.add(new DataField(6, "_VALUE_STATS", SimpleStats.SCHEMA));
fields.add(new DataField(7, "_MIN_SEQUENCE_NUMBER", new BigIntType(false)));
fields.add(new DataField(8, "_MAX_SEQUENCE_NUMBER", new BigIntType(false)));
fields.add(new DataField(9, "_SCHEMA_ID", new BigIntType(false)));
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/system/AuditLogTable.java b/paimon-core/src/main/java/org/apache/paimon/table/system/AuditLogTable.java
index 526a71d10..8a48353b2 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/system/AuditLogTable.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/system/AuditLogTable.java
@@ -27,6 +27,8 @@ import org.apache.paimon.data.InternalRow;
import org.apache.paimon.disk.IOManager;
import org.apache.paimon.fs.FileIO;
import org.apache.paimon.fs.Path;
+import org.apache.paimon.manifest.ManifestEntry;
+import org.apache.paimon.manifest.ManifestFileMeta;
import org.apache.paimon.manifest.PartitionEntry;
import org.apache.paimon.metrics.MetricRegistry;
import org.apache.paimon.predicate.LeafPredicate;
@@ -55,6 +57,7 @@ import org.apache.paimon.types.VarCharType;
import org.apache.paimon.utils.BranchManager;
import org.apache.paimon.utils.Filter;
import org.apache.paimon.utils.ProjectedRow;
+import org.apache.paimon.utils.SimpleFileReader;
import org.apache.paimon.utils.SnapshotManager;
import org.apache.paimon.utils.TagManager;
@@ -111,6 +114,16 @@ public class AuditLogTable implements DataTable, ReadonlyTable {
return wrapped.snapshot(snapshotId);
}
+ @Override
+ public SimpleFileReader<ManifestFileMeta> manifestListReader() {
+ return wrapped.manifestListReader();
+ }
+
+ @Override
+ public SimpleFileReader<ManifestEntry> manifestFileReader() {
+ return wrapped.manifestFileReader();
+ }
+
@Override
public String name() {
return wrapped.name() + SYSTEM_TABLE_SPLITTER + AUDIT_LOG;
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/system/BucketsTable.java b/paimon-core/src/main/java/org/apache/paimon/table/system/BucketsTable.java
index 282839e43..be3d210f2 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/system/BucketsTable.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/system/BucketsTable.java
@@ -28,6 +28,8 @@ import org.apache.paimon.fs.FileIO;
import org.apache.paimon.fs.Path;
import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.io.DataFileMetaSerializer;
+import org.apache.paimon.manifest.ManifestEntry;
+import org.apache.paimon.manifest.ManifestFileMeta;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.reader.RecordReader;
import org.apache.paimon.table.DataTable;
@@ -47,6 +49,7 @@ import org.apache.paimon.types.RowType;
import org.apache.paimon.types.VarCharType;
import org.apache.paimon.utils.BranchManager;
import org.apache.paimon.utils.IteratorRecordReader;
+import org.apache.paimon.utils.SimpleFileReader;
import org.apache.paimon.utils.SnapshotManager;
import org.apache.paimon.utils.TagManager;
@@ -115,6 +118,16 @@ public class BucketsTable implements DataTable, ReadonlyTable {
return wrapped.snapshot(snapshotId);
}
+ @Override
+ public SimpleFileReader<ManifestFileMeta> manifestListReader() {
+ return wrapped.manifestListReader();
+ }
+
+ @Override
+ public SimpleFileReader<ManifestEntry> manifestFileReader() {
+ return wrapped.manifestFileReader();
+ }
+
@Override
public Path location() {
return wrapped.location();
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/system/FileMonitorTable.java b/paimon-core/src/main/java/org/apache/paimon/table/system/FileMonitorTable.java
index cc3d7b5c3..45e8e7478 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/system/FileMonitorTable.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/system/FileMonitorTable.java
@@ -29,6 +29,8 @@ import org.apache.paimon.fs.FileIO;
import org.apache.paimon.fs.Path;
import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.io.DataFileMetaSerializer;
+import org.apache.paimon.manifest.ManifestEntry;
+import org.apache.paimon.manifest.ManifestFileMeta;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.reader.RecordReader;
import org.apache.paimon.table.DataTable;
@@ -47,6 +49,7 @@ import org.apache.paimon.types.IntType;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.BranchManager;
import org.apache.paimon.utils.IteratorRecordReader;
+import org.apache.paimon.utils.SimpleFileReader;
import org.apache.paimon.utils.SnapshotManager;
import org.apache.paimon.utils.TagManager;
@@ -102,6 +105,16 @@ public class FileMonitorTable implements DataTable, ReadonlyTable {
return wrapped.snapshot(snapshotId);
}
+ @Override
+ public SimpleFileReader<ManifestFileMeta> manifestListReader() {
+ return wrapped.manifestListReader();
+ }
+
+ @Override
+ public SimpleFileReader<ManifestEntry> manifestFileReader() {
+ return wrapped.manifestFileReader();
+ }
+
@Override
public Path location() {
return wrapped.location();
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/system/ReadOptimizedTable.java b/paimon-core/src/main/java/org/apache/paimon/table/system/ReadOptimizedTable.java
index 9cb9abe1a..cc7e7619c 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/system/ReadOptimizedTable.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/system/ReadOptimizedTable.java
@@ -22,6 +22,8 @@ import org.apache.paimon.CoreOptions;
import org.apache.paimon.Snapshot;
import org.apache.paimon.fs.FileIO;
import org.apache.paimon.fs.Path;
+import org.apache.paimon.manifest.ManifestEntry;
+import org.apache.paimon.manifest.ManifestFileMeta;
import org.apache.paimon.operation.DefaultValueAssigner;
import org.apache.paimon.table.DataTable;
import org.apache.paimon.table.FileStoreTable;
@@ -34,6 +36,7 @@ import org.apache.paimon.table.source.StreamDataTableScan;
import org.apache.paimon.table.source.snapshot.SnapshotReader;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.BranchManager;
+import org.apache.paimon.utils.SimpleFileReader;
import org.apache.paimon.utils.SnapshotManager;
import org.apache.paimon.utils.TagManager;
@@ -72,6 +75,16 @@ public class ReadOptimizedTable implements DataTable, ReadonlyTable {
return wrapped.snapshot(snapshotId);
}
+ @Override
+ public SimpleFileReader<ManifestFileMeta> manifestListReader() {
+ return wrapped.manifestListReader();
+ }
+
+ @Override
+ public SimpleFileReader<ManifestEntry> manifestFileReader() {
+ return wrapped.manifestFileReader();
+ }
+
@Override
public String name() {
return wrapped.name() + SYSTEM_TABLE_SPLITTER + READ_OPTIMIZED;
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 41f18f448..5150d2d65 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
@@ -39,7 +39,7 @@ import java.util.List;
import static org.apache.paimon.utils.FileUtils.checkExists;
/** A file which contains several {@link T}s, provides read and write. */
-public class ObjectsFile<T> {
+public class ObjectsFile<T> implements SimpleFileReader<T> {
protected final FileIO fileIO;
protected final ObjectSerializer<T> serializer;
@@ -88,6 +88,7 @@ public class ObjectsFile<T> {
}
}
+ @Override
public List<T> read(String fileName) {
return read(fileName, null);
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/manifest/FileKind.java b/paimon-core/src/main/java/org/apache/paimon/utils/SimpleFileReader.java
similarity index 56%
copy from paimon-core/src/main/java/org/apache/paimon/manifest/FileKind.java
copy to paimon-core/src/main/java/org/apache/paimon/utils/SimpleFileReader.java
index a18d65f15..0792bd16c 100644
--- a/paimon-core/src/main/java/org/apache/paimon/manifest/FileKind.java
+++ b/paimon-core/src/main/java/org/apache/paimon/utils/SimpleFileReader.java
@@ -16,33 +16,19 @@
* limitations under the License.
*/
-package org.apache.paimon.manifest;
+package org.apache.paimon.utils;
-/** Kind of a file. */
-public enum FileKind {
- ADD((byte) 0),
+import org.apache.paimon.annotation.Public;
- DELETE((byte) 1);
+import java.util.List;
- private final byte value;
-
- FileKind(byte value) {
- this.value = value;
- }
-
- public byte toByteValue() {
- return value;
- }
+/**
+ * A reader for reading file to entries.
+ *
+ * @since 0.9.0
+ */
+@Public
+public interface SimpleFileReader<T> {
- public static FileKind fromByteValue(byte value) {
- switch (value) {
- case 0:
- return ADD;
- case 1:
- return DELETE;
- default:
- throw new UnsupportedOperationException(
- "Unsupported byte value '" + value + "' for value kind.");
- }
- }
+ List<T> read(String fileName);
}
diff --git a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileMetaTest.java b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileMetaTest.java
index 06563a624..89ed09fb5 100644
--- a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileMetaTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileMetaTest.java
@@ -20,6 +20,7 @@ package org.apache.paimon.manifest;
import org.apache.paimon.fs.Path;
import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.operation.ManifestFileMerger;
import org.apache.paimon.types.IntType;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.FailingFileIO;
@@ -71,7 +72,7 @@ public class ManifestFileMetaTest extends ManifestFileMetaTestBase {
// no trigger Full Compaction
List<ManifestFileMeta> actual =
- ManifestFileMeta.merge(
+ ManifestFileMerger.merge(
input, manifestFile, 500, 3, Long.MAX_VALUE, getPartitionType(), null);
assertThat(actual).hasSameSizeAs(expected);
@@ -104,7 +105,7 @@ public class ManifestFileMetaTest extends ManifestFileMetaTestBase {
ManifestFile failingManifestFile =
createManifestFile(FailingFileIO.getFailingPath(failingName, tempDir.toString()));
try {
- ManifestFileMeta.merge(
+ ManifestFileMerger.merge(
input,
failingManifestFile,
500,
@@ -143,7 +144,8 @@ public class ManifestFileMetaTest extends ManifestFileMetaTestBase {
addDeltaManifests(input, true);
// trigger full compaction
List<ManifestFileMeta> merged =
- ManifestFileMeta.merge(input, manifestFile, 500, 3, 200, getPartitionType(), null);
+ ManifestFileMerger.merge(
+ input, manifestFile, 500, 3, 200, getPartitionType(), null);
// 1st Manifest don't need to Merge
assertSameContent(input.get(0), merged.get(0), manifestFile);
@@ -159,7 +161,8 @@ public class ManifestFileMetaTest extends ManifestFileMetaTestBase {
List<ManifestFileMeta> input = createBaseManifestFileMetas(true);
List<ManifestFileMeta> merged =
- ManifestFileMeta.merge(input, manifestFile, 500, 3, 200, getPartitionType(), null);
+ ManifestFileMerger.merge(
+ input, manifestFile, 500, 3, 200, getPartitionType(), null);
assertEquivalentEntries(input, merged);
assertThat(merged).hasSameElementsAs(input);
@@ -171,7 +174,8 @@ public class ManifestFileMetaTest extends ManifestFileMetaTestBase {
input1.add(delta);
List<ManifestFileMeta> merged1 =
- ManifestFileMeta.merge(input1, manifestFile, 500, 3, 200, getPartitionType(), null);
+ ManifestFileMerger.merge(
+ input1, manifestFile, 500, 3, 200, getPartitionType(), null);
assertThat(base).hasSameElementsAs(merged1);
assertEquivalentEntries(input1, merged1);
@@ -182,7 +186,8 @@ public class ManifestFileMetaTest extends ManifestFileMetaTestBase {
List<ManifestFileMeta> input = new ArrayList<>();
addDeltaManifests(input, true);
List<ManifestFileMeta> merged =
- ManifestFileMeta.merge(input, manifestFile, 500, 3, 200, getPartitionType(), null);
+ ManifestFileMerger.merge(
+ input, manifestFile, 500, 3, 200, getPartitionType(), null);
assertEquivalentEntries(input, merged);
}
@@ -208,7 +213,8 @@ public class ManifestFileMetaTest extends ManifestFileMetaTestBase {
input.add(makeManifest(makeEntry(true, "G")));
List<ManifestFileMeta> merged =
- ManifestFileMeta.merge(input, manifestFile, 500, 3, 200, getPartitionType(), null);
+ ManifestFileMerger.merge(
+ input, manifestFile, 500, 3, 200, getPartitionType(), null);
assertEquivalentEntries(input, merged);
}
@@ -239,7 +245,7 @@ public class ManifestFileMetaTest extends ManifestFileMetaTestBase {
// no trigger for delta size
List<ManifestFileMeta> newMetas2 = new ArrayList<>();
Optional<List<ManifestFileMeta>> fullCompacted =
- ManifestFileMeta.tryFullCompaction(
+ ManifestFileMerger.tryFullCompaction(
input,
newMetas2,
manifestFile,
@@ -253,7 +259,7 @@ public class ManifestFileMetaTest extends ManifestFileMetaTestBase {
// trigger full compaction
List<ManifestFileMeta> newMetas3 = new ArrayList<>();
List<ManifestFileMeta> merged =
- ManifestFileMeta.tryFullCompaction(
+ ManifestFileMerger.tryFullCompaction(
input, newMetas3, manifestFile, 500, 100, getPartitionType(), null)
.get();
@@ -278,7 +284,7 @@ public class ManifestFileMetaTest extends ManifestFileMetaTestBase {
List<ManifestFileMeta> newMetas = new ArrayList<>();
List<ManifestFileMeta> mergedManifest =
- ManifestFileMeta.tryFullCompaction(
+ ManifestFileMerger.tryFullCompaction(
input, newMetas, manifestFile, 500, 100, getPartitionType(), null)
.get();
diff --git a/paimon-core/src/test/java/org/apache/paimon/manifest/NoPartitionManifestFileMetaTest.java b/paimon-core/src/test/java/org/apache/paimon/manifest/NoPartitionManifestFileMetaTest.java
index 1fc27aa97..9978f9ee3 100644
--- a/paimon-core/src/test/java/org/apache/paimon/manifest/NoPartitionManifestFileMetaTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/manifest/NoPartitionManifestFileMetaTest.java
@@ -18,6 +18,7 @@
package org.apache.paimon.manifest;
+import org.apache.paimon.operation.ManifestFileMerger;
import org.apache.paimon.types.RowType;
import org.junit.jupiter.api.BeforeEach;
@@ -46,7 +47,8 @@ public class NoPartitionManifestFileMetaTest extends ManifestFileMetaTestBase {
addDeltaManifests(input, false);
List<ManifestFileMeta> merged =
- ManifestFileMeta.merge(input, manifestFile, 500, 3, 200, getPartitionType(), null);
+ ManifestFileMerger.merge(
+ input, manifestFile, 500, 3, 200, getPartitionType(), null);
assertEquivalentEntries(input, merged);
// the first one is not deleted, it should not be merged