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..ef1665ce350c 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,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 partitionFields = + getPartitionFields(table.schema().partitionKeys(), icebergSchema); + Map 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 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 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 variantFields = new LinkedHashSet<>(); 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..3fc1c7f2b2ad 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 @@ -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 { 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 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 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 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 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 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 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 rollingWrite( public SingleFileWriter 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 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); + } }