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..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 @@ -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. @@ -337,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); } @@ -348,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 new file mode 100644 index 000000000000..35a2cfa0cc09 --- /dev/null +++ b/paimon-core/src/test/java/org/apache/paimon/schema/FieldIdOneBasedTest.java @@ -0,0 +1,141 @@ +/* + * 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; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** 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()); + // 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()); + 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(), "false")); + assertThat(topLevelIds(unchanged)).containsExactly(0, 1, 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; + } +}