rdblue commented on code in PR #16936:
URL: https://github.com/apache/iceberg/pull/16936#discussion_r4021164143
##########
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:
This is no longer needed. The default implementation returns 0 for older
manifests.
--
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]
