From ea3560c3caf6411fc3423e78139b762f683f6712 Mon Sep 17 00:00:00 2001 From: Victor Babenko <37556649+vbabenkoru@users.noreply.github.com> Date: Fri, 21 Aug 2026 10:27:39 -0700 Subject: [PATCH 1/2] [core] Add field-id.one-based option for strictly positive Iceberg field ids Snowflake's external Iceberg reader rejects Paimon-generated Iceberg metadata with 'external field ID in metadata file cannot be zero': Paimon assigns table field ids from 0, and those ids flow verbatim into both the Iceberg schema and the Parquet footers. A metadata-only remap is not safe, because Iceberg readers resolve embedded Parquet field ids before name mapping, so metadata and physical ids must move together. Add an immutable, create-time-only boolean table option 'field-id.one-based' (default false, unchanged behavior). When enabled, all field ids of a NEW table - top-level and nested - are shifted by one at creation, so schema files, Parquet/ORC footers, Iceberg metadata, partition source ids and stats keys all agree on the same strictly positive id space. Schema evolution continues from the persisted highestFieldId, so later columns keep 1-based ids with no further changes. Existing tables cannot be migrated with this option: their data files already embed 0-based ids, so they must be rewritten into a new table created with field-id.one-based=true. Tests: default stays 0-based; shifted creation and evolution; immutability; end-to-end strict-mode primary-key + DV table on format-version 3 verifying 1-based ids in Iceberg metadata, identical ids in Parquet footers (top-level and nested), partition source-id, non-null manifest-list first_row_id, Apache Iceberg 1.11 read with the DV applied, and Paimon read-back. --- docs/generated/core_configuration.html | 6 + .../java/org/apache/paimon/CoreOptions.java | 19 + .../org/apache/paimon/types/ShiftFieldId.java | 88 ++++ .../apache/paimon/schema/SchemaManager.java | 14 + .../paimon/schema/FieldIdOneBasedTest.java | 121 ++++++ ...ebergFieldIdOneBasedCompatibilityTest.java | 377 ++++++++++++++++++ 6 files changed, 625 insertions(+) create mode 100644 paimon-api/src/main/java/org/apache/paimon/types/ShiftFieldId.java create mode 100644 paimon-core/src/test/java/org/apache/paimon/schema/FieldIdOneBasedTest.java create mode 100644 paimon-iceberg/src/test/java/org/apache/paimon/core/IcebergFieldIdOneBasedCompatibilityTest.java diff --git a/docs/generated/core_configuration.html b/docs/generated/core_configuration.html index 3fb25ebce15f..3964fa67c330 100644 --- a/docs/generated/core_configuration.html +++ b/docs/generated/core_configuration.html @@ -662,6 +662,12 @@ Boolean Optional endInput check partition expire used in case of batch mode or bounded stream. + +
field-id.one-based
+ false + Boolean + Whether to assign field ids starting from 1 instead of 0 when creating a table (ids of all columns, including nested ones, are shifted by one). Paimon historically starts field ids at 0, but some external Iceberg readers (e.g. Snowflake) reject field id 0 in Iceberg metadata. Enable when creating tables with Iceberg compatibility for such readers. Only affects table creation; existing tables (and their data files, which embed field ids) keep their original ids. +
fields.default-aggregate-function
(none) diff --git a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java index a19d1818c247..6dc500ac34ba 100644 --- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java +++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java @@ -2486,6 +2486,21 @@ public String toString() { .defaultValue(false) .withDescription("Whether enable unique row id for append table."); + @Immutable + public static final ConfigOption FIELD_ID_ONE_BASED = + key("field-id.one-based") + .booleanType() + .defaultValue(false) + .withDescription( + "Whether to assign field ids starting from 1 instead of 0 when creating " + + "a table (ids of all columns, including nested ones, are " + + "shifted by one). Paimon historically starts field ids at 0, " + + "but some external Iceberg readers (e.g. Snowflake) reject " + + "field id 0 in Iceberg metadata. Enable when creating tables " + + "with Iceberg compatibility for such readers. Only affects " + + "table creation; existing tables (and their data files, which " + + "embed field ids) keep their original ids."); + public static final ConfigOption ROW_TRACKING_PARTITION_GROUP_ON_COMMIT = key("row-tracking.partition-group-on-commit") .booleanType() @@ -4257,6 +4272,10 @@ public boolean rowTrackingEnabled() { return options.get(ROW_TRACKING_ENABLED); } + public boolean fieldIdOneBased() { + return options.get(FIELD_ID_ONE_BASED); + } + public boolean rowTrackingPartitionGroupOnCommit() { return options.get(ROW_TRACKING_PARTITION_GROUP_ON_COMMIT); } diff --git a/paimon-api/src/main/java/org/apache/paimon/types/ShiftFieldId.java b/paimon-api/src/main/java/org/apache/paimon/types/ShiftFieldId.java new file mode 100644 index 000000000000..84ea384b50f0 --- /dev/null +++ b/paimon-api/src/main/java/org/apache/paimon/types/ShiftFieldId.java @@ -0,0 +1,88 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.paimon.types; + +import java.util.List; +import java.util.stream.Collectors; + +/** + * Shift every field id in a type by a fixed offset. Unlike {@link ReassignFieldId} this preserves + * the relative order and gaps of the existing ids, so the result is exactly the original id space + * translated by {@code offset}. + */ +public class ShiftFieldId extends DataTypeDefaultVisitor { + + private final int offset; + + public ShiftFieldId(int offset) { + this.offset = offset; + } + + public static DataType shift(DataType input, int offset) { + return input.accept(new ShiftFieldId(offset)); + } + + @Override + public DataType visit(ArrayType arrayType) { + return new ArrayType(arrayType.isNullable(), arrayType.getElementType().accept(this)); + } + + @Override + public DataType visit(VectorType vectorType) { + return new VectorType( + vectorType.isNullable(), + vectorType.getLength(), + vectorType.getElementType().accept(this)); + } + + @Override + public DataType visit(MultisetType multisetType) { + return new MultisetType( + multisetType.isNullable(), multisetType.getElementType().accept(this)); + } + + @Override + public DataType visit(MapType mapType) { + return new MapType( + mapType.isNullable(), + mapType.getKeyType().accept(this), + mapType.getValueType().accept(this)); + } + + @Override + public DataType visit(RowType rowType) { + List fields = + rowType.getFields().stream() + .map( + f -> + new DataField( + f.id() + offset, + f.name(), + f.type().accept(this), + f.description(), + f.defaultValue())) + .collect(Collectors.toList()); + return new RowType(rowType.isNullable(), fields); + } + + @Override + protected DataType defaultMethod(DataType dataType) { + return dataType; + } +} diff --git a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaManager.java b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaManager.java index 73e947d9fa0e..6a3d809b5358 100644 --- a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaManager.java +++ b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaManager.java @@ -47,6 +47,7 @@ import org.apache.paimon.types.MapType; import org.apache.paimon.types.ReassignFieldId; import org.apache.paimon.types.RowType; +import org.apache.paimon.types.ShiftFieldId; import org.apache.paimon.utils.BranchManager; import org.apache.paimon.utils.ChangelogManager; import org.apache.paimon.utils.LazyField; @@ -205,6 +206,7 @@ public TableSchema createTable(Schema schema, boolean externalTable) throws Exce } schema = applyDirectives(schema); + schema = applyFieldIdOneBased(schema); TableSchema newSchema = TableSchema.create(0, schema); // validate table from creating table @@ -217,6 +219,18 @@ public TableSchema createTable(Schema schema, boolean externalTable) throws Exce } } + /** + * Shift all field ids of a new table by one when {@link CoreOptions#FIELD_ID_ONE_BASED} is set. + * Applied only at table creation: data files embed these ids (Parquet footers, Iceberg + * metadata), so the id space of an existing table must never be re-based. + */ + private static Schema applyFieldIdOneBased(Schema schema) { + if (!CoreOptions.fromMap(schema.options()).fieldIdOneBased()) { + return schema; + } + return schema.copy((RowType) ShiftFieldId.shift(schema.rowType(), 1)); + } + private void checkSchemaForExternalTable(Schema existsSchema, Schema newSchema) { // When creating an external table, if the table already exists in the location, we can // choose not to specify the fields. diff --git a/paimon-core/src/test/java/org/apache/paimon/schema/FieldIdOneBasedTest.java b/paimon-core/src/test/java/org/apache/paimon/schema/FieldIdOneBasedTest.java new file mode 100644 index 000000000000..11f814392f37 --- /dev/null +++ b/paimon-core/src/test/java/org/apache/paimon/schema/FieldIdOneBasedTest.java @@ -0,0 +1,121 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.paimon.schema; + +import org.apache.paimon.CoreOptions; +import org.apache.paimon.fs.Path; +import org.apache.paimon.fs.local.LocalFileIO; +import org.apache.paimon.types.ArrayType; +import org.apache.paimon.types.DataField; +import org.apache.paimon.types.DataTypes; +import org.apache.paimon.types.MapType; +import org.apache.paimon.types.RowType; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.util.List; +import java.util.UUID; +import java.util.stream.Collectors; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Tests for {@link CoreOptions#FIELD_ID_ONE_BASED} at table creation and evolution. */ +public class FieldIdOneBasedTest { + + @TempDir java.nio.file.Path tempDir; + + private Schema.Builder schemaBuilder() { + return Schema.newBuilder() + .column("a", DataTypes.INT()) + .column( + "s", + DataTypes.ROW( + DataTypes.FIELD(0, "x", DataTypes.INT()), + DataTypes.FIELD(0, "y", DataTypes.STRING()))) + .column("m", DataTypes.MAP(DataTypes.STRING(), DataTypes.ARRAY(DataTypes.INT()))); + } + + private SchemaManager newSchemaManager(String name) { + return new SchemaManager( + LocalFileIO.create(), + new Path(tempDir.toString() + "/" + name + UUID.randomUUID())); + } + + @Test + public void testDefaultRemainsZeroBased() throws Exception { + TableSchema schema = newSchemaManager("t").createTable(schemaBuilder().build()); + assertThat(topLevelIds(schema)).containsExactly(0, 1, 4); + RowType nested = (RowType) schema.fields().get(1).type(); + assertThat(nested.getFields().get(0).id()).isEqualTo(2); + assertThat(nested.getFields().get(1).id()).isEqualTo(3); + assertThat(schema.highestFieldId()).isEqualTo(4); + } + + @Test + public void testOneBasedShiftsAllIds() throws Exception { + TableSchema schema = + newSchemaManager("t") + .createTable( + schemaBuilder() + .option(CoreOptions.FIELD_ID_ONE_BASED.key(), "true") + .build()); + assertThat(topLevelIds(schema)).containsExactly(1, 2, 5); + RowType nested = (RowType) schema.fields().get(1).type(); + assertThat(nested.getFields().get(0).id()).isEqualTo(3); + assertThat(nested.getFields().get(1).id()).isEqualTo(4); + // map/array types carry no ids of their own; ensure the structure survived the shift + MapType map = (MapType) schema.fields().get(2).type(); + assertThat(map.getValueType()).isInstanceOf(ArrayType.class); + assertThat(schema.highestFieldId()).isEqualTo(5); + } + + @Test + public void testEvolutionContinuesFromShiftedIds() throws Exception { + SchemaManager manager = newSchemaManager("t"); + manager.createTable( + schemaBuilder().option(CoreOptions.FIELD_ID_ONE_BASED.key(), "true").build()); + TableSchema evolved = manager.commitChanges(SchemaChange.addColumn("z", DataTypes.INT())); + DataField added = + evolved.fields().stream() + .filter(f -> f.name().equals("z")) + .findFirst() + .orElseThrow(IllegalStateException::new); + assertThat(added.id()).isEqualTo(6); + assertThat(evolved.highestFieldId()).isEqualTo(6); + } + + @Test + public void testOneBasedImmutableAndCreateTimeOnly() throws Exception { + // registered as immutable, so ALTER is rejected once the table has snapshots + assertThat(CoreOptions.IMMUTABLE_OPTIONS).contains(CoreOptions.FIELD_ID_ONE_BASED.key()); + // a pre-snapshot option change must not re-base ids: assignment happens at creation only + SchemaManager manager = newSchemaManager("t"); + manager.createTable(schemaBuilder().build()); + TableSchema altered = + manager.commitChanges( + SchemaChange.setOption(CoreOptions.FIELD_ID_ONE_BASED.key(), "true")); + assertThat(topLevelIds(altered)).containsExactly(0, 1, 4); + assertThat(altered.highestFieldId()).isEqualTo(4); + } + + private static List topLevelIds(TableSchema schema) { + return schema.fields().stream().map(DataField::id).collect(Collectors.toList()); + } +} diff --git a/paimon-iceberg/src/test/java/org/apache/paimon/core/IcebergFieldIdOneBasedCompatibilityTest.java b/paimon-iceberg/src/test/java/org/apache/paimon/core/IcebergFieldIdOneBasedCompatibilityTest.java new file mode 100644 index 000000000000..8aca8e8e7f37 --- /dev/null +++ b/paimon-iceberg/src/test/java/org/apache/paimon/core/IcebergFieldIdOneBasedCompatibilityTest.java @@ -0,0 +1,377 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.paimon.core; + +import org.apache.paimon.CoreOptions; +import org.apache.paimon.catalog.FileSystemCatalog; +import org.apache.paimon.catalog.Identifier; +import org.apache.paimon.data.BinaryRow; +import org.apache.paimon.data.BinaryRowWriter; +import org.apache.paimon.data.BinaryString; +import org.apache.paimon.data.GenericRow; +import org.apache.paimon.data.InternalRow; +import org.apache.paimon.disk.IOManagerImpl; +import org.apache.paimon.fs.Path; +import org.apache.paimon.fs.local.LocalFileIO; +import org.apache.paimon.iceberg.IcebergOptions; +import org.apache.paimon.iceberg.IcebergPathFactory; +import org.apache.paimon.iceberg.manifest.IcebergManifestFileMeta; +import org.apache.paimon.iceberg.manifest.IcebergManifestList; +import org.apache.paimon.iceberg.metadata.IcebergDataField; +import org.apache.paimon.iceberg.metadata.IcebergMetadata; +import org.apache.paimon.iceberg.metadata.IcebergPartitionField; +import org.apache.paimon.iceberg.metadata.IcebergSchema; +import org.apache.paimon.iceberg.metadata.IcebergStructType; +import org.apache.paimon.options.MemorySize; +import org.apache.paimon.options.Options; +import org.apache.paimon.reader.RecordReader; +import org.apache.paimon.schema.Schema; +import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.table.sink.TableCommitImpl; +import org.apache.paimon.table.sink.TableWriteImpl; +import org.apache.paimon.table.source.ReadBuilder; +import org.apache.paimon.table.source.Split; +import org.apache.paimon.types.DataField; +import org.apache.paimon.types.DataTypes; +import org.apache.paimon.types.RowKind; +import org.apache.paimon.types.RowType; + +import org.apache.paimon.shade.org.apache.parquet.hadoop.ParquetFileReader; +import org.apache.paimon.shade.org.apache.parquet.schema.GroupType; +import org.apache.paimon.shade.org.apache.parquet.schema.MessageType; + +import org.apache.hadoop.conf.Configuration; +import org.apache.iceberg.Table; +import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.iceberg.data.IcebergGenerics; +import org.apache.iceberg.data.Record; +import org.apache.iceberg.hadoop.HadoopCatalog; +import org.apache.iceberg.io.CloseableIterable; +import org.apache.iceberg.types.Types; +import org.apache.iceberg.util.StructLikeSet; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.io.IOException; +import java.io.UncheckedIOException; +import java.nio.file.Files; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.UUID; +import java.util.stream.Collectors; +import java.util.stream.Stream; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Tests for {@link CoreOptions#FIELD_ID_ONE_BASED} with Iceberg compatibility: a table created with + * 1-based field ids must emit strictly positive ids in Iceberg metadata that exactly match the ids + * embedded in the Parquet data files (readers like Snowflake reject field id 0 and resolve columns + * by the physical ids). + */ +public class IcebergFieldIdOneBasedCompatibilityTest { + + @TempDir java.nio.file.Path tempDir; + + private RowType rowType() { + return new RowType( + Arrays.asList( + new DataField(0, "pt", DataTypes.INT().notNull()), + new DataField(1, "k", DataTypes.INT().notNull()), + new DataField(2, "v", DataTypes.STRING()), + new DataField( + 3, + "nested", + new RowType( + Arrays.asList( + new DataField(4, "a", DataTypes.INT()), + new DataField(5, "b", DataTypes.STRING())))))); + } + + @Test + public void testStrictModePrimaryKeyDvTable() throws Exception { + Map customOptions = new HashMap<>(); + customOptions.put(CoreOptions.FIELD_ID_ONE_BASED.key(), "true"); + customOptions.put(CoreOptions.DELETION_VECTORS_ENABLED.key(), "true"); + customOptions.put(CoreOptions.DELETION_VECTOR_BITMAP64.key(), "true"); + customOptions.put(IcebergOptions.FORMAT_VERSION.key(), "3"); + + FileStoreTable table = createPaimonTable(customOptions); + + // the Paimon schema itself is 1-based, top-level and nested alike + List fields = table.schema().fields(); + assertThat(fields.stream().map(DataField::id)).containsExactly(1, 2, 3, 4); + RowType nested = (RowType) fields.get(3).type(); + assertThat(nested.getFields().stream().map(DataField::id)).containsExactly(5, 6); + assertThat(table.schema().highestFieldId()).isEqualTo(6); + + String commitUser = UUID.randomUUID().toString(); + TableWriteImpl write = + table.newWrite(commitUser) + .withIOManager(new IOManagerImpl(tempDir.toString() + "/tmp")); + TableCommitImpl commit = table.newCommit(commitUser); + + write.write(row(RowKind.INSERT, 1, 1, "a", 10, "x")); + write.write(row(RowKind.INSERT, 1, 2, "b", 20, "y")); + commit.commit(1, write.prepareCommit(false, 1)); + + write.write(row(RowKind.DELETE, 1, 2, "b", 20, "y")); + commit.commit(2, write.prepareCommit(false, 2)); + + // produce a deletion vector + write.compact(partition(1), 0, false); + commit.commit(3, write.prepareCommit(true, 3)); + write.close(); + commit.close(); + + IcebergMetadata metadata = readLatestIcebergMetadata(table); + + // 1) metadata field ids are strictly positive and match the Paimon schema + assertThat(metadata.formatVersion()).isEqualTo(3); + for (IcebergSchema schema : metadata.schemas()) { + assertThat(collectAllFieldIds(schema.fields())).allMatch(id -> id >= 1); + } + IcebergSchema currentSchema = metadata.schemas().get(metadata.currentSchemaId()); + assertThat(currentSchema.fields().stream().map(IcebergDataField::id)) + .containsExactly(1, 2, 3, 4); + IcebergStructType nestedType = (IcebergStructType) currentSchema.fields().get(3).type(); + assertThat(nestedType.fields().stream().map(IcebergDataField::id)).containsExactly(5, 6); + assertThat(metadata.lastColumnId()).isEqualTo(4); + + // 2) partition source ids reference the shifted ids + assertThat(metadata.partitionSpecs()).hasSize(1); + List partitionFields = metadata.partitionSpecs().get(0).fields(); + assertThat(partitionFields).hasSize(1); + assertThat(partitionFields.get(0).sourceId()).isEqualTo(1); + assertThat(partitionFields.get(0).fieldId()) + .isGreaterThanOrEqualTo(IcebergPartitionField.FIRST_FIELD_ID); + + // 3) v3 data manifest-list entries carry a non-null first_row_id + IcebergPathFactory paths = new IcebergPathFactory(new Path(table.location(), "metadata")); + IcebergManifestList manifestList = IcebergManifestList.create(table, paths); + List metas = + manifestList.read(new Path(metadata.currentSnapshot().manifestList()).getName()); + assertThat(metas).isNotEmpty(); + assertThat(metas.stream().filter(m -> m.content() == IcebergManifestFileMeta.Content.DATA)) + .allMatch(m -> m.firstRowId() != null); + + // 4) Parquet footers embed exactly the metadata ids + List parquetFiles = dataParquetFiles(table); + assertThat(parquetFiles).isNotEmpty(); + for (java.nio.file.Path file : parquetFiles) { + MessageType parquetSchema = readParquetSchema(file); + assertFieldId(parquetSchema, "pt", 1); + assertFieldId(parquetSchema, "k", 2); + assertFieldId(parquetSchema, "v", 3); + assertFieldId(parquetSchema, "nested", 4); + GroupType nestedGroup = parquetSchema.getType("nested").asGroupType(); + assertFieldId(nestedGroup, "a", 5); + assertFieldId(nestedGroup, "b", 6); + } + + // 5) Apache Iceberg reads the table (schema ids and data, with the DV applied) + HadoopCatalog icebergCatalog = new HadoopCatalog(new Configuration(), tempDir.toString()); + Table icebergTable = icebergCatalog.loadTable(TableIdentifier.of("mydb.db", "t")); + assertThat(icebergTable.schema().findField("pt").fieldId()).isEqualTo(1); + assertThat(icebergTable.schema().findField("k").fieldId()).isEqualTo(2); + assertThat(icebergTable.schema().findField("v").fieldId()).isEqualTo(3); + assertThat(icebergTable.schema().findField("nested").fieldId()).isEqualTo(4); + assertThat(icebergTable.schema().findField("nested.a").fieldId()).isEqualTo(5); + assertThat(icebergTable.schema().findField("nested.b").fieldId()).isEqualTo(6); + + Types.StructType structType = icebergTable.schema().asStruct(); + StructLikeSet actual = StructLikeSet.create(structType); + try (CloseableIterable reader = IcebergGenerics.read(icebergTable).build()) { + reader.forEach(actual::add); + } + org.apache.iceberg.data.GenericRecord expected = + org.apache.iceberg.data.GenericRecord.create(structType); + expected.set(0, 1); + expected.set(1, 1); + expected.set(2, "a"); + org.apache.iceberg.data.GenericRecord expectedNested = + org.apache.iceberg.data.GenericRecord.create( + structType.fieldType("nested").asStructType()); + expectedNested.set(0, 10); + expectedNested.set(1, "x"); + expected.set(3, expectedNested); + StructLikeSet expectedSet = StructLikeSet.create(structType); + expectedSet.add(expected); + assertThat(actual).isEqualTo(expectedSet); + + // 6) Paimon itself reads the strict-mode files + List rows = readPaimonRows(table); + assertThat(rows).hasSize(1); + InternalRow row = rows.get(0); + assertThat(row.getInt(0)).isEqualTo(1); + assertThat(row.getInt(1)).isEqualTo(1); + assertThat(row.getString(2).toString()).isEqualTo("a"); + assertThat(row.getRow(3, 2).getInt(0)).isEqualTo(10); + assertThat(row.getRow(3, 2).getString(1).toString()).isEqualTo("x"); + } + + @Test + public void testDefaultRemainsZeroBased() throws Exception { + Map customOptions = new HashMap<>(); + customOptions.put(IcebergOptions.FORMAT_VERSION.key(), "3"); + FileStoreTable table = createPaimonTable(customOptions); + + assertThat(table.schema().fields().stream().map(DataField::id)).containsExactly(0, 1, 2, 3); + + String commitUser = UUID.randomUUID().toString(); + TableWriteImpl write = + table.newWrite(commitUser) + .withIOManager(new IOManagerImpl(tempDir.toString() + "/tmp")); + TableCommitImpl commit = table.newCommit(commitUser); + write.write(row(RowKind.INSERT, 1, 1, "a", 10, "x")); + commit.commit(1, write.prepareCommit(false, 1)); + write.close(); + commit.close(); + + IcebergMetadata metadata = readLatestIcebergMetadata(table); + IcebergSchema currentSchema = metadata.schemas().get(metadata.currentSchemaId()); + assertThat(currentSchema.fields().stream().map(IcebergDataField::id)) + .containsExactly(0, 1, 2, 3); + } + + private FileStoreTable createPaimonTable(Map customOptions) throws Exception { + LocalFileIO fileIO = LocalFileIO.create(); + Path path = new Path(tempDir.toString()); + + Options options = new Options(customOptions); + options.set(CoreOptions.BUCKET, 1); + options.set( + IcebergOptions.METADATA_ICEBERG_STORAGE, IcebergOptions.StorageType.TABLE_LOCATION); + options.set(CoreOptions.FILE_FORMAT, "parquet"); + options.set(CoreOptions.TARGET_FILE_SIZE, MemorySize.ofKibiBytes(32)); + + Schema schema = + new Schema( + rowType().getFields(), + Collections.singletonList("pt"), + Arrays.asList("pt", "k"), + options.toMap(), + ""); + + try (FileSystemCatalog paimonCatalog = new FileSystemCatalog(fileIO, path)) { + paimonCatalog.createDatabase("mydb", false); + Identifier paimonIdentifier = Identifier.create("mydb", "t"); + paimonCatalog.createTable(paimonIdentifier, schema, false); + return (FileStoreTable) paimonCatalog.getTable(paimonIdentifier); + } + } + + private static GenericRow row(RowKind kind, int pt, int k, String v, int a, String b) { + return GenericRow.ofKind( + kind, + pt, + k, + BinaryString.fromString(v), + GenericRow.of(a, BinaryString.fromString(b))); + } + + private static BinaryRow partition(int pt) { + BinaryRow partition = new BinaryRow(1); + BinaryRowWriter writer = new BinaryRowWriter(partition); + writer.writeInt(0, pt); + writer.complete(); + return partition; + } + + private IcebergMetadata readLatestIcebergMetadata(FileStoreTable table) throws IOException { + java.nio.file.Path metadataDir = + java.nio.file.Paths.get(new Path(table.location(), "metadata").toUri().getPath()); + java.nio.file.Path latest; + try (Stream files = Files.list(metadataDir)) { + latest = + files.filter(f -> f.getFileName().toString().endsWith(".metadata.json")) + .max( + java.util.Comparator.comparingLong( + f -> + Long.parseLong( + f.getFileName() + .toString() + .replaceAll("[^0-9]", "")))) + .orElseThrow(() -> new IllegalStateException("no metadata.json found")); + } + return IcebergMetadata.fromPath(LocalFileIO.create(), new Path(latest.toUri())); + } + + private List dataParquetFiles(FileStoreTable table) throws IOException { + java.nio.file.Path tableDir = java.nio.file.Paths.get(table.location().toUri().getPath()); + try (Stream files = Files.walk(tableDir)) { + return files.filter(f -> f.getFileName().toString().endsWith(".parquet")) + .filter(f -> !f.toString().contains("/metadata/")) + .collect(Collectors.toList()); + } + } + + private MessageType readParquetSchema(java.nio.file.Path file) { + try (ParquetFileReader reader = + org.apache.paimon.format.parquet.ParquetUtil.getParquetReader( + LocalFileIO.create(), + new Path(file.toUri()), + Files.size(file), + new Options())) { + return reader.getFooter().getFileMetaData().getSchema(); + } catch (IOException e) { + throw new UncheckedIOException(e); + } + } + + private static void assertFieldId(GroupType parent, String name, int expectedId) { + org.apache.paimon.shade.org.apache.parquet.schema.Type field = parent.getType(name); + assertThat(field.getId()).as("field id of '%s'", name).isNotNull(); + assertThat(field.getId().intValue()).as("field id of '%s'", name).isEqualTo(expectedId); + } + + private static List collectAllFieldIds(List fields) { + List ids = new ArrayList<>(); + for (IcebergDataField field : fields) { + ids.add(field.id()); + if (field.type() instanceof IcebergStructType) { + ids.addAll(collectAllFieldIds(((IcebergStructType) field.type()).fields())); + } + } + return ids; + } + + private static List readPaimonRows(FileStoreTable table) throws Exception { + ReadBuilder readBuilder = table.newReadBuilder(); + List splits = readBuilder.newScan().plan().splits(); + List rows = new ArrayList<>(); + try (RecordReader reader = readBuilder.newRead().createReader(splits)) { + reader.forEachRemaining( + r -> + rows.add( + GenericRow.of( + r.getInt(0), + r.getInt(1), + r.getString(2).copy(), + GenericRow.of( + r.getRow(3, 2).getInt(0), + r.getRow(3, 2).getString(1).copy())))); + } + return rows; + } +} From 0998cf2908e774569764e417c8c2f84198e793cc Mon Sep 17 00:00:00 2001 From: Victor Babenko <37556649+vbabenkoru@users.noreply.github.com> Date: Mon, 24 Aug 2026 07:55:11 -0700 Subject: [PATCH 2/2] [core] Reject field-id.one-based changes unconditionally The general ALTER validation enforces immutable options only once the table has a snapshot, so between creation and the first write the value could be toggled: field ids are assigned once at creation and were correctly left untouched, but the stored option then permanently contradicted the actual id base. The value is now rejected exactly like 'type', regardless of snapshots; restating the effective value (e.g. an explicit default) stays allowed, and removing the option is rejected when the table was created one-based. Reported by JingsongLi in review. --- .../apache/paimon/schema/SchemaManager.java | 18 +++++++++++ .../paimon/schema/FieldIdOneBasedTest.java | 30 +++++++++++++++---- 2 files changed, 43 insertions(+), 5 deletions(-) diff --git a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaManager.java b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaManager.java index 6a3d809b5358..7c603b932fa4 100644 --- a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaManager.java +++ b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaManager.java @@ -351,6 +351,16 @@ public static TableSchema generateTableSchema( if (!unchanged && CoreOptions.TYPE.key().equals(setOption.key())) { throw new UnsupportedOperationException("Change 'type' is not supported yet."); } + // reject even without snapshots: field ids are assigned once at creation, + // so changing the value later only makes the option lie about the schema + // (restating the effective value, e.g. an explicit default, stays allowed) + if (CoreOptions.FIELD_ID_ONE_BASED.key().equals(setOption.key()) + && Boolean.parseBoolean(oldValue) != Boolean.parseBoolean(newValue)) { + throw new UnsupportedOperationException( + "Change '" + + CoreOptions.FIELD_ID_ONE_BASED.key() + + "' is not supported."); + } if (hasSnapshots.get() && !unchanged) { checkAlterTableOption(oldOptions, setOption.key(), oldValue, newValue); } @@ -362,6 +372,14 @@ public static TableSchema generateTableSchema( if (CoreOptions.TYPE.key().equals(removeOption.key())) { throw new UnsupportedOperationException("Change 'type' is not supported yet."); } + if (CoreOptions.FIELD_ID_ONE_BASED.key().equals(removeOption.key()) + && Boolean.parseBoolean(oldOptions.get(removeOption.key()))) { + // removing the option while it is true changes the effective value + throw new UnsupportedOperationException( + "Change '" + + CoreOptions.FIELD_ID_ONE_BASED.key() + + "' is not supported."); + } if (hasSnapshots.get()) { checkResetTableOption(oldOptions, removeOption.key()); } diff --git a/paimon-core/src/test/java/org/apache/paimon/schema/FieldIdOneBasedTest.java b/paimon-core/src/test/java/org/apache/paimon/schema/FieldIdOneBasedTest.java index 11f814392f37..35a2cfa0cc09 100644 --- a/paimon-core/src/test/java/org/apache/paimon/schema/FieldIdOneBasedTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/schema/FieldIdOneBasedTest.java @@ -35,6 +35,7 @@ import java.util.stream.Collectors; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; /** Tests for {@link CoreOptions#FIELD_ID_ONE_BASED} at table creation and evolution. */ public class FieldIdOneBasedTest { @@ -105,14 +106,33 @@ public void testEvolutionContinuesFromShiftedIds() throws Exception { public void testOneBasedImmutableAndCreateTimeOnly() throws Exception { // registered as immutable, so ALTER is rejected once the table has snapshots assertThat(CoreOptions.IMMUTABLE_OPTIONS).contains(CoreOptions.FIELD_ID_ONE_BASED.key()); - // a pre-snapshot option change must not re-base ids: assignment happens at creation only + // ids are assigned once at creation, so changing the value is rejected even before the + // first snapshot: the ids would keep their base while the option claims another one SchemaManager manager = newSchemaManager("t"); manager.createTable(schemaBuilder().build()); - TableSchema altered = + assertThatThrownBy( + () -> + manager.commitChanges( + SchemaChange.setOption( + CoreOptions.FIELD_ID_ONE_BASED.key(), "true"))) + .isInstanceOf(UnsupportedOperationException.class) + .hasMessageContaining(CoreOptions.FIELD_ID_ONE_BASED.key()); + // removing the option from a one-based table would change the effective value back + SchemaManager oneBased = newSchemaManager("t2"); + oneBased.createTable( + schemaBuilder().option(CoreOptions.FIELD_ID_ONE_BASED.key(), "true").build()); + assertThatThrownBy( + () -> + oneBased.commitChanges( + SchemaChange.removeOption( + CoreOptions.FIELD_ID_ONE_BASED.key()))) + .isInstanceOf(UnsupportedOperationException.class) + .hasMessageContaining(CoreOptions.FIELD_ID_ONE_BASED.key()); + // re-stating the current value is a no-op, not a change, and stays allowed + TableSchema unchanged = manager.commitChanges( - SchemaChange.setOption(CoreOptions.FIELD_ID_ONE_BASED.key(), "true")); - assertThat(topLevelIds(altered)).containsExactly(0, 1, 4); - assertThat(altered.highestFieldId()).isEqualTo(4); + SchemaChange.setOption(CoreOptions.FIELD_ID_ONE_BASED.key(), "false")); + assertThat(topLevelIds(unchanged)).containsExactly(0, 1, 4); } private static List topLevelIds(TableSchema schema) {