Core: Use avro compression properties from table properties when writing manifests and manifest lists (API change)

.palantir/revapi.yml

@@ -802,10 +802,6 @@ acceptedBreaks:
       old: "method org.apache.iceberg.view.ViewBuilder org.apache.iceberg.view.ViewBuilder::withQueryColumnNames(java.util.List<java.lang.String>)"
       justification: "Acceptable break due to updating View APIs and the View Spec"
     org.apache.iceberg:iceberg-core:
-    - code: "java.method.visibilityReduced"
-      old: "method void org.apache.iceberg.encryption.Ciphers::<init>()"
-      new: "method void org.apache.iceberg.encryption.Ciphers::<init>()"
-      justification: "Static utility class - should not have public constructor"
     - code: "java.class.removed"
       old: "class org.apache.iceberg.actions.BaseDeleteOrphanFilesActionResult"
       justification: "Removing deprecated code"
@@ -833,6 +829,13 @@ acceptedBreaks:
     - code: "java.class.removed"
       old: "interface org.apache.iceberg.actions.RewritePositionDeleteStrategy"
       justification: "Removing deprecated code"
+    - code: "java.method.abstractMethodAdded"
+      new: "method org.apache.iceberg.io.FileAppender<org.apache.iceberg.ManifestEntry<F>>\
+        \ org.apache.iceberg.ManifestWriter<F extends org.apache.iceberg.ContentFile<F\
+        \ extends org.apache.iceberg.ContentFile<F>>>::newAppender(org.apache.iceberg.PartitionSpec,\
+        \ org.apache.iceberg.io.OutputFile, java.lang.String, java.lang.Integer)"
+      justification: "Allow adding a new method to the abstract class - old method\
+        \ is deprecated"
     - code: "java.method.removed"
       old: "method java.util.List<org.apache.iceberg.DataFile> org.apache.iceberg.MergingSnapshotProducer<ThisT>::addedFiles()\
         \ @ org.apache.iceberg.BaseOverwriteFiles"
@@ -857,6 +860,10 @@ acceptedBreaks:
       old: "method void org.apache.iceberg.MergingSnapshotProducer<ThisT>::setNewFilesSequenceNumber(long)\
         \ @ org.apache.iceberg.StreamingDelete"
       justification: "Removing deprecated code"
+    - code: "java.method.visibilityReduced"
+      old: "method void org.apache.iceberg.encryption.Ciphers::<init>()"
+      new: "method void org.apache.iceberg.encryption.Ciphers::<init>()"
+      justification: "Static utility class - should not have public constructor"
   apache-iceberg-0.14.0:
     org.apache.iceberg:iceberg-api:
     - code: "java.class.defaultSerializationChanged"

build.gradle

@@ -358,6 +358,7 @@ project(':iceberg-core') {
     testImplementation libs.esotericsoftware.kryo
     testImplementation libs.guava.testlib
     testImplementation libs.awaitility
+    testRuntimeOnly libs.zstd.jni
   }
 }
 

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

@@ -170,7 +170,9 @@ private ManifestFile copyManifest(ManifestFile manifest) {
         specsById,
         newFile,
         snapshotId(),
-        summaryBuilder);
+        summaryBuilder,
+        current.properties().get(TableProperties.AVRO_COMPRESSION),
+        current.propertyAsNullableInt(TableProperties.AVRO_COMPRESSION_LEVEL));
   }
 
   @Override

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

@@ -138,7 +138,9 @@ private ManifestFile copyManifest(ManifestFile manifest) {
         current.specsById(),
         newManifestPath,
         snapshotId(),
-        summaryBuilder);
+        summaryBuilder,
+        current.properties().get(TableProperties.AVRO_COMPRESSION),
+        current.propertyAsNullableInt(TableProperties.AVRO_COMPRESSION_LEVEL));
   }
 
   @Override

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

@@ -157,11 +157,34 @@ public static ManifestWriter<DataFile> write(PartitionSpec spec, OutputFile outp
    */
   public static ManifestWriter<DataFile> write(
       int formatVersion, PartitionSpec spec, OutputFile outputFile, Long snapshotId) {
+    return write(formatVersion, spec, outputFile, snapshotId, null, null);
+  }
+
+  /**
+   * Create a new {@link ManifestWriter} for the given format version.
+   *
+   * @param formatVersion a target format version
+   * @param spec a {@link PartitionSpec}
+   * @param outputFile an {@link OutputFile} where the manifest will be written
+   * @param snapshotId a snapshot ID for the manifest entries, or null for an inherited ID
+   * @param compressionCodec compression codec for the manifest file
+   * @param compressionLevel compression level of the compressionCodec
+   * @return a manifest writer
+   */
+  public static ManifestWriter<DataFile> write(
+      int formatVersion,
+      PartitionSpec spec,
+      OutputFile outputFile,
+      Long snapshotId,
+      String compressionCodec,
+      Integer compressionLevel) {
     switch (formatVersion) {
       case 1:
-        return new ManifestWriter.V1Writer(spec, outputFile, snapshotId);
+        return new ManifestWriter.V1Writer(
+            spec, outputFile, snapshotId, compressionCodec, compressionLevel);
       case 2:
-        return new ManifestWriter.V2Writer(spec, outputFile, snapshotId);
+        return new ManifestWriter.V2Writer(
+            spec, outputFile, snapshotId, compressionCodec, compressionLevel);
     }
     throw new UnsupportedOperationException(
         "Cannot write manifest for table version: " + formatVersion);
@@ -198,11 +221,33 @@ public static ManifestReader<DeleteFile> readDeleteManifest(
    */
   public static ManifestWriter<DeleteFile> writeDeleteManifest(
       int formatVersion, PartitionSpec spec, OutputFile outputFile, Long snapshotId) {
+    return writeDeleteManifest(formatVersion, spec, outputFile, snapshotId, null, null);
+  }
+
+  /**
+   * Create a new {@link ManifestWriter} for the given format version.
+   *
+   * @param formatVersion a target format version
+   * @param spec a {@link PartitionSpec}
+   * @param outputFile an {@link OutputFile} where the manifest will be written
+   * @param snapshotId a snapshot ID for the manifest entries, or null for an inherited ID
+   * @param compressionCodec compression codec for the manifest file
+   * @param compressionLevel compression level of the compressionCodec
+   * @return a manifest writer
+   */
+  public static ManifestWriter<DeleteFile> writeDeleteManifest(
+      int formatVersion,
+      PartitionSpec spec,
+      OutputFile outputFile,
+      Long snapshotId,
+      String compressionCodec,
+      Integer compressionLevel) {
     switch (formatVersion) {
       case 1:
         throw new IllegalArgumentException("Cannot write delete files in a v1 table");
       case 2:
-        return new ManifestWriter.V2DeleteWriter(spec, outputFile, snapshotId);
+        return new ManifestWriter.V2DeleteWriter(
+            spec, outputFile, snapshotId, compressionCodec, compressionLevel);
     }
     throw new UnsupportedOperationException(
         "Cannot write manifest for table version: " + formatVersion);
@@ -256,7 +301,9 @@ static ManifestFile copyAppendManifest(
       Map<Integer, PartitionSpec> specsById,
       OutputFile outputFile,
       long snapshotId,
-      SnapshotSummary.Builder summaryBuilder) {
+      SnapshotSummary.Builder summaryBuilder,
+      String compressionCodec,
+      Integer compressionLevel) {
     // use metadata that will add the current snapshot's ID for the rewrite
     InheritableMetadata inheritableMetadata = InheritableMetadataFactory.forCopy(snapshotId);
     try (ManifestReader<DataFile> reader =
@@ -267,7 +314,9 @@ static ManifestFile copyAppendManifest(
           outputFile,
           snapshotId,
           summaryBuilder,
-          ManifestEntry.Status.ADDED);
+          ManifestEntry.Status.ADDED,
+          compressionCodec,
+          compressionLevel);
     } catch (IOException e) {
       throw new RuntimeIOException(e, "Failed to close manifest: %s", toCopy.location());
     }
@@ -280,7 +329,9 @@ static ManifestFile copyRewriteManifest(
       Map<Integer, PartitionSpec> specsById,
       OutputFile outputFile,
       long snapshotId,
-      SnapshotSummary.Builder summaryBuilder) {
+      SnapshotSummary.Builder summaryBuilder,
+      String compressionCodec,
+      Integer compressionLevel) {
     // for a rewritten manifest all snapshot ids should be set. use empty metadata to throw an
     // exception if it is not
     InheritableMetadata inheritableMetadata = InheritableMetadataFactory.empty();
@@ -292,7 +343,9 @@ static ManifestFile copyRewriteManifest(
           outputFile,
           snapshotId,
           summaryBuilder,
-          ManifestEntry.Status.EXISTING);
+          ManifestEntry.Status.EXISTING,
+          compressionCodec,
+          compressionLevel);
     } catch (IOException e) {
       throw new RuntimeIOException(e, "Failed to close manifest: %s", toCopy.location());
     }
@@ -305,8 +358,17 @@ private static ManifestFile copyManifestInternal(
       OutputFile outputFile,
       long snapshotId,
       SnapshotSummary.Builder summaryBuilder,
-      ManifestEntry.Status allowedEntryStatus) {
-    ManifestWriter<DataFile> writer = write(formatVersion, reader.spec(), outputFile, snapshotId);
+      ManifestEntry.Status allowedEntryStatus,
+      String compressionCodec,
+      Integer compressionLevel) {
+    ManifestWriter<DataFile> writer =
+        write(
+            formatVersion,
+            reader.spec(),
+            outputFile,
+            snapshotId,
+            compressionCodec,
+            compressionLevel);
     boolean threw = true;
     try {
       for (ManifestEntry<DataFile> entry : reader.entries()) {

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

@@ -31,14 +31,18 @@
 abstract class ManifestListWriter implements FileAppender<ManifestFile> {
   private final FileAppender<ManifestFile> writer;
 
-  private ManifestListWriter(OutputFile file, Map<String, String> meta) {
-    this.writer = newAppender(file, meta);
+  private ManifestListWriter(
+      OutputFile file,
+      Map<String, String> meta,
+      String compressionCodec,
+      Integer compressionLevel) {
+    this.writer = newAppender(file, meta, compressionCodec, compressionLevel);
   }
 
   protected abstract ManifestFile prepare(ManifestFile manifest);
 
   protected abstract FileAppender<ManifestFile> newAppender(
-      OutputFile file, Map<String, String> meta);
+      OutputFile file, Map<String, String> meta, String compressionCodec, Integer compressionLevel);
 
   @Override
   public void add(ManifestFile manifest) {
@@ -73,14 +77,22 @@ public long length() {
   static class V2Writer extends ManifestListWriter {
     private final V2Metadata.IndexedManifestFile wrapper;
 
-    V2Writer(OutputFile snapshotFile, long snapshotId, Long parentSnapshotId, long sequenceNumber) {
+    V2Writer(
+        OutputFile snapshotFile,
+        long snapshotId,
+        Long parentSnapshotId,
+        long sequenceNumber,
+        String compressionCodec,
+        Integer compressionLevel) {
       super(
           snapshotFile,
           ImmutableMap.of(
               "snapshot-id", String.valueOf(snapshotId),
               "parent-snapshot-id", String.valueOf(parentSnapshotId),
               "sequence-number", String.valueOf(sequenceNumber),
-              "format-version", "2"));
+              "format-version", "2"),
+          compressionCodec,
+          compressionLevel);
       this.wrapper = new V2Metadata.IndexedManifestFile(snapshotId, sequenceNumber);
     }
 
@@ -90,15 +102,28 @@ protected ManifestFile prepare(ManifestFile manifest) {
     }
 
     @Override
-    protected FileAppender<ManifestFile> newAppender(OutputFile file, Map<String, String> meta) {
+    protected FileAppender<ManifestFile> newAppender(
+        OutputFile file,
+        Map<String, String> meta,
+        String compressionCodec,
+        Integer compressionLevel) {
       try {
-        return Avro.write(file)
-            .schema(V2Metadata.MANIFEST_LIST_SCHEMA)
-            .named("manifest_file")
-            .meta(meta)
-            .overwrite()
-            .build();
-
+        Avro.WriteBuilder builder =
+            Avro.write(file)
+                .schema(V2Metadata.MANIFEST_LIST_SCHEMA)
+                .named("manifest_file")
+                .meta(meta)
+                .overwrite();
+
+        if (compressionCodec != null) {
+          builder.set(TableProperties.AVRO_COMPRESSION, compressionCodec);
+        }
+
+        if (compressionLevel != null) {
+          builder.set(TableProperties.AVRO_COMPRESSION_LEVEL, compressionLevel.toString());
+        }
+
+        return builder.build();
       } catch (IOException e) {
         throw new RuntimeIOException(e, "Failed to create snapshot list writer for path: %s", file);
       }
@@ -108,13 +133,20 @@ protected FileAppender<ManifestFile> newAppender(OutputFile file, Map<String, St
   static class V1Writer extends ManifestListWriter {
     private final V1Metadata.IndexedManifestFile wrapper = new V1Metadata.IndexedManifestFile();
 
-    V1Writer(OutputFile snapshotFile, long snapshotId, Long parentSnapshotId) {
+    V1Writer(
+        OutputFile snapshotFile,
+        long snapshotId,
+        Long parentSnapshotId,
+        String compressionCodec,
+        Integer compressionLevel) {
       super(
           snapshotFile,
           ImmutableMap.of(
               "snapshot-id", String.valueOf(snapshotId),
               "parent-snapshot-id", String.valueOf(parentSnapshotId),
-              "format-version", "1"));
+              "format-version", "1"),
+          compressionCodec,
+          compressionLevel);
     }
 
     @Override
@@ -126,15 +158,28 @@ protected ManifestFile prepare(ManifestFile manifest) {
     }
 
     @Override
-    protected FileAppender<ManifestFile> newAppender(OutputFile file, Map<String, String> meta) {
+    protected FileAppender<ManifestFile> newAppender(
+        OutputFile file,
+        Map<String, String> meta,
+        String compressionCodec,
+        Integer compressionLevel) {
       try {
-        return Avro.write(file)
-            .schema(V1Metadata.MANIFEST_LIST_SCHEMA)
-            .named("manifest_file")
-            .meta(meta)
-            .overwrite()
-            .build();
-
+        Avro.WriteBuilder builder =
+            Avro.write(file)
+                .schema(V1Metadata.MANIFEST_LIST_SCHEMA)
+                .named("manifest_file")
+                .meta(meta)
+                .overwrite();
+
+        if (compressionCodec != null) {
+          builder.set(TableProperties.AVRO_COMPRESSION, compressionCodec);
+        }
+
+        if (compressionLevel != null) {
+          builder.set(TableProperties.AVRO_COMPRESSION_LEVEL, compressionLevel.toString());
+        }
+
+        return builder.build();
       } catch (IOException e) {
         throw new RuntimeIOException(e, "Failed to create snapshot list writer for path: %s", file);
       }