[core] Add field-id.one-based option for strictly positive Iceberg field ids

docs/generated/core_configuration.html

@@ -662,6 +662,12 @@
             <td>Boolean</td>
             <td>Optional endInput check partition expire used in case of batch mode or bounded stream.</td>
         </tr>
+        <tr>
+            <td><h5>field-id.one-based</h5></td>
+            <td style="word-wrap: break-word;">false</td>
+            <td>Boolean</td>
+            <td>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.</td>
+        </tr>
         <tr>
             <td><h5>fields.default-aggregate-function</h5></td>
             <td style="word-wrap: break-word;">(none)</td>

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<Boolean> 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<Boolean> 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);
     }

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<DataType> {
+
+    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<DataField> 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;
+    }
+}

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());
                 }

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<Integer> topLevelIds(TableSchema schema) {
+        return schema.fields().stream().map(DataField::id).collect(Collectors.toList());
+    }
+}