Core: Basic fields and schemas for column files

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();
+}

core/src/main/java/org/apache/iceberg/ColumnFileStruct.java

@@ -0,0 +1,160 @@
+/*
+ * 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;
+  }
+
+  @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();
+  }
+}

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 @@ PARTITION_ID, PARTITION_NAME, typeOrUnknown(partitionType), PARTITION_DOC),
         DELETION_VECTOR,
         MANIFEST_INFO,
         KEY_METADATA,
-        SPLIT_OFFSETS);
+        SPLIT_OFFSETS,
+        COLUMN_FILES);
   }
 
   private static Type typeOrUnknown(Types.StructType structType) {
@@ -178,6 +182,9 @@ static Schema readSchema(Types.StructType partitionType, Types.StructType conten
   /** 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();
 

core/src/main/java/org/apache/iceberg/TrackedFileAdapters.java

@@ -648,6 +648,11 @@ public List<Long> splitOffsets() {
       return file.splitOffsets();
     }
 
+    @Override
+    public List<ColumnFile> columnFiles() {
+      return null;
+    }
+
     @Override
     public TrackedFile copy() {
       throw new UnsupportedOperationException("copy is not implemented");
@@ -770,6 +775,11 @@ public List<Long> splitOffsets() {
       return null;
     }
 
+    @Override
+    public List<ColumnFile> columnFiles() {
+      return null;
+    }
+
     @Override
     public TrackedFile copy() {
       throw new UnsupportedOperationException("copy is not implemented");

core/src/main/java/org/apache/iceberg/TrackedFileStruct.java

@@ -21,11 +21,13 @@
 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 @@ private TrackedFileStruct(TrackedFileStruct toCopy, Set<Integer> statsIds) {
         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 @@ public List<Long> splitOffsets() {
     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 @@ private Object getByPos(int pos) {
       case 11 -> manifestInfo;
       case 12 -> keyMetadata();
       case 13 -> splitOffsets();
+      case 14 -> columnFiles();
       default -> throw new UnsupportedOperationException("Unknown field ordinal: " + pos);
     };
   }
@@ -298,6 +319,7 @@ protected <T> void internalSet(int pos, T value) {
       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 @@ public String toString() {
         .add("manifest_info", manifestInfo)
         .add("key_metadata", keyMetadata == null ? "null" : "(redacted)")
         .add("split_offsets", splitOffsets == null ? "null" : splitOffsets())
+        .add("column_files", columnFiles)
         .toString();
   }
 }