rdblue commented on code in PR #16936:
URL: https://github.com/apache/iceberg/pull/16936#discussion_r4020986512
##########
core/src/main/java/org/apache/iceberg/TrackedFileAdapters.java:
##########
@@ -552,6 +609,511 @@ public ManifestFile copy() {
}
}
+ /** Shared base for content-file (DATA / EQUALITY_DELETES) write-direction
wrappers. */
+ abstract static class ContentTrackedFile<F extends ContentFile<F>>
+ implements TrackedFile, StructLike {
+ private final Types.StructType partitionType;
+ private final MapBackedContentStats statsWrapper;
+
+ private Tracking tracking;
+ private F file;
+ private StructProjection partition;
+ private ContentStats stats;
+
+ ContentTrackedFile(Schema writeSchema) {
+ Preconditions.checkArgument(writeSchema != null, "Invalid write schema:
null");
+ Types.StructType statsType = nestedStruct(writeSchema,
TrackedFile.CONTENT_STATS_ID);
+ this.partitionType = nestedStruct(writeSchema, TrackedFile.PARTITION_ID);
+ this.statsWrapper = statsType != null ? new
MapBackedContentStats(statsType) : null;
+ }
+
+ TrackedFile wrapWithTracking(F newFile, Tracking newTracking) {
+ Preconditions.checkArgument(newFile != null, "Invalid file: null");
+ Preconditions.checkArgument(newTracking != null, "Invalid tracking:
null");
+ validateContent(newFile);
+
+ if (newFile instanceof TrackedContentFile) {
+ return ((TrackedContentFile<?>) newFile).file();
+ }
+
+ this.file = newFile;
+ this.partition = partitionType != null ? projectPartition(newFile,
partitionType) : null;
+ this.stats = statsWrapper != null ? statsWrapper.wrap(newFile) : null;
+ this.tracking = newTracking;
+ return this;
+ }
+
+ /** 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 TableMetadata.MIN_FORMAT_VERSION_PARQUET_MANIFESTS;
+ }
+
+ @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 null;
+ }
+
+ @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(TrackedFileStruct.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(Schema writeSchema) {
+ super(writeSchema);
+ }
+
+ /** Re-points this wrapper at {@code newFile} in place. */
+ public TrackedFile wrap(DataFile newFile, Tracking tracking) {
+ return wrapWithTracking(newFile, tracking);
+ }
+
+ @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;
+ }
+
+ @Override
+ public Integer sortOrderId() {
+ return file().sortOrderId();
+ }
+ }
+
+ /** Wraps an equality {@link DeleteFile} as a {@link TrackedFile} row. */
+ static class EqualityDeleteTrackedFile extends
ContentTrackedFile<DeleteFile> {
+ EqualityDeleteTrackedFile(Schema writeSchema) {
+ super(writeSchema);
+ }
+
+ /** Re-points this wrapper at {@code newFile} in place. */
+ public TrackedFile wrap(DeleteFile newFile, Tracking tracking) {
+ return wrapWithTracking(newFile, tracking);
+ }
+
+ @Override
+ void validateContent(DeleteFile newFile) {
+ 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 List<Integer> equalityIds() {
+ return file().equalityFieldIds();
+ }
+ }
+
+ /** Wraps a {@link ManifestFile} as a v4+ leaf manifest row. */
+ static class ManifestTrackedFile implements TrackedFile, StructLike {
+ private final WrappedManifestInfo manifestInfo = new WrappedManifestInfo();
+ private Tracking tracking;
+ private ManifestFile manifest;
+ private long recordCount;
+ private FileContent contentType;
+
+ ManifestTrackedFile() {}
+
+ /**
+ * Re-points this wrapper at {@code newManifest} in place.
+ *
+ * @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
+ * @return this wrapper, or the original {@link TrackedFile} if {@code
newManifest} is already
+ * adapted from a tracked file
+ */
+ public TrackedFile wrap(ManifestFile newManifest, EntryStatus status, Long
firstRowId) {
+ Preconditions.checkArgument(newManifest != null, "Invalid manifest file:
null");
+ Preconditions.checkArgument(status != null, "Invalid status: null");
+
+ if (newManifest instanceof TrackedManifestFile) {
+ return ((TrackedManifestFile) newManifest).file();
+ }
+
+ 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() {
+ // newly written ManifestFile instances have no persisted version;
carry-over rows keep
+ // theirs via unwrap of TrackedManifestFile
+ return TableMetadata.MIN_FORMAT_VERSION_PARQUET_MANIFESTS;
+ }
+
+ @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();
Review Comment:
I think this is a reasonable choice, but it deserves a comment since we
don't expect this to be consistent for all files in a manifest. If it is
consistent, should it be required?
@amogh-jahagirdar FYI - this should probably be in the spec.
--
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]
