[Bug] Fix Iceberg metadata unreadable by Snowflake - add Avro schema/partition-spec metadata to manifest files

paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergCommitCallback.java

@@ -72,6 +72,7 @@
 import org.apache.paimon.types.RowType;
 import org.apache.paimon.utils.DataFilePathFactories;
 import org.apache.paimon.utils.FileStorePathFactory;
+import org.apache.paimon.utils.JsonSerdeUtil;
 import org.apache.paimon.utils.ManifestReadThreadPool;
 import org.apache.paimon.utils.Pair;
 import org.apache.paimon.utils.Preconditions;
@@ -189,8 +190,6 @@ public IcebergCommitCallback(FileStoreTable table, String commitUser) {
                 metadataCommitterFactory == null ? null : metadataCommitterFactory.create(table);
 
         this.fileStorePathFactory = table.store().pathFactory();
-        this.manifestFile = IcebergManifestFile.create(table, pathFactory);
-        this.manifestList = IcebergManifestList.create(table, pathFactory);
 
         this.formatVersion =
                 table.coreOptions().toConfiguration().get(IcebergOptions.FORMAT_VERSION);
@@ -200,6 +199,25 @@ public IcebergCommitCallback(FileStoreTable table, String commitUser) {
                 "Unsupported iceberg format version! Only version 2 or version 3 is valid, but current version is ",
                 formatVersion);
 
+        // Compute Iceberg schema and partition spec for Avro manifest metadata.
+        // Snowflake and other Iceberg readers require these in the manifest file header.
+        // Iceberg field IDs must be positive. Paimon column IDs start at 0, which Iceberg
+        // readers reject (see #9012), so remap top-level fields to start from 1. Nested type
+        // IDs are left as-is for a follow-up.
+        IcebergSchema icebergSchema = withPositiveFieldIds(IcebergSchema.create(table.schema()));
+        List<IcebergPartitionField> partitionFields =
+                getPartitionFields(table.schema().partitionKeys(), icebergSchema);
+        Map<String, String> avroMetadata = new HashMap<>();
+        avroMetadata.put("schema", icebergSchema.toJson());
+        // Iceberg manifest "partition-spec" metadata is the JSON array of partition fields,
+        // not the whole spec object (PartitionSpecParser.toJsonFields semantics).
+        avroMetadata.put("partition-spec", JsonSerdeUtil.toJson(partitionFields));
+        avroMetadata.put("partition-spec-id", String.valueOf(IcebergPartitionSpec.SPEC_ID));
+        avroMetadata.put("format-version", String.valueOf(formatVersion));
+        this.manifestFile = IcebergManifestFile.create(table, pathFactory, avroMetadata);
+
+        this.manifestList = IcebergManifestList.create(table, pathFactory);
+
         this.indexFileHandler = table.store().newIndexFileHandler();
         this.needAddDvToIceberg = needAddDvToIceberg();
     }
@@ -735,6 +753,28 @@ private List<IcebergPartitionField> getPartitionFields(
         return result;
     }
 
+    /**
+     * Rebuilds the schema with positive field IDs starting from 1 for the Avro manifest header. See
+     * PR #9497 review: `Schema.Builder` starts Paimon column IDs at 0 and Iceberg readers like
+     * Snowflake reject them.
+     */
+    private static IcebergSchema withPositiveFieldIds(IcebergSchema schema) {
+        int[] nextId = {1};
+        List<IcebergDataField> fields =
+                schema.fields().stream()
+                        .map(
+                                field ->
+                                        new IcebergDataField(
+                                                nextId[0]++,
+                                                field.name(),
+                                                field.required(),
+                                                field.type(),
+                                                field.dataType(),
+                                                field.doc()))
+                        .collect(Collectors.toList());
+        return new IcebergSchema(schema.schemaId(), fields);
+    }
+
     /** VARIANT is an Iceberg format-version-3 type; reject publishing it into v2 metadata. */
     static void checkVariantNotPublishable(RowType rowType) {
         Collection<String> variantFields = new LinkedHashSet<>();

paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergManifestFile.java

@@ -26,6 +26,7 @@
 import org.apache.paimon.format.FormatWriterFactory;
 import org.apache.paimon.format.SimpleColStats;
 import org.apache.paimon.format.SimpleStatsCollector;
+import org.apache.paimon.format.avro.AvroFileFormat;
 import org.apache.paimon.fs.FileIO;
 import org.apache.paimon.fs.Path;
 import org.apache.paimon.iceberg.IcebergOptions;
@@ -52,8 +53,10 @@
 
 import java.io.IOException;
 import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.Iterator;
 import java.util.List;
+import java.util.Map;
 import java.util.Objects;
 
 import static org.apache.paimon.iceberg.manifest.IcebergConversions.toByteBuffer;
@@ -66,7 +69,24 @@ public class IcebergManifestFile extends ObjectsFile<IcebergManifestEntry> {
 
     private static final long UNASSIGNED_SEQ = -1L;
 
+    private static final String ROW_NAME_MAPPING =
+            "org.apache.paimon.avro.generated.record:manifest_entry,"
+                    + "iceberg:true,"
+                    + "manifest_entry_data_file:r2,"
+                    + "r2_partition:r102,"
+                    + "kv_name_r2_null_value_counts:k121_v122,"
+                    + "k_id_k121_v122:121,"
+                    + "v_id_k121_v122:122,"
+                    + "kv_name_r2_lower_bounds:k126_v127,"
+                    + "k_id_k126_v127:126,"
+                    + "v_id_k126_v127:127,"
+                    + "kv_name_r2_upper_bounds:k129_v130,"
+                    + "k_id_k129_v130:129,"
+                    + "v_id_k129_v130:130";
+
+    private final Map<Content, FormatWriterFactory> writerFactories;
     private final RowType partitionType;
+    // Default (DATA) writer factory, kept for the base ObjectsFile.
     private final FormatWriterFactory writerFactory;
     private final MemorySize targetFileSize;
 
@@ -75,7 +95,31 @@ public IcebergManifestFile(
             RowType partitionType,
             boolean withFirstRowId,
             FormatReaderFactory readerFactory,
-            FormatWriterFactory writerFactory,
+            Map<Content, FormatWriterFactory> writerFactories,
+            String compression,
+            PathFactory pathFactory,
+            MemorySize targetFileSize) {
+        // Java forbids `this.` assignments before super(); route the DATA factory through
+        // the parameter list instead.
+        this(
+                fileIO,
+                partitionType,
+                withFirstRowId,
+                readerFactory,
+                writerFactories,
+                writerFactories.get(Content.DATA),
+                compression,
+                pathFactory,
+                targetFileSize);
+    }
+
+    private IcebergManifestFile(
+            FileIO fileIO,
+            RowType partitionType,
+            boolean withFirstRowId,
+            FormatReaderFactory readerFactory,
+            Map<Content, FormatWriterFactory> writerFactories,
+            FormatWriterFactory defaultWriterFactory,
             String compression,
             PathFactory pathFactory,
             MemorySize targetFileSize) {
@@ -84,12 +128,13 @@ public IcebergManifestFile(
                 new IcebergManifestEntrySerializer(partitionType, withFirstRowId),
                 IcebergManifestEntry.schema(partitionType, withFirstRowId),
                 readerFactory,
-                writerFactory,
+                defaultWriterFactory,
                 compression,
                 pathFactory,
                 null);
         this.partitionType = partitionType;
-        this.writerFactory = writerFactory;
+        this.writerFactories = writerFactories;
+        this.writerFactory = defaultWriterFactory;
         this.targetFileSize = targetFileSize;
     }
 
@@ -99,34 +144,43 @@ public String compression() {
     }
 
     public static IcebergManifestFile create(FileStoreTable table, IcebergPathFactory pathFactory) {
+        return create(table, pathFactory, new HashMap<>());
+    }
+
+    public static IcebergManifestFile create(
+            FileStoreTable table,
+            IcebergPathFactory pathFactory,
+            Map<String, String> avroMetadata) {
         RowType partitionType = table.schema().logicalPartitionType();
         Options avroOptions = Options.fromMap(table.options());
         boolean withFirstRowId =
                 avroOptions.get(IcebergOptions.FORMAT_VERSION) >= IcebergMetadata.FORMAT_VERSION_V3;
         RowType entryType = IcebergManifestEntry.schema(partitionType, withFirstRowId);
         // https://github.com/apache/iceberg/blob/main/core/src/main/java/org/apache/iceberg/ManifestReader.java
-        avroOptions.set(
-                "avro.row-name-mapping",
-                "org.apache.paimon.avro.generated.record:manifest_entry,"
-                        + "iceberg:true,"
-                        + "manifest_entry_data_file:r2,"
-                        + "r2_partition:r102,"
-                        + "kv_name_r2_null_value_counts:k121_v122,"
-                        + "k_id_k121_v122:121,"
-                        + "v_id_k121_v122:122,"
-                        + "kv_name_r2_lower_bounds:k126_v127,"
-                        + "k_id_k126_v127:126,"
-                        + "v_id_k126_v127:127,"
-                        + "kv_name_r2_upper_bounds:k129_v130,"
-                        + "k_id_k129_v130:129,"
-                        + "v_id_k129_v130:130");
-        FileFormat manifestFileAvro = FileFormat.fromIdentifier("avro", avroOptions);
+        avroOptions.set("avro.row-name-mapping", ROW_NAME_MAPPING);
+        // The "content" Avro header differs per manifest (data vs deletes), so build one
+        // writer factory per content. See PR #9497 review.
+        Map<Content, FormatWriterFactory> writerFactories = new HashMap<>();
+        FormatReaderFactory readerFactory = null;
+        for (Content content : Content.values()) {
+            Options contentOptions = Options.fromMap(table.options());
+            contentOptions.set("avro.row-name-mapping", ROW_NAME_MAPPING);
+            Map<String, String> contentMetadata = new HashMap<>(avroMetadata);
+            contentMetadata.put("content", content == Content.DATA ? "data" : "deletes");
+            AvroFileFormat.setAvroMetadata(contentOptions, contentMetadata);
+            FileFormat contentAvro = FileFormat.fromIdentifier("avro", contentOptions);
+            writerFactories.put(content, contentAvro.createWriterFactory(entryType));
+            if (content == Content.DATA) {
+                readerFactory =
+                        contentAvro.createReaderFactory(entryType, entryType, new ArrayList<>());
+            }
+        }
         return new IcebergManifestFile(
                 table.fileIO(),
                 partitionType,
                 withFirstRowId,
-                manifestFileAvro.createReaderFactory(entryType, entryType, new ArrayList<>()),
-                manifestFileAvro.createWriterFactory(entryType),
+                readerFactory,
+                writerFactories,
                 avroOptions.get(IcebergOptions.MANIFEST_COMPRESSION),
                 pathFactory.manifestFileFactory(),
                 table.coreOptions().manifestTargetSize());
@@ -195,7 +249,11 @@ public List<IcebergManifestFileMeta> rollingWrite(
     public SingleFileWriter<IcebergManifestEntry, IcebergManifestFileMeta> createWriter(
             long sequenceNumber, Content content) {
         return new IcebergManifestEntryWriter(
-                writerFactory, pathFactory.newPath(), compression, sequenceNumber, content);
+                writerFactories.get(content),
+                pathFactory.newPath(),
+                compression,
+                sequenceNumber,
+                content);
     }
 
     private class IcebergManifestEntryWriter

paimon-format/src/main/java/org/apache/paimon/format/avro/AvroFileFormat.java

@@ -63,6 +63,9 @@ public class AvroFileFormat extends FileFormat {
     private static final ConfigOption<Map<String, String>> AVRO_ROW_NAME_MAPPING =
             ConfigOptions.key("avro.row-name-mapping").mapType().defaultValue(new HashMap<>());
 
+    private static final ConfigOption<Map<String, String>> AVRO_METADATA =
+            ConfigOptions.key("avro.metadata").mapType().defaultValue(new HashMap<>());
+
     private final Options options;
     private final int zstdLevel;
 
@@ -92,6 +95,12 @@ public AvroBlockWriter createBlockWriter(
                 AvroSchemaConverter.convertToSchema(rowType, options.get(AVRO_ROW_NAME_MAPPING));
         AvroRowDatumWriter datumWriter = new AvroRowDatumWriter(rowType);
         DataFileWriter<InternalRow> writer = new DataFileWriter<>(datumWriter);
+        Map<String, String> metadata = options.get(AVRO_METADATA);
+        if (metadata != null) {
+            for (Map.Entry<String, String> entry : metadata.entrySet()) {
+                writer.setMeta(entry.getKey(), entry.getValue());
+            }
+        }
         writer.setCodec(createCodecFactory(compression));
         writer.setFlushOnEveryBlock(false);
         writer.create(schema, new CloseShieldOutputStream(out));
@@ -138,4 +147,13 @@ public FormatWriter create(PositionOutputStream out, String compression)
             return createBlockWriter(out, rowType, compression);
         }
     }
+
+    /**
+     * Sets Avro file-level metadata key-value pairs on the given options. These metadata are
+     * written into the Avro container file header and are visible to Iceberg-compatible readers
+     * (e.g. Snowflake).
+     */
+    public static void setAvroMetadata(Options options, Map<String, String> metadata) {
+        options.set(AVRO_METADATA, metadata);
+    }
 }