diff --git a/paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergCommitCallback.java b/paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergCommitCallback.java index 5b0a78b828fd..7232ec40487e 100644 --- a/paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergCommitCallback.java +++ b/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,21 @@ 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. + IcebergSchema icebergSchema = IcebergSchema.create(table.schema()); + List partitionFields = + getPartitionFields(table.schema().partitionKeys(), icebergSchema); + IcebergPartitionSpec partitionSpec = new IcebergPartitionSpec(partitionFields); + Map avroMetadata = new HashMap<>(); + avroMetadata.put("schema", icebergSchema.toJson()); + avroMetadata.put("partition-spec", JsonSerdeUtil.toJson(partitionSpec)); + 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(); } diff --git a/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergManifestFile.java b/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergManifestFile.java index 5e43a2e2fa36..01392aaa9c0a 100644 --- a/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergManifestFile.java +++ b/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergManifestFile.java @@ -48,12 +48,16 @@ import org.apache.paimon.utils.ObjectsFile; import org.apache.paimon.utils.PathFactory; +import org.apache.paimon.format.avro.AvroFileFormat; + import javax.annotation.Nullable; 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; @@ -99,6 +103,11 @@ 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 avroMetadata) { RowType partitionType = table.schema().logicalPartitionType(); Options avroOptions = Options.fromMap(table.options()); boolean withFirstRowId = @@ -120,6 +129,7 @@ public static IcebergManifestFile create(FileStoreTable table, IcebergPathFactor + "kv_name_r2_upper_bounds:k129_v130," + "k_id_k129_v130:129," + "v_id_k129_v130:130"); + AvroFileFormat.setAvroMetadata(avroOptions, avroMetadata); FileFormat manifestFileAvro = FileFormat.fromIdentifier("avro", avroOptions); return new IcebergManifestFile( table.fileIO(), 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 69512d5ee7cf..0f6a023b9c20 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 @@ -63,6 +63,9 @@ public class AvroFileFormat extends FileFormat { private static final ConfigOption> AVRO_ROW_NAME_MAPPING = ConfigOptions.key("avro.row-name-mapping").mapType().defaultValue(new HashMap<>()); + private static final ConfigOption> 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 writer = new DataFileWriter<>(datumWriter); + Map metadata = options.get(AVRO_METADATA); + if (metadata != null) { + for (Map.Entry 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 metadata) { + options.set(AVRO_METADATA, metadata); + } }