laskoviymishka commented on code in PR #16285:
URL: https://github.com/apache/iceberg/pull/16285#discussion_r4075323746
##########
core/src/main/java/org/apache/iceberg/TrackedFileStruct.java:
##########
@@ -313,6 +337,7 @@ protected <T> void internalSet(int pos, T value) {
case 13 -> this.keyMetadata = ByteBuffers.toByteArray((ByteBuffer)
value);
case 14 -> this.splitOffsets = ArrayUtil.toLongArray((List<Long>) value);
case 15 -> this.equalityIds = ArrayUtil.toIntArray((List<Integer>)
value);
+ case 16 -> this.columnFiles = (List<ColumnFile>) value;
Review Comment:
Every other collection field in this class copies on the way in — the
constructor does `Lists.newArrayList(columnFiles)`, and
`splitOffsets`/`equalityIds` go through `ArrayUtil.toLongArray`/`toIntArray`,
which always allocate. `internalSet` case 16 stores the caller's `List` by
reference, and `columnFiles()` hands back an unmodifiable view of it, so if a
reader reuses or later mutates that list the struct changes underneath callers.
It's inert today since nothing calls this with a non-null value yet, but I'd
copy here to match the constructor: `this.columnFiles = value == null ? null :
Lists.newArrayList((List<ColumnFile>) value);`
##########
core/src/main/java/org/apache/iceberg/ColumnFileStruct.java:
##########
@@ -0,0 +1,271 @@
+/*
+ * 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 org.apache.iceberg.avro.SupportsIndexProjection;
+import org.apache.iceberg.relocated.com.google.common.base.MoreObjects;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+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.FORMAT_VERSION,
+ ColumnFile.FIELD_IDS,
+ ColumnFile.LOCATION,
+ ColumnFile.FILE_FORMAT,
+ ColumnFile.FILE_SIZE_IN_BYTES,
+ ColumnFile.KEY_METADATA,
+ ColumnFile.SPLIT_OFFSETS);
+
+ private int formatVersion = -1;
+ private int[] fieldIds = null;
+ private String location = null;
+ private FileFormat fileFormat = null;
+ private long fileSizeInBytes = -1L;
+ private byte[] keyMetadata = null;
+ private long[] splitOffsets = null;
+
+ /** Used by internal readers to instantiate this class with a projection
schema. */
+ ColumnFileStruct(Types.StructType projection) {
+ super(BASE_TYPE, projection);
+ }
+
+ ColumnFileStruct(
+ int formatVersion,
+ List<Integer> fieldIds,
+ String location,
+ FileFormat fileFormat,
+ long fileSizeInBytes,
+ ByteBuffer keyMetadata,
+ List<Long> splitOffsets) {
+ super(BASE_TYPE.fields().size());
+ this.formatVersion = formatVersion;
+ this.fieldIds = ArrayUtil.toIntArray(fieldIds);
+ this.location = location;
+ this.fileFormat = fileFormat;
+ this.fileSizeInBytes = fileSizeInBytes;
+ this.keyMetadata = ByteBuffers.toByteArray(keyMetadata);
+ this.splitOffsets = ArrayUtil.toLongArray(splitOffsets);
+ }
+
+ /** Copy constructor. */
+ private ColumnFileStruct(ColumnFileStruct toCopy) {
+ super(toCopy);
+ this.formatVersion = toCopy.formatVersion;
+ this.fieldIds =
+ toCopy.fieldIds != null ? Arrays.copyOf(toCopy.fieldIds,
toCopy.fieldIds.length) : null;
+ this.location = toCopy.location;
+ this.fileFormat = toCopy.fileFormat;
+ this.fileSizeInBytes = toCopy.fileSizeInBytes;
+ this.keyMetadata =
+ toCopy.keyMetadata != null
+ ? Arrays.copyOf(toCopy.keyMetadata, toCopy.keyMetadata.length)
+ : null;
+ this.splitOffsets =
+ toCopy.splitOffsets != null
+ ? Arrays.copyOf(toCopy.splitOffsets, toCopy.splitOffsets.length)
+ : null;
+ }
+
+ /** Constructor for Java serialization. */
+ ColumnFileStruct() {
+ super(BASE_TYPE.fields().size());
+ }
+
+ @Override
+ public int formatVersion() {
+ return formatVersion;
+ }
+
+ @Override
+ public List<Integer> fieldIds() {
+ return fieldIds != null ? ArrayUtil.toUnmodifiableIntList(fieldIds) : null;
+ }
+
+ @Override
+ public String location() {
+ return location;
+ }
+
+ @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 List<Long> splitOffsets() {
+ return splitOffsets != null ?
ArrayUtil.toUnmodifiableLongList(splitOffsets) : 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 -> formatVersion;
+ case 1 -> fieldIds();
+ case 2 -> location;
+ case 3 -> fileFormat != null ? fileFormat.toString() : null;
+ case 4 -> fileSizeInBytes;
+ case 5 -> keyMetadata();
+ case 6 -> splitOffsets();
+ default -> throw new UnsupportedOperationException("Unknown field
ordinal: " + pos);
+ };
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ protected <T> void internalSet(int pos, T value) {
+ switch (pos) {
+ case 0 -> this.formatVersion = (int) value;
+ case 1 -> this.fieldIds = ArrayUtil.toIntArray((List<Integer>) value);
+ // always coerce to String for Serializable
+ case 2 -> this.location = value.toString();
+ case 3 -> this.fileFormat = FileFormat.fromString(value.toString());
+ case 4 -> this.fileSizeInBytes = (long) value;
+ case 5 -> this.keyMetadata = ByteBuffers.toByteArray((ByteBuffer) value);
+ case 6 -> this.splitOffsets = ArrayUtil.toLongArray((List<Long>) value);
+ default -> {
+ // ignore the object, it must be from a newer version of the format
+ }
+ }
+ }
+
+ static Builder builder() {
+ return new Builder();
+ }
+
+ @Override
+ public String toString() {
+ return MoreObjects.toStringHelper(this)
+ .add("format_version", formatVersion)
+ .add("field_ids", fieldIds)
+ .add("location", location)
+ .add("file_format", fileFormat)
+ .add("file_size_in_bytes", fileSizeInBytes)
+ .add("key_metadata", keyMetadata == null ? "null" : "(redacted)")
+ .add("split_offsets", splitOffsets == null ? "null" : splitOffsets())
+ .toString();
+ }
+
+ static class Builder {
+ private Integer formatVersion = null;
+ private List<Integer> fieldIds = null;
+ private String location = null;
+ private FileFormat fileFormat = null;
+ private Long fileSizeInBytes = null;
+ private ByteBuffer keyMetadata = null;
+ private List<Long> splitOffsets = null;
+
+ Builder formatVersion(int newFormatVersion) {
+ Preconditions.checkArgument(
+ newFormatVersion >= 0, "Invalid format version: %s (must be >= 0)",
newFormatVersion);
+ this.formatVersion = newFormatVersion;
+ return this;
+ }
+
+ Builder fieldIds(List<Integer> newFieldIds) {
+ Preconditions.checkArgument(newFieldIds != null, "Invalid field IDs:
null");
+ Preconditions.checkArgument(!newFieldIds.isEmpty(), "Invalid field IDs:
empty");
+ Preconditions.checkArgument(
+ Sets.newHashSet(newFieldIds).size() == newFieldIds.size(),
+ "Invalid field IDs: duplicated IDs found in: %s",
+ newFieldIds);
+ this.fieldIds = newFieldIds;
+ return this;
+ }
+
+ Builder location(String newLocation) {
+ Preconditions.checkArgument(newLocation != null, "Invalid location:
null");
+ Preconditions.checkArgument(!newLocation.isEmpty(), "Invalid location:
empty");
+ this.location = newLocation;
+ return this;
+ }
+
+ Builder fileFormat(FileFormat newFileFormat) {
+ Preconditions.checkArgument(newFileFormat != null, "Invalid file format:
null");
+ this.fileFormat = newFileFormat;
+ return this;
+ }
+
+ Builder fileSizeInBytes(long newFileSizeInBytes) {
+ Preconditions.checkArgument(
+ newFileSizeInBytes >= 0,
+ "Invalid file size in bytes: %s (must be >= 0)",
+ newFileSizeInBytes);
+ this.fileSizeInBytes = newFileSizeInBytes;
+ return this;
+ }
+
+ Builder keyMetadata(ByteBuffer newKeyMetadata) {
+ Preconditions.checkArgument(newKeyMetadata != null, "Invalid key
metadata: null");
+ this.keyMetadata = newKeyMetadata;
+ return this;
+ }
+
+ Builder splitOffsets(List<Long> newSplitOffsets) {
+ Preconditions.checkArgument(newSplitOffsets != null, "Invalid split
offsets: null");
+ this.splitOffsets = newSplitOffsets;
+ return this;
+ }
+
+ ColumnFile build() {
+ Preconditions.checkArgument(formatVersion != null, "Missing required
value: formatVersion");
Review Comment:
These messages use the raw field identifiers (`formatVersion`, `fieldIds`,
`fileSizeInBytes`), but the sibling builders and even the "Invalid ..."
messages in this same file use spaced-out phrasing (`file size in bytes`).
Small thing, but the tests lock the wording in as contract — I'd switch to
`format version` / `field IDs` / `file size in bytes` to match.
##########
core/src/main/java/org/apache/iceberg/ColumnFile.java:
##########
@@ -0,0 +1,94 @@
+/*
+ * 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.List;
+import org.apache.iceberg.types.Types;
+
+interface ColumnFile {
+ Types.NestedField FORMAT_VERSION =
+ Types.NestedField.required(
+ 161, "format_version", Types.IntegerType.get(), "Format version of
this column file");
+ Types.NestedField FIELD_IDS =
+ Types.NestedField.required(
+ 162,
+ "field_ids",
+ Types.ListType.ofRequired(163, Types.IntegerType.get()),
+ "Live field IDs in this column file");
+ Types.NestedField LOCATION =
+ Types.NestedField.required(
+ 164, "location", Types.StringType.get(), "Location of the column
file");
+ Types.NestedField FILE_FORMAT =
+ Types.NestedField.required(
+ 165,
+ "file_format",
+ Types.StringType.get(),
+ "String file format name for this column file");
+ Types.NestedField FILE_SIZE_IN_BYTES =
+ Types.NestedField.required(
+ 166, "file_size_in_bytes", Types.LongType.get(), "Total column file
size in bytes");
+ Types.NestedField KEY_METADATA =
+ Types.NestedField.optional(
+ 167,
+ "key_metadata",
+ Types.BinaryType.get(),
+ "Implementation-specific key metadata for encryption");
+ Types.NestedField SPLIT_OFFSETS =
+ Types.NestedField.optional(
+ 168,
+ "split_offsets",
+ Types.ListType.ofRequired(169, Types.LongType.get()),
+ "Split offsets for the data file");
Review Comment:
"Split offsets for the data file" looks copy-pasted from `TrackedFile` —
this is a column file, not a data file. Minor, but since it's the doc string
that ships on the schema field: "Split offsets for the column file".
##########
core/src/test/java/org/apache/iceberg/TestV4ManifestReader.java:
##########
@@ -167,7 +167,8 @@ public void readDataFile(FileFormat format) throws
IOException {
null, // manifest info
ByteBuffer.wrap(new byte[] {1, 2, 3}), // key metadata
ImmutableList.of(50L, 100L),
- null); // equality field IDs
+ null, // equality field IDs
+ null); // column files
Review Comment:
Every touched call site here passes null for `columnFiles`, so nothing
actually round-trips the new field through a real manifest write/read. This is
the one suite that exercises the true Avro writer plus `V4ManifestReader`, and
`column_files` is the trickiest shape in the schema — an optional list of a
struct that itself holds an optional binary and an optional list-of-long.
`TestColumnFileStruct.serializationRoundTrip` only covers Java/Kryo, which
never touches the Avro conversion path. Since laying down the persisted schema
is the whole point of this PR, I'd add at least one case that writes a
`TrackedFile` with a populated `column_files` list and reads it back here,
including a projected read. Otherwise we won't know the schema actually
persists until a later PR wires up the writer.
##########
core/src/main/java/org/apache/iceberg/TrackedFileStruct.java:
##########
@@ -167,6 +174,17 @@ private TrackedFileStruct(TrackedFileStruct toCopy,
Set<Integer> statsIds) {
toCopy.equalityIds != null
? Arrays.copyOf(toCopy.equalityIds, toCopy.equalityIds.length)
: null;
+
+ if (toCopy.columnFiles != null) {
+ this.columnFiles =
Lists.newArrayListWithCapacity(toCopy.columnFiles.size());
+ for (ColumnFile columnFile : toCopy.columnFiles) {
+ if (columnFile != null) {
Review Comment:
The `if (columnFile != null)` guard means a list with a null element comes
back shorter after `copy()` — the clone silently has fewer elements than the
original, which breaks the size/element correspondence every other field in
this constructor preserves.
`column_files` is `ofRequired` so nulls shouldn't happen, but the explicit
guard suggests some doubt, and a silent size change is a worse failure than a
fast NPE. I'd drop the guard and add unconditionally
(`this.columnFiles.add(columnFile != null ? columnFile.copy() : null)`),
matching the ternary the sibling fields use.
##########
core/src/main/java/org/apache/iceberg/TrackingBuilder.java:
##########
@@ -115,6 +118,19 @@ TrackingBuilder dvUpdated() {
return this;
}
+ /** Indicates that the column files list has been updated for the new
Tracking. */
+ TrackingBuilder columnFilesUpdated() {
+ this.latestColumnFileSnapshotId = 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;
Review Comment:
I'd hold on this one before it lands, since everything downstream inherits
the semantics. `columnFilesUpdated()` resets `dataSequenceNumber` to null, and
the `inheritFrom()` change (TrackingStruct.java:138) now lets MODIFIED entries
re-inherit it from the manifest's file sequence number — so attaching a column
file bumps the entry's data sequence number forward even though the row content
didn't change.
The way I look at it, a column file is a positionally-aligned overlay of a
subset of columns onto the base file — the rows and their identity don't
change, so nothing that scopes delete applicability should move with it. But
`sequence_number` (field 3) is exactly that scoping key: it's been
immutable-after-add since v2, and `DeleteFileIndex` keys off it directly
(`applySequenceNumber = dataSequenceNumber() - 1`). If a file's data sequence
number moves forward as a side effect of a column-file update, any delete whose
sequence number falls between the old and new value stops being applied, and
previously-deleted rows can resurface.
The comment says it's safe because writers must rewrite v2 position/equality
deletes to DVs on column update, but nothing here enforces that —
`columnFilesUpdated()` is only reachable from its own unit tests today. I'd
either not touch `dataSequenceNumber` for MODIFIED at all, or, if that rewrite
precondition really is load-bearing, enforce it at this call site and consider
a dedicated v4 field rather than overloading the spec-locked one. I think this
wants settling here, ideally with a companion spec note, before the writer PRs
build on it. wdyt?
--
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]
