From 9f5e3a12717f15d9e9addb430724942ca89675fe Mon Sep 17 00:00:00 2001
From: Jingsong <jingsonglee0@gmail.com>
Date: Tue, 3 Dec 2024 22:35:20 +0700
Subject: [PATCH] [avro] Fix compression not work in writer

---
 docs/content/migration/iceberg-compatibility.md      |  2 +-
 .../org/apache/paimon/iceberg/IcebergOptions.java    |  2 +-
 .../apache/paimon/format/avro/AvroFileFormat.java    |  2 +-
 .../paimon/format/avro/AvroFileFormatTest.java       | 12 ++++++++++++
 4 files changed, 15 insertions(+), 3 deletions(-)

diff --git a/docs/content/migration/iceberg-compatibility.md b/docs/content/migration/iceberg-compatibility.md
index 01a03a45264d..8e4d3c90176b 100644
--- a/docs/content/migration/iceberg-compatibility.md
+++ b/docs/content/migration/iceberg-compatibility.md
@@ -373,7 +373,7 @@ you also need to set some (or all) of the following table options when creating
     </tr>
     <tr>
       <td><h5>metadata.iceberg.manifest-compression</h5></td>
-      <td style="word-wrap: break-word;">gzip</td>
+      <td style="word-wrap: break-word;">snappy</td>
       <td>String</td>
       <td>Compression for Iceberg manifest files.</td>
     </tr>
diff --git a/paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergOptions.java b/paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergOptions.java
index 4b59e29c8c33..55fbab5158fa 100644
--- a/paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergOptions.java
+++ b/paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergOptions.java
@@ -74,7 +74,7 @@ public class IcebergOptions {
             key("metadata.iceberg.manifest-compression")
                     .stringType()
                     .defaultValue(
-                            "gzip") // some Iceberg reader cannot support zstd, for example DuckDB
+                            "snappy") // some Iceberg reader cannot support zstd, for example DuckDB
                     .withDescription("Compression for Iceberg manifest files.");
 
     public static final ConfigOption<Boolean> MANIFEST_LEGACY_VERSION =
diff --git a/paimon-format/src/main/java/org/apache/paimon/format/avro/AvroFileFormat.java b/paimon-format/src/main/java/org/apache/paimon/format/avro/AvroFileFormat.java
index 63a51c0a13a9..fcce9ae50530 100644
--- a/paimon-format/src/main/java/org/apache/paimon/format/avro/AvroFileFormat.java
+++ b/paimon-format/src/main/java/org/apache/paimon/format/avro/AvroFileFormat.java
@@ -105,7 +105,7 @@ private CodecFactory createCodecFactory(String compression) {
         if (compression.equalsIgnoreCase("zstd")) {
             return CodecFactory.zstandardCodec(zstdLevel);
         }
-        return CodecFactory.fromString(options.get(AVRO_OUTPUT_CODEC));
+        return CodecFactory.fromString(compression);
     }
 
     /** A {@link FormatWriterFactory} to write {@link InternalRow}. */
diff --git a/paimon-format/src/test/java/org/apache/paimon/format/avro/AvroFileFormatTest.java b/paimon-format/src/test/java/org/apache/paimon/format/avro/AvroFileFormatTest.java
index 3f6486baaef2..9c0dbb43fe62 100644
--- a/paimon-format/src/test/java/org/apache/paimon/format/avro/AvroFileFormatTest.java
+++ b/paimon-format/src/test/java/org/apache/paimon/format/avro/AvroFileFormatTest.java
@@ -198,4 +198,16 @@ private void checkException() throws IOException {
                 .isInstanceOf(IOException.class)
                 .hasMessageContaining("Artificial exception");
     }
+
+    @Test
+    void testCompression() throws IOException {
+        RowType rowType = DataTypes.ROW(DataTypes.INT().notNull());
+        AvroFileFormat format = new AvroFileFormat(new FormatContext(new Options(), 1024, 1024));
+        LocalFileIO localFileIO = LocalFileIO.create();
+        Path file = new Path(new Path(tempPath.toUri()), UUID.randomUUID().toString());
+        try (PositionOutputStream out = localFileIO.newOutputStream(file, false)) {
+            assertThatThrownBy(() -> format.createWriterFactory(rowType).create(out, "unsupported"))
+                    .hasMessageContaining("Unrecognized codec: unsupported");
+        }
+    }
 }