Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
zhang-arvin wants to merge 1 commit into
base: master
Choose a base branch
from
Open

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 16 additions & 2 deletions paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergCommitCallback.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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);
Expand All @@ -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());

@JingsongLi JingsongLi Aug 31, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P1] This still writes field ID 0 into the Iceberg schema. Schema.Builder assigns the first Paimon column ID 0, and IcebergDataField(DataField) preserves it. I verified that a manifest produced by this PR has "id" : 0 in its schema header, which is the incompatibility reported in #9012. Adding the header therefore does not demonstrate that Snowflake can read the table. Please introduce a consistent positive-ID mapping everywhere Iceberg IDs are emitted (schema, partition source IDs, metrics maps, and any physical schema IDs), and cover it with a compatibility regression test.

List<IcebergPartitionField> partitionFields =
getPartitionFields(table.schema().partitionKeys(), icebergSchema);
IcebergPartitionSpec partitionSpec = new IcebergPartitionSpec(partitionFields);
Map<String, String> avroMetadata = new HashMap<>();

@JingsongLi JingsongLi Aug 31, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P1] Iceberg v2/v3 manifests require a content header whose value is data or deletes, but this map omits it; the generated manifest has content = null. A single constructor-level value would also be insufficient because this IcebergManifestFile writes both Content.DATA and Content.DELETES, selected only by rollingWrite. Please build the metadata per writer from its Content (or use separate writer factories), and test both data and delete manifests.

avroMetadata.put("schema", icebergSchema.toJson());
avroMetadata.put("partition-spec", JsonSerdeUtil.toJson(partitionSpec));

@JingsongLi JingsongLi Aug 31, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P1] Iceberg's partition-spec manifest metadata is a JSON array of partition fields, not the complete partition-spec object. This serializes an unpartitioned spec as {"spec-id":0,"fields":[]}. I reproduced the resulting failure with Iceberg 1.6.1 ManifestFiles.read: Cannot parse partition spec fields, not an array. Please use the equivalent of PartitionSpecParser.toJsonFields(spec) here, keep partition-spec-id separate, and add a test that opens the generated manifest through Iceberg without supplying an external spec map.

avroMetadata.put("partition-spec-id", String.valueOf(IcebergPartitionSpec.SPEC_ID));
avroMetadata.put("format-version", String.valueOf(formatVersion));
this.manifestFile = IcebergManifestFile.create(table, pathFactory, avroMetadata);

@JingsongLi JingsongLi Aug 31, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P2] This only adds metadata to manifests created after the upgrade. createMetadataWithBase retains baseDataManifestFileMetas for add-only commits and retains existing DV manifests when there is no new index, so an already affected table remains a mixture of new and legacy headerless manifests and Snowflake still has to traverse the legacy files. Please provide a one-time manifest rewrite/migration path (or an explicit operational migration) and add an upgrade test starting from existing manifests.


this.manifestList = IcebergManifestList.create(table, pathFactory);

this.indexFileHandler = table.store().newIndexFileHandler();
this.needAddDvToIceberg = needAddDvToIceberg();
}
Expand Down
10 changes: 10 additions & 0 deletions paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergManifestFile.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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<String, String> avroMetadata) {
RowType partitionType = table.schema().logicalPartitionType();
Options avroOptions = Options.fromMap(table.options());
boolean withFirstRowId =
Expand All @@ -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(),
Expand Down
18 changes: 18 additions & 0 deletions paimon-format/src/main/java/org/apache/paimon/format/avro/AvroFileFormat.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;

Expand Down Expand Up @@ -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));
Expand Down Expand Up @@ -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);
}
}
Loading