gaborkaszab commented on code in PR #16936:
URL: https://github.com/apache/iceberg/pull/16936#discussion_r3613889001
##########
core/src/main/java/org/apache/iceberg/GenericManifestFile.java:
##########
@@ -60,6 +60,10 @@ public class GenericManifestFile extends
SupportsIndexProjection
private PartitionFieldSummary[] partitions = null;
private byte[] keyMetadata = null;
private Long firstRowId = null;
+ // v4+ root-manifest entry count; null for pre-v4 manifest list entries.
+ private Long recordCount = null;
+ // LEGACY_FORMAT_VERSION (0) for pre-v4 manifests; the table format version
for v4+.
Review Comment:
Since having similar comments for the new methods on the API, not sure if
these comments are needed.
##########
api/src/main/java/org/apache/iceberg/ManifestFile.java:
##########
@@ -210,6 +229,16 @@ default Long firstRowId() {
return null;
}
+ /** Returns the number of records in the manifest file, or null for pre-v4
manifests. */
Review Comment:
Maybe emphasize this is not the sum of the referenced data files' record
count, but the number of entries in the manifest file.
How about this?
`/** Returns the number of entries in the manifest file, or null for pre-v4
manifests. */`
##########
core/src/main/java/org/apache/iceberg/GenericManifestFile.java:
##########
@@ -112,6 +116,51 @@ public GenericManifestFile(Schema avroSchema) {
Integer deletedFilesCount,
Long deletedRowsCount,
Long firstRowId) {
+ this(
+ path,
+ length,
+ specId,
+ content,
+ sequenceNumber,
+ minSequenceNumber,
+ snapshotId,
+ partitions,
+ keyMetadata,
+ addedFilesCount,
+ addedRowsCount,
+ existingFilesCount,
+ existingRowsCount,
+ deletedFilesCount,
+ deletedRowsCount,
+ firstRowId,
+ null /* recordCount */,
Review Comment:
Instead of calling the other constructor with the default values spelled
explicitly here, can't we just leave this variant of the constructor intact?
With this implementation we spell the default values twice: Once when
introducing the members, once here when passing params to the other constructor.
##########
api/src/main/java/org/apache/iceberg/ManifestFile.java:
##########
@@ -29,6 +29,9 @@
public interface ManifestFile {
int PARTITION_SUMMARIES_ELEMENT_ID = 508;
+ /** Value of {@link #formatVersion()} for pre-v4 manifest files (v1, v2, and
v3). */
Review Comment:
Simply `Format version for pre-v4 manifest files.` ?
##########
core/src/main/java/org/apache/iceberg/GenericManifestFile.java:
##########
@@ -112,6 +116,51 @@ public GenericManifestFile(Schema avroSchema) {
Integer deletedFilesCount,
Long deletedRowsCount,
Long firstRowId) {
+ this(
+ path,
+ length,
+ specId,
+ content,
+ sequenceNumber,
+ minSequenceNumber,
+ snapshotId,
+ partitions,
+ keyMetadata,
+ addedFilesCount,
+ addedRowsCount,
+ existingFilesCount,
+ existingRowsCount,
+ deletedFilesCount,
+ deletedRowsCount,
+ firstRowId,
+ null /* recordCount */,
+ LEGACY_FORMAT_VERSION);
+ }
+
+ /**
Review Comment:
Too much information in the comment IMO. Isn't it enough to say that this
constructor is the one for V4+?
##########
core/src/main/java/org/apache/iceberg/GenericManifestFile.java:
##########
Review Comment:
With this PR this constructor is for pre-v4 versions. Maybe write a comment
here that calls out this?
##########
core/src/main/java/org/apache/iceberg/TrackedFileAdapters.java:
##########
@@ -405,6 +486,740 @@ public DeleteFile copyWithStats(Set<Integer>
requestedColumnIds) {
}
}
+ /** Shared base for content-file (DATA / EQUALITY_DELETES) write-direction
wrappers. */
+ abstract static class ContentTrackedFile<F extends ContentFile<F>>
+ implements TrackedFile, StructLike {
+ private final int formatVersion;
+ private final Types.StructType partitionType;
+ private final MapBackedContentStats statsWrapper;
+
+ private Tracking tracking;
+ private F file;
+ private StructProjection partition;
+ private ContentStats stats;
+
+ ContentTrackedFile(
+ int formatVersion,
+ Schema tableSchema,
+ MetricsConfig metricsConfig,
+ Types.StructType partitionType) {
+ Preconditions.checkArgument(
Review Comment:
Just for my understanding: this wrapper is not supposed to support wrapping
pre-v4 entries and then write them as v4 entries, right?
##########
core/src/main/java/org/apache/iceberg/GenericManifestFile.java:
##########
@@ -60,6 +60,10 @@ public class GenericManifestFile extends
SupportsIndexProjection
private PartitionFieldSummary[] partitions = null;
private byte[] keyMetadata = null;
private Long firstRowId = null;
+ // v4+ root-manifest entry count; null for pre-v4 manifest list entries.
Review Comment:
Could you explain this `v4+ root-manifest entry count`?
I figured this is for leaf manifests too.
##########
core/src/main/java/org/apache/iceberg/TrackedFileAdapters.java:
##########
@@ -405,6 +486,740 @@ public DeleteFile copyWithStats(Set<Integer>
requestedColumnIds) {
}
}
+ /** Shared base for content-file (DATA / EQUALITY_DELETES) write-direction
wrappers. */
+ abstract static class ContentTrackedFile<F extends ContentFile<F>>
+ implements TrackedFile, StructLike {
+ private final int formatVersion;
+ private final Types.StructType partitionType;
+ private final MapBackedContentStats statsWrapper;
+
+ private Tracking tracking;
+ private F file;
+ private StructProjection partition;
+ private ContentStats stats;
+
+ ContentTrackedFile(
+ int formatVersion,
+ Schema tableSchema,
+ MetricsConfig metricsConfig,
+ Types.StructType partitionType) {
+ Preconditions.checkArgument(
+ formatVersion >=
TableMetadata.MIN_FORMAT_VERSION_ADAPTIVE_MANIFEST_TREE,
+ "Invalid format version for adaptive manifest tree: %s (must be >=
%s)",
+ formatVersion,
+ TableMetadata.MIN_FORMAT_VERSION_ADAPTIVE_MANIFEST_TREE);
+ Preconditions.checkArgument(tableSchema != null, "Invalid table schema:
null");
+ Preconditions.checkArgument(metricsConfig != null, "Invalid metrics
config: null");
+ Preconditions.checkArgument(partitionType != null, "Invalid partition
type: null");
+ this.formatVersion = formatVersion;
+ this.partitionType = partitionType;
+ this.statsWrapper = new MapBackedContentStats(tableSchema,
metricsConfig);
+ }
+
+ void wrapWithTracking(F newFile, Tracking newTracking) {
+ Preconditions.checkArgument(newFile != null, "Invalid file: null");
+ Preconditions.checkArgument(newTracking != null, "Invalid tracking:
null");
+ validateContent(newFile);
+
+ this.file = newFile;
+ this.partition = projectPartition(newFile, partitionType);
+ this.stats = statsWrapper.wrap(newFile);
+ this.tracking = newTracking;
+ }
+
+ /** Content-type-specific validation of the wrapped file. */
+ abstract void validateContent(F newFile);
+
+ protected F file() {
+ return file;
+ }
+
+ @Override
+ public Tracking tracking() {
+ return tracking;
+ }
+
+ @Override
+ public int formatVersion() {
+ return formatVersion;
+ }
+
+ @Override
+ public String location() {
+ return file.location();
+ }
+
+ @Override
+ public FileFormat fileFormat() {
+ return file.format();
+ }
+
+ @Override
+ public long recordCount() {
+ return file.recordCount();
+ }
+
+ @Override
+ public long fileSizeInBytes() {
+ return file.fileSizeInBytes();
+ }
+
+ @Override
+ public Integer specId() {
+ return file.specId();
+ }
+
+ @Override
+ public StructLike partition() {
+ return partition;
+ }
+
+ @Override
+ public ContentStats contentStats() {
+ return stats;
+ }
+
+ @Override
+ public Integer sortOrderId() {
+ return file.sortOrderId();
+ }
+
+ @Override
+ public DeletionVector deletionVector() {
+ return null;
+ }
+
+ @Override
+ public ManifestInfo manifestInfo() {
+ return null;
+ }
+
+ @Override
+ public ByteBuffer keyMetadata() {
+ return file.keyMetadata();
+ }
+
+ @Override
+ public List<Long> splitOffsets() {
+ return file.splitOffsets();
+ }
+
+ @Override
+ public List<Integer> equalityIds() {
+ return null;
+ }
+
+ @Override
+ public TrackedFile copy() {
+ throw new UnsupportedOperationException(
+ "Reusable content-file wrapper does not support copy(); materialize
via a writer instead");
+ }
+
+ @Override
+ public TrackedFile copyWithStats(Set<Integer> requestedColumnIds) {
+ throw new UnsupportedOperationException(
+ "Reusable content-file wrapper does not support copyWithStats()");
+ }
+
+ @Override
+ public int size() {
+ return TRACKED_FILE_FIELD_COUNT;
+ }
+
+ @Override
+ public <T> T get(int pos, Class<T> javaClass) {
+ return javaClass.cast(getByPos(this, pos));
+ }
+
+ @Override
+ public <T> void set(int pos, T value) {
+ throw new UnsupportedOperationException(
+ "Reusable content-file wrapper does not support set()");
+ }
+ }
+
+ /** Wraps a {@link DataFile} as a {@link TrackedFile} row. */
+ static class DataTrackedFile extends ContentTrackedFile<DataFile> {
+ DataTrackedFile(
+ int formatVersion,
+ Schema tableSchema,
+ MetricsConfig metricsConfig,
+ Types.StructType partitionType) {
+ super(formatVersion, tableSchema, metricsConfig, partitionType);
+ }
+
+ /**
+ * Wraps a data file with caller-supplied tracking. The caller builds the
{@link Tracking} to
+ * encode the entry's status and sequence numbers.
+ */
+ public DataTrackedFile wrap(DataFile newFile, Tracking tracking) {
+ wrapWithTracking(newFile, tracking);
+ return this;
Review Comment:
I find it misleading that a `wrap()` function returns an object, but in fact
this is not a new one and in fact wrapping is an operation that replaces some
internal state in-place.
I'd also like to drill down into this "reusable wrapper per writer +
in-place adjustment for each file" approach. I see the other direction where we
adapt `TrackedFile` as `ContentFile` creates a new object per each
`TrackedFile` and I'm wondering about the motivation not to do the same on the
write path. Is the number of object this would generate an issue?
##########
core/src/main/java/org/apache/iceberg/TrackedFileAdapters.java:
##########
@@ -405,6 +486,740 @@ public DeleteFile copyWithStats(Set<Integer>
requestedColumnIds) {
}
}
+ /** Shared base for content-file (DATA / EQUALITY_DELETES) write-direction
wrappers. */
+ abstract static class ContentTrackedFile<F extends ContentFile<F>>
+ implements TrackedFile, StructLike {
+ private final int formatVersion;
+ private final Types.StructType partitionType;
+ private final MapBackedContentStats statsWrapper;
+
+ private Tracking tracking;
+ private F file;
+ private StructProjection partition;
+ private ContentStats stats;
+
+ ContentTrackedFile(
+ int formatVersion,
+ Schema tableSchema,
+ MetricsConfig metricsConfig,
+ Types.StructType partitionType) {
+ Preconditions.checkArgument(
+ formatVersion >=
TableMetadata.MIN_FORMAT_VERSION_ADAPTIVE_MANIFEST_TREE,
+ "Invalid format version for adaptive manifest tree: %s (must be >=
%s)",
+ formatVersion,
+ TableMetadata.MIN_FORMAT_VERSION_ADAPTIVE_MANIFEST_TREE);
+ Preconditions.checkArgument(tableSchema != null, "Invalid table schema:
null");
+ Preconditions.checkArgument(metricsConfig != null, "Invalid metrics
config: null");
+ Preconditions.checkArgument(partitionType != null, "Invalid partition
type: null");
+ this.formatVersion = formatVersion;
+ this.partitionType = partitionType;
+ this.statsWrapper = new MapBackedContentStats(tableSchema,
metricsConfig);
+ }
+
+ void wrapWithTracking(F newFile, Tracking newTracking) {
+ Preconditions.checkArgument(newFile != null, "Invalid file: null");
+ Preconditions.checkArgument(newTracking != null, "Invalid tracking:
null");
+ validateContent(newFile);
+
+ this.file = newFile;
+ this.partition = projectPartition(newFile, partitionType);
+ this.stats = statsWrapper.wrap(newFile);
+ this.tracking = newTracking;
+ }
+
+ /** Content-type-specific validation of the wrapped file. */
+ abstract void validateContent(F newFile);
+
+ protected F file() {
+ return file;
+ }
+
+ @Override
+ public Tracking tracking() {
+ return tracking;
+ }
+
+ @Override
+ public int formatVersion() {
+ return formatVersion;
+ }
+
+ @Override
+ public String location() {
+ return file.location();
+ }
+
+ @Override
+ public FileFormat fileFormat() {
+ return file.format();
+ }
+
+ @Override
+ public long recordCount() {
+ return file.recordCount();
+ }
+
+ @Override
+ public long fileSizeInBytes() {
+ return file.fileSizeInBytes();
+ }
+
+ @Override
+ public Integer specId() {
+ return file.specId();
+ }
+
+ @Override
+ public StructLike partition() {
+ return partition;
+ }
+
+ @Override
+ public ContentStats contentStats() {
+ return stats;
+ }
+
+ @Override
+ public Integer sortOrderId() {
+ return file.sortOrderId();
+ }
+
+ @Override
+ public DeletionVector deletionVector() {
+ return null;
+ }
+
+ @Override
+ public ManifestInfo manifestInfo() {
+ return null;
+ }
+
+ @Override
+ public ByteBuffer keyMetadata() {
+ return file.keyMetadata();
+ }
+
+ @Override
+ public List<Long> splitOffsets() {
+ return file.splitOffsets();
+ }
+
+ @Override
+ public List<Integer> equalityIds() {
+ return null;
+ }
+
+ @Override
+ public TrackedFile copy() {
+ throw new UnsupportedOperationException(
+ "Reusable content-file wrapper does not support copy(); materialize
via a writer instead");
+ }
+
+ @Override
+ public TrackedFile copyWithStats(Set<Integer> requestedColumnIds) {
+ throw new UnsupportedOperationException(
+ "Reusable content-file wrapper does not support copyWithStats()");
+ }
+
+ @Override
+ public int size() {
+ return TRACKED_FILE_FIELD_COUNT;
+ }
+
+ @Override
+ public <T> T get(int pos, Class<T> javaClass) {
+ return javaClass.cast(getByPos(this, pos));
+ }
+
+ @Override
+ public <T> void set(int pos, T value) {
+ throw new UnsupportedOperationException(
+ "Reusable content-file wrapper does not support set()");
+ }
+ }
+
+ /** Wraps a {@link DataFile} as a {@link TrackedFile} row. */
+ static class DataTrackedFile extends ContentTrackedFile<DataFile> {
+ DataTrackedFile(
+ int formatVersion,
+ Schema tableSchema,
+ MetricsConfig metricsConfig,
+ Types.StructType partitionType) {
+ super(formatVersion, tableSchema, metricsConfig, partitionType);
+ }
+
+ /**
+ * Wraps a data file with caller-supplied tracking. The caller builds the
{@link Tracking} to
+ * encode the entry's status and sequence numbers.
+ */
+ public DataTrackedFile wrap(DataFile newFile, Tracking tracking) {
+ wrapWithTracking(newFile, tracking);
+ return this;
+ }
+
+ @Override
+ void validateContent(DataFile newFile) {
+ Preconditions.checkArgument(
+ newFile.content() == FileContent.DATA,
+ "Invalid content for data file: %s",
+ newFile.content());
+ }
+
+ @Override
+ public FileContent contentType() {
+ return FileContent.DATA;
+ }
+ }
+
+ /** Wraps an equality {@link DeleteFile} as a {@link TrackedFile} row. */
+ static class EqualityDeleteTrackedFile extends
ContentTrackedFile<DeleteFile> {
+ EqualityDeleteTrackedFile(
+ int formatVersion,
+ Schema tableSchema,
+ MetricsConfig metricsConfig,
+ Types.StructType partitionType) {
+ super(formatVersion, tableSchema, metricsConfig, partitionType);
+ }
+
+ /**
+ * Wraps an equality delete file with caller-supplied tracking. The caller
builds the {@link
+ * Tracking} to encode the entry's status and sequence numbers.
+ */
+ public EqualityDeleteTrackedFile wrap(DeleteFile newFile, Tracking
tracking) {
+ wrapWithTracking(newFile, tracking);
+ return this;
+ }
+
+ @Override
+ void validateContent(DeleteFile newFile) {
+ // v4+ leaf delete manifests carry only equality deletes. DVs colocate
on the data file's
+ // TrackedFile row; v2 position delete files have no v4 leaf
representation.
+ Preconditions.checkArgument(
+ newFile.content() == FileContent.EQUALITY_DELETES,
+ "Invalid content for delete file: %s",
+ newFile.content());
+ }
+
+ @Override
+ public FileContent contentType() {
+ return FileContent.EQUALITY_DELETES;
+ }
+
+ @Override
+ public Integer sortOrderId() {
+ return null;
+ }
+
+ @Override
+ public List<Integer> equalityIds() {
+ return file().equalityFieldIds();
+ }
+ }
+
+ /** Wraps a {@link ManifestFile} as a v4+ leaf manifest row. */
+ static class ManifestTrackedFile implements TrackedFile, StructLike {
+ private Tracking tracking;
+ private final WrappedManifestInfo manifestInfo = new WrappedManifestInfo();
+ private ManifestFile manifest;
+ private long recordCount;
+ private FileContent contentType;
+
+ ManifestTrackedFile() {}
+
+ /**
+ * Wraps a manifest reference row.
+ *
+ * @param newManifest manifest file being referenced; must carry an
assigned {@code
+ * sequence_number} and {@code min_sequence_number}
+ * @param status entry status for the reference
+ * @param firstRowId first-row-id resolved by the caller for a DATA
manifest reference, or null
+ * for a DELETE manifest reference
+ */
+ public ManifestTrackedFile wrap(ManifestFile newManifest, EntryStatus
status, Long firstRowId) {
+ Preconditions.checkArgument(newManifest != null, "Invalid manifest file:
null");
+ Preconditions.checkArgument(status != null, "Invalid status: null");
+ int formatVersion = newManifest.formatVersion();
+ Preconditions.checkArgument(
Review Comment:
Hmm, opposed to ContentFiles, here we allow pre-v4 versions too. Is this
intentional?
##########
core/src/main/java/org/apache/iceberg/TrackedFileAdapters.java:
##########
@@ -18,15 +18,46 @@
*/
package org.apache.iceberg;
+import java.io.Serializable;
import java.nio.ByteBuffer;
import java.util.List;
import java.util.Map;
import java.util.Set;
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
-
-/** Adapts {@link TrackedFile} entries to the {@link DataFile} and {@link
DeleteFile} APIs. */
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.types.Conversions;
+import org.apache.iceberg.types.Type;
+import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.StructProjection;
+
+/**
+ * Adapts between the {@link TrackedFile} row and the {@link DataFile} /
{@link DeleteFile} / {@link
+ * ManifestFile} APIs in both directions.
+ *
+ * <p>Read direction: {@link #asDataFile}, {@link #asDVDeleteFile}, {@link
#asEqualityDeleteFile}
Review Comment:
I think the above sentence describes the content of this class pretty well:
it's a collection if adapters between pre-v4 and pos-v4 metadata structures.
Everything from here and below is too much explanation IMO, covering
expectations which adapter to be used in the read or write path and
expectations how the readers and writers should work. I don't think we need
those comments here.
And in general scrolling through this file quickly I have the impression
that there is too much comment compared to other implementation we have through
the project.
##########
core/src/main/java/org/apache/iceberg/TrackedFileAdapters.java:
##########
@@ -18,15 +18,46 @@
*/
package org.apache.iceberg;
+import java.io.Serializable;
import java.nio.ByteBuffer;
import java.util.List;
import java.util.Map;
import java.util.Set;
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
-
-/** Adapts {@link TrackedFile} entries to the {@link DataFile} and {@link
DeleteFile} APIs. */
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.types.Conversions;
+import org.apache.iceberg.types.Type;
+import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.StructProjection;
+
+/**
+ * Adapts between the {@link TrackedFile} row and the {@link DataFile} /
{@link DeleteFile} / {@link
+ * ManifestFile} APIs in both directions.
+ *
+ * <p>Read direction: {@link #asDataFile}, {@link #asDVDeleteFile}, {@link
#asEqualityDeleteFile}
+ * present a {@link TrackedFile} row read from a v4 manifest as the
corresponding legacy content
+ * file.
+ *
+ * <p>Write direction: {@link #forDataFile}, {@link #forEqualityDeleteFile},
{@link
+ * #forManifestReference} return reusable wrappers that present a source
object as a {@link
+ * TrackedFile}. Each writer holds a single instance and rebinds the wrapped
source via {@code
+ * wrap(...)} for every row, matching the {@link
V3Metadata.ManifestEntryWrapper} pattern used by
+ * {@code ManifestWriter.V3Writer} / {@code V4Writer}.
+ */
class TrackedFileAdapters {
+ // TrackedFile field count for the wrapper's StructLike view, derived from
the persisted schema so
Review Comment:
nit: I don't think we need this comment. The code and the name of the member
describes well what it is.
##########
core/src/main/java/org/apache/iceberg/TrackedFileAdapters.java:
##########
@@ -405,6 +486,740 @@ public DeleteFile copyWithStats(Set<Integer>
requestedColumnIds) {
}
}
+ /** Shared base for content-file (DATA / EQUALITY_DELETES) write-direction
wrappers. */
+ abstract static class ContentTrackedFile<F extends ContentFile<F>>
+ implements TrackedFile, StructLike {
+ private final int formatVersion;
+ private final Types.StructType partitionType;
+ private final MapBackedContentStats statsWrapper;
+
+ private Tracking tracking;
+ private F file;
+ private StructProjection partition;
+ private ContentStats stats;
+
+ ContentTrackedFile(
+ int formatVersion,
+ Schema tableSchema,
+ MetricsConfig metricsConfig,
+ Types.StructType partitionType) {
+ Preconditions.checkArgument(
+ formatVersion >=
TableMetadata.MIN_FORMAT_VERSION_ADAPTIVE_MANIFEST_TREE,
+ "Invalid format version for adaptive manifest tree: %s (must be >=
%s)",
+ formatVersion,
+ TableMetadata.MIN_FORMAT_VERSION_ADAPTIVE_MANIFEST_TREE);
+ Preconditions.checkArgument(tableSchema != null, "Invalid table schema:
null");
+ Preconditions.checkArgument(metricsConfig != null, "Invalid metrics
config: null");
+ Preconditions.checkArgument(partitionType != null, "Invalid partition
type: null");
+ this.formatVersion = formatVersion;
+ this.partitionType = partitionType;
+ this.statsWrapper = new MapBackedContentStats(tableSchema,
metricsConfig);
+ }
+
+ void wrapWithTracking(F newFile, Tracking newTracking) {
+ Preconditions.checkArgument(newFile != null, "Invalid file: null");
+ Preconditions.checkArgument(newTracking != null, "Invalid tracking:
null");
+ validateContent(newFile);
+
+ this.file = newFile;
+ this.partition = projectPartition(newFile, partitionType);
+ this.stats = statsWrapper.wrap(newFile);
+ this.tracking = newTracking;
+ }
+
+ /** Content-type-specific validation of the wrapped file. */
+ abstract void validateContent(F newFile);
+
+ protected F file() {
+ return file;
+ }
+
+ @Override
+ public Tracking tracking() {
+ return tracking;
+ }
+
+ @Override
+ public int formatVersion() {
+ return formatVersion;
+ }
+
+ @Override
+ public String location() {
+ return file.location();
+ }
+
+ @Override
+ public FileFormat fileFormat() {
+ return file.format();
+ }
+
+ @Override
+ public long recordCount() {
+ return file.recordCount();
+ }
+
+ @Override
+ public long fileSizeInBytes() {
+ return file.fileSizeInBytes();
+ }
+
+ @Override
+ public Integer specId() {
+ return file.specId();
+ }
+
+ @Override
+ public StructLike partition() {
+ return partition;
+ }
+
+ @Override
+ public ContentStats contentStats() {
+ return stats;
+ }
+
+ @Override
+ public Integer sortOrderId() {
+ return file.sortOrderId();
Review Comment:
"Can only be set if content_type is 0."
I see for EQ-deletes we override this to return null. Maybe reverse? Return
null here and override where it makes sense? We do that for `deletionVector`
for instance.
##########
core/src/main/java/org/apache/iceberg/TrackedFileAdapters.java:
##########
@@ -405,6 +486,740 @@ public DeleteFile copyWithStats(Set<Integer>
requestedColumnIds) {
}
}
+ /** Shared base for content-file (DATA / EQUALITY_DELETES) write-direction
wrappers. */
+ abstract static class ContentTrackedFile<F extends ContentFile<F>>
+ implements TrackedFile, StructLike {
+ private final int formatVersion;
+ private final Types.StructType partitionType;
+ private final MapBackedContentStats statsWrapper;
+
+ private Tracking tracking;
+ private F file;
+ private StructProjection partition;
+ private ContentStats stats;
+
+ ContentTrackedFile(
+ int formatVersion,
+ Schema tableSchema,
+ MetricsConfig metricsConfig,
+ Types.StructType partitionType) {
+ Preconditions.checkArgument(
+ formatVersion >=
TableMetadata.MIN_FORMAT_VERSION_ADAPTIVE_MANIFEST_TREE,
+ "Invalid format version for adaptive manifest tree: %s (must be >=
%s)",
+ formatVersion,
+ TableMetadata.MIN_FORMAT_VERSION_ADAPTIVE_MANIFEST_TREE);
+ Preconditions.checkArgument(tableSchema != null, "Invalid table schema:
null");
+ Preconditions.checkArgument(metricsConfig != null, "Invalid metrics
config: null");
+ Preconditions.checkArgument(partitionType != null, "Invalid partition
type: null");
+ this.formatVersion = formatVersion;
+ this.partitionType = partitionType;
+ this.statsWrapper = new MapBackedContentStats(tableSchema,
metricsConfig);
+ }
+
+ void wrapWithTracking(F newFile, Tracking newTracking) {
+ Preconditions.checkArgument(newFile != null, "Invalid file: null");
+ Preconditions.checkArgument(newTracking != null, "Invalid tracking:
null");
+ validateContent(newFile);
+
+ this.file = newFile;
+ this.partition = projectPartition(newFile, partitionType);
+ this.stats = statsWrapper.wrap(newFile);
+ this.tracking = newTracking;
+ }
+
+ /** Content-type-specific validation of the wrapped file. */
+ abstract void validateContent(F newFile);
+
+ protected F file() {
+ return file;
+ }
+
+ @Override
+ public Tracking tracking() {
+ return tracking;
+ }
+
+ @Override
+ public int formatVersion() {
+ return formatVersion;
+ }
+
+ @Override
+ public String location() {
+ return file.location();
+ }
+
+ @Override
+ public FileFormat fileFormat() {
+ return file.format();
+ }
+
+ @Override
+ public long recordCount() {
+ return file.recordCount();
+ }
+
+ @Override
+ public long fileSizeInBytes() {
+ return file.fileSizeInBytes();
+ }
+
+ @Override
+ public Integer specId() {
+ return file.specId();
+ }
+
+ @Override
+ public StructLike partition() {
+ return partition;
+ }
+
+ @Override
+ public ContentStats contentStats() {
+ return stats;
+ }
+
+ @Override
+ public Integer sortOrderId() {
+ return file.sortOrderId();
+ }
+
+ @Override
+ public DeletionVector deletionVector() {
+ return null;
+ }
+
+ @Override
+ public ManifestInfo manifestInfo() {
+ return null;
+ }
+
+ @Override
+ public ByteBuffer keyMetadata() {
+ return file.keyMetadata();
+ }
+
+ @Override
+ public List<Long> splitOffsets() {
+ return file.splitOffsets();
+ }
+
+ @Override
+ public List<Integer> equalityIds() {
+ return null;
+ }
+
+ @Override
+ public TrackedFile copy() {
+ throw new UnsupportedOperationException(
+ "Reusable content-file wrapper does not support copy(); materialize
via a writer instead");
+ }
+
+ @Override
+ public TrackedFile copyWithStats(Set<Integer> requestedColumnIds) {
+ throw new UnsupportedOperationException(
+ "Reusable content-file wrapper does not support copyWithStats()");
+ }
+
+ @Override
+ public int size() {
+ return TRACKED_FILE_FIELD_COUNT;
+ }
+
+ @Override
+ public <T> T get(int pos, Class<T> javaClass) {
+ return javaClass.cast(getByPos(this, pos));
+ }
+
+ @Override
+ public <T> void set(int pos, T value) {
+ throw new UnsupportedOperationException(
+ "Reusable content-file wrapper does not support set()");
+ }
+ }
+
+ /** Wraps a {@link DataFile} as a {@link TrackedFile} row. */
+ static class DataTrackedFile extends ContentTrackedFile<DataFile> {
+ DataTrackedFile(
+ int formatVersion,
+ Schema tableSchema,
+ MetricsConfig metricsConfig,
+ Types.StructType partitionType) {
+ super(formatVersion, tableSchema, metricsConfig, partitionType);
+ }
+
+ /**
+ * Wraps a data file with caller-supplied tracking. The caller builds the
{@link Tracking} to
+ * encode the entry's status and sequence numbers.
+ */
+ public DataTrackedFile wrap(DataFile newFile, Tracking tracking) {
+ wrapWithTracking(newFile, tracking);
+ return this;
+ }
+
+ @Override
+ void validateContent(DataFile newFile) {
+ Preconditions.checkArgument(
+ newFile.content() == FileContent.DATA,
+ "Invalid content for data file: %s",
+ newFile.content());
+ }
+
+ @Override
+ public FileContent contentType() {
+ return FileContent.DATA;
+ }
+ }
+
+ /** Wraps an equality {@link DeleteFile} as a {@link TrackedFile} row. */
+ static class EqualityDeleteTrackedFile extends
ContentTrackedFile<DeleteFile> {
+ EqualityDeleteTrackedFile(
+ int formatVersion,
+ Schema tableSchema,
+ MetricsConfig metricsConfig,
+ Types.StructType partitionType) {
+ super(formatVersion, tableSchema, metricsConfig, partitionType);
+ }
+
+ /**
+ * Wraps an equality delete file with caller-supplied tracking. The caller
builds the {@link
+ * Tracking} to encode the entry's status and sequence numbers.
+ */
+ public EqualityDeleteTrackedFile wrap(DeleteFile newFile, Tracking
tracking) {
+ wrapWithTracking(newFile, tracking);
+ return this;
+ }
+
+ @Override
+ void validateContent(DeleteFile newFile) {
+ // v4+ leaf delete manifests carry only equality deletes. DVs colocate
on the data file's
+ // TrackedFile row; v2 position delete files have no v4 leaf
representation.
+ Preconditions.checkArgument(
+ newFile.content() == FileContent.EQUALITY_DELETES,
+ "Invalid content for delete file: %s",
+ newFile.content());
+ }
+
+ @Override
+ public FileContent contentType() {
+ return FileContent.EQUALITY_DELETES;
+ }
+
+ @Override
+ public Integer sortOrderId() {
+ return null;
+ }
+
+ @Override
+ public List<Integer> equalityIds() {
+ return file().equalityFieldIds();
+ }
+ }
+
+ /** Wraps a {@link ManifestFile} as a v4+ leaf manifest row. */
+ static class ManifestTrackedFile implements TrackedFile, StructLike {
+ private Tracking tracking;
+ private final WrappedManifestInfo manifestInfo = new WrappedManifestInfo();
+ private ManifestFile manifest;
+ private long recordCount;
+ private FileContent contentType;
+
+ ManifestTrackedFile() {}
+
+ /**
+ * Wraps a manifest reference row.
+ *
+ * @param newManifest manifest file being referenced; must carry an
assigned {@code
+ * sequence_number} and {@code min_sequence_number}
+ * @param status entry status for the reference
+ * @param firstRowId first-row-id resolved by the caller for a DATA
manifest reference, or null
+ * for a DELETE manifest reference
+ */
+ public ManifestTrackedFile wrap(ManifestFile newManifest, EntryStatus
status, Long firstRowId) {
+ Preconditions.checkArgument(newManifest != null, "Invalid manifest file:
null");
+ Preconditions.checkArgument(status != null, "Invalid status: null");
+ int formatVersion = newManifest.formatVersion();
+ Preconditions.checkArgument(
+ formatVersion == ManifestFile.LEGACY_FORMAT_VERSION
+ || formatVersion >=
TableMetadata.MIN_FORMAT_VERSION_ADAPTIVE_MANIFEST_TREE,
Review Comment:
Do we guard against this being 1, 2 or 3 here? Would it make sense to
silently use the legacy value for them?
##########
core/src/main/java/org/apache/iceberg/TrackedFileAdapters.java:
##########
@@ -405,6 +486,740 @@ public DeleteFile copyWithStats(Set<Integer>
requestedColumnIds) {
}
}
+ /** Shared base for content-file (DATA / EQUALITY_DELETES) write-direction
wrappers. */
+ abstract static class ContentTrackedFile<F extends ContentFile<F>>
+ implements TrackedFile, StructLike {
+ private final int formatVersion;
+ private final Types.StructType partitionType;
+ private final MapBackedContentStats statsWrapper;
+
+ private Tracking tracking;
+ private F file;
+ private StructProjection partition;
+ private ContentStats stats;
+
+ ContentTrackedFile(
+ int formatVersion,
+ Schema tableSchema,
+ MetricsConfig metricsConfig,
+ Types.StructType partitionType) {
+ Preconditions.checkArgument(
+ formatVersion >=
TableMetadata.MIN_FORMAT_VERSION_ADAPTIVE_MANIFEST_TREE,
+ "Invalid format version for adaptive manifest tree: %s (must be >=
%s)",
+ formatVersion,
+ TableMetadata.MIN_FORMAT_VERSION_ADAPTIVE_MANIFEST_TREE);
+ Preconditions.checkArgument(tableSchema != null, "Invalid table schema:
null");
+ Preconditions.checkArgument(metricsConfig != null, "Invalid metrics
config: null");
+ Preconditions.checkArgument(partitionType != null, "Invalid partition
type: null");
+ this.formatVersion = formatVersion;
+ this.partitionType = partitionType;
+ this.statsWrapper = new MapBackedContentStats(tableSchema,
metricsConfig);
+ }
+
+ void wrapWithTracking(F newFile, Tracking newTracking) {
+ Preconditions.checkArgument(newFile != null, "Invalid file: null");
+ Preconditions.checkArgument(newTracking != null, "Invalid tracking:
null");
+ validateContent(newFile);
+
+ this.file = newFile;
+ this.partition = projectPartition(newFile, partitionType);
+ this.stats = statsWrapper.wrap(newFile);
+ this.tracking = newTracking;
+ }
+
+ /** Content-type-specific validation of the wrapped file. */
+ abstract void validateContent(F newFile);
+
+ protected F file() {
+ return file;
+ }
+
+ @Override
+ public Tracking tracking() {
+ return tracking;
+ }
+
+ @Override
+ public int formatVersion() {
+ return formatVersion;
+ }
+
+ @Override
+ public String location() {
+ return file.location();
+ }
+
+ @Override
+ public FileFormat fileFormat() {
+ return file.format();
+ }
+
+ @Override
+ public long recordCount() {
+ return file.recordCount();
+ }
+
+ @Override
+ public long fileSizeInBytes() {
+ return file.fileSizeInBytes();
+ }
+
+ @Override
+ public Integer specId() {
+ return file.specId();
+ }
+
+ @Override
+ public StructLike partition() {
+ return partition;
+ }
+
+ @Override
+ public ContentStats contentStats() {
+ return stats;
+ }
+
+ @Override
+ public Integer sortOrderId() {
+ return file.sortOrderId();
+ }
+
+ @Override
+ public DeletionVector deletionVector() {
+ return null;
+ }
+
+ @Override
+ public ManifestInfo manifestInfo() {
+ return null;
+ }
+
+ @Override
+ public ByteBuffer keyMetadata() {
+ return file.keyMetadata();
+ }
+
+ @Override
+ public List<Long> splitOffsets() {
+ return file.splitOffsets();
+ }
+
+ @Override
+ public List<Integer> equalityIds() {
+ return null;
+ }
+
+ @Override
+ public TrackedFile copy() {
+ throw new UnsupportedOperationException(
+ "Reusable content-file wrapper does not support copy(); materialize
via a writer instead");
+ }
+
+ @Override
+ public TrackedFile copyWithStats(Set<Integer> requestedColumnIds) {
+ throw new UnsupportedOperationException(
+ "Reusable content-file wrapper does not support copyWithStats()");
+ }
+
+ @Override
+ public int size() {
+ return TRACKED_FILE_FIELD_COUNT;
+ }
+
+ @Override
+ public <T> T get(int pos, Class<T> javaClass) {
+ return javaClass.cast(getByPos(this, pos));
+ }
+
+ @Override
+ public <T> void set(int pos, T value) {
+ throw new UnsupportedOperationException(
+ "Reusable content-file wrapper does not support set()");
+ }
+ }
+
+ /** Wraps a {@link DataFile} as a {@link TrackedFile} row. */
+ static class DataTrackedFile extends ContentTrackedFile<DataFile> {
+ DataTrackedFile(
+ int formatVersion,
+ Schema tableSchema,
+ MetricsConfig metricsConfig,
+ Types.StructType partitionType) {
+ super(formatVersion, tableSchema, metricsConfig, partitionType);
+ }
+
+ /**
+ * Wraps a data file with caller-supplied tracking. The caller builds the
{@link Tracking} to
+ * encode the entry's status and sequence numbers.
+ */
+ public DataTrackedFile wrap(DataFile newFile, Tracking tracking) {
+ wrapWithTracking(newFile, tracking);
+ return this;
+ }
+
+ @Override
+ void validateContent(DataFile newFile) {
+ Preconditions.checkArgument(
+ newFile.content() == FileContent.DATA,
+ "Invalid content for data file: %s",
+ newFile.content());
+ }
+
+ @Override
+ public FileContent contentType() {
+ return FileContent.DATA;
+ }
+ }
+
+ /** Wraps an equality {@link DeleteFile} as a {@link TrackedFile} row. */
+ static class EqualityDeleteTrackedFile extends
ContentTrackedFile<DeleteFile> {
+ EqualityDeleteTrackedFile(
+ int formatVersion,
+ Schema tableSchema,
+ MetricsConfig metricsConfig,
+ Types.StructType partitionType) {
+ super(formatVersion, tableSchema, metricsConfig, partitionType);
+ }
+
+ /**
+ * Wraps an equality delete file with caller-supplied tracking. The caller
builds the {@link
+ * Tracking} to encode the entry's status and sequence numbers.
+ */
+ public EqualityDeleteTrackedFile wrap(DeleteFile newFile, Tracking
tracking) {
+ wrapWithTracking(newFile, tracking);
+ return this;
+ }
+
+ @Override
+ void validateContent(DeleteFile newFile) {
+ // v4+ leaf delete manifests carry only equality deletes. DVs colocate
on the data file's
+ // TrackedFile row; v2 position delete files have no v4 leaf
representation.
+ Preconditions.checkArgument(
+ newFile.content() == FileContent.EQUALITY_DELETES,
+ "Invalid content for delete file: %s",
+ newFile.content());
+ }
+
+ @Override
+ public FileContent contentType() {
+ return FileContent.EQUALITY_DELETES;
+ }
+
+ @Override
+ public Integer sortOrderId() {
+ return null;
+ }
+
+ @Override
+ public List<Integer> equalityIds() {
+ return file().equalityFieldIds();
+ }
+ }
+
+ /** Wraps a {@link ManifestFile} as a v4+ leaf manifest row. */
+ static class ManifestTrackedFile implements TrackedFile, StructLike {
+ private Tracking tracking;
+ private final WrappedManifestInfo manifestInfo = new WrappedManifestInfo();
+ private ManifestFile manifest;
+ private long recordCount;
+ private FileContent contentType;
+
+ ManifestTrackedFile() {}
+
+ /**
+ * Wraps a manifest reference row.
+ *
+ * @param newManifest manifest file being referenced; must carry an
assigned {@code
+ * sequence_number} and {@code min_sequence_number}
+ * @param status entry status for the reference
+ * @param firstRowId first-row-id resolved by the caller for a DATA
manifest reference, or null
+ * for a DELETE manifest reference
+ */
+ public ManifestTrackedFile wrap(ManifestFile newManifest, EntryStatus
status, Long firstRowId) {
+ Preconditions.checkArgument(newManifest != null, "Invalid manifest file:
null");
+ Preconditions.checkArgument(status != null, "Invalid status: null");
+ int formatVersion = newManifest.formatVersion();
+ Preconditions.checkArgument(
+ formatVersion == ManifestFile.LEGACY_FORMAT_VERSION
+ || formatVersion >=
TableMetadata.MIN_FORMAT_VERSION_ADAPTIVE_MANIFEST_TREE,
+ "Invalid manifest format_version: %s (must be %s for pre-v4 or >= %s
for v4+)",
+ formatVersion,
+ ManifestFile.LEGACY_FORMAT_VERSION,
+ TableMetadata.MIN_FORMAT_VERSION_ADAPTIVE_MANIFEST_TREE);
+ Long manifestSnapshotId = newManifest.snapshotId();
+ Preconditions.checkArgument(manifestSnapshotId != null, "Invalid
manifest snapshot id: null");
+ long manifestSeq = newManifest.sequenceNumber();
+ Preconditions.checkArgument(
+ manifestSeq != ManifestWriter.UNASSIGNED_SEQ,
+ "Invalid manifest reference %s: sequence_number is unassigned",
+ newManifest.path());
+ Preconditions.checkArgument(
+ newManifest.minSequenceNumber() != ManifestWriter.UNASSIGNED_SEQ,
+ "Invalid manifest reference %s: min_sequence_number is unassigned",
+ newManifest.path());
+ Preconditions.checkArgument(
+ firstRowId == null || newManifest.content() == ManifestContent.DATA,
+ "firstRowId is only valid for DATA manifests, but content is %s",
+ newManifest.content());
+
+ this.manifest = newManifest;
+ this.contentType =
+ newManifest.content() == ManifestContent.DATA
+ ? FileContent.DATA_MANIFEST
+ : FileContent.DELETE_MANIFEST;
+ this.recordCount = resolveRecordCount(newManifest);
+ this.tracking =
Review Comment:
I thought that `Tracking` is an input for these adapters, even for manifests.
##########
core/src/main/java/org/apache/iceberg/TrackedFileAdapters.java:
##########
@@ -405,6 +486,740 @@ public DeleteFile copyWithStats(Set<Integer>
requestedColumnIds) {
}
}
+ /** Shared base for content-file (DATA / EQUALITY_DELETES) write-direction
wrappers. */
+ abstract static class ContentTrackedFile<F extends ContentFile<F>>
+ implements TrackedFile, StructLike {
+ private final int formatVersion;
+ private final Types.StructType partitionType;
+ private final MapBackedContentStats statsWrapper;
+
+ private Tracking tracking;
+ private F file;
+ private StructProjection partition;
+ private ContentStats stats;
+
+ ContentTrackedFile(
+ int formatVersion,
+ Schema tableSchema,
+ MetricsConfig metricsConfig,
+ Types.StructType partitionType) {
+ Preconditions.checkArgument(
+ formatVersion >=
TableMetadata.MIN_FORMAT_VERSION_ADAPTIVE_MANIFEST_TREE,
+ "Invalid format version for adaptive manifest tree: %s (must be >=
%s)",
+ formatVersion,
+ TableMetadata.MIN_FORMAT_VERSION_ADAPTIVE_MANIFEST_TREE);
+ Preconditions.checkArgument(tableSchema != null, "Invalid table schema:
null");
+ Preconditions.checkArgument(metricsConfig != null, "Invalid metrics
config: null");
+ Preconditions.checkArgument(partitionType != null, "Invalid partition
type: null");
+ this.formatVersion = formatVersion;
+ this.partitionType = partitionType;
+ this.statsWrapper = new MapBackedContentStats(tableSchema,
metricsConfig);
+ }
+
+ void wrapWithTracking(F newFile, Tracking newTracking) {
+ Preconditions.checkArgument(newFile != null, "Invalid file: null");
+ Preconditions.checkArgument(newTracking != null, "Invalid tracking:
null");
+ validateContent(newFile);
+
+ this.file = newFile;
+ this.partition = projectPartition(newFile, partitionType);
+ this.stats = statsWrapper.wrap(newFile);
+ this.tracking = newTracking;
+ }
+
+ /** Content-type-specific validation of the wrapped file. */
+ abstract void validateContent(F newFile);
+
+ protected F file() {
+ return file;
+ }
+
+ @Override
+ public Tracking tracking() {
+ return tracking;
+ }
+
+ @Override
+ public int formatVersion() {
+ return formatVersion;
+ }
+
+ @Override
+ public String location() {
+ return file.location();
+ }
+
+ @Override
+ public FileFormat fileFormat() {
+ return file.format();
+ }
+
+ @Override
+ public long recordCount() {
+ return file.recordCount();
+ }
+
+ @Override
+ public long fileSizeInBytes() {
+ return file.fileSizeInBytes();
+ }
+
+ @Override
+ public Integer specId() {
+ return file.specId();
+ }
+
+ @Override
+ public StructLike partition() {
+ return partition;
+ }
+
+ @Override
+ public ContentStats contentStats() {
+ return stats;
+ }
+
+ @Override
+ public Integer sortOrderId() {
+ return file.sortOrderId();
+ }
+
+ @Override
+ public DeletionVector deletionVector() {
+ return null;
+ }
+
+ @Override
+ public ManifestInfo manifestInfo() {
+ return null;
+ }
+
+ @Override
+ public ByteBuffer keyMetadata() {
+ return file.keyMetadata();
+ }
+
+ @Override
+ public List<Long> splitOffsets() {
+ return file.splitOffsets();
+ }
+
+ @Override
+ public List<Integer> equalityIds() {
+ return null;
+ }
+
+ @Override
+ public TrackedFile copy() {
+ throw new UnsupportedOperationException(
+ "Reusable content-file wrapper does not support copy(); materialize
via a writer instead");
+ }
+
+ @Override
+ public TrackedFile copyWithStats(Set<Integer> requestedColumnIds) {
+ throw new UnsupportedOperationException(
+ "Reusable content-file wrapper does not support copyWithStats()");
+ }
+
+ @Override
+ public int size() {
+ return TRACKED_FILE_FIELD_COUNT;
+ }
+
+ @Override
+ public <T> T get(int pos, Class<T> javaClass) {
+ return javaClass.cast(getByPos(this, pos));
+ }
+
+ @Override
+ public <T> void set(int pos, T value) {
+ throw new UnsupportedOperationException(
+ "Reusable content-file wrapper does not support set()");
+ }
+ }
+
+ /** Wraps a {@link DataFile} as a {@link TrackedFile} row. */
+ static class DataTrackedFile extends ContentTrackedFile<DataFile> {
+ DataTrackedFile(
+ int formatVersion,
+ Schema tableSchema,
+ MetricsConfig metricsConfig,
+ Types.StructType partitionType) {
+ super(formatVersion, tableSchema, metricsConfig, partitionType);
+ }
+
+ /**
+ * Wraps a data file with caller-supplied tracking. The caller builds the
{@link Tracking} to
+ * encode the entry's status and sequence numbers.
+ */
+ public DataTrackedFile wrap(DataFile newFile, Tracking tracking) {
+ wrapWithTracking(newFile, tracking);
+ return this;
+ }
+
+ @Override
+ void validateContent(DataFile newFile) {
+ Preconditions.checkArgument(
+ newFile.content() == FileContent.DATA,
+ "Invalid content for data file: %s",
+ newFile.content());
+ }
+
+ @Override
+ public FileContent contentType() {
+ return FileContent.DATA;
+ }
+ }
+
+ /** Wraps an equality {@link DeleteFile} as a {@link TrackedFile} row. */
+ static class EqualityDeleteTrackedFile extends
ContentTrackedFile<DeleteFile> {
+ EqualityDeleteTrackedFile(
+ int formatVersion,
+ Schema tableSchema,
+ MetricsConfig metricsConfig,
+ Types.StructType partitionType) {
+ super(formatVersion, tableSchema, metricsConfig, partitionType);
+ }
+
+ /**
+ * Wraps an equality delete file with caller-supplied tracking. The caller
builds the {@link
+ * Tracking} to encode the entry's status and sequence numbers.
+ */
+ public EqualityDeleteTrackedFile wrap(DeleteFile newFile, Tracking
tracking) {
+ wrapWithTracking(newFile, tracking);
+ return this;
+ }
+
+ @Override
+ void validateContent(DeleteFile newFile) {
+ // v4+ leaf delete manifests carry only equality deletes. DVs colocate
on the data file's
Review Comment:
I don't think we need to explain why we require the content to be eq-delete
in a wrapper class for eq-deletes
##########
core/src/main/java/org/apache/iceberg/TrackedFileAdapters.java:
##########
@@ -405,6 +486,740 @@ public DeleteFile copyWithStats(Set<Integer>
requestedColumnIds) {
}
}
+ /** Shared base for content-file (DATA / EQUALITY_DELETES) write-direction
wrappers. */
+ abstract static class ContentTrackedFile<F extends ContentFile<F>>
+ implements TrackedFile, StructLike {
+ private final int formatVersion;
+ private final Types.StructType partitionType;
+ private final MapBackedContentStats statsWrapper;
+
+ private Tracking tracking;
+ private F file;
+ private StructProjection partition;
+ private ContentStats stats;
+
+ ContentTrackedFile(
+ int formatVersion,
+ Schema tableSchema,
+ MetricsConfig metricsConfig,
+ Types.StructType partitionType) {
+ Preconditions.checkArgument(
+ formatVersion >=
TableMetadata.MIN_FORMAT_VERSION_ADAPTIVE_MANIFEST_TREE,
+ "Invalid format version for adaptive manifest tree: %s (must be >=
%s)",
+ formatVersion,
+ TableMetadata.MIN_FORMAT_VERSION_ADAPTIVE_MANIFEST_TREE);
+ Preconditions.checkArgument(tableSchema != null, "Invalid table schema:
null");
+ Preconditions.checkArgument(metricsConfig != null, "Invalid metrics
config: null");
+ Preconditions.checkArgument(partitionType != null, "Invalid partition
type: null");
+ this.formatVersion = formatVersion;
+ this.partitionType = partitionType;
+ this.statsWrapper = new MapBackedContentStats(tableSchema,
metricsConfig);
+ }
+
+ void wrapWithTracking(F newFile, Tracking newTracking) {
+ Preconditions.checkArgument(newFile != null, "Invalid file: null");
+ Preconditions.checkArgument(newTracking != null, "Invalid tracking:
null");
+ validateContent(newFile);
+
+ this.file = newFile;
+ this.partition = projectPartition(newFile, partitionType);
+ this.stats = statsWrapper.wrap(newFile);
+ this.tracking = newTracking;
+ }
+
+ /** Content-type-specific validation of the wrapped file. */
+ abstract void validateContent(F newFile);
+
+ protected F file() {
+ return file;
+ }
+
+ @Override
+ public Tracking tracking() {
+ return tracking;
+ }
+
+ @Override
+ public int formatVersion() {
+ return formatVersion;
+ }
+
+ @Override
+ public String location() {
+ return file.location();
+ }
+
+ @Override
+ public FileFormat fileFormat() {
+ return file.format();
+ }
+
+ @Override
+ public long recordCount() {
+ return file.recordCount();
+ }
+
+ @Override
+ public long fileSizeInBytes() {
+ return file.fileSizeInBytes();
+ }
+
+ @Override
+ public Integer specId() {
+ return file.specId();
+ }
+
+ @Override
+ public StructLike partition() {
+ return partition;
+ }
+
+ @Override
+ public ContentStats contentStats() {
+ return stats;
+ }
+
+ @Override
+ public Integer sortOrderId() {
+ return file.sortOrderId();
+ }
+
+ @Override
+ public DeletionVector deletionVector() {
+ return null;
+ }
+
+ @Override
+ public ManifestInfo manifestInfo() {
+ return null;
+ }
+
+ @Override
+ public ByteBuffer keyMetadata() {
+ return file.keyMetadata();
+ }
+
+ @Override
+ public List<Long> splitOffsets() {
+ return file.splitOffsets();
+ }
+
+ @Override
+ public List<Integer> equalityIds() {
+ return null;
+ }
+
+ @Override
+ public TrackedFile copy() {
+ throw new UnsupportedOperationException(
+ "Reusable content-file wrapper does not support copy(); materialize
via a writer instead");
+ }
+
+ @Override
+ public TrackedFile copyWithStats(Set<Integer> requestedColumnIds) {
+ throw new UnsupportedOperationException(
+ "Reusable content-file wrapper does not support copyWithStats()");
+ }
+
+ @Override
+ public int size() {
+ return TRACKED_FILE_FIELD_COUNT;
+ }
+
+ @Override
+ public <T> T get(int pos, Class<T> javaClass) {
+ return javaClass.cast(getByPos(this, pos));
+ }
+
+ @Override
+ public <T> void set(int pos, T value) {
+ throw new UnsupportedOperationException(
+ "Reusable content-file wrapper does not support set()");
+ }
+ }
+
+ /** Wraps a {@link DataFile} as a {@link TrackedFile} row. */
+ static class DataTrackedFile extends ContentTrackedFile<DataFile> {
+ DataTrackedFile(
+ int formatVersion,
+ Schema tableSchema,
+ MetricsConfig metricsConfig,
+ Types.StructType partitionType) {
+ super(formatVersion, tableSchema, metricsConfig, partitionType);
+ }
+
+ /**
+ * Wraps a data file with caller-supplied tracking. The caller builds the
{@link Tracking} to
+ * encode the entry's status and sequence numbers.
+ */
+ public DataTrackedFile wrap(DataFile newFile, Tracking tracking) {
+ wrapWithTracking(newFile, tracking);
+ return this;
+ }
+
+ @Override
+ void validateContent(DataFile newFile) {
+ Preconditions.checkArgument(
+ newFile.content() == FileContent.DATA,
+ "Invalid content for data file: %s",
+ newFile.content());
+ }
+
+ @Override
+ public FileContent contentType() {
+ return FileContent.DATA;
+ }
+ }
+
+ /** Wraps an equality {@link DeleteFile} as a {@link TrackedFile} row. */
+ static class EqualityDeleteTrackedFile extends
ContentTrackedFile<DeleteFile> {
+ EqualityDeleteTrackedFile(
+ int formatVersion,
+ Schema tableSchema,
+ MetricsConfig metricsConfig,
+ Types.StructType partitionType) {
+ super(formatVersion, tableSchema, metricsConfig, partitionType);
+ }
+
+ /**
+ * Wraps an equality delete file with caller-supplied tracking. The caller
builds the {@link
+ * Tracking} to encode the entry's status and sequence numbers.
+ */
+ public EqualityDeleteTrackedFile wrap(DeleteFile newFile, Tracking
tracking) {
+ wrapWithTracking(newFile, tracking);
+ return this;
+ }
+
+ @Override
+ void validateContent(DeleteFile newFile) {
+ // v4+ leaf delete manifests carry only equality deletes. DVs colocate
on the data file's
+ // TrackedFile row; v2 position delete files have no v4 leaf
representation.
+ Preconditions.checkArgument(
+ newFile.content() == FileContent.EQUALITY_DELETES,
+ "Invalid content for delete file: %s",
+ newFile.content());
+ }
+
+ @Override
+ public FileContent contentType() {
+ return FileContent.EQUALITY_DELETES;
+ }
+
+ @Override
+ public Integer sortOrderId() {
+ return null;
+ }
+
+ @Override
+ public List<Integer> equalityIds() {
+ return file().equalityFieldIds();
+ }
+ }
+
+ /** Wraps a {@link ManifestFile} as a v4+ leaf manifest row. */
+ static class ManifestTrackedFile implements TrackedFile, StructLike {
+ private Tracking tracking;
+ private final WrappedManifestInfo manifestInfo = new WrappedManifestInfo();
+ private ManifestFile manifest;
+ private long recordCount;
+ private FileContent contentType;
+
+ ManifestTrackedFile() {}
+
+ /**
+ * Wraps a manifest reference row.
+ *
+ * @param newManifest manifest file being referenced; must carry an
assigned {@code
+ * sequence_number} and {@code min_sequence_number}
+ * @param status entry status for the reference
+ * @param firstRowId first-row-id resolved by the caller for a DATA
manifest reference, or null
+ * for a DELETE manifest reference
+ */
+ public ManifestTrackedFile wrap(ManifestFile newManifest, EntryStatus
status, Long firstRowId) {
+ Preconditions.checkArgument(newManifest != null, "Invalid manifest file:
null");
+ Preconditions.checkArgument(status != null, "Invalid status: null");
+ int formatVersion = newManifest.formatVersion();
+ Preconditions.checkArgument(
+ formatVersion == ManifestFile.LEGACY_FORMAT_VERSION
+ || formatVersion >=
TableMetadata.MIN_FORMAT_VERSION_ADAPTIVE_MANIFEST_TREE,
+ "Invalid manifest format_version: %s (must be %s for pre-v4 or >= %s
for v4+)",
+ formatVersion,
+ ManifestFile.LEGACY_FORMAT_VERSION,
+ TableMetadata.MIN_FORMAT_VERSION_ADAPTIVE_MANIFEST_TREE);
+ Long manifestSnapshotId = newManifest.snapshotId();
+ Preconditions.checkArgument(manifestSnapshotId != null, "Invalid
manifest snapshot id: null");
+ long manifestSeq = newManifest.sequenceNumber();
+ Preconditions.checkArgument(
+ manifestSeq != ManifestWriter.UNASSIGNED_SEQ,
+ "Invalid manifest reference %s: sequence_number is unassigned",
+ newManifest.path());
+ Preconditions.checkArgument(
+ newManifest.minSequenceNumber() != ManifestWriter.UNASSIGNED_SEQ,
+ "Invalid manifest reference %s: min_sequence_number is unassigned",
+ newManifest.path());
+ Preconditions.checkArgument(
+ firstRowId == null || newManifest.content() == ManifestContent.DATA,
+ "firstRowId is only valid for DATA manifests, but content is %s",
+ newManifest.content());
+
+ this.manifest = newManifest;
+ this.contentType =
+ newManifest.content() == ManifestContent.DATA
+ ? FileContent.DATA_MANIFEST
+ : FileContent.DELETE_MANIFEST;
+ this.recordCount = resolveRecordCount(newManifest);
+ this.tracking =
+ new TrackingStruct(
+ status, manifestSnapshotId, manifestSeq, manifestSeq, null,
firstRowId, null, null);
+ this.manifestInfo.wrap(newManifest);
+ return this;
+ }
+
+ @Override
+ public Tracking tracking() {
+ return tracking;
+ }
+
+ @Override
+ public FileContent contentType() {
+ return contentType;
+ }
+
+ @Override
+ public int formatVersion() {
+ return manifest.formatVersion();
+ }
+
+ @Override
+ public String location() {
+ return manifest.path();
+ }
+
+ @Override
+ public FileFormat fileFormat() {
+ return FileFormat.fromFileName(manifest.path());
+ }
+
+ @Override
+ public long recordCount() {
+ return recordCount;
+ }
+
+ @Override
+ public long fileSizeInBytes() {
+ return manifest.length();
+ }
+
+ @Override
+ public Integer specId() {
+ return manifest.partitionSpecId();
+ }
+
+ @Override
+ public StructLike partition() {
+ return null;
+ }
+
+ @Override
+ public ContentStats contentStats() {
+ return null;
+ }
+
+ @Override
+ public Integer sortOrderId() {
+ return null;
+ }
+
+ @Override
+ public DeletionVector deletionVector() {
+ return null;
+ }
+
+ @Override
+ public ManifestInfo manifestInfo() {
+ return manifestInfo;
+ }
+
+ @Override
+ public ByteBuffer keyMetadata() {
+ return manifest.keyMetadata();
+ }
+
+ @Override
+ public List<Long> splitOffsets() {
+ return null;
+ }
+
+ @Override
+ public List<Integer> equalityIds() {
+ return null;
+ }
+
+ @Override
+ public TrackedFile copy() {
+ throw new UnsupportedOperationException(
+ "Reusable manifest-reference wrapper does not support copy()");
+ }
+
+ @Override
+ public TrackedFile copyWithStats(Set<Integer> requestedColumnIds) {
+ throw new UnsupportedOperationException(
+ "Reusable manifest-reference wrapper does not support
copyWithStats()");
+ }
+
+ @Override
+ public int size() {
+ return TRACKED_FILE_FIELD_COUNT;
+ }
+
+ @Override
+ public <T> T get(int pos, Class<T> javaClass) {
+ return javaClass.cast(getByPos(this, pos));
+ }
+
+ @Override
+ public <T> void set(int pos, T value) {
+ throw new UnsupportedOperationException(
+ "Reusable manifest-reference wrapper does not support set()");
+ }
+ }
+
+ /** Reusable {@link ManifestInfo} view over a {@link ManifestFile}'s counts.
*/
+ private static class WrappedManifestInfo implements ManifestInfo, StructLike
{
+ private ManifestFile manifest;
+
+ void wrap(ManifestFile newManifest) {
+ this.manifest = newManifest;
+ }
+
+ @Override
+ public int addedFilesCount() {
+ return zeroIfNull(manifest.addedFilesCount());
+ }
+
+ @Override
+ public int existingFilesCount() {
+ return zeroIfNull(manifest.existingFilesCount());
+ }
+
+ @Override
+ public int deletedFilesCount() {
+ return zeroIfNull(manifest.deletedFilesCount());
+ }
+
+ @Override
+ public int replacedFilesCount() {
+ return zeroIfNull(manifest.replacedFilesCount());
Review Comment:
Maybe I miss something, but it seems that `manifest.replacedFilesCount()`
and `manifest.replacedRowsCount()` are always null.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
