diff --git a/core/src/main/java/org/apache/iceberg/TableMetadata.java b/core/src/main/java/org/apache/iceberg/TableMetadata.java index 89427e4c6943..fb883a686578 100644 --- a/core/src/main/java/org/apache/iceberg/TableMetadata.java +++ b/core/src/main/java/org/apache/iceberg/TableMetadata.java @@ -43,7 +43,9 @@ import org.apache.iceberg.relocated.com.google.common.collect.Maps; import org.apache.iceberg.relocated.com.google.common.collect.Sets; import org.apache.iceberg.transforms.Transforms; +import org.apache.iceberg.types.Type; import org.apache.iceberg.types.TypeUtil; +import org.apache.iceberg.types.Types; import org.apache.iceberg.util.LocationUtil; import org.apache.iceberg.util.Pair; import org.apache.iceberg.util.PartitionUtil; @@ -85,6 +87,51 @@ public static TableMetadata newTableMetadata( return newTableMetadata(schema, spec, SortOrder.unsorted(), location, properties); } + /** + * Creates metadata for a new table, optionally preserving the field IDs assigned on the provided + * schema instead of reassigning fresh ones. + * + *

By default (and in the other {@code newTableMetadata} overloads) the schema's field IDs are + * reassigned fresh and sequentially to guarantee a consistent, canonical ID layout for a new + * table. Passing {@code reassignFieldIds = false} keeps the IDs already present on {@code schema} + * exactly as-is. This is intended for advanced use cases where the IDs are already meaningful — + * for example migrating or re-registering a table whose data files were written against a + * specific ID assignment. The provided IDs must be positive; otherwise an {@link + * IllegalArgumentException} is thrown. + * + *

The partition spec and sort order are always rebuilt against the resulting schema with a new + * table's initial spec and sort-order IDs. Their source field IDs follow the schema, so they are + * preserved as well when {@code reassignFieldIds = false}. + * + * @param schema a schema + * @param spec a partition spec + * @param sortOrder a sort order + * @param location a location for the table; may be null + * @param properties a string map of table properties + * @param reassignFieldIds whether to reassign fresh field IDs ({@code true}, matching the other + * overloads) or preserve the schema's existing field IDs ({@code false}) + * @return the new table metadata + */ + public static TableMetadata newTableMetadata( + Schema schema, + PartitionSpec spec, + SortOrder sortOrder, + String location, + Map properties, + boolean reassignFieldIds) { + int formatVersion = + PropertyUtil.propertyAsInt( + properties, TableProperties.FORMAT_VERSION, DEFAULT_TABLE_FORMAT_VERSION); + return newTableMetadata( + schema, + spec, + sortOrder, + location, + persistedProperties(properties), + formatVersion, + reassignFieldIds); + } + private static Map unreservedProperties(Map rawProperties) { return rawProperties.entrySet().stream() .filter(e -> !TableProperties.RESERVED_PROPERTIES.contains(e.getKey())) @@ -113,21 +160,42 @@ static TableMetadata newTableMetadata( String location, Map properties, int formatVersion) { + return newTableMetadata(schema, spec, sortOrder, location, properties, formatVersion, true); + } + + private static TableMetadata newTableMetadata( + Schema schema, + PartitionSpec spec, + SortOrder sortOrder, + String location, + Map properties, + int formatVersion, + boolean reassignFieldIds) { Preconditions.checkArgument( properties.keySet().stream().noneMatch(TableProperties.RESERVED_PROPERTIES::contains), "Table properties should not contain reserved properties, but got %s", properties); - // reassign all column ids to ensure consistency - AtomicInteger lastColumnId = new AtomicInteger(0); - Schema freshSchema = - TypeUtil.assignFreshIds(INITIAL_SCHEMA_ID, schema, lastColumnId::incrementAndGet); + Schema freshSchema; + int lastColumnId; + if (reassignFieldIds) { + // reassign all column ids to ensure consistency + AtomicInteger lastColumnIdTracker = new AtomicInteger(0); + freshSchema = + TypeUtil.assignFreshIds(INITIAL_SCHEMA_ID, schema, lastColumnIdTracker::incrementAndGet); + lastColumnId = lastColumnIdTracker.get(); + } else { + // preserve the field ids already assigned on the provided schema + validateFieldIdsForReuse(schema); + freshSchema = new Schema(INITIAL_SCHEMA_ID, schema.columns(), schema.identifierFieldIds()); + lastColumnId = schema.highestFieldId(); + } - // rebuild the partition spec using the new column ids + // rebuild the partition spec using the fresh schema's column ids PartitionSpec.Builder specBuilder = PartitionSpec.builderFor(freshSchema).withSpecId(INITIAL_SPEC_ID); for (PartitionField field : spec.fields()) { - // look up the name of the source field in the old schema to get the new schema's id + // look up the name of the source field in the old schema to get the fresh schema's id String sourceName = schema.findColumnName(field.sourceId()); // reassign all partition fields with fresh partition field Ids to ensure consistency specBuilder.add(freshSchema.findField(sourceName).fieldId(), field.name(), field.transform()); @@ -148,7 +216,7 @@ static TableMetadata newTableMetadata( return new Builder() .setInitialFormatVersion(formatVersion) - .setCurrentSchema(freshSchema, lastColumnId.get()) + .setCurrentSchema(freshSchema, lastColumnId) .setDefaultPartitionSpec(freshSpec) .setDefaultSortOrder(freshSortOrder) .setLocation(location) @@ -156,6 +224,28 @@ static TableMetadata newTableMetadata( .build(); } + private static void validateFieldIdsForReuse(Schema schema) { + // Schema construction already guarantees field ids are unique; ensure they are also positive so + // the preserved ids are valid for a new table. + List fieldIds = Lists.newArrayList(); + collectFieldIds(schema.asStruct(), fieldIds); + for (int fieldId : fieldIds) { + Preconditions.checkArgument( + fieldId > 0, + "Cannot preserve field IDs: field ID must be positive, but found %s", + fieldId); + } + } + + private static void collectFieldIds(Type type, List fieldIds) { + if (type.isNestedType()) { + for (Types.NestedField field : type.asNestedType().fields()) { + fieldIds.add(field.fieldId()); + collectFieldIds(field.type(), fieldIds); + } + } + } + public static class SnapshotLogEntry implements HistoryEntry { private final long timestampMillis; private final long snapshotId; diff --git a/core/src/test/java/org/apache/iceberg/TestTableMetadata.java b/core/src/test/java/org/apache/iceberg/TestTableMetadata.java index 02c23405033a..9032ceeae1ff 100644 --- a/core/src/test/java/org/apache/iceberg/TestTableMetadata.java +++ b/core/src/test/java/org/apache/iceberg/TestTableMetadata.java @@ -1283,6 +1283,90 @@ public void testNewTableMetadataReassignmentAllIds() throws Exception { assertThat(metadata.spec()).isEqualTo(expected); } + @Test + public void testNewTableMetadataPreservesFieldIds() { + Schema schema = + new Schema( + Types.NestedField.required(3, "x", Types.LongType.get()), + Types.NestedField.required(4, "y", Types.LongType.get()), + Types.NestedField.required( + 5, + "point", + Types.StructType.of( + Types.NestedField.required(6, "lat", Types.DoubleType.get()), + Types.NestedField.required(7, "long", Types.DoubleType.get())))); + + PartitionSpec spec = + PartitionSpec.builderFor(schema) + .withSpecId(5) + .add(3, 1005, "x_partition", Transforms.bucket(4)) + .build(); + String location = "file://tmp/db/table"; + + TableMetadata metadata = + TableMetadata.newTableMetadata( + schema, spec, SortOrder.unsorted(), location, ImmutableMap.of(), false); + + // field ids are preserved exactly, including nested fields + assertThat(metadata.schema().asStruct()).isEqualTo(schema.asStruct()); + + // the schema id is reset to the initial id for a new table + assertThat(metadata.schema().schemaId()).isEqualTo(0); + + // last-column-id tracks the highest preserved id so future columns don't collide + assertThat(metadata.lastColumnId()).isEqualTo(7); + + // the partition spec keeps the original source id, with a new table's spec and field ids + PartitionSpec expected = + PartitionSpec.builderFor(metadata.schema()) + .withSpecId(0) + .add(3, 1000, "x_partition", Transforms.bucket(4)) + .build(); + assertThat(metadata.spec()).isEqualTo(expected); + } + + @Test + public void testNewTableMetadataReassignFieldIdsTrueMatchesDefault() { + Schema schema = + new Schema( + Types.NestedField.required(3, "x", Types.LongType.get()), + Types.NestedField.required(4, "y", Types.LongType.get()), + Types.NestedField.required(5, "z", Types.LongType.get())); + String location = "file://tmp/db/table"; + + TableMetadata reassigned = + TableMetadata.newTableMetadata( + schema, + PartitionSpec.unpartitioned(), + SortOrder.unsorted(), + location, + ImmutableMap.of(), + true); + + // reassignFieldIds=true matches the existing fresh, sequential id assignment + assertThat(reassigned.schema().columns()) + .extracting(Types.NestedField::fieldId) + .containsExactly(1, 2, 3); + assertThat(reassigned.lastColumnId()).isEqualTo(3); + } + + @Test + public void testNewTableMetadataPreserveRejectsNonPositiveFieldIds() { + Schema schema = new Schema(Types.NestedField.required(0, "x", Types.LongType.get())); + + assertThatThrownBy( + () -> + TableMetadata.newTableMetadata( + schema, + PartitionSpec.unpartitioned(), + SortOrder.unsorted(), + "file://tmp/db/table", + ImmutableMap.of(), + false)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("field ID must be positive"); + } + @Test public void testInvalidUpdatePartitionSpecForV1Table() throws Exception { Schema schema = new Schema(Types.NestedField.required(1, "x", Types.LongType.get()));