rdblue commented on code in PR #16936:
URL: https://github.com/apache/iceberg/pull/16936#discussion_r4170346457
##########
core/src/main/java/org/apache/iceberg/TrackedFileAdapters.java:
##########
@@ -529,6 +566,427 @@ public ManifestFile copy() {
}
}
+ /** Adapts a {@link DataFile} to {@link TrackedFile}. */
+ static class DataTrackedFile implements TrackedFile {
+ private final MapBackedContentStats statsWrapper;
+ private Tracking tracking;
+ private DataFile file;
+ private ContentStats stats;
+
+ DataTrackedFile(Schema tableSchema, MetricsConfig metricsConfig) {
+ this.statsWrapper = new MapBackedContentStats(tableSchema,
metricsConfig);
+ }
+
+ /** Re-points this adapter at a {@link DataFile} from the public API.
Tracking is unset. */
+ public TrackedFile wrap(DataFile newFile) {
+ return wrapFile(newFile, null);
+ }
+
+ /**
+ * Re-points this adapter at a {@link ManifestEntry}. Converts the
contained data file and the
+ * entry's tracking fields.
+ */
+ public TrackedFile wrap(ManifestEntry<DataFile> entry) {
+ Preconditions.checkArgument(entry != null, "Invalid entry: null");
+ return wrapFile(entry.file(), trackingFrom(entry, entry.file()));
+ }
+
+ private TrackedFile wrapFile(DataFile newFile, Tracking newTracking) {
+ if (newFile instanceof TrackedDataFile tracked) {
+ return tracked.file();
+ }
+
+ Preconditions.checkArgument(newFile != null, "Invalid file: null");
+ Preconditions.checkArgument(
+ newFile.content() == FileContent.DATA,
+ "Invalid content for data file: %s",
+ newFile.content());
+
+ this.file = newFile;
+ this.stats = hasContentStats(newFile) ? statsWrapper.wrap(newFile) :
null;
+ this.tracking = newTracking;
+ return this;
+ }
+
+ @Override
+ public Tracking tracking() {
+ return tracking;
+ }
+
+ @Override
+ public FileContent contentType() {
+ return FileContent.DATA;
+ }
+
+ @Override
+ public int formatVersion() {
+ throw new IllegalStateException("Format version is assigned at write
time");
+ }
+
+ @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() {
+ // Files in one manifest may use different specs; this is the spec for
this data file only.
+ return file.specId();
+ }
+
+ @Override
+ public StructLike partition() {
+ return file.partition();
+ }
+
+ @Override
+ public ContentStats contentStats() {
+ return stats;
+ }
+
+ @Override
+ public Integer sortOrderId() {
+ return file.sortOrderId();
+ }
+
+ @Override
+ public DeletionVector deletionVector() {
+ return file.deletionVector();
+ }
+
+ @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("copy is not implemented");
+ }
+
+ @Override
+ public TrackedFile copyWithStats(Set<Integer> requestedColumnIds) {
+ throw new UnsupportedOperationException("copy is not implemented");
+ }
+ }
+
+ /** Adapts a {@link ManifestFile} to {@link TrackedFile}. */
+ static class ManifestTrackedFile implements TrackedFile {
+ private final WrappedManifestInfo manifestInfo = new WrappedManifestInfo();
+ private Tracking tracking;
+ private ManifestFile manifest;
+ private long recordCount;
+ private FileContent contentType;
+
+ ManifestTrackedFile() {}
+
+ /**
+ * Re-points this adapter at {@code newManifest}. Converts the manifest's
own fields only;
+ * write-time tracking updates are applied by the versioned writer.
+ */
+ public TrackedFile wrap(ManifestFile newManifest) {
+ if (newManifest instanceof TrackedManifestFile tracked) {
+ return tracked.file();
+ }
+
+ Preconditions.checkArgument(newManifest != null, "Invalid manifest file:
null");
+
+ this.manifest = newManifest;
+ this.contentType =
+ newManifest.content() == ManifestContent.DATA
+ ? FileContent.DATA_MANIFEST
+ : FileContent.DELETE_MANIFEST;
+ this.recordCount = manifestRecordCount(newManifest);
+ this.tracking = trackingFrom(newManifest);
+ 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() {
+ // Number of TrackedFile rows stored in the manifest.
+ return recordCount;
+ }
+
+ @Override
+ public long fileSizeInBytes() {
+ return manifest.length();
+ }
+
+ @Override
+ public Integer specId() {
+ // Spec the wrapped manifest was written with. Data file entries in a v4
manifest file may
+ // use different specs.
+ 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("copy is not implemented");
+ }
+
+ @Override
+ public TrackedFile copyWithStats(Set<Integer> requestedColumnIds) {
+ throw new UnsupportedOperationException("copy is not implemented");
+ }
+ }
+
+ /** Reusable {@link ManifestInfo} view over a {@link ManifestFile}'s counts.
*/
+ private static class WrappedManifestInfo implements ManifestInfo {
+ private ManifestFile manifest;
+
+ void wrap(ManifestFile newManifest) {
+ this.manifest = newManifest;
+ }
+
+ @Override
+ public int addedFilesCount() {
+ return manifest.addedFilesCount();
+ }
+
+ @Override
+ public int existingFilesCount() {
+ return manifest.existingFilesCount();
+ }
+
+ @Override
+ public int deletedFilesCount() {
+ return manifest.deletedFilesCount();
+ }
+
+ @Override
+ public int replacedFilesCount() {
+ return manifest.replacedFilesCount();
+ }
+
+ @Override
+ public int modifiedFilesCount() {
+ return manifest.modifiedFilesCount();
+ }
+
+ @Override
+ public long addedRowsCount() {
+ return manifest.addedRowsCount();
+ }
+
+ @Override
+ public long existingRowsCount() {
+ return manifest.existingRowsCount();
+ }
+
+ @Override
+ public long deletedRowsCount() {
+ return manifest.deletedRowsCount();
+ }
+
+ @Override
+ public long replacedRowsCount() {
+ return manifest.replacedRowsCount();
+ }
+
+ @Override
+ public long modifiedRowsCount() {
+ return manifest.modifiedRowsCount();
+ }
+
+ @Override
+ public long minSequenceNumber() {
+ return manifest.minSequenceNumber();
+ }
+
+ @Override
+ public ManifestBitmap manifestDeletionVector() {
+ return manifest.manifestDeletionVector();
+ }
+
+ @Override
+ public ManifestInfo copy() {
+ throw new UnsupportedOperationException("copy is not implemented");
+ }
+ }
+
+ private static Tracking trackingFrom(ManifestEntry<?> entry, ContentFile<?>
file) {
+ return new TrackingStruct(
+ entryStatus(entry.status()),
+ entry.snapshotId(),
+ entry.dataSequenceNumber(),
+ entry.fileSequenceNumber(),
+ null,
Review Comment:
```suggestion
null, // data file has no DV
```
It's nice to have comments that call out what is being passed null. In this
case, it's the DV snapshot ID because this data file has no DV. We may need a
wrapper that combines a DV and a data file later that will set this to
`dvDeleteFileEntry.snapshotId()`.
--
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]
