diff --git a/.palantir/revapi.yml b/.palantir/revapi.yml index a41d3ddfb8df..808a19299055 100644 --- a/.palantir/revapi.yml +++ b/.palantir/revapi.yml @@ -1018,6 +1018,12 @@ acceptedBreaks: old: "method void org.apache.iceberg.PositionDeletesTable.PositionDeletesBatchScan::(org.apache.iceberg.Table,\ \ org.apache.iceberg.Schema, org.apache.iceberg.TableScanContext)" justification: "Removing deprecated code" + "1.5.0": + org.apache.iceberg:iceberg-api: + - code: "java.class.defaultSerializationChanged" + old: "class org.apache.iceberg.types.Types.NestedField" + new: "class org.apache.iceberg.types.Types.NestedField" + justification: "new Constructor added" apache-iceberg-0.14.0: org.apache.iceberg:iceberg-api: - code: "java.class.defaultSerializationChanged" diff --git a/api/src/main/java/org/apache/iceberg/PartitionSpec.java b/api/src/main/java/org/apache/iceberg/PartitionSpec.java index 4fcb110db87c..8f1df794030a 100644 --- a/api/src/main/java/org/apache/iceberg/PartitionSpec.java +++ b/api/src/main/java/org/apache/iceberg/PartitionSpec.java @@ -27,6 +27,7 @@ import java.util.Map; import java.util.Set; import java.util.concurrent.atomic.AtomicInteger; +import java.util.stream.Collectors; import org.apache.iceberg.exceptions.ValidationException; import org.apache.iceberg.relocated.com.google.common.base.Preconditions; import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList; @@ -60,6 +61,7 @@ public class PartitionSpec implements Serializable { private transient volatile ListMultimap fieldsBySourceId = null; private transient volatile Class[] lazyJavaClasses = null; private transient volatile StructType lazyPartitionType = null; + private transient volatile StructType lazyRawPartitionType = null; private transient volatile List fieldList = null; private final int lastAssignedFieldId; @@ -140,6 +142,30 @@ public StructType partitionType() { return lazyPartitionType; } + /** + * Returns a struct matching partition information as written into manifest files. See {@link + * #partitionType()} for a struct with field ID's potentially re-assigned to avoid conflict. + */ + public StructType rawPartitionType() { + if (schema.idsToOriginal().isEmpty()) { + // not re-assigned. + return partitionType(); + } + if (lazyRawPartitionType == null) { + synchronized (this) { + if (lazyRawPartitionType == null) { + this.lazyRawPartitionType = + StructType.of( + partitionType().fields().stream() + .map(f -> f.withFieldId(schema.idsToOriginal().get(f.fieldId()))) + .collect(Collectors.toList())); + } + } + } + + return lazyRawPartitionType; + } + public Class[] javaClasses() { if (lazyJavaClasses == null) { synchronized (this) { diff --git a/api/src/main/java/org/apache/iceberg/Schema.java b/api/src/main/java/org/apache/iceberg/Schema.java index 5e024b7c1c29..d5ec3f250982 100644 --- a/api/src/main/java/org/apache/iceberg/Schema.java +++ b/api/src/main/java/org/apache/iceberg/Schema.java @@ -21,6 +21,7 @@ import java.io.Serializable; import java.util.Arrays; import java.util.Collection; +import java.util.Collections; import java.util.Deque; import java.util.List; import java.util.Locale; @@ -34,6 +35,7 @@ import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet; import org.apache.iceberg.relocated.com.google.common.collect.Lists; +import org.apache.iceberg.relocated.com.google.common.collect.Maps; import org.apache.iceberg.relocated.com.google.common.collect.Sets; import org.apache.iceberg.relocated.com.google.common.primitives.Ints; import org.apache.iceberg.types.Type; @@ -65,6 +67,8 @@ public class Schema implements Serializable { private transient Map> idToAccessor = null; private transient Map idToName = null; private transient Set identifierFieldIdSet = null; + private transient Map idsToReassigned; + private transient Map idsToOriginal; public Schema(List columns, Map aliases) { this(columns, aliases, ImmutableSet.of()); @@ -83,12 +87,24 @@ public Schema(List columns, Set identifierFieldIds) { this(DEFAULT_SCHEMA_ID, columns, identifierFieldIds); } + public Schema(List columns, Set identifierFieldIds, TypeUtil.GetID getId) { + this(DEFAULT_SCHEMA_ID, columns, identifierFieldIds, getId); + } + public Schema(int schemaId, List columns) { this(schemaId, columns, ImmutableSet.of()); } public Schema(int schemaId, List columns, Set identifierFieldIds) { - this(schemaId, columns, null, identifierFieldIds); + this(schemaId, columns, null, identifierFieldIds, null); + } + + public Schema( + int schemaId, + List columns, + Set identifierFieldIds, + TypeUtil.GetID getId) { + this(schemaId, columns, null, identifierFieldIds, getId); } public Schema( @@ -96,8 +112,22 @@ public Schema( List columns, Map aliases, Set identifierFieldIds) { + this(schemaId, columns, aliases, identifierFieldIds, null); + } + + public Schema( + int schemaId, + List columns, + Map aliases, + Set identifierFieldIds, + TypeUtil.GetID getID) { this.schemaId = schemaId; - this.struct = StructType.of(columns); + + this.idsToOriginal = Maps.newHashMap(); + this.idsToReassigned = Maps.newHashMap(); + List finalColumns = reassignIds(columns, getID); + + this.struct = StructType.of(finalColumns); this.aliasToId = aliases != null ? ImmutableBiMap.copyOf(aliases) : null; // validate IdentifierField @@ -507,4 +537,40 @@ public String toString() { .map(this::identifierFieldToString) .collect(Collectors.toList()))); } + + /** + * The ID's of some fields will be re-assigned if GetID is specified for the Schema. + * + * @return map of original to reassigned field ids + */ + public Map idsToReassigned() { + return idsToReassigned != null ? idsToReassigned : Collections.emptyMap(); + } + + /** + * The ID's of some fields will be re-assigned if GetID is specified for the Schema. + * + * @return map of reassigned to original field ids + */ + public Map idsToOriginal() { + return idsToOriginal != null ? idsToOriginal : Collections.emptyMap(); + } + + private List reassignIds(List columns, TypeUtil.GetID getID) { + if (getID == null) { + return columns; + } + Type res = + TypeUtil.assignIds( + StructType.of(columns), + oldId -> { + int newId = getID.get(oldId); + if (newId != oldId) { + idsToReassigned.put(oldId, newId); + idsToOriginal.put(newId, oldId); + } + return newId; + }); + return res.asStructType().fields(); + } } diff --git a/api/src/main/java/org/apache/iceberg/types/AssignIds.java b/api/src/main/java/org/apache/iceberg/types/AssignIds.java new file mode 100644 index 000000000000..68588f581adc --- /dev/null +++ b/api/src/main/java/org/apache/iceberg/types/AssignIds.java @@ -0,0 +1,99 @@ +/* + * 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.iceberg.types; + +import java.util.Iterator; +import java.util.List; +import java.util.function.Supplier; +import org.apache.iceberg.Schema; +import org.apache.iceberg.relocated.com.google.common.collect.Lists; + +class AssignIds extends TypeUtil.CustomOrderSchemaVisitor { + private final TypeUtil.GetID getID; + + AssignIds(TypeUtil.GetID getID) { + this.getID = getID; + } + + private int idFor(int id) { + return getID.get(id); + } + + @Override + public Type schema(Schema schema, Supplier future) { + return future.get(); + } + + @Override + public Type struct(Types.StructType struct, Iterable futures) { + List fields = struct.fields(); + int length = struct.fields().size(); + + // assign IDs for this struct's fields first + List newIds = Lists.newArrayListWithExpectedSize(length); + for (Types.NestedField field : fields) { + newIds.add(idFor(field.fieldId())); + } + + List newFields = Lists.newArrayListWithExpectedSize(length); + Iterator types = futures.iterator(); + for (int i = 0; i < length; i += 1) { + Types.NestedField field = fields.get(i); + Type type = types.next(); + if (field.isOptional()) { + newFields.add(Types.NestedField.optional(newIds.get(i), field.name(), type, field.doc())); + } else { + newFields.add(Types.NestedField.required(newIds.get(i), field.name(), type, field.doc())); + } + } + + return Types.StructType.of(newFields); + } + + @Override + public Type field(Types.NestedField field, Supplier future) { + return future.get(); + } + + @Override + public Type list(Types.ListType list, Supplier future) { + int newId = idFor(list.elementId()); + if (list.isElementOptional()) { + return Types.ListType.ofOptional(newId, future.get()); + } else { + return Types.ListType.ofRequired(newId, future.get()); + } + } + + @Override + public Type map(Types.MapType map, Supplier keyFuture, Supplier valueFuture) { + int newKeyId = idFor(map.keyId()); + int newValueId = idFor(map.valueId()); + if (map.isValueOptional()) { + return Types.MapType.ofOptional(newKeyId, newValueId, keyFuture.get(), valueFuture.get()); + } else { + return Types.MapType.ofRequired(newKeyId, newValueId, keyFuture.get(), valueFuture.get()); + } + } + + @Override + public Type primitive(Type.PrimitiveType primitive) { + return primitive; + } +} diff --git a/api/src/main/java/org/apache/iceberg/types/TypeUtil.java b/api/src/main/java/org/apache/iceberg/types/TypeUtil.java index 7c13d6094084..07d06dcc5a89 100644 --- a/api/src/main/java/org/apache/iceberg/types/TypeUtil.java +++ b/api/src/main/java/org/apache/iceberg/types/TypeUtil.java @@ -355,6 +355,17 @@ public static Schema reassignOrRefreshIds( return new Schema(struct.fields(), refreshIdentifierFields(struct, schema)); } + /** + * Assigns fresh ids from the {@link GetID getId function} for all fields in a type. + * + * @param type a type + * @param getId an id assignment function + * @return an structurally identical type with new ids assigned by the getId function + */ + public static Type assignIds(Type type, GetID getId) { + return TypeUtil.visit(type, new AssignIds(getId)); + } + public static Type find(Schema schema, Predicate predicate) { return visit(schema, new FindTypeVisitor(predicate)); } @@ -521,6 +532,11 @@ public interface NextID { int get(); } + /** Interface for passing a function that assigns column IDs from the previous Id. */ + public interface GetID { + int get(int oldId); + } + public static class SchemaVisitor { public void beforeField(Types.NestedField field) {} diff --git a/api/src/main/java/org/apache/iceberg/types/Types.java b/api/src/main/java/org/apache/iceberg/types/Types.java index dda842c9e161..ce6caa4721df 100644 --- a/api/src/main/java/org/apache/iceberg/types/Types.java +++ b/api/src/main/java/org/apache/iceberg/types/Types.java @@ -475,6 +475,10 @@ public NestedField asRequired() { return new NestedField(false, id, name, type, doc); } + public NestedField withFieldId(int newId) { + return new NestedField(isOptional, newId, name, type, doc); + } + public int fieldId() { return id; } diff --git a/core/src/main/java/org/apache/iceberg/BaseMetadataTable.java b/core/src/main/java/org/apache/iceberg/BaseMetadataTable.java index 57a6386093d6..e1e138109f8e 100644 --- a/core/src/main/java/org/apache/iceberg/BaseMetadataTable.java +++ b/core/src/main/java/org/apache/iceberg/BaseMetadataTable.java @@ -74,8 +74,11 @@ static PartitionSpec transformSpec(Schema metadataTableSchema, PartitionSpec spe .withSpecId(spec.specId()) .checkConflicts(false); + Map reassignedFields = metadataTableSchema.idsToReassigned(); + for (PartitionField field : spec.fields()) { - builder.add(field.fieldId(), field.fieldId(), field.name(), Transforms.identity()); + int newFieldId = reassignedFields.getOrDefault(field.fieldId(), field.fieldId()); + builder.add(newFieldId, newFieldId, field.name(), Transforms.identity()); } return builder.build(); } diff --git a/core/src/main/java/org/apache/iceberg/ManifestReader.java b/core/src/main/java/org/apache/iceberg/ManifestReader.java index 4ee51aa60c31..b5f85813dd2f 100644 --- a/core/src/main/java/org/apache/iceberg/ManifestReader.java +++ b/core/src/main/java/org/apache/iceberg/ManifestReader.java @@ -114,7 +114,7 @@ protected ManifestReader( this.spec = readPartitionSpec(file); } - this.fileSchema = new Schema(DataFile.getType(spec.partitionType()).fields()); + this.fileSchema = new Schema(DataFile.getType(spec.rawPartitionType()).fields()); } private > PartitionSpec readPartitionSpec(InputFile inputFile) { diff --git a/core/src/main/java/org/apache/iceberg/PositionDeletesTable.java b/core/src/main/java/org/apache/iceberg/PositionDeletesTable.java index 56270635809b..382ad663a8d1 100644 --- a/core/src/main/java/org/apache/iceberg/PositionDeletesTable.java +++ b/core/src/main/java/org/apache/iceberg/PositionDeletesTable.java @@ -24,6 +24,8 @@ import java.util.Collections; import java.util.List; import java.util.Map; +import java.util.Set; +import java.util.concurrent.atomic.AtomicInteger; import java.util.function.Function; import java.util.stream.Collectors; import org.apache.iceberg.expressions.Expression; @@ -33,6 +35,8 @@ import org.apache.iceberg.expressions.ResidualEvaluator; import org.apache.iceberg.io.CloseableIterable; import org.apache.iceberg.io.CloseableIterator; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet; import org.apache.iceberg.relocated.com.google.common.collect.Sets; import org.apache.iceberg.types.TypeUtil; import org.apache.iceberg.types.Types; @@ -107,8 +111,8 @@ public Map properties() { private Schema calculateSchema() { Types.StructType partitionType = Partitioning.partitionType(table()); - Schema result = - new Schema( + List columns = + ImmutableList.of( MetadataColumns.DELETE_FILE_PATH, MetadataColumns.DELETE_FILE_POS, Types.NestedField.optional( @@ -132,6 +136,35 @@ private Schema calculateSchema() { Types.StringType.get(), MetadataColumns.FILE_PATH_COLUMN_DOC)); + // Calculate used ids (for de-conflict) + Set currentlyUsedIds = + Collections.unmodifiableSet(TypeUtil.indexById(Types.StructType.of(columns)).keySet()); + Set allUsedIds = + table().schemas().values().stream() + .map(currSchema -> TypeUtil.indexById(currSchema.asStruct()).keySet()) + .reduce(currentlyUsedIds, Sets::union); + + // Calculate ids to reassign + Set idsToReassign = + partitionType.fields().stream().map(Types.NestedField::fieldId).collect(Collectors.toSet()); + + // Reassign selected ids to de-conflict with used ids. + AtomicInteger nextId = new AtomicInteger(); + Schema result = + new Schema( + columns, + ImmutableSet.of(), + oldId -> { + if (!idsToReassign.contains(oldId)) { + return oldId; + } + int candidate = nextId.incrementAndGet(); + while (allUsedIds.contains(candidate)) { + candidate = nextId.incrementAndGet(); + } + return candidate; + }); + if (!partitionType.fields().isEmpty()) { return result; } else { diff --git a/core/src/test/java/org/apache/iceberg/TestMetadataTableScans.java b/core/src/test/java/org/apache/iceberg/TestMetadataTableScans.java index 4c5f1d240f57..df314f6a802f 100644 --- a/core/src/test/java/org/apache/iceberg/TestMetadataTableScans.java +++ b/core/src/test/java/org/apache/iceberg/TestMetadataTableScans.java @@ -42,6 +42,7 @@ import org.apache.iceberg.relocated.com.google.common.collect.Iterators; import org.apache.iceberg.relocated.com.google.common.collect.Lists; import org.apache.iceberg.relocated.com.google.common.collect.Streams; +import org.apache.iceberg.types.TypeUtil; import org.apache.iceberg.types.Types; import org.apache.iceberg.util.StructLikeWrapper; import org.junit.jupiter.api.TestTemplate; @@ -1266,7 +1267,7 @@ public void testPositionDeletesWithFilter() { ScanTask task = tasks.get(0); assertThat(task).isInstanceOf(PositionDeletesScanTask.class); - Types.StructType partitionType = Partitioning.partitionType(table); + Types.StructType partitionType = positionDeletesTable.spec().partitionType(); PositionDeletesScanTask posDeleteTask = (PositionDeletesScanTask) task; int filePartition = posDeleteTask.file().partition().get(0, Integer.class); @@ -1333,7 +1334,7 @@ private void testPositionDeletesBaseTableFilter(boolean transactional) { ScanTask task = tasks.get(0); assertThat(task).isInstanceOf(PositionDeletesScanTask.class); - Types.StructType partitionType = Partitioning.partitionType(table); + Types.StructType partitionType = positionDeletesTable.spec().partitionType(); PositionDeletesScanTask posDeleteTask = (PositionDeletesScanTask) task; // base table filter should only be used to evaluate partitions @@ -1415,7 +1416,7 @@ public void testPositionDeletesWithBaseTableFilterNot() { ScanTask task = tasks.get(0); assertThat(task).isInstanceOf(PositionDeletesScanTask.class); - Types.StructType partitionType = Partitioning.partitionType(table); + Types.StructType partitionType = positionDeletesTable.spec().partitionType(); PositionDeletesScanTask posDeleteTask = (PositionDeletesScanTask) task; // base table filter should only be used to evaluate partitions @@ -1426,7 +1427,7 @@ public void testPositionDeletesWithBaseTableFilterNot() { (StructLike) constantsMap(posDeleteTask, partitionType).get(MetadataColumns.PARTITION_COLUMN_ID); int taskPartition = - taskPartitionStruct.get(1, Integer.class); // new partition field in position 1 + taskPartitionStruct.get(0, Integer.class); // new partition field in position 0 assertThat(filePartition).as("Expected correct partition on task's file").isEqualTo(1); assertThat(taskPartition).as("Expected correct partition on task's column").isEqualTo(1); @@ -1564,4 +1565,66 @@ public void testPositionDeletesUnpartitioned() { assertThat(scanTask1Partition).isEqualTo(expected); assertThat(scanTask2Partition).isEqualTo(expected); } + + @TestTemplate + public void testPositionDeletesManyColumns() { + assumeThat(formatVersion).as("Position deletes supported only for v2 tables").isEqualTo(2); + + UpdateSchema updateSchema = table.updateSchema(); + for (int i = 0; i <= 2000; i++) { + updateSchema.addColumn(String.valueOf(i), Types.IntegerType.get()); + } + updateSchema.commit(); + + DataFile dataFile1 = + DataFiles.builder(table.spec()) + .withPath("/path/to/data1.parquet") + .withFileSizeInBytes(10) + .withRecordCount(1) + .build(); + DataFile dataFile2 = + DataFiles.builder(table.spec()) + .withPath("/path/to/data2.parquet") + .withFileSizeInBytes(10) + .withRecordCount(1) + .build(); + table.newAppend().appendFile(dataFile1).appendFile(dataFile2).commit(); + + DeleteFile delete1 = + FileMetadata.deleteFileBuilder(table.spec()) + .ofPositionDeletes() + .withPath("/path/to/delete1.parquet") + .withFileSizeInBytes(10) + .withRecordCount(1) + .build(); + DeleteFile delete2 = + FileMetadata.deleteFileBuilder(table.spec()) + .ofPositionDeletes() + .withPath("/path/to/delete2.parquet") + .withFileSizeInBytes(10) + .withRecordCount(1) + .build(); + table.newRowDelta().addDeletes(delete1).addDeletes(delete2).commit(); + + PositionDeletesTable positionDeletesTable = new PositionDeletesTable(table); + assertThat(TypeUtil.indexById(positionDeletesTable.schema().asStruct()).size()).isEqualTo(2010); + + BatchScan scan = positionDeletesTable.newBatchScan(); + assertThat(scan).isInstanceOf(PositionDeletesTable.PositionDeletesBatchScan.class); + PositionDeletesTable.PositionDeletesBatchScan deleteScan = + (PositionDeletesTable.PositionDeletesBatchScan) scan; + + List scanTasks = + Lists.newArrayList( + Iterators.transform( + deleteScan.planFiles().iterator(), + task -> { + assertThat(task).isInstanceOf(PositionDeletesScanTask.class); + return (PositionDeletesScanTask) task; + })); + assertThat(scanTasks).hasSize(2); + scanTasks.sort(Comparator.comparing(f -> f.file().path().toString())); + assertThat(scanTasks.get(0).file().path().toString()).isEqualTo("/path/to/delete1.parquet"); + assertThat(scanTasks.get(1).file().path().toString()).isEqualTo("/path/to/delete2.parquet"); + } } diff --git a/core/src/test/java/org/apache/iceberg/TestMetadataTableScansWithPartitionEvolution.java b/core/src/test/java/org/apache/iceberg/TestMetadataTableScansWithPartitionEvolution.java index faccdcb3dd95..a2e5386d29df 100644 --- a/core/src/test/java/org/apache/iceberg/TestMetadataTableScansWithPartitionEvolution.java +++ b/core/src/test/java/org/apache/iceberg/TestMetadataTableScansWithPartitionEvolution.java @@ -21,7 +21,6 @@ import static org.apache.iceberg.types.Types.NestedField.optional; import static org.apache.iceberg.types.Types.NestedField.required; import static org.assertj.core.api.Assertions.assertThat; -import static org.assertj.core.api.Assertions.entry; import static org.assertj.core.api.Assumptions.assumeThat; import java.io.File; @@ -186,7 +185,7 @@ public void testPositionDeletesPartitionSpecRemoval() { ScanTask task = tasks.get(0); assertThat(task).isInstanceOf(PositionDeletesScanTask.class); - Types.StructType partitionType = Partitioning.partitionType(table); + Types.StructType partitionType = positionDeletesTable.spec().partitionType(); PositionDeletesScanTask posDeleteTask = (PositionDeletesScanTask) task; int filePartition = posDeleteTask.file().partition().get(0, Integer.class); @@ -196,15 +195,13 @@ public void testPositionDeletesPartitionSpecRemoval() { int taskConstantPartition = ((StructLike) constantsMap(posDeleteTask, partitionType).get(MetadataColumns.PARTITION_COLUMN_ID)) - .get(1, Integer.class); + .get(0, Integer.class); assertThat(taskConstantPartition) .as("Expected correct partition on constant column") .isEqualTo(1); - assertThat(posDeleteTask.spec().fields().get(0).fieldId()) .as("Expected correct partition field id on task's spec") - .isEqualTo(table.ops().current().spec().partitionType().fields().get(0).fieldId()); - + .isEqualTo(partitionType.fields().get(0).fieldId()); assertThat(posDeleteTask.file().specId()) .as("Expected correct partition spec id on task") .isEqualTo(table.ops().current().spec().specId()); diff --git a/spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/actions/RewritePositionDeleteFilesSparkAction.java b/spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/actions/RewritePositionDeleteFilesSparkAction.java index f3dfd2dcc364..ea1c52940175 100644 --- a/spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/actions/RewritePositionDeleteFilesSparkAction.java +++ b/spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/actions/RewritePositionDeleteFilesSparkAction.java @@ -137,10 +137,12 @@ public RewritePositionDeleteFiles.Result execute() { } private StructLikeMap>> planFileGroups() { - CloseableIterable fileTasks = planFiles(); + Table deletesTable = + MetadataTableUtils.createMetadataTableInstance(table, MetadataTableType.POSITION_DELETES); + CloseableIterable fileTasks = planFiles(deletesTable); try { - StructType partitionType = Partitioning.partitionType(table); + StructType partitionType = Partitioning.partitionType(deletesTable); StructLikeMap> fileTasksByPartition = groupByPartition(partitionType, fileTasks); return fileGroupsByPartition(fileTasksByPartition); @@ -153,10 +155,7 @@ private StructLikeMap>> planFileGroups() { } } - private CloseableIterable planFiles() { - Table deletesTable = - MetadataTableUtils.createMetadataTableInstance(table, MetadataTableType.POSITION_DELETES); - + private CloseableIterable planFiles(Table deletesTable) { PositionDeletesBatchScan scan = (PositionDeletesBatchScan) deletesTable.newBatchScan(); return CloseableIterable.transform( scan.baseTableFilter(filter).ignoreResiduals().planFiles(), diff --git a/spark/v3.3/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewritePositionDeleteFilesAction.java b/spark/v3.3/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewritePositionDeleteFilesAction.java index 77800e2ea007..aa2817e8753d 100644 --- a/spark/v3.3/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewritePositionDeleteFilesAction.java +++ b/spark/v3.3/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewritePositionDeleteFilesAction.java @@ -19,6 +19,8 @@ package org.apache.iceberg.spark.actions; import static org.apache.iceberg.types.Types.NestedField.optional; +import static org.apache.spark.sql.functions.expr; +import static org.assertj.core.api.Assertions.assertThat; import java.io.IOException; import java.util.Arrays; @@ -69,6 +71,7 @@ import org.apache.iceberg.util.StructLikeMap; import org.apache.spark.sql.Dataset; import org.apache.spark.sql.Row; +import org.apache.spark.sql.types.StructType; import org.junit.After; import org.junit.Assert; import org.junit.Rule; @@ -581,6 +584,69 @@ public void testSchemaEvolution() throws Exception { assertEquals("Rows must match", expectedRecords, actualRecords); } + @Test + public void testRewriteManyColumns() throws Exception { + List fields = + Lists.newArrayList(Types.NestedField.optional(0, "id", Types.LongType.get())); + List additionalCols = + IntStream.range(1, 1010) + .mapToObj(i -> Types.NestedField.optional(i, "c" + i, Types.StringType.get())) + .collect(Collectors.toList()); + fields.addAll(additionalCols); + Schema schema = new Schema(fields); + PartitionSpec spec = PartitionSpec.builderFor(schema).bucket("id", 2).build(); + Table table = + validationCatalog.createTable( + TableIdentifier.of("default", TABLE_NAME), schema, spec, tableProperties()); + + Dataset df = + spark + .range(4) + .withColumns( + IntStream.range(1, 1010) + .boxed() + .collect(Collectors.toMap(i -> "c" + i, i -> expr("CAST(id as STRING)")))); + StructType sparkSchema = spark.table(name(table)).schema(); + spark + .createDataFrame(df.rdd(), sparkSchema) + .coalesce(1) + .write() + .format("iceberg") + .mode("append") + .save(name(table)); + + List dataFiles = TestHelpers.dataFiles(table); + writePosDeletesForFiles(table, 1, 1, dataFiles); + assertThat(dataFiles).hasSize(2); + + List deleteFiles = deleteFiles(table); + assertThat(deleteFiles).hasSize(2); + + List expectedRecords = records(table); + List expectedDeletes = deleteRecords(table); + assertThat(expectedRecords).hasSize(2); + assertThat(expectedDeletes).hasSize(2); + + Result result = + SparkActions.get(spark) + .rewritePositionDeletes(table) + .option(SizeBasedFileRewriter.REWRITE_ALL, "true") + .option(SizeBasedFileRewriter.TARGET_FILE_SIZE_BYTES, Long.toString(Long.MAX_VALUE - 1)) + .execute(); + + List newDeleteFiles = deleteFiles(table); + assertThat(newDeleteFiles).hasSize(2); + assertNotContains(deleteFiles, newDeleteFiles); + assertLocallySorted(newDeleteFiles); + checkResult(result, deleteFiles, newDeleteFiles, 2); + checkSequenceNumbers(table, deleteFiles, newDeleteFiles); + + List actualRecords = records(table); + List actualDeletes = deleteRecords(table); + assertEquals("Rows must match", expectedRecords, actualRecords); + assertEquals("Position deletes must match", expectedDeletes, actualDeletes); + } + private Table createTablePartitioned(int partitions, int files, int numRecords) { PartitionSpec spec = PartitionSpec.builderFor(SCHEMA).identity("c1").build(); Table table = diff --git a/spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/actions/RewritePositionDeleteFilesSparkAction.java b/spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/actions/RewritePositionDeleteFilesSparkAction.java index f3dfd2dcc364..bdb0ee35273f 100644 --- a/spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/actions/RewritePositionDeleteFilesSparkAction.java +++ b/spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/actions/RewritePositionDeleteFilesSparkAction.java @@ -137,10 +137,12 @@ public RewritePositionDeleteFiles.Result execute() { } private StructLikeMap>> planFileGroups() { - CloseableIterable fileTasks = planFiles(); + Table deletesTable = + MetadataTableUtils.createMetadataTableInstance(table, MetadataTableType.POSITION_DELETES); + CloseableIterable fileTasks = planFiles(deletesTable); try { - StructType partitionType = Partitioning.partitionType(table); + StructType partitionType = Partitioning.partitionType(deletesTable); StructLikeMap> fileTasksByPartition = groupByPartition(partitionType, fileTasks); return fileGroupsByPartition(fileTasksByPartition); @@ -153,11 +155,9 @@ private StructLikeMap>> planFileGroups() { } } - private CloseableIterable planFiles() { - Table deletesTable = - MetadataTableUtils.createMetadataTableInstance(table, MetadataTableType.POSITION_DELETES); - + private CloseableIterable planFiles(Table deletesTable) { PositionDeletesBatchScan scan = (PositionDeletesBatchScan) deletesTable.newBatchScan(); + return CloseableIterable.transform( scan.baseTableFilter(filter).ignoreResiduals().planFiles(), task -> (PositionDeletesScanTask) task); diff --git a/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewritePositionDeleteFilesAction.java b/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewritePositionDeleteFilesAction.java index db2d42501d04..7be300e84fc6 100644 --- a/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewritePositionDeleteFilesAction.java +++ b/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewritePositionDeleteFilesAction.java @@ -19,6 +19,7 @@ package org.apache.iceberg.spark.actions; import static org.apache.iceberg.types.Types.NestedField.optional; +import static org.apache.spark.sql.functions.expr; import static org.assertj.core.api.Assertions.assertThat; import java.io.IOException; @@ -71,6 +72,7 @@ import org.apache.iceberg.util.StructLikeMap; import org.apache.spark.sql.Dataset; import org.apache.spark.sql.Row; +import org.apache.spark.sql.types.StructType; import org.junit.After; import org.junit.Assert; import org.junit.Rule; @@ -618,6 +620,69 @@ public void testSchemaEvolution() throws Exception { assertEquals("Rows must match", expectedRecords, actualRecords); } + @Test + public void testRewriteManyColumns() throws Exception { + List fields = + Lists.newArrayList(Types.NestedField.optional(0, "id", Types.LongType.get())); + List additionalCols = + IntStream.range(1, 1010) + .mapToObj(i -> Types.NestedField.optional(i, "c" + i, Types.StringType.get())) + .collect(Collectors.toList()); + fields.addAll(additionalCols); + Schema schema = new Schema(fields); + PartitionSpec spec = PartitionSpec.builderFor(schema).bucket("id", 2).build(); + Table table = + validationCatalog.createTable( + TableIdentifier.of("default", TABLE_NAME), schema, spec, tableProperties()); + + Dataset df = + spark + .range(4) + .withColumns( + IntStream.range(1, 1010) + .boxed() + .collect(Collectors.toMap(i -> "c" + i, i -> expr("CAST(id as STRING)")))); + StructType sparkSchema = spark.table(name(table)).schema(); + spark + .createDataFrame(df.rdd(), sparkSchema) + .coalesce(1) + .write() + .format("iceberg") + .mode("append") + .save(name(table)); + + List dataFiles = TestHelpers.dataFiles(table); + writePosDeletesForFiles(table, 1, 1, dataFiles); + assertThat(dataFiles).hasSize(2); + + List deleteFiles = deleteFiles(table); + assertThat(deleteFiles).hasSize(2); + + List expectedRecords = records(table); + List expectedDeletes = deleteRecords(table); + assertThat(expectedRecords).hasSize(2); + assertThat(expectedDeletes).hasSize(2); + + Result result = + SparkActions.get(spark) + .rewritePositionDeletes(table) + .option(SizeBasedFileRewriter.REWRITE_ALL, "true") + .option(SizeBasedFileRewriter.TARGET_FILE_SIZE_BYTES, Long.toString(Long.MAX_VALUE - 1)) + .execute(); + + List newDeleteFiles = deleteFiles(table); + assertThat(newDeleteFiles).hasSize(2); + assertNotContains(deleteFiles, newDeleteFiles); + assertLocallySorted(newDeleteFiles); + checkResult(result, deleteFiles, newDeleteFiles, 2); + checkSequenceNumbers(table, deleteFiles, newDeleteFiles); + + List actualRecords = records(table); + List actualDeletes = deleteRecords(table); + assertEquals("Rows must match", expectedRecords, actualRecords); + assertEquals("Position deletes must match", expectedDeletes, actualDeletes); + } + private Table createTablePartitioned(int partitions, int files, int numRecords) { PartitionSpec spec = PartitionSpec.builderFor(SCHEMA).identity("c1").build(); Table table = diff --git a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/RewritePositionDeleteFilesSparkAction.java b/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/RewritePositionDeleteFilesSparkAction.java index 539f6de92007..1166740f441a 100644 --- a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/RewritePositionDeleteFilesSparkAction.java +++ b/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/RewritePositionDeleteFilesSparkAction.java @@ -137,10 +137,12 @@ public RewritePositionDeleteFiles.Result execute() { } private StructLikeMap>> planFileGroups() { - CloseableIterable fileTasks = planFiles(); + Table deletesTable = + MetadataTableUtils.createMetadataTableInstance(table, MetadataTableType.POSITION_DELETES); + CloseableIterable fileTasks = planFiles(deletesTable); try { - StructType partitionType = Partitioning.partitionType(table); + StructType partitionType = Partitioning.partitionType(deletesTable); StructLikeMap> fileTasksByPartition = groupByPartition(partitionType, fileTasks); return fileGroupsByPartition(fileTasksByPartition); @@ -153,10 +155,7 @@ private StructLikeMap>> planFileGroups() { } } - private CloseableIterable planFiles() { - Table deletesTable = - MetadataTableUtils.createMetadataTableInstance(table, MetadataTableType.POSITION_DELETES); - + private CloseableIterable planFiles(Table deletesTable) { PositionDeletesBatchScan scan = (PositionDeletesBatchScan) deletesTable.newBatchScan(); return CloseableIterable.transform( scan.baseTableFilter(filter).ignoreResiduals().planFiles(), diff --git a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewritePositionDeleteFilesAction.java b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewritePositionDeleteFilesAction.java index 89c44dbfccf8..37b6cd86fb92 100644 --- a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewritePositionDeleteFilesAction.java +++ b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewritePositionDeleteFilesAction.java @@ -19,6 +19,7 @@ package org.apache.iceberg.spark.actions; import static org.apache.iceberg.types.Types.NestedField.optional; +import static org.apache.spark.sql.functions.expr; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.fail; @@ -77,6 +78,7 @@ import org.apache.iceberg.util.StructLikeMap; import org.apache.spark.sql.Dataset; import org.apache.spark.sql.Row; +import org.apache.spark.sql.types.StructType; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.TestTemplate; import org.junit.jupiter.api.io.TempDir; @@ -647,6 +649,69 @@ public void testSnapshotProperty() throws Exception { assertThat(table.currentSnapshot().summary()).containsKeys(commitMetricsKeys); } + @TestTemplate + public void testRewriteManyColumns() throws Exception { + List fields = + Lists.newArrayList(Types.NestedField.required(0, "id", Types.LongType.get())); + List additionalCols = + IntStream.range(1, 1010) + .mapToObj(i -> Types.NestedField.optional(i, "c" + i, Types.StringType.get())) + .collect(Collectors.toList()); + fields.addAll(additionalCols); + Schema schema = new Schema(fields); + PartitionSpec spec = PartitionSpec.builderFor(schema).bucket("id", 2).build(); + Table table = + validationCatalog.createTable( + TableIdentifier.of("default", TABLE_NAME), schema, spec, tableProperties()); + + Dataset df = + spark + .range(4) + .withColumns( + IntStream.range(1, 1010) + .boxed() + .collect(Collectors.toMap(i -> "c" + i, i -> expr("CAST(id as STRING)")))); + StructType sparkSchema = spark.table(name(table)).schema(); + spark + .createDataFrame(df.rdd(), sparkSchema) + .coalesce(1) + .write() + .format("iceberg") + .mode("append") + .save(name(table)); + + List dataFiles = TestHelpers.dataFiles(table); + writePosDeletesForFiles(table, 1, 1, dataFiles); + assertThat(dataFiles).hasSize(2); + + List deleteFiles = deleteFiles(table); + assertThat(deleteFiles).hasSize(2); + + List expectedRecords = records(table); + List expectedDeletes = deleteRecords(table); + assertThat(expectedRecords).hasSize(2); + assertThat(expectedDeletes).hasSize(2); + + Result result = + SparkActions.get(spark) + .rewritePositionDeletes(table) + .option(SizeBasedFileRewriter.REWRITE_ALL, "true") + .option(SizeBasedFileRewriter.TARGET_FILE_SIZE_BYTES, Long.toString(Long.MAX_VALUE - 1)) + .execute(); + + List newDeleteFiles = deleteFiles(table); + assertThat(newDeleteFiles).hasSize(2); + assertNotContains(deleteFiles, newDeleteFiles); + assertLocallySorted(newDeleteFiles); + checkResult(result, deleteFiles, newDeleteFiles, 2); + checkSequenceNumbers(table, deleteFiles, newDeleteFiles); + + List actualRecords = records(table); + List actualDeletes = deleteRecords(table); + assertEquals("Rows must match", expectedRecords, actualRecords); + assertEquals("Position deletes must match", expectedDeletes, actualDeletes); + } + private Table createTablePartitioned(int partitions, int files, int numRecords) { PartitionSpec spec = PartitionSpec.builderFor(SCHEMA).identity("c1").build(); Table table =