stevenzwu commented on code in PR #16936:
URL: https://github.com/apache/iceberg/pull/16936#discussion_r3615590336
##########
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:
Will simplify to `/** Format version for pre-v4 manifest files. */`.
##########
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:
Will update to `Returns the number of entries in the manifest file, or null
for pre-v4 manifests.` — the field is the entry count of the manifest, not the
sum of record counts of its data files.
##########
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:
On naming: `wrap(...)` returning `this` mirrors the existing writer-side
wrapper pattern in `V2Metadata`/`V3Metadata.ManifestEntryWrapper` — enables
`writer.append(wrapper.wrap(file, tracking))`. The mutation semantics are
documented on the method. Open to `rebind(...)` if that reads better.
On write-vs-read asymmetry: the write path streams every row through a
single writer instance, so per-row wrapper allocation is on the hot path. JMH
on this branch: reusable-wrap is ~1.4–2.0x faster than convert with much less
allocation (see the [benchmark
doc](https://docs.google.com/document/d/1FcUY068W-RKqT2xaKeFjxyBuO66oOeaepc_lIsLLask/edit)).
The read path allocates one adapter per file scanned (not per row), so
per-file allocation isn't in the same regime — different pressures, different
shape.
##########
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:
The `formatVersion >= 4` check gates the target manifest format (this
wrapper produces a v4+ TrackedFile row, only meaningful for v4+ writers). The
source `DataFile`/`DeleteFile` is version-agnostic at the Java API level —
wrapping a pre-v4 file into a v4+ manifest row is in fact the primary use case
(migration, and ongoing writes where existing pre-v4 files are carried into a
new v4 manifest). Will add a Javadoc note to make that explicit.
##########
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:
Content files (`DataFile`/`DeleteFile`) accept caller-built `Tracking`
because status and sequence numbers depend on writer/commit-time context that
lives outside the file. Manifest references have their tracking fully
determined by the referenced manifest itself (`snapshot_id`, `sequence_number`,
`min_sequence_number`), so `wrap(...)` builds it here. Open to taking
`Tracking` as input for API consistency, but that shifts the reconstruction
burden to callers with no new information.
##########
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:
Rejecting 1/2/3 is deliberate — those aren't valid values for the
`format_version` field (v4+ only). A `ManifestFile` reporting 1/2/3 would
signal a reader bug, not a legacy input; silently coercing would mask that.
Pre-v4 inputs correctly report `LEGACY_FORMAT_VERSION` (0) and are covered by
the first branch.
##########
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:
I am fine either way. but it seems that your described pattern seems more
common in the existing code base. I will update.
##########
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:
Will trim to the one-line intent. The read/write direction bullets and the
reusable-wrapper expectation belong on the individual `asDataFile` /
`forDataFile` methods where they're actionable.
##########
core/src/main/java/org/apache/iceberg/GenericManifestFile.java:
##########
Review Comment:
Will update the comment to note this is the pre-v4 constructor (kept for
callers on older format versions). The Javadoc for the new constructor covers
both. I can move the pre-v4 part here.
##########
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:
Wording is misleading — this field is the entry count of the manifest
referenced by this manifest_list entry, applicable to any v4+ manifest (leaf or
root in a split tree). Will drop the field-level comment; the interface Javadoc
on `recordCount()` is authoritative.
##########
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:
Will shorten to `/** v4+ constructor variant that accepts recordCount and
formatVersion. */`.
##########
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:
Will drop.
##########
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:
Will invert to match `deletionVector`: base returns null, `DataTrackedFile`
overrides with `file().sortOrderId()`, `EqualityDeleteTrackedFile` drops its
override.
##########
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:
Will remove — the class name and the check already convey it.
##########
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:
Intentional. A v4+ root manifest_list can reference either a v4+ manifest or
a pre-v4 manifest that's still live (during migration, or when older manifests
haven't been rewritten yet). Pre-v4 `ManifestFile` instances report
`LEGACY_FORMAT_VERSION` (0) because the field is v4+ only. Both are accepted
here.
##########
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:
Correct — `GenericManifestFile` doesn't override the interface defaults for
`replacedFilesCount`/`replacedRowsCount`, so both resolve to null and
`zeroIfNull` folds that to 0. Plumbing REPLACED aggregates from leaf-manifest
writers up to the manifest_list entry is a follow-up; will add a TODO here so
it's not forgotten.
##########
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:
Will drop the field-level comment — the interface Javadoc on
`formatVersion()` covers it.
--
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]
