This is an automated email from the ASF dual-hosted git repository.
RussellSpitzer pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/main by this push:
new b56b4bf9d6 Core: Basic fields and schemas for column files (#16285)
b56b4bf9d6 is described below
commit b56b4bf9d6f3f4547fd3773e8fd78239869fad4b
Author: gaborkaszab <[email protected]>
AuthorDate: Thu Nov 5 20:12:47 2026 +0100
Core: Basic fields and schemas for column files (#16285)
---
.../main/java/org/apache/iceberg/ColumnFile.java | 72 +++++++++
.../java/org/apache/iceberg/ColumnFileStruct.java | 166 +++++++++++++++++++++
.../main/java/org/apache/iceberg/TrackedFile.java | 9 +-
.../org/apache/iceberg/TrackedFileAdapters.java | 10 ++
.../java/org/apache/iceberg/TrackedFileStruct.java | 27 +++-
.../java/org/apache/iceberg/TrackingBuilder.java | 15 +-
.../java/org/apache/iceberg/TrackingStruct.java | 3 +-
.../java/org/apache/iceberg/V4ManifestReader.java | 13 ++
.../java/org/apache/iceberg/StatsTestUtil.java | 1 +
.../org/apache/iceberg/TestColumnFileStruct.java | 132 ++++++++++++++++
.../java/org/apache/iceberg/TestTrackedFile.java | 5 +-
.../apache/iceberg/TestTrackedFileAdapters.java | 13 +-
.../org/apache/iceberg/TestTrackedFileStruct.java | 34 ++++-
.../org/apache/iceberg/TestTrackingBuilder.java | 39 +++++
.../org/apache/iceberg/TestTrackingStruct.java | 14 +-
.../org/apache/iceberg/TestV4ManifestReader.java | 141 ++++++++++++++++-
.../java/org/apache/iceberg/V4TestComparators.java | 38 ++++-
.../java/org/apache/iceberg/V4TestHelpers.java | 63 ++++++--
18 files changed, 757 insertions(+), 38 deletions(-)
diff --git a/core/src/main/java/org/apache/iceberg/ColumnFile.java
b/core/src/main/java/org/apache/iceberg/ColumnFile.java
new file mode 100644
index 0000000000..7bdab8d88d
--- /dev/null
+++ b/core/src/main/java/org/apache/iceberg/ColumnFile.java
@@ -0,0 +1,72 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg;
+
+import java.nio.ByteBuffer;
+import java.util.Set;
+import org.apache.iceberg.types.Types;
+
+interface ColumnFile {
+ Types.NestedField LOCATION =
+ Types.NestedField.required(
+ 160, "location", Types.StringType.get(), "Location of the column
file");
+ Types.NestedField FIELD_IDS =
+ Types.NestedField.required(
+ 161,
+ "field_ids",
+ Types.ListType.ofRequired(162, Types.IntegerType.get()),
+ "Live field IDs in this column file");
+ Types.NestedField FILE_FORMAT =
+ Types.NestedField.required(
+ 163,
+ "file_format",
+ Types.StringType.get(),
+ "String file format name for this column file");
+ Types.NestedField FILE_SIZE_IN_BYTES =
+ Types.NestedField.required(
+ 164, "file_size_in_bytes", Types.LongType.get(), "Total column file
size in bytes");
+ Types.NestedField KEY_METADATA =
+ Types.NestedField.optional(
+ 165,
+ "key_metadata",
+ Types.BinaryType.get(),
+ "Key metadata for encryption; specific to the encryption scheme.");
+
+ static Types.StructType schema() {
+ return Types.StructType.of(LOCATION, FIELD_IDS, FILE_FORMAT,
FILE_SIZE_IN_BYTES, KEY_METADATA);
+ }
+
+ /** Returns the location of this column file. */
+ String location();
+
+ /** Returns the field IDs contained in this column file. */
+ Set<Integer> fieldIds();
+
+ /** Returns the format of this column file. */
+ FileFormat fileFormat();
+
+ /** Returns the total size of this column file in bytes. */
+ long fileSizeInBytes();
+
+ /** Returns encryption key metadata, or null if this column file is not
encrypted. */
+ ByteBuffer keyMetadata();
+
+ /** Copies this column file. */
+ ColumnFile copy();
+}
diff --git a/core/src/main/java/org/apache/iceberg/ColumnFileStruct.java
b/core/src/main/java/org/apache/iceberg/ColumnFileStruct.java
new file mode 100644
index 0000000000..5115be121b
--- /dev/null
+++ b/core/src/main/java/org/apache/iceberg/ColumnFileStruct.java
@@ -0,0 +1,166 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg;
+
+import java.io.Serializable;
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Set;
+import org.apache.iceberg.avro.SupportsIndexProjection;
+import org.apache.iceberg.relocated.com.google.common.base.MoreObjects;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
+import org.apache.iceberg.relocated.com.google.common.primitives.Ints;
+import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.ArrayUtil;
+import org.apache.iceberg.util.ByteBuffers;
+
+/** Mutable {@link StructLike} implementation of {@link ColumnFile}. */
+class ColumnFileStruct extends SupportsIndexProjection implements ColumnFile,
Serializable {
+ private static final Types.StructType BASE_TYPE =
+ Types.StructType.of(
+ ColumnFile.LOCATION,
+ ColumnFile.FIELD_IDS,
+ ColumnFile.FILE_FORMAT,
+ ColumnFile.FILE_SIZE_IN_BYTES,
+ ColumnFile.KEY_METADATA);
+
+ private String location = null;
+ private int[] fieldIds = null;
+ private FileFormat fileFormat = null;
+ private long fileSizeInBytes = -1L;
+ private byte[] keyMetadata = null;
+
+ /** Used by internal readers to instantiate this class with a projection
schema. */
+ ColumnFileStruct(Types.StructType projection) {
+ super(BASE_TYPE, projection);
+ }
+
+ ColumnFileStruct(
+ String location,
+ Set<Integer> fieldIds,
+ FileFormat fileFormat,
+ long fileSizeInBytes,
+ ByteBuffer keyMetadata) {
+ super(BASE_TYPE.fields().size());
+ this.location = location;
+ this.fieldIds = fieldIds != null ? Ints.toArray(fieldIds) : null;
+ this.fileFormat = fileFormat;
+ this.fileSizeInBytes = fileSizeInBytes;
+ this.keyMetadata = ByteBuffers.toByteArray(keyMetadata);
+ }
+
+ /** Copy constructor. */
+ private ColumnFileStruct(ColumnFileStruct toCopy) {
+ super(toCopy);
+ this.location = toCopy.location;
+ this.fieldIds =
+ toCopy.fieldIds != null ? Arrays.copyOf(toCopy.fieldIds,
toCopy.fieldIds.length) : null;
+ this.fileFormat = toCopy.fileFormat;
+ this.fileSizeInBytes = toCopy.fileSizeInBytes;
+ this.keyMetadata =
+ toCopy.keyMetadata != null
+ ? Arrays.copyOf(toCopy.keyMetadata, toCopy.keyMetadata.length)
+ : null;
+ }
+
+ /** Constructor for Java serialization. */
+ ColumnFileStruct() {
+ super(BASE_TYPE.fields().size());
+ }
+
+ @Override
+ public String location() {
+ return location;
+ }
+
+ // Package-private only so the manifest reader can store the location
resolved against the
+ // table location; other callers must go through construction.
+ void setLocation(String newLocation) {
+ this.location = newLocation;
+ }
+
+ @Override
+ public Set<Integer> fieldIds() {
+ return fieldIds != null ? ImmutableSet.copyOf(Ints.asList(fieldIds)) :
null;
+ }
+
+ @Override
+ public FileFormat fileFormat() {
+ return fileFormat;
+ }
+
+ @Override
+ public long fileSizeInBytes() {
+ return fileSizeInBytes;
+ }
+
+ @Override
+ public ByteBuffer keyMetadata() {
+ return keyMetadata != null ? ByteBuffer.wrap(keyMetadata) : null;
+ }
+
+ @Override
+ public ColumnFile copy() {
+ return new ColumnFileStruct(this);
+ }
+
+ @Override
+ protected <T> T internalGet(int pos, Class<T> javaClass) {
+ return javaClass.cast(getByPos(pos));
+ }
+
+ private Object getByPos(int pos) {
+ return switch (pos) {
+ case 0 -> location;
+ case 1 -> ArrayUtil.toUnmodifiableIntList(fieldIds);
+ case 2 -> fileFormat != null ? fileFormat.toString() : null;
+ case 3 -> fileSizeInBytes;
+ case 4 -> keyMetadata();
+ default -> throw new UnsupportedOperationException("Unknown field
ordinal: " + pos);
+ };
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ protected <T> void internalSet(int pos, T value) {
+ switch (pos) {
+ // always coerce to String for Serializable
+ case 0 -> this.location = value.toString();
+ case 1 -> this.fieldIds = ArrayUtil.toIntArray((List<Integer>) value);
+ case 2 -> this.fileFormat = FileFormat.fromString(value.toString());
+ case 3 -> this.fileSizeInBytes = (long) value;
+ case 4 -> this.keyMetadata = ByteBuffers.toByteArray((ByteBuffer) value);
+ default -> {
+ // ignore the object, it must be from a newer version of the format
+ }
+ }
+ }
+
+ @Override
+ public String toString() {
+ return MoreObjects.toStringHelper(this)
+ .add("location", location)
+ .add("field_ids", fieldIds)
+ .add("file_format", fileFormat)
+ .add("file_size_in_bytes", fileSizeInBytes)
+ .add("key_metadata", keyMetadata == null ? "null" : "(redacted)")
+ .toString();
+ }
+}
diff --git a/core/src/main/java/org/apache/iceberg/TrackedFile.java
b/core/src/main/java/org/apache/iceberg/TrackedFile.java
index 32757d235b..3581a22ec3 100644
--- a/core/src/main/java/org/apache/iceberg/TrackedFile.java
+++ b/core/src/main/java/org/apache/iceberg/TrackedFile.java
@@ -87,6 +87,9 @@ interface TrackedFile {
"split_offsets",
Types.ListType.ofRequired(133, Types.LongType.get()),
"Split offsets for the data file");
+ Types.NestedField COLUMN_FILES =
+ Types.NestedField.optional(
+ 158, "column_files", Types.ListType.ofRequired(159,
ColumnFile.schema()), "Column files");
private static List<Types.NestedField> fields(
Types.StructType partitionType, Types.StructType contentStatsType) {
@@ -109,7 +112,8 @@ interface TrackedFile {
DELETION_VECTOR,
MANIFEST_INFO,
KEY_METADATA,
- SPLIT_OFFSETS);
+ SPLIT_OFFSETS,
+ COLUMN_FILES);
}
private static Type typeOrUnknown(Types.StructType structType) {
@@ -178,6 +182,9 @@ interface TrackedFile {
/** Returns the list of recommended split locations, or null. */
List<Long> splitOffsets();
+ /** Returns the column files for this file. */
+ List<ColumnFile> columnFiles();
+
/** Copies this tracked file. */
TrackedFile copy();
diff --git a/core/src/main/java/org/apache/iceberg/TrackedFileAdapters.java
b/core/src/main/java/org/apache/iceberg/TrackedFileAdapters.java
index d1d074c116..77ad3b2139 100644
--- a/core/src/main/java/org/apache/iceberg/TrackedFileAdapters.java
+++ b/core/src/main/java/org/apache/iceberg/TrackedFileAdapters.java
@@ -652,6 +652,11 @@ class TrackedFileAdapters {
return file.splitOffsets();
}
+ @Override
+ public List<ColumnFile> columnFiles() {
+ return null;
+ }
+
@Override
public TrackedFile copy() {
throw new UnsupportedOperationException("copy is not implemented");
@@ -774,6 +779,11 @@ class TrackedFileAdapters {
return null;
}
+ @Override
+ public List<ColumnFile> columnFiles() {
+ return null;
+ }
+
@Override
public TrackedFile copy() {
throw new UnsupportedOperationException("copy is not implemented");
diff --git a/core/src/main/java/org/apache/iceberg/TrackedFileStruct.java
b/core/src/main/java/org/apache/iceberg/TrackedFileStruct.java
index e19ab7ab0e..f6a96edeb5 100644
--- a/core/src/main/java/org/apache/iceberg/TrackedFileStruct.java
+++ b/core/src/main/java/org/apache/iceberg/TrackedFileStruct.java
@@ -21,11 +21,13 @@ package org.apache.iceberg;
import java.io.Serializable;
import java.nio.ByteBuffer;
import java.util.Arrays;
+import java.util.Collections;
import java.util.List;
import java.util.Set;
import org.apache.iceberg.avro.SupportsIndexProjection;
import org.apache.iceberg.exceptions.ValidationException;
import org.apache.iceberg.relocated.com.google.common.base.MoreObjects;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.apache.iceberg.types.Types;
import org.apache.iceberg.util.ArrayUtil;
import org.apache.iceberg.util.ByteBuffers;
@@ -59,7 +61,8 @@ class TrackedFileStruct extends SupportsIndexProjection
implements TrackedFile,
TrackedFile.DELETION_VECTOR,
TrackedFile.MANIFEST_INFO,
TrackedFile.KEY_METADATA,
- TrackedFile.SPLIT_OFFSETS);
+ TrackedFile.SPLIT_OFFSETS,
+ TrackedFile.COLUMN_FILES);
private FileContent contentType = null;
private String location = null;
@@ -77,6 +80,7 @@ class TrackedFileStruct extends SupportsIndexProjection
implements TrackedFile,
private ManifestInfo manifestInfo = null;
private byte[] keyMetadata = null;
private long[] splitOffsets = null;
+ private List<ColumnFile> columnFiles = null;
private transient StructProjection partitionProjection = null;
@@ -104,7 +108,8 @@ class TrackedFileStruct extends SupportsIndexProjection
implements TrackedFile,
DeletionVector deletionVector,
ManifestInfo manifestInfo,
ByteBuffer keyMetadata,
- List<Long> splitOffsets) {
+ List<Long> splitOffsets,
+ List<ColumnFile> columnFiles) {
super(BASE_TYPE.fields().size());
this.tracking = tracking;
this.contentType = contentType;
@@ -120,9 +125,11 @@ class TrackedFileStruct extends SupportsIndexProjection
implements TrackedFile,
this.manifestInfo = manifestInfo;
this.keyMetadata = ByteBuffers.toByteArray(keyMetadata);
this.splitOffsets = ArrayUtil.toLongArray(splitOffsets);
+ this.columnFiles = columnFiles != null ? Lists.newArrayList(columnFiles) :
null;
}
/** Copy constructor. */
+ @SuppressWarnings("CyclomaticComplexity")
private TrackedFileStruct(TrackedFileStruct toCopy, Set<Integer> statsIds) {
super(toCopy);
this.contentType = toCopy.contentType;
@@ -152,6 +159,14 @@ class TrackedFileStruct extends SupportsIndexProjection
implements TrackedFile,
toCopy.splitOffsets != null
? Arrays.copyOf(toCopy.splitOffsets, toCopy.splitOffsets.length)
: null;
+ if (toCopy.columnFiles != null) {
+ this.columnFiles =
Lists.newArrayListWithCapacity(toCopy.columnFiles.size());
+ for (ColumnFile columnFile : toCopy.columnFiles) {
+ this.columnFiles.add(columnFile.copy());
+ }
+ } else {
+ this.columnFiles = null;
+ }
}
@Override
@@ -244,6 +259,11 @@ class TrackedFileStruct extends SupportsIndexProjection
implements TrackedFile,
return splitOffsets != null ?
ArrayUtil.toUnmodifiableLongList(splitOffsets) : null;
}
+ @Override
+ public List<ColumnFile> columnFiles() {
+ return columnFiles != null ? Collections.unmodifiableList(columnFiles) :
null;
+ }
+
@Override
public TrackedFile copy() {
return new TrackedFileStruct(this, null);
@@ -275,6 +295,7 @@ class TrackedFileStruct extends SupportsIndexProjection
implements TrackedFile,
case 11 -> manifestInfo;
case 12 -> keyMetadata();
case 13 -> splitOffsets();
+ case 14 -> columnFiles();
default -> throw new UnsupportedOperationException("Unknown field
ordinal: " + pos);
};
}
@@ -298,6 +319,7 @@ class TrackedFileStruct extends SupportsIndexProjection
implements TrackedFile,
case 11 -> this.manifestInfo = (ManifestInfo) value;
case 12 -> this.keyMetadata = ByteBuffers.toByteArray((ByteBuffer)
value);
case 13 -> this.splitOffsets = ArrayUtil.toLongArray((List<Long>) value);
+ case 14 -> this.columnFiles = (List<ColumnFile>) value;
default -> {
// ignore the object, it must be from a newer version of the format
}
@@ -321,6 +343,7 @@ class TrackedFileStruct extends SupportsIndexProjection
implements TrackedFile,
.add("manifest_info", manifestInfo)
.add("key_metadata", keyMetadata == null ? "null" : "(redacted)")
.add("split_offsets", splitOffsets == null ? "null" : splitOffsets())
+ .add("column_files", columnFiles)
.toString();
}
}
diff --git a/core/src/main/java/org/apache/iceberg/TrackingBuilder.java
b/core/src/main/java/org/apache/iceberg/TrackingBuilder.java
index c076531fcb..64e7cae043 100644
--- a/core/src/main/java/org/apache/iceberg/TrackingBuilder.java
+++ b/core/src/main/java/org/apache/iceberg/TrackingBuilder.java
@@ -25,10 +25,10 @@ import org.apache.iceberg.util.ByteBuffers;
class TrackingBuilder {
private final long newSnapshotId;
private final Long snapshotId;
- private final Long dataSequenceNumber;
private final Long fileSequenceNumber;
private final Long firstRowId;
private EntryStatus status;
+ private Long dataSequenceNumber;
private Long modifiedSnapshotId;
private byte[] deletedPositions;
private byte[] replacedPositions;
@@ -115,6 +115,19 @@ class TrackingBuilder {
return this;
}
+ /** Indicates that the column files list has been updated for the new
Tracking. */
+ TrackingBuilder columnFilesUpdated() {
+ this.modifiedSnapshotId = newSnapshotId;
+ if (status == EntryStatus.EXISTING) {
+ this.status = EntryStatus.MODIFIED;
+ }
+ // Reset to null to inherit from the new snapshot sequence number. It is
safe to bump up the
+ // dataSequenceNumber as writers are required to rewrite v2 equality and
position deletes to DVs
+ // when applying column update.
+ this.dataSequenceNumber = null;
+ return this;
+ }
+
/** Sets the positions deleted by this commit for a manifest entry. */
TrackingBuilder deletedPositions(ByteBuffer positions) {
Preconditions.checkState(
diff --git a/core/src/main/java/org/apache/iceberg/TrackingStruct.java
b/core/src/main/java/org/apache/iceberg/TrackingStruct.java
index c804497d82..d06eae71c5 100644
--- a/core/src/main/java/org/apache/iceberg/TrackingStruct.java
+++ b/core/src/main/java/org/apache/iceberg/TrackingStruct.java
@@ -141,8 +141,9 @@ class TrackingStruct extends SupportsIndexProjection
implements Tracking, Serial
}
boolean isAdded = status == EntryStatus.ADDED;
+ boolean isModified = status == EntryStatus.MODIFIED;
- if (null == dataSequenceNumber && (isAdded || manifestSeqNumber == 0)) {
+ if (null == dataSequenceNumber && (isAdded || isModified ||
manifestSeqNumber == 0)) {
this.dataSequenceNumber = manifestSeqNumber;
}
diff --git a/core/src/main/java/org/apache/iceberg/V4ManifestReader.java
b/core/src/main/java/org/apache/iceberg/V4ManifestReader.java
index b1009a8ea6..a91c32d4ff 100644
--- a/core/src/main/java/org/apache/iceberg/V4ManifestReader.java
+++ b/core/src/main/java/org/apache/iceberg/V4ManifestReader.java
@@ -19,6 +19,7 @@
package org.apache.iceberg;
import java.util.Arrays;
+import java.util.List;
import java.util.Map;
import java.util.Set;
import org.apache.iceberg.expressions.Binder;
@@ -227,6 +228,8 @@ class V4ManifestReader extends CloseableGroup implements
CloseableIterable<Track
.setCustomType(TrackedFile.TRACKING.fieldId(),
TrackingStruct.class)
.setCustomType(TrackedFile.DELETION_VECTOR.fieldId(),
DeletionVectorStruct.class)
.setCustomType(TrackedFile.MANIFEST_INFO.fieldId(),
ManifestInfoStruct.class)
+ .setCustomType(
+ TrackedFile.COLUMN_FILES.type().asListType().elementId(),
ColumnFileStruct.class)
.reuseContainers();
Types.NestedField partitionField =
readSchema.findField(TrackedFile.PARTITION_ID);
@@ -283,6 +286,16 @@ class V4ManifestReader extends CloseableGroup implements
CloseableIterable<Track
.setLocation(LocationUtil.resolveLocation(tableLocation,
deletionVector.location()));
}
+ List<ColumnFile> columnFiles = copy.columnFiles();
+ if (columnFiles != null) {
+ for (ColumnFile columnFile : columnFiles) {
+ if (columnFile.location() != null) {
+ ((ColumnFileStruct) columnFile)
+ .setLocation(LocationUtil.resolveLocation(tableLocation,
columnFile.location()));
+ }
+ }
+ }
+
return copy;
}
diff --git a/core/src/test/java/org/apache/iceberg/StatsTestUtil.java
b/core/src/test/java/org/apache/iceberg/StatsTestUtil.java
index 2e0307fe9f..9f06e17c4f 100644
--- a/core/src/test/java/org/apache/iceberg/StatsTestUtil.java
+++ b/core/src/test/java/org/apache/iceberg/StatsTestUtil.java
@@ -40,6 +40,7 @@ class StatsTestUtil {
null,
null,
null,
+ null,
null);
}
diff --git a/core/src/test/java/org/apache/iceberg/TestColumnFileStruct.java
b/core/src/test/java/org/apache/iceberg/TestColumnFileStruct.java
new file mode 100644
index 0000000000..8e7b4ae06e
--- /dev/null
+++ b/core/src/test/java/org/apache/iceberg/TestColumnFileStruct.java
@@ -0,0 +1,132 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Set;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
+import org.apache.iceberg.types.Comparators;
+import org.apache.iceberg.types.Types;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.MethodSource;
+
+class TestColumnFileStruct {
+
+ private static final String LOCATION = "s3://bucket/data/column.parquet";
+ private static final Set<Integer> FIELD_IDS = ImmutableSet.of(1, 2, 3);
+ private static final FileFormat FILE_FORMAT = FileFormat.PARQUET;
+ private static final long FILE_SIZE_IN_BYTES = 1024L;
+ private static final ByteBuffer KEY_METADATA = ByteBuffer.wrap(new byte[]
{1, 2, 3});
+ private static final Comparator<StructLike> COLUMN_FILE_COMPARATOR =
+ Comparators.forType(ColumnFile.schema());
+
+ @Test
+ void fieldAccess() {
+ ColumnFile columnFile =
+ new ColumnFileStruct(LOCATION, FIELD_IDS, FILE_FORMAT,
FILE_SIZE_IN_BYTES, KEY_METADATA);
+
+ assertThat(columnFile.location()).isEqualTo(LOCATION);
+ assertThat(columnFile.fieldIds()).isEqualTo(FIELD_IDS);
+ assertThat(columnFile.fileFormat()).isEqualTo(FILE_FORMAT);
+ assertThat(columnFile.fileSizeInBytes()).isEqualTo(FILE_SIZE_IN_BYTES);
+ assertThat(columnFile.keyMetadata()).isEqualTo(KEY_METADATA);
+ }
+
+ @Test
+ void copy() {
+ ColumnFile columnFile =
+ new ColumnFileStruct(LOCATION, FIELD_IDS, FILE_FORMAT,
FILE_SIZE_IN_BYTES, KEY_METADATA);
+
+ ColumnFile copy = columnFile.copy();
+
+ assertThat((StructLike)
copy).usingComparator(COLUMN_FILE_COMPARATOR).isEqualTo(columnFile);
+
+ // verify deep copy of key metadata
+
assertThat(copy.keyMetadata().array()).isNotSameAs(columnFile.keyMetadata().array());
+ }
+
+ @Test
+ void setFieldsByOrdinals() {
+ ColumnFileStruct columnFile = new ColumnFileStruct();
+
+ columnFile.set(0, LOCATION);
+ columnFile.set(1, ImmutableList.copyOf(FIELD_IDS));
+ columnFile.set(2, FILE_FORMAT.toString());
+ columnFile.set(3, FILE_SIZE_IN_BYTES);
+ columnFile.set(4, KEY_METADATA);
+
+ assertThat(columnFile.location()).isEqualTo(LOCATION);
+ assertThat(columnFile.fieldIds()).isEqualTo(FIELD_IDS);
+ assertThat(columnFile.fileFormat()).isEqualTo(FILE_FORMAT);
+ assertThat(columnFile.fileSizeInBytes()).isEqualTo(FILE_SIZE_IN_BYTES);
+ assertThat(columnFile.keyMetadata()).isEqualTo(KEY_METADATA);
+ }
+
+ @Test
+ void getFieldsByOrdinals() {
+ ColumnFileStruct columnFile =
+ new ColumnFileStruct(LOCATION, FIELD_IDS, FILE_FORMAT,
FILE_SIZE_IN_BYTES, KEY_METADATA);
+
+ assertThat(columnFile.get(0, String.class)).isEqualTo(LOCATION);
+ assertThat(columnFile.get(1,
List.class)).containsExactlyElementsOf(FIELD_IDS);
+ assertThat(columnFile.get(2,
String.class)).isEqualTo(FILE_FORMAT.toString());
+ assertThat(columnFile.get(3, Long.class)).isEqualTo(FILE_SIZE_IN_BYTES);
+ assertThat(columnFile.get(4, ByteBuffer.class)).isEqualTo(KEY_METADATA);
+ }
+
+ @Test
+ void projectedStructLike() {
+ Types.StructType projection =
+ Types.StructType.of(ColumnFile.LOCATION,
ColumnFile.FILE_SIZE_IN_BYTES);
+
+ ColumnFileStruct columnFile = new ColumnFileStruct(projection);
+ assertThat(columnFile.size()).isEqualTo(2);
+
+ // projected position 0 maps to internal position of location
+ // projected position 1 maps to internal position of file_size_in_bytes
+ columnFile.set(0, LOCATION);
+ columnFile.set(1, FILE_SIZE_IN_BYTES);
+
+ assertThat(columnFile.location()).isEqualTo(LOCATION);
+ assertThat(columnFile.fileSizeInBytes()).isEqualTo(FILE_SIZE_IN_BYTES);
+ assertThat(columnFile.get(0, String.class)).isEqualTo(LOCATION);
+ assertThat(columnFile.get(1, Long.class)).isEqualTo(FILE_SIZE_IN_BYTES);
+ }
+
+ @ParameterizedTest
+ @MethodSource("org.apache.iceberg.TestHelpers#serializers")
+ void serializationRoundTrip(TestHelpers.RoundTripSerializer<ColumnFile>
roundTripSerializer)
+ throws IOException, ClassNotFoundException {
+ ColumnFile columnFile =
+ new ColumnFileStruct(LOCATION, FIELD_IDS, FILE_FORMAT,
FILE_SIZE_IN_BYTES, KEY_METADATA);
+
+ ColumnFile deserialized = roundTripSerializer.apply(columnFile);
+
+ assertThat((StructLike) deserialized)
+ .usingComparator(COLUMN_FILE_COMPARATOR)
+ .isEqualTo(columnFile);
+ }
+}
diff --git a/core/src/test/java/org/apache/iceberg/TestTrackedFile.java
b/core/src/test/java/org/apache/iceberg/TestTrackedFile.java
index dbb589ee36..d4c4e467e8 100644
--- a/core/src/test/java/org/apache/iceberg/TestTrackedFile.java
+++ b/core/src/test/java/org/apache/iceberg/TestTrackedFile.java
@@ -59,7 +59,8 @@ public class TestTrackedFile {
"deletion_vector",
"manifest_info",
"key_metadata",
- "split_offsets");
+ "split_offsets",
+ "column_files");
}
@Test
@@ -69,7 +70,7 @@ public class TestTrackedFile {
assertThat(fields)
.extracting(Types.NestedField::fieldId)
- .containsExactly(147, 134, 100, 101, 103, 104, 141, 102, 146, 140,
148, 150, 131, 132);
+ .containsExactly(147, 134, 100, 101, 103, 104, 141, 102, 146, 140,
148, 150, 131, 132, 158);
}
@Test
diff --git a/core/src/test/java/org/apache/iceberg/TestTrackedFileAdapters.java
b/core/src/test/java/org/apache/iceberg/TestTrackedFileAdapters.java
index 1d72a884e7..9ccdd8753e 100644
--- a/core/src/test/java/org/apache/iceberg/TestTrackedFileAdapters.java
+++ b/core/src/test/java/org/apache/iceberg/TestTrackedFileAdapters.java
@@ -214,7 +214,8 @@ class TestTrackedFileAdapters {
dv,
null,
ByteBuffer.wrap(new byte[] {1, 2, 3}),
- ImmutableList.of(50L, 100L));
+ ImmutableList.of(50L, 100L),
+ null);
DataFile dataFile = TrackedFileAdapters.asDataFile(file,
specsById(PARTITIONED_SPEC));
@@ -301,6 +302,7 @@ class TestTrackedFileAdapters {
dv,
null,
null,
+ null,
null);
DeleteFile dvFile = TrackedFileAdapters.asDVDeleteFile(file,
specsById(PARTITIONED_SPEC));
@@ -431,7 +433,8 @@ class TestTrackedFileAdapters {
dv, // deletionVector
null, // manifestInfo
KEY_METADATA,
- ImmutableList.of(50L, 100L)); // splitOffsets
+ ImmutableList.of(50L, 100L), // splitOffsets
+ null); // columnFiles
}
@ParameterizedTest
@@ -473,7 +476,8 @@ class TestTrackedFileAdapters {
null, // deletionVector
MANIFEST_INFO,
MANIFEST_KEY_METADATA,
- null); // splitOffsets
+ null, // splitOffsets
+ null); // columnFiles
ManifestFile manifest = TrackedFileAdapters.asManifestFile(file);
@@ -579,6 +583,7 @@ class TestTrackedFileAdapters {
deletionVector(),
null,
null,
+ null,
null);
assertNullTrackingFields(TrackedFileAdapters.asDVDeleteFile(fileWithDV,
UNPARTITIONED));
}
@@ -631,6 +636,7 @@ class TestTrackedFileAdapters {
null,
null,
null,
+ null,
null);
assertThatThrownBy(() -> TrackedFileAdapters.asDataFile(file,
ImmutableMap.of()))
@@ -934,6 +940,7 @@ class TestTrackedFileAdapters {
null,
null,
null,
+ null,
null);
}
diff --git a/core/src/test/java/org/apache/iceberg/TestTrackedFileStruct.java
b/core/src/test/java/org/apache/iceberg/TestTrackedFileStruct.java
index 91919ca248..f65ce4941e 100644
--- a/core/src/test/java/org/apache/iceberg/TestTrackedFileStruct.java
+++ b/core/src/test/java/org/apache/iceberg/TestTrackedFileStruct.java
@@ -57,11 +57,18 @@ class TestTrackedFileStruct {
private static final ManifestInfo MANIFEST_INFO =
Mockito.mock(ManifestInfo.class);
private static final ManifestInfo MANIFEST_INFO_COPY =
Mockito.mock(ManifestInfo.class);
+ private static final ColumnFile COLUMN_FILE_1 =
Mockito.mock(ColumnFile.class);
+ private static final ColumnFile COLUMN_FILE_2 =
Mockito.mock(ColumnFile.class);
+ private static final ColumnFile COLUMN_FILE_1_COPY =
Mockito.mock(ColumnFile.class);
+ private static final ColumnFile COLUMN_FILE_2_COPY =
Mockito.mock(ColumnFile.class);
+
static {
Mockito.when(TRACKING.copy()).thenReturn(TRACKING_COPY);
Mockito.when(CONTENT_STATS.copy()).thenReturn(CONTENT_STATS_COPY);
Mockito.when(DELETION_VECTOR.copy()).thenReturn(DELETION_VECTOR_COPY);
Mockito.when(MANIFEST_INFO.copy()).thenReturn(MANIFEST_INFO_COPY);
+ Mockito.when(COLUMN_FILE_1.copy()).thenReturn(COLUMN_FILE_1_COPY);
+ Mockito.when(COLUMN_FILE_2.copy()).thenReturn(COLUMN_FILE_2_COPY);
}
@Test
@@ -81,7 +88,8 @@ class TestTrackedFileStruct {
DELETION_VECTOR,
MANIFEST_INFO,
ByteBuffer.wrap(new byte[] {1, 2, 3}),
- ImmutableList.of(100L, 200L));
+ ImmutableList.of(100L, 200L),
+ ImmutableList.of(COLUMN_FILE_1, COLUMN_FILE_2));
assertThat(file.tracking()).isSameAs(TRACKING);
assertThat(file.contentType()).isEqualTo(FileContent.DATA);
@@ -97,6 +105,7 @@ class TestTrackedFileStruct {
assertThat(file.manifestInfo()).isSameAs(MANIFEST_INFO);
assertThat(file.keyMetadata()).isEqualTo(ByteBuffer.wrap(new byte[] {1, 2,
3}));
assertThat(file.splitOffsets()).containsExactly(100L, 200L);
+ assertThat(file.columnFiles()).containsExactly(COLUMN_FILE_1,
COLUMN_FILE_2);
}
@Test
@@ -150,7 +159,8 @@ class TestTrackedFileStruct {
DELETION_VECTOR,
MANIFEST_INFO,
ByteBuffer.wrap(new byte[] {1, 2, 3}),
- ImmutableList.of(100L, 200L));
+ ImmutableList.of(100L, 200L),
+ ImmutableList.of(COLUMN_FILE_1, COLUMN_FILE_2));
assertThat(file.get(pos("tracking"), Tracking.class)).isSameAs(TRACKING);
assertThat(file.get(pos("content_type"),
Integer.class)).isEqualTo(FileContent.DATA.id());
@@ -187,7 +197,8 @@ class TestTrackedFileStruct {
DELETION_VECTOR,
MANIFEST_INFO,
ByteBuffer.wrap(new byte[] {1, 2, 3}),
- ImmutableList.of(100L, 200L));
+ ImmutableList.of(100L, 200L),
+ ImmutableList.of(COLUMN_FILE_1, COLUMN_FILE_2));
TrackedFile copy = file.copy();
@@ -206,6 +217,7 @@ class TestTrackedFileStruct {
assertThat(copy.keyMetadata()).isEqualTo(ByteBuffer.wrap(new byte[] {1, 2,
3}));
assertThat(copy.splitOffsets()).containsExactly(100L, 200L);
assertThat(copy.partition()).isNotSameAs(PARTITION);
+ assertThat(copy.columnFiles()).containsExactly(COLUMN_FILE_1_COPY,
COLUMN_FILE_2_COPY);
// mutable fields are deep-copied, not shared with the original
assertThat(copy.keyMetadata()).isNotSameAs(file.keyMetadata());
@@ -232,7 +244,8 @@ class TestTrackedFileStruct {
DELETION_VECTOR,
MANIFEST_INFO,
ByteBuffer.wrap(new byte[] {1, 2, 3}),
- ImmutableList.of(100L, 200L));
+ ImmutableList.of(100L, 200L),
+ ImmutableList.of(COLUMN_FILE_1, COLUMN_FILE_2));
TrackedFile copy = file.copyWithStats(ImmutableSet.of(1));
@@ -251,6 +264,7 @@ class TestTrackedFileStruct {
assertThat(copy.keyMetadata()).isEqualTo(ByteBuffer.wrap(new byte[] {1, 2,
3}));
assertThat(copy.splitOffsets()).containsExactly(100L, 200L);
assertThat(copy.partition()).isNotSameAs(PARTITION);
+ assertThat(copy.columnFiles()).containsExactly(COLUMN_FILE_1_COPY,
COLUMN_FILE_2_COPY);
// mutable fields are deep-copied, not shared with the original
assertThat(copy.keyMetadata()).isNotSameAs(file.keyMetadata());
@@ -275,7 +289,8 @@ class TestTrackedFileStruct {
DELETION_VECTOR,
MANIFEST_INFO,
ByteBuffer.wrap(new byte[] {1, 2, 3}),
- ImmutableList.of(100L, 200L));
+ ImmutableList.of(100L, 200L),
+ ImmutableList.of(COLUMN_FILE_1, COLUMN_FILE_2));
TrackedFile copy = file.copyWithoutStats();
@@ -297,9 +312,11 @@ class TestTrackedFileStruct {
assertThat(copy.keyMetadata()).isEqualTo(ByteBuffer.wrap(new byte[] {1, 2,
3}));
assertThat(copy.splitOffsets()).containsExactly(100L, 200L);
assertThat(copy.partition()).isNotSameAs(PARTITION);
+ assertThat(copy.columnFiles()).containsExactly(COLUMN_FILE_1_COPY,
COLUMN_FILE_2_COPY);
// mutable fields are deep-copied, not shared with the original
assertThat(copy.keyMetadata()).isNotSameAs(file.keyMetadata());
+ assertThat(copy.columnFiles()).isNotSameAs(file.columnFiles());
}
@Test
@@ -421,7 +438,8 @@ class TestTrackedFileStruct {
null, // DeletionVector has its own serialization tests
null, // ManifestInfo has its own serialization tests
ByteBuffer.wrap(new byte[] {1, 2, 3}),
- ImmutableList.of(50L));
+ ImmutableList.of(50L),
+ null); // ColumnFile has its own serialization tests
TrackedFileStruct deserialized = serializer.apply(file);
@@ -440,6 +458,7 @@ class TestTrackedFileStruct {
assertThat(deserialized.manifestInfo()).isNull();
assertThat(deserialized.keyMetadata()).isEqualTo(ByteBuffer.wrap(new
byte[] {1, 2, 3}));
assertThat(deserialized.splitOffsets()).containsExactly(50L);
+ assertThat(deserialized.columnFiles()).isNull();
}
@ParameterizedTest
@@ -494,7 +513,8 @@ class TestTrackedFileStruct {
null, // deletionVector
null, // manifestInfo
null, // keyMetadata
- null); // splitOffsets
+ null, // splitOffsets
+ null); // column files
}
private static int pos(String fieldName) {
diff --git a/core/src/test/java/org/apache/iceberg/TestTrackingBuilder.java
b/core/src/test/java/org/apache/iceberg/TestTrackingBuilder.java
index 6fac40ecc9..ae1a13570f 100644
--- a/core/src/test/java/org/apache/iceberg/TestTrackingBuilder.java
+++ b/core/src/test/java/org/apache/iceberg/TestTrackingBuilder.java
@@ -122,6 +122,18 @@ class TestTrackingBuilder {
assertThat(modified.modifiedSnapshotId()).isEqualTo(999L);
}
+ @Test
+ void columnFilesUpdatedProducesModifiedAndResetsDataSequenceNumber() {
+ Tracking modified =
+ TrackingBuilder.from(SOURCE_TRACKING_ADDED,
999L).columnFilesUpdated().build();
+
+ assertThat(modified.status()).isEqualTo(EntryStatus.MODIFIED);
+ assertThat(modified.modifiedSnapshotId()).isEqualTo(999L);
+ // data sequence number is reset to inherit the new snapshot's sequence
number
+ assertThat(modified.dataSequenceNumber()).isNull();
+
assertThat(modified.fileSequenceNumber()).isEqualTo(SOURCE_TRACKING_ADDED.fileSequenceNumber());
+ }
+
@Test
void manifestDVMutatorsRejectedOnAdded() {
assertThatThrownBy(
@@ -280,4 +292,31 @@ class TestTrackingBuilder {
assertThat(modified.modifiedSnapshotId()).isEqualTo(999L);
assertThat(modified.deletedPositions()).isEqualTo(deletedBytes);
}
+
+ @Test
+ void manifestPositionsWithColumnFilesUpdated() {
+ ByteBuffer deletedBytes = ByteBuffer.wrap(new byte[] {1});
+ Tracking withDeletedPositions =
+ TrackingBuilder.from(SOURCE_TRACKING_ADDED, 999L)
+ .columnFilesUpdated()
+ .deletedPositions(deletedBytes)
+ .build();
+
+ assertThat(withDeletedPositions.status()).isEqualTo(EntryStatus.MODIFIED);
+ assertThat(withDeletedPositions.modifiedSnapshotId()).isEqualTo(999L);
+
assertThat(withDeletedPositions.deletedPositions()).isEqualTo(deletedBytes);
+ assertThat(withDeletedPositions.dataSequenceNumber()).isNull();
+
+ ByteBuffer replacedBytes = ByteBuffer.wrap(new byte[] {2});
+ Tracking withReplacedPositions =
+ TrackingBuilder.from(SOURCE_TRACKING_ADDED, 999L)
+ .columnFilesUpdated()
+ .replacedPositions(replacedBytes)
+ .build();
+
+ assertThat(withReplacedPositions.status()).isEqualTo(EntryStatus.MODIFIED);
+ assertThat(withReplacedPositions.modifiedSnapshotId()).isEqualTo(999L);
+
assertThat(withReplacedPositions.replacedPositions()).isEqualTo(replacedBytes);
+ assertThat(withReplacedPositions.dataSequenceNumber()).isNull();
+ }
}
diff --git a/core/src/test/java/org/apache/iceberg/TestTrackingStruct.java
b/core/src/test/java/org/apache/iceberg/TestTrackingStruct.java
index 3975706e5c..eea6e4e0a7 100644
--- a/core/src/test/java/org/apache/iceberg/TestTrackingStruct.java
+++ b/core/src/test/java/org/apache/iceberg/TestTrackingStruct.java
@@ -189,9 +189,19 @@ class TestTrackingStruct {
assertThat(tracking.fileSequenceNumber()).isEqualTo(60L);
}
+ @Test
+ void inheritanceModifiedEntriesInheritDataSequenceNumber() {
+ TrackingStruct tracking =
+ new TrackingStruct(EntryStatus.MODIFIED, 42L, null, null, null, null,
null, null);
+
+ tracking.inherit(100L, 60L);
+
+ assertThat(tracking.dataSequenceNumber()).isEqualTo(60L);
+ assertThat(tracking.fileSequenceNumber()).isNull();
+ }
+
private static final List<EntryStatus> NON_INHERITING_STATUSES =
- List.of(
- EntryStatus.EXISTING, EntryStatus.MODIFIED, EntryStatus.DELETED,
EntryStatus.REPLACED);
+ List.of(EntryStatus.EXISTING, EntryStatus.DELETED, EntryStatus.REPLACED);
@ParameterizedTest
@FieldSource("NON_INHERITING_STATUSES")
diff --git a/core/src/test/java/org/apache/iceberg/TestV4ManifestReader.java
b/core/src/test/java/org/apache/iceberg/TestV4ManifestReader.java
index 36f9712a81..864a0e1eec 100644
--- a/core/src/test/java/org/apache/iceberg/TestV4ManifestReader.java
+++ b/core/src/test/java/org/apache/iceberg/TestV4ManifestReader.java
@@ -23,17 +23,21 @@ import static
org.apache.iceberg.V4TestHelpers.FILE_SIZE_IN_BYTES;
import static org.apache.iceberg.V4TestHelpers.FORMAT_VERSION_V4;
import static org.apache.iceberg.V4TestHelpers.RECORD_COUNT;
import static org.apache.iceberg.V4TestHelpers.SNAPSHOT_ID;
+import static org.apache.iceberg.V4TestHelpers.columnFile;
import static org.apache.iceberg.V4TestHelpers.dataFile;
+import static org.apache.iceberg.V4TestHelpers.dataFileWithColumnFiles;
import static org.apache.iceberg.V4TestHelpers.dataFileWithDV;
import static org.apache.iceberg.V4TestHelpers.dataFileWithStats;
import static org.apache.iceberg.V4TestHelpers.deletionVector;
import static org.apache.iceberg.V4TestHelpers.manifestRef;
+import static org.apache.iceberg.V4TestHelpers.manifestRefWithColumnFiles;
import static org.apache.iceberg.V4TestHelpers.manifestRefWithStats;
import static org.apache.iceberg.V4TestHelpers.partition;
import static org.apache.iceberg.V4TestHelpers.writeTrackedFiles;
import static org.apache.iceberg.types.Types.NestedField.optional;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.assertj.core.api.Assertions.tuple;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
@@ -155,6 +159,19 @@ class TestV4ManifestReader {
private static final InMemoryFileIO IO = new InMemoryFileIO();
+ private static final ColumnFile COLUMN_FILE_1 =
+ columnFile(
+ "s3a://bucket/column_file_1.parquet",
+ Set.of(1, 3),
+ ByteBuffer.wrap(new byte[] {1, 2, 3}));
+ private static final ColumnFile COLUMN_FILE_2 =
+ columnFile(
+ "s3a://bucket/column_file_2.parquet",
+ Set.of(2),
+ ByteBuffer.wrap(new byte[] {1, 2, 3, 4}));
+ private static final List<ColumnFile> COLUMN_FILES =
+ ImmutableList.of(COLUMN_FILE_1, COLUMN_FILE_2);
+
@ParameterizedTest
@FieldSource("MANIFEST_FORMATS")
public void readDataFile(FileFormat format) throws IOException {
@@ -173,7 +190,8 @@ class TestV4ManifestReader {
DV,
null, // manifest info
ByteBuffer.wrap(new byte[] {1, 2, 3}), // key metadata
- ImmutableList.of(50L, 100L));
+ ImmutableList.of(50L, 100L),
+ COLUMN_FILES);
ManifestFile manifest = writeManifest(format, ID_PARTITIONED_TYPE, file);
@@ -203,7 +221,8 @@ class TestV4ManifestReader {
DV,
null, // manifest info
ByteBuffer.wrap(new byte[] {1, 2, 3}), // key metadata
- ImmutableList.of(50L, 100L));
+ ImmutableList.of(50L, 100L),
+ COLUMN_FILES);
ManifestFile manifest = writeManifest(format, ID_PARTITIONED_TYPE, file);
@@ -235,7 +254,8 @@ class TestV4ManifestReader {
DV,
null, // manifest info
ByteBuffer.wrap(new byte[] {1, 2, 3}), // key metadata
- ImmutableList.of(50L, 100L));
+ ImmutableList.of(50L, 100L),
+ COLUMN_FILES);
ManifestFile manifest = writeManifest(format, ID_PARTITIONED_TYPE, file);
@@ -269,7 +289,8 @@ class TestV4ManifestReader {
null, // dv
MANIFEST_INFO,
ByteBuffer.wrap(new byte[] {1, 2, 3}), // key metadata
- null); // split offsets
+ null, // split offsets
+ COLUMN_FILES);
ManifestFile manifest = writeManifest(format, ID_PARTITIONED_TYPE,
manifestRef);
@@ -567,6 +588,42 @@ class TestV4ManifestReader {
.isEqualTo(expected);
}
+ @ParameterizedTest
+ @FieldSource("MANIFEST_FORMATS")
+ public void inheritanceModifiedDataSequenceNumber(FileFormat format) throws
IOException {
+ TrackedFile withoutDataSeq =
+ dataFileWithColumnFiles(
+ new TrackingStruct(
+ EntryStatus.MODIFIED, SNAPSHOT_ID, null, 3L, 34L, 5_000L,
null, null),
+ "s3://bucket/table/file-a.parquet",
+ COLUMN_FILES);
+
+ ManifestFile manifest =
+ writeManifest(format, UNPARTITIONED_TYPE,
ImmutableList.of(withoutDataSeq));
+
+ when(manifest.firstRowId()).thenReturn(10_000L);
+ when(manifest.snapshotId()).thenReturn(34L);
+ when(manifest.sequenceNumber()).thenReturn(5L);
+
+ V4ManifestReader.Builder builder =
+ V4ManifestReader.builder(manifest, IO, TABLE_SCHEMA,
ID_PARTITIONING_SPECS)
+ .metricsConfig(METRICS_CONFIG);
+ List<TrackedFile> actual = read(builder);
+
+ // inherits data sequence number
+ List<TrackedFile> expected =
+ List.of(
+ dataFileWithColumnFiles(
+ new TrackingStruct(
+ EntryStatus.MODIFIED, SNAPSHOT_ID, 5L, 3L, 34L, 5_000L,
null, null),
+ "s3://bucket/table/file-a.parquet",
+ COLUMN_FILES));
+
+ assertThat(actual)
+ .usingComparatorForType(FILE_AND_TRACKING_COMPARATOR,
TrackedFile.class)
+ .isEqualTo(expected);
+ }
+
@ParameterizedTest
@FieldSource("MANIFEST_FORMATS")
public void inheritanceFirstRowId(FileFormat format) throws IOException {
@@ -1128,7 +1185,8 @@ class TestV4ManifestReader {
null,
null,
null,
- List.of(4L));
+ List.of(4L),
+ null); // column files
ManifestFile manifest =
writeManifest(format, UNPARTITIONED_TYPE,
ImmutableList.of(emptyTrackedFile, FILE_D));
@@ -1163,7 +1221,8 @@ class TestV4ManifestReader {
null,
null,
null,
- List.of(4L));
+ List.of(4L),
+ null); // column files
ManifestFile manifest =
writeManifest(format, UNPARTITIONED_TYPE,
ImmutableList.of(invalidRecordCountFile, FILE_D));
@@ -1261,6 +1320,33 @@ class TestV4ManifestReader {
assertThat(actualFiles).extracting(TrackedFile::location).containsExactly(FILE_D.location());
}
+ @ParameterizedTest
+ @FieldSource("MANIFEST_FORMATS")
+ public void statsFilterWithColumnFilesProjection(FileFormat format) throws
IOException {
+ // FILE_C has stats {id in [0, 99], data in [a, z]}, the file with column
files has no stats
+ TrackedFile fileWithColumnFiles =
+ dataFileWithColumnFiles("s3://bucket/table/file-e.parquet",
COLUMN_FILES);
+ ManifestFile manifest =
+ writeManifest(format, UNPARTITIONED_TYPE, ImmutableList.of(FILE_C,
fileWithColumnFiles));
+
+ V4ManifestReader.Builder builder =
+ V4ManifestReader.builder(manifest, IO, TABLE_SCHEMA,
UNPARTITIONED_SPECS)
+ .filter(Expressions.equal("id", 105)) // eliminates FILE_C
+ .metricsConfig(METRICS_CONFIG)
+ .select("location", "column_files.field_ids");
+
+ TrackedFile actual = readOne(builder);
+
+ assertThat(actual.location()).isEqualTo(fileWithColumnFiles.location());
+ assertThat(actual.columnFiles())
+ .extracting(ColumnFile::fieldIds)
+ .containsExactly(COLUMN_FILE_1.fieldIds(), COLUMN_FILE_2.fieldIds());
+ assertThat(actual.columnFiles())
+ .as("Unselected column file fields should not be projected")
+ .extracting(ColumnFile::location, ColumnFile::fileFormat,
ColumnFile::keyMetadata)
+ .containsOnly(tuple(null, null, null));
+ }
+
@ParameterizedTest
@FieldSource("MANIFEST_FORMATS")
public void statsFilterCaseSensitivity(FileFormat format) throws IOException
{
@@ -1346,7 +1432,8 @@ class TestV4ManifestReader {
null, // dv
null, // manifest info
null, // key metadata
- List.of(4L));
+ List.of(4L),
+ null); // column files
ManifestFile manifest =
writeManifest(
@@ -1743,6 +1830,44 @@ class TestV4ManifestReader {
"s3://other/abs.parquet",
"s3://bucket/db/table/data/dv.puffin"));
}
+ @ParameterizedTest
+ @FieldSource("MANIFEST_FORMATS")
+ public void resolutionRelativeColumnFileLocation(FileFormat format) throws
IOException {
+ ManifestFile manifest =
+ writeManifest(
+ format,
+ UNPARTITIONED_TYPE,
+ ImmutableList.of(
+ dataFileWithColumnFiles(
+ "s3://other/abs.parquet",
+ List.of(columnFile("data/column.parquet", Set.of(1)),
COLUMN_FILE_2)),
+ manifestRefWithColumnFiles(
+ FileContent.DATA_MANIFEST,
+ "s3://other/leaf.avro",
+ MANIFEST_INFO,
+ List.of(columnFile("metadata/leaf-column.parquet",
Set.of(1))))));
+
+ V4ManifestReader.Builder builder =
+ V4ManifestReader.builder(manifest, IO, TABLE_SCHEMA,
UNPARTITIONED_SPECS)
+ .tableLocation("s3://bucket/db/table")
+ .metricsConfig(METRICS_CONFIG);
+
+ assertThat(read(builder))
+ .usingComparatorForType(FILE_COMPARATOR, TrackedFile.class)
+ .containsExactly(
+ dataFileWithColumnFiles(
+ "s3://other/abs.parquet",
+ List.of(
+ columnFile("s3://bucket/db/table/data/column.parquet",
Set.of(1)),
+ COLUMN_FILE_2)),
+ manifestRefWithColumnFiles(
+ FileContent.DATA_MANIFEST,
+ "s3://other/leaf.avro",
+ MANIFEST_INFO,
+ List.of(
+
columnFile("s3://bucket/db/table/metadata/leaf-column.parquet", Set.of(1)))));
+ }
+
@ParameterizedTest
@FieldSource("MANIFEST_FORMATS")
public void resolutionResolvesOnlyRelativePaths(FileFormat format) throws
IOException {
@@ -1892,7 +2017,7 @@ class TestV4ManifestReader {
SNAPSHOT_ID,
3L, // data sequence number
3L, // file sequence number
- null, // dv snapshot id
+ null, // modified snapshot id
null, // first row id
null, // deleted positions
null); // replaced positions
diff --git a/core/src/test/java/org/apache/iceberg/V4TestComparators.java
b/core/src/test/java/org/apache/iceberg/V4TestComparators.java
index 85ac12138c..5c999fcd7d 100644
--- a/core/src/test/java/org/apache/iceberg/V4TestComparators.java
+++ b/core/src/test/java/org/apache/iceberg/V4TestComparators.java
@@ -24,6 +24,7 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.apache.iceberg.relocated.com.google.common.collect.Maps;
import org.apache.iceberg.relocated.com.google.common.collect.Sets;
import org.apache.iceberg.types.Comparators;
@@ -66,7 +67,8 @@ class V4TestComparators {
.thenComparing(TrackedFile::deletionVector, DELETION_VECTOR)
.thenComparing(TrackedFile::manifestInfo, MANIFEST_INFO)
.thenComparing(TrackedFile::keyMetadata, BYTES)
- .thenComparing(TrackedFile::splitOffsets, SPLIT_OFFSETS));
+ .thenComparing(TrackedFile::splitOffsets, SPLIT_OFFSETS)
+ .thenComparing(TrackedFile::columnFiles, COLUMN_FILES));
}
// convenience method for a null-safe natural order comparator
@@ -81,6 +83,22 @@ class V4TestComparators {
private static final Comparator<List<Long>> SPLIT_OFFSETS =
Comparator.nullsFirst(Comparators.forType(TrackedFile.SPLIT_OFFSETS.type().asListType()));
+ // field IDs are a set, so compare them in sorted order to ignore iteration
order
+ private static final Comparator<Set<Integer>> FIELD_IDS =
+ Comparator.nullsFirst(
+ Comparator.comparing(
+ ids -> Lists.newArrayList(Sets.newTreeSet(ids)),
+ Comparators.forType(ColumnFile.FIELD_IDS.type().asListType())));
+ private static final Comparator<ColumnFile> COLUMN_FILE =
+ Comparator.nullsFirst(
+ Comparator.comparing(ColumnFile::location, natural())
+ .thenComparing(ColumnFile::fieldIds, FIELD_IDS)
+ .thenComparing(ColumnFile::fileFormat, natural())
+ .thenComparingLong(ColumnFile::fileSizeInBytes)
+ .thenComparing(ColumnFile::keyMetadata, BYTES));
+ private static final Comparator<List<ColumnFile>> COLUMN_FILES =
+ Comparator.nullsFirst(columnFilesComparator());
+
static final Comparator<Tracking> TRACKING =
Comparator.nullsFirst(
Comparator.comparing(Tracking::status, natural())
@@ -145,6 +163,24 @@ class V4TestComparators {
}
}
+ private static Comparator<List<ColumnFile>> columnFilesComparator() {
+ return (left, right) -> {
+ int sizeCmp = Integer.compare(left.size(), right.size());
+ if (sizeCmp != 0) {
+ return sizeCmp;
+ }
+
+ for (int i = 0; i < left.size(); i += 1) {
+ int cmp = COLUMN_FILE.compare(left.get(i), right.get(i));
+ if (cmp != 0) {
+ return cmp;
+ }
+ }
+
+ return 0;
+ };
+ }
+
private static Map<Integer, FieldStats<?>> statsById(ContentStats stats) {
Map<Integer, FieldStats<?>> statsById = Maps.newHashMap();
for (FieldStats<?> fieldStats : stats.fieldStats()) {
diff --git a/core/src/test/java/org/apache/iceberg/V4TestHelpers.java
b/core/src/test/java/org/apache/iceberg/V4TestHelpers.java
index b8cf5164d4..9829f3e679 100644
--- a/core/src/test/java/org/apache/iceberg/V4TestHelpers.java
+++ b/core/src/test/java/org/apache/iceberg/V4TestHelpers.java
@@ -19,6 +19,9 @@
package org.apache.iceberg;
import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.List;
+import java.util.Set;
import org.apache.iceberg.io.FileAppender;
import org.apache.iceberg.io.FileIO;
import org.apache.iceberg.io.OutputFile;
@@ -35,11 +38,11 @@ class V4TestHelpers {
static final Tracking ADDED_TRACKING =
TrackingBuilder.added(SNAPSHOT_ID).build();
static TrackedFile dataFile(Tracking tracking, String location) {
- return dataFile(tracking, location, null, null, null, null);
+ return dataFile(tracking, location, null, null, null, null, null);
}
static TrackedFile dataFile(String location, Integer specId, PartitionData
partition) {
- return dataFile(ADDED_TRACKING, location, specId, partition, null, null);
+ return dataFile(ADDED_TRACKING, location, specId, partition, null, null,
null);
}
static TrackedFile dataFile(
@@ -48,15 +51,24 @@ class V4TestHelpers {
PartitionData partition,
ContentStats stats,
DeletionVector dv) {
- return dataFile(ADDED_TRACKING, location, specId, partition, stats, dv);
+ return dataFile(ADDED_TRACKING, location, specId, partition, stats, dv,
null);
}
static TrackedFile dataFileWithStats(String location, ContentStats stats) {
- return dataFile(ADDED_TRACKING, location, null, null, stats, null);
+ return dataFile(ADDED_TRACKING, location, null, null, stats, null, null);
}
static TrackedFile dataFileWithDV(String location, DeletionVector dv) {
- return dataFile(ADDED_TRACKING, location, null, null, null, dv);
+ return dataFile(ADDED_TRACKING, location, null, null, null, dv, null);
+ }
+
+ static TrackedFile dataFileWithColumnFiles(String location, List<ColumnFile>
columnFiles) {
+ return dataFile(ADDED_TRACKING, location, null, null, null, null,
columnFiles);
+ }
+
+ static TrackedFile dataFileWithColumnFiles(
+ Tracking tracking, String location, List<ColumnFile> columnFiles) {
+ return dataFile(tracking, location, null, null, null, null, columnFiles);
}
private static TrackedFile dataFile(
@@ -65,7 +77,8 @@ class V4TestHelpers {
Integer specId,
PartitionData partition,
ContentStats stats,
- DeletionVector dv) {
+ DeletionVector dv,
+ List<ColumnFile> columnFiles) {
return trackedFile(
tracking,
FileContent.DATA,
@@ -74,7 +87,8 @@ class V4TestHelpers {
partition,
stats,
dv,
- null /* manifestInfo */);
+ null /* manifestInfo */,
+ columnFiles);
}
static TrackedFile manifestRef(FileContent content, String location,
ManifestInfo manifestInfo) {
@@ -91,7 +105,25 @@ class V4TestHelpers {
null, // partition
stats,
null, // dv
- manifestInfo);
+ manifestInfo,
+ null); // columnFiles
+ }
+
+ static TrackedFile manifestRefWithColumnFiles(
+ FileContent content,
+ String location,
+ ManifestInfo manifestInfo,
+ List<ColumnFile> columnFiles) {
+ return trackedFile(
+ ADDED_TRACKING,
+ content,
+ location,
+ null, // specId
+ null, // partition
+ null, // stats
+ null, // dv
+ manifestInfo,
+ columnFiles);
}
private static TrackedFile trackedFile(
@@ -102,7 +134,8 @@ class V4TestHelpers {
PartitionData partition,
ContentStats stats,
DeletionVector dv,
- ManifestInfo manifestInfo) {
+ ManifestInfo manifestInfo,
+ List<ColumnFile> columnFiles) {
return new TrackedFileStruct(
tracking,
content,
@@ -117,7 +150,8 @@ class V4TestHelpers {
dv,
manifestInfo,
null, // keyMetadata
- null); // splitOffsets
+ null, // splitOffsets
+ columnFiles);
}
static DeletionVector deletionVector(String location) {
@@ -134,6 +168,15 @@ class V4TestHelpers {
.build();
}
+ static ColumnFile columnFile(String location, Set<Integer> fieldIds) {
+ return columnFile(location, fieldIds, null);
+ }
+
+ static ColumnFile columnFile(String location, Set<Integer> fieldIds,
ByteBuffer keyMetadata) {
+ return new ColumnFileStruct(
+ location, fieldIds, FileFormat.fromFileName(location),
FILE_SIZE_IN_BYTES, keyMetadata);
+ }
+
static PartitionData partition(PartitionSpec spec, Object... values) {
PartitionData data = new PartitionData(spec.partitionType());
for (int i = 0; i < values.length; i += 1) {
