From c8288b0f01ecee745b31dc77d900e21b518079cf Mon Sep 17 00:00:00 2001 From: Szehon Ho Date: Thu, 21 Mar 2024 13:47:43 -0700 Subject: [PATCH 01/15] Core: Calling rewrite_position_delete_files fails on tables with more than 1k columns --- .../main/java/org/apache/iceberg/Schema.java | 9 ++ .../apache/iceberg/types/AssignFreshIds.java | 83 +----------- .../org/apache/iceberg/types/AssignIds.java | 37 ++++++ .../apache/iceberg/types/BaseAssignIds.java | 120 ++++++++++++++++++ .../org/apache/iceberg/types/TypeUtil.java | 11 ++ .../apache/iceberg/PositionDeletesTable.java | 30 ++++- .../TestRewritePositionDeleteFilesAction.java | 65 ++++++++++ 7 files changed, 275 insertions(+), 80 deletions(-) create mode 100644 api/src/main/java/org/apache/iceberg/types/AssignIds.java create mode 100644 api/src/main/java/org/apache/iceberg/types/BaseAssignIds.java diff --git a/api/src/main/java/org/apache/iceberg/Schema.java b/api/src/main/java/org/apache/iceberg/Schema.java index 5e024b7c1c29..e8ed1c3d109b 100644 --- a/api/src/main/java/org/apache/iceberg/Schema.java +++ b/api/src/main/java/org/apache/iceberg/Schema.java @@ -242,6 +242,15 @@ public Map idToName() { return lazyIdToName(); } + /** + * Returns a map for this schema between qualified field names and field id + * + * @return a map of qualified field names to field id + */ + public Map nameToId() { + return lazyNameToId(); + } + /** * Returns the underlying {@link StructType struct type} for this schema. * diff --git a/api/src/main/java/org/apache/iceberg/types/AssignFreshIds.java b/api/src/main/java/org/apache/iceberg/types/AssignFreshIds.java index e58f76a8de56..d4ef83c56067 100644 --- a/api/src/main/java/org/apache/iceberg/types/AssignFreshIds.java +++ b/api/src/main/java/org/apache/iceberg/types/AssignFreshIds.java @@ -18,19 +18,13 @@ */ 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 AssignFreshIds extends TypeUtil.CustomOrderSchemaVisitor { - private final Schema visitingSchema; +class AssignFreshIds extends BaseAssignIds { private final Schema baseSchema; private final TypeUtil.NextID nextId; AssignFreshIds(TypeUtil.NextID nextId) { - this.visitingSchema = null; this.baseSchema = null; this.nextId = nextId; } @@ -43,12 +37,13 @@ class AssignFreshIds extends TypeUtil.CustomOrderSchemaVisitor { * @param nextId new id assigner */ AssignFreshIds(Schema visitingSchema, Schema baseSchema, TypeUtil.NextID nextId) { - this.visitingSchema = visitingSchema; + super(visitingSchema); this.baseSchema = baseSchema; this.nextId = nextId; } - private int idFor(String fullName) { + @Override + protected int idFor(String fullName) { if (baseSchema != null && fullName != null) { Types.NestedField field = baseSchema.findField(fullName); if (field != null) { @@ -58,74 +53,4 @@ private int idFor(String fullName) { return nextId.get(); } - - private String name(int id) { - if (visitingSchema != null) { - return visitingSchema.findColumnName(id); - } - - return null; - } - - @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 (int i = 0; i < length; i += 1) { - newIds.add(idFor(name(fields.get(i).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(name(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(name(map.keyId())); - int newValueId = idFor(name(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/AssignIds.java b/api/src/main/java/org/apache/iceberg/types/AssignIds.java new file mode 100644 index 000000000000..5c778ae0ac6f --- /dev/null +++ b/api/src/main/java/org/apache/iceberg/types/AssignIds.java @@ -0,0 +1,37 @@ +/* + * 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.Map; +import org.apache.iceberg.Schema; + +class AssignIds extends BaseAssignIds { + + private Map fieldIdMap; + + AssignIds(Schema visitingSchema, Map fieldIdMap) { + super(visitingSchema); + this.fieldIdMap = fieldIdMap; + } + + @Override + protected int idFor(String fullName) { + return fieldIdMap.get(fullName); + } +} diff --git a/api/src/main/java/org/apache/iceberg/types/BaseAssignIds.java b/api/src/main/java/org/apache/iceberg/types/BaseAssignIds.java new file mode 100644 index 000000000000..4f33a4d6775f --- /dev/null +++ b/api/src/main/java/org/apache/iceberg/types/BaseAssignIds.java @@ -0,0 +1,120 @@ +/* + * 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; + +abstract class BaseAssignIds extends TypeUtil.CustomOrderSchemaVisitor { + private final Schema visitingSchema; + + BaseAssignIds() { + this.visitingSchema = null; + } + + /** + * Replaces the ids in a schema with ids from a base schema, or uses nextId to assign a fresh ids. + * + * @param visitingSchema current schema that will have ids replaced (for id to name lookup) + */ + BaseAssignIds(Schema visitingSchema) { + this.visitingSchema = visitingSchema; + } + + /** + * Return an id for qualified field name + * + * @param fullName qualified field name + * @return id to assign on field + */ + protected abstract int idFor(String fullName); + + private String name(int id) { + if (visitingSchema != null) { + return visitingSchema.findColumnName(id); + } + + return null; + } + + @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 (int i = 0; i < length; i += 1) { + newIds.add(idFor(name(fields.get(i).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(name(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(name(map.keyId())); + int newValueId = idFor(name(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..602457d20b83 100644 --- a/api/src/main/java/org/apache/iceberg/types/TypeUtil.java +++ b/api/src/main/java/org/apache/iceberg/types/TypeUtil.java @@ -251,6 +251,17 @@ public static Schema assignFreshIds(Schema schema, Schema baseSchema, NextID nex return new Schema(struct.fields(), refreshIdentifierFields(struct, schema)); } + /** + * Assigns ids from the provided field to id map for all fields in a type. + * + * @param schema a schema + * @param fieldIdMap map of qualified field names to their ids + * @return a structurally identical type with new ids assigned by the provided map + */ + public static Type assignIds(Schema schema, Map fieldIdMap) { + return TypeUtil.visit(schema.asStruct(), new AssignIds(schema, fieldIdMap)); + } + /** * Get the identifier fields in the fresh schema based on the identifier fields in the base * schema. diff --git a/core/src/main/java/org/apache/iceberg/PositionDeletesTable.java b/core/src/main/java/org/apache/iceberg/PositionDeletesTable.java index 56270635809b..3b4eb32c7abd 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; @@ -34,6 +36,7 @@ import org.apache.iceberg.io.CloseableIterable; import org.apache.iceberg.io.CloseableIterator; import org.apache.iceberg.relocated.com.google.common.collect.Sets; +import org.apache.iceberg.types.Type; import org.apache.iceberg.types.TypeUtil; import org.apache.iceberg.types.Types; import org.apache.iceberg.util.ParallelIterable; @@ -114,7 +117,7 @@ private Schema calculateSchema() { Types.NestedField.optional( MetadataColumns.DELETE_FILE_ROW_FIELD_ID, MetadataColumns.DELETE_FILE_ROW_FIELD_NAME, - table().schema().asStruct(), + dedupSchemaFieldIds(partitionType), MetadataColumns.DELETE_FILE_ROW_DOC), Types.NestedField.required( MetadataColumns.PARTITION_COLUMN_ID, @@ -141,6 +144,31 @@ private Schema calculateSchema() { } } + // Handle collisions between table field and partition field ids + private Type dedupSchemaFieldIds(Types.StructType partitionType) { + Map originalIds = table().schema().nameToId(); + Set partitionFieldIds = new Schema(partitionType.fields()).idToName().keySet(); + AtomicInteger nextId = new AtomicInteger(table().schema().highestFieldId()); + + Map fieldToIds = + originalIds.entrySet().stream() + .collect( + Collectors.toMap( + Map.Entry::getKey, + e -> { + if (partitionFieldIds.contains(e.getValue())) { + int candidate = nextId.incrementAndGet(); + while (partitionFieldIds.contains(candidate)) { + candidate = nextId.incrementAndGet(); + } + return candidate; + } else { + return e.getValue(); + } + })); + return TypeUtil.assignIds(table().schema(), fieldToIds); + } + public static class PositionDeletesBatchScan extends SnapshotScan> implements BatchScan { 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..71e74ee4df23 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", 4).build(); + Table table = + validationCatalog.createTable( + TableIdentifier.of("default", TABLE_NAME), schema, spec, tableProperties()); + + Dataset df = + spark + .range(12) + .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(4); + + List deleteFiles = deleteFiles(table); + assertThat(deleteFiles).hasSize(4); + + List expectedRecords = records(table); + List expectedDeletes = deleteRecords(table); + assertThat(expectedRecords).hasSize(8); + assertThat(expectedDeletes).hasSize(4); + + 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(4); + assertNotContains(deleteFiles, newDeleteFiles); + assertLocallySorted(newDeleteFiles); + checkResult(result, deleteFiles, newDeleteFiles, 4); + 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 = From 3efefc5fa3c5d503ab570b171bc95a70383526b1 Mon Sep 17 00:00:00 2001 From: Szehon Ho Date: Fri, 22 Mar 2024 16:15:12 -0700 Subject: [PATCH 02/15] Make test smaller (not sure if it OOM'ed) --- .../TestRewritePositionDeleteFilesAction.java | 16 ++++++++-------- 1 file changed, 8 insertions(+), 8 deletions(-) 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 71e74ee4df23..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 @@ -659,14 +659,14 @@ public void testRewriteManyColumns() throws Exception { .collect(Collectors.toList()); fields.addAll(additionalCols); Schema schema = new Schema(fields); - PartitionSpec spec = PartitionSpec.builderFor(schema).bucket("id", 4).build(); + 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(12) + .range(4) .withColumns( IntStream.range(1, 1010) .boxed() @@ -682,15 +682,15 @@ public void testRewriteManyColumns() throws Exception { List dataFiles = TestHelpers.dataFiles(table); writePosDeletesForFiles(table, 1, 1, dataFiles); - assertThat(dataFiles).hasSize(4); + assertThat(dataFiles).hasSize(2); List deleteFiles = deleteFiles(table); - assertThat(deleteFiles).hasSize(4); + assertThat(deleteFiles).hasSize(2); List expectedRecords = records(table); List expectedDeletes = deleteRecords(table); - assertThat(expectedRecords).hasSize(8); - assertThat(expectedDeletes).hasSize(4); + assertThat(expectedRecords).hasSize(2); + assertThat(expectedDeletes).hasSize(2); Result result = SparkActions.get(spark) @@ -700,10 +700,10 @@ public void testRewriteManyColumns() throws Exception { .execute(); List newDeleteFiles = deleteFiles(table); - assertThat(newDeleteFiles).hasSize(4); + assertThat(newDeleteFiles).hasSize(2); assertNotContains(deleteFiles, newDeleteFiles); assertLocallySorted(newDeleteFiles); - checkResult(result, deleteFiles, newDeleteFiles, 4); + checkResult(result, deleteFiles, newDeleteFiles, 2); checkSequenceNumbers(table, deleteFiles, newDeleteFiles); List actualRecords = records(table); From 7ff8c9d9e20efdebad58e5bc247766fb79a090b4 Mon Sep 17 00:00:00 2001 From: Szehon Ho Date: Tue, 26 Mar 2024 15:50:02 -0700 Subject: [PATCH 03/15] Redo change to reassign ids on partition instead of row --- .../org/apache/iceberg/BaseMetadataTable.java | 21 ++- .../apache/iceberg/PositionDeletesTable.java | 136 ++++++++++++------ ...RewritePositionDeleteFilesSparkAction.java | 5 +- 3 files changed, 113 insertions(+), 49 deletions(-) diff --git a/core/src/main/java/org/apache/iceberg/BaseMetadataTable.java b/core/src/main/java/org/apache/iceberg/BaseMetadataTable.java index 57a6386093d6..9b2062ed6c6b 100644 --- a/core/src/main/java/org/apache/iceberg/BaseMetadataTable.java +++ b/core/src/main/java/org/apache/iceberg/BaseMetadataTable.java @@ -68,18 +68,24 @@ protected BaseMetadataTable(Table table, String name) { * @return a spec used to rewrite the metadata table filters to partition filters using an * inclusive projection */ - static PartitionSpec transformSpec(Schema metadataTableSchema, PartitionSpec spec) { + static PartitionSpec transformSpec( + Schema metadataTableSchema, PartitionSpec spec, Map fieldMap) { PartitionSpec.Builder builder = PartitionSpec.builderFor(metadataTableSchema) .withSpecId(spec.specId()) .checkConflicts(false); for (PartitionField field : spec.fields()) { - builder.add(field.fieldId(), field.fieldId(), field.name(), Transforms.identity()); + int newFieldId = fieldMap.getOrDefault(field.fieldId(), field.fieldId()); + builder.add(newFieldId, newFieldId, field.name(), Transforms.identity()); } return builder.build(); } + static PartitionSpec transformSpec(Schema metadataTableSchema, PartitionSpec spec) { + return transformSpec(metadataTableSchema, spec, ImmutableMap.of()); + } + /** * This method transforms the given partition specs to specs that are used to rewrite the * user-provided filter expression against the given metadata table. @@ -92,12 +98,19 @@ static PartitionSpec transformSpec(Schema metadataTableSchema, PartitionSpec spe * inclusive projection */ static Map transformSpecs( - Schema metadataTableSchema, Map specs) { + Schema metadataTableSchema, + Map specs, + Map fieldMap) { return specs.values().stream() - .map(spec -> transformSpec(metadataTableSchema, spec)) + .map(spec -> transformSpec(metadataTableSchema, spec, fieldMap)) .collect(Collectors.toMap(PartitionSpec::specId, spec -> spec)); } + static Map transformSpecs( + Schema metadataTableSchema, Map specs) { + return transformSpecs(metadataTableSchema, specs, ImmutableMap.of()); + } + abstract MetadataTableType metadataTableType(); public BaseTable table() { diff --git a/core/src/main/java/org/apache/iceberg/PositionDeletesTable.java b/core/src/main/java/org/apache/iceberg/PositionDeletesTable.java index 3b4eb32c7abd..68e24653cda2 100644 --- a/core/src/main/java/org/apache/iceberg/PositionDeletesTable.java +++ b/core/src/main/java/org/apache/iceberg/PositionDeletesTable.java @@ -24,10 +24,10 @@ 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.Evaluator; import org.apache.iceberg.expressions.Expression; import org.apache.iceberg.expressions.Expressions; import org.apache.iceberg.expressions.ManifestEvaluator; @@ -35,8 +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.ImmutableMap; import org.apache.iceberg.relocated.com.google.common.collect.Sets; -import org.apache.iceberg.types.Type; import org.apache.iceberg.types.TypeUtil; import org.apache.iceberg.types.Types; import org.apache.iceberg.util.ParallelIterable; @@ -55,6 +55,7 @@ public class PositionDeletesTable extends BaseMetadataTable { private final Schema schema; private final int defaultSpecId; private final Map specs; + private final Map fieldMap; PositionDeletesTable(Table table) { this(table, table.name() + ".position_deletes"); @@ -62,9 +63,11 @@ public class PositionDeletesTable extends BaseMetadataTable { PositionDeletesTable(Table table, String name) { super(table, name); - this.schema = calculateSchema(); + Types.StructType partitionType = Partitioning.partitionType(table()); + this.fieldMap = partitionFieldMap(table.schema(), partitionType); + this.schema = calculateSchema(partitionType, fieldMap); this.defaultSpecId = table.spec().specId(); - this.specs = transformSpecs(schema(), table.specs()); + this.specs = transformSpecs(schema(), table.specs(), fieldMap); } @Override @@ -80,7 +83,7 @@ public TableScan newScan() { @Override public BatchScan newBatchScan() { - return new PositionDeletesBatchScan(table(), schema()); + return new PositionDeletesBatchScan(table(), schema(), fieldMap); } @Override @@ -108,8 +111,8 @@ public Map properties() { .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue))); } - private Schema calculateSchema() { - Types.StructType partitionType = Partitioning.partitionType(table()); + private Schema calculateSchema( + Types.StructType partitionType, Map partfieldMap) { Schema result = new Schema( MetadataColumns.DELETE_FILE_PATH, @@ -117,12 +120,12 @@ private Schema calculateSchema() { Types.NestedField.optional( MetadataColumns.DELETE_FILE_ROW_FIELD_ID, MetadataColumns.DELETE_FILE_ROW_FIELD_NAME, - dedupSchemaFieldIds(partitionType), + table().schema().asStruct(), MetadataColumns.DELETE_FILE_ROW_DOC), Types.NestedField.required( MetadataColumns.PARTITION_COLUMN_ID, PARTITION, - partitionType, + partitionType(partitionType, partfieldMap), "Partition that position delete row belongs to"), Types.NestedField.required( MetadataColumns.SPEC_ID_COLUMN_ID, @@ -144,50 +147,88 @@ private Schema calculateSchema() { } } + /** + * Handle collisions between table and partition field ids, as both need to be part of position + * deletes table + * + * @param tableSchema original table schema + * @param partitionType original table's partition type + * @return partition type with reassigned field ids + */ + public static Types.StructType partitionType(Schema tableSchema, Types.StructType partitionType) { + Map fieldMap = partitionFieldMap(tableSchema, partitionType); + return partitionType(partitionType, fieldMap); + } + // Handle collisions between table field and partition field ids - private Type dedupSchemaFieldIds(Types.StructType partitionType) { - Map originalIds = table().schema().nameToId(); - Set partitionFieldIds = new Schema(partitionType.fields()).idToName().keySet(); - AtomicInteger nextId = new AtomicInteger(table().schema().highestFieldId()); - - Map fieldToIds = - originalIds.entrySet().stream() - .collect( - Collectors.toMap( - Map.Entry::getKey, - e -> { - if (partitionFieldIds.contains(e.getValue())) { - int candidate = nextId.incrementAndGet(); - while (partitionFieldIds.contains(candidate)) { - candidate = nextId.incrementAndGet(); - } - return candidate; - } else { - return e.getValue(); - } - })); - return TypeUtil.assignIds(table().schema(), fieldToIds); + static Map partitionFieldMap( + Schema tableSchema, Types.StructType partitionType) { + AtomicInteger nextId = new AtomicInteger(tableSchema.highestFieldId()); + return partitionType.fields().stream() + .collect(Collectors.toMap(Types.NestedField::fieldId, f -> nextId.incrementAndGet())); + } + + static Types.StructType partitionType( + Types.StructType partitionType, Map fieldMap) { + return Types.StructType.of( + partitionType.fields().stream() + .map( + f -> + Types.NestedField.of( + fieldMap.get(f.fieldId()), f.isOptional(), f.name(), f.type(), f.doc())) + .collect(Collectors.toList())); } public static class PositionDeletesBatchScan extends SnapshotScan> implements BatchScan { private Expression baseTableFilter = Expressions.alwaysTrue(); + private final Map fieldMap; + + protected PositionDeletesBatchScan(Table table, Schema schema, Map fieldMap) { + super(table, schema, TableScanContext.empty()); + this.fieldMap = fieldMap; + } + + protected PositionDeletesBatchScan( + Table table, + Schema schema, + TableScanContext context, + Expression baseTableFilter, + Map fieldMap) { + super(table, schema, context); + this.baseTableFilter = baseTableFilter; + this.fieldMap = fieldMap; + } + /** @deprecated since 1.5.0, will be removed in 1.6.0; use fieldMap constructor instead. */ + @Deprecated protected PositionDeletesBatchScan(Table table, Schema schema) { super(table, schema, TableScanContext.empty()); + this.fieldMap = ImmutableMap.of(); + } + + /** @deprecated since 1.5.0, will be removed in 1.6.0; use fieldMap constructor instead. */ + @Deprecated + protected PositionDeletesBatchScan(Table table, Schema schema, TableScanContext context) { + super(table, schema, context); + this.fieldMap = ImmutableMap.of(); } + /** @deprecated since 1.5.0, will be removed in 1.6.0; use fieldMap constructor instead. */ + @Deprecated protected PositionDeletesBatchScan( Table table, Schema schema, TableScanContext context, Expression baseTableFilter) { super(table, schema, context); this.baseTableFilter = baseTableFilter; + this.fieldMap = ImmutableMap.of(); } @Override protected PositionDeletesBatchScan newRefinedScan( Table newTable, Schema newSchema, TableScanContext newContext) { - return new PositionDeletesBatchScan(newTable, newSchema, newContext, baseTableFilter); + return new PositionDeletesBatchScan( + newTable, newSchema, newContext, baseTableFilter, fieldMap); } @Override @@ -224,15 +265,14 @@ protected List scanColumns() { */ public BatchScan baseTableFilter(Expression expr) { return new PositionDeletesBatchScan( - table(), schema(), context(), Expressions.and(baseTableFilter, expr)); + table(), schema(), context(), Expressions.and(baseTableFilter, expr), fieldMap); } @Override protected CloseableIterable doPlanFiles() { String schemaString = SchemaParser.toJson(tableSchema()); - - // prepare transformed partition specs and caches - Map transformedSpecs = transformSpecs(tableSchema(), table().specs()); + Map transformedSpecs = + transformSpecs(tableSchema(), table().specs(), fieldMap); LoadingCache specStringCache = partitionCacheOf(transformedSpecs, PartitionSpecParser::toJson); @@ -276,7 +316,6 @@ protected CloseableIterable doPlanFiles() { manifest, table().specs().get(manifest.partitionSpecId()), schemaString, - transformedSpecs, residualCache, specStringCache)); @@ -291,7 +330,6 @@ private CloseableIterable posDeletesScanTasks( ManifestFile manifest, PartitionSpec spec, String schemaString, - Map transformedSpecs, LoadingCache residualCache, LoadingCache specStringCache) { return new CloseableIterable() { @@ -306,28 +344,38 @@ public void close() throws IOException { @Override public CloseableIterator iterator() { + // Partition filter by base table filter Expression partitionFilter = Projections.inclusive(spec, isCaseSensitive()).project(baseTableFilter); - // Filter partitions + // Read manifests (use original table's partition ids to de-serialize partition values) CloseableIterable> deleteFileEntries = - ManifestFiles.readDeleteManifest(manifest, table().io(), transformedSpecs) + ManifestFiles.readDeleteManifest(manifest, table().io(), table().specs()) .caseSensitive(isCaseSensitive()) .select(scanColumns()) - .filterRows(filter()) .filterPartitions(partitionFilter) .scanMetrics(scanMetrics()) .liveEntries(); - // Filter delete file type - CloseableIterable> positionDeleteEntries = + // Partition Filter by metadata table filter (on transformed spec/schema) + PartitionSpec transformedSpec = transformSpec(tableSchema(), spec, fieldMap); + Expression projected = + Projections.inclusive(transformedSpec, isCaseSensitive()).project(filter()); + Evaluator eval = + new Evaluator(transformedSpec.partitionType(), projected, isCaseSensitive()); + deleteFileEntries = + CloseableIterable.filter( + deleteFileEntries, entry -> eval.eval(entry.file().partition())); + + // Filter by delete file type + deleteFileEntries = CloseableIterable.filter( deleteFileEntries, entry -> entry.file().content().equals(FileContent.POSITION_DELETES)); this.iterable = CloseableIterable.transform( - positionDeleteEntries, + deleteFileEntries, entry -> { int specId = entry.file().specId(); return new BasePositionDeletesScanTask( 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..7ec27063b51f 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 @@ -35,6 +35,7 @@ import org.apache.iceberg.MetadataTableUtils; import org.apache.iceberg.Partitioning; import org.apache.iceberg.PositionDeletesScanTask; +import org.apache.iceberg.PositionDeletesTable; import org.apache.iceberg.PositionDeletesTable.PositionDeletesBatchScan; import org.apache.iceberg.RewriteJobOrder; import org.apache.iceberg.StructLike; @@ -59,6 +60,7 @@ import org.apache.iceberg.relocated.com.google.common.math.IntMath; import org.apache.iceberg.relocated.com.google.common.util.concurrent.MoreExecutors; import org.apache.iceberg.relocated.com.google.common.util.concurrent.ThreadFactoryBuilder; +import org.apache.iceberg.types.Types; import org.apache.iceberg.types.Types.StructType; import org.apache.iceberg.util.PartitionUtil; import org.apache.iceberg.util.PropertyUtil; @@ -458,6 +460,7 @@ public int totalGroupCount() { } private StructLike coercePartition(PositionDeletesScanTask task, StructType partitionType) { - return PartitionUtil.coercePartition(partitionType, task.spec(), task.partition()); + Types.StructType dedupType = PositionDeletesTable.partitionType(table.schema(), partitionType); + return PartitionUtil.coercePartition(dedupType, task.spec(), task.partition()); } } From cfbb0a6542bf07b047c10d14a320c0f0b576c833 Mon Sep 17 00:00:00 2001 From: Szehon Ho Date: Tue, 26 Mar 2024 15:57:27 -0700 Subject: [PATCH 04/15] Remove TypeUtil changes from previous attempt --- .../main/java/org/apache/iceberg/Schema.java | 9 -- .../apache/iceberg/types/AssignFreshIds.java | 83 +++++++++++- .../org/apache/iceberg/types/AssignIds.java | 37 ------ .../apache/iceberg/types/BaseAssignIds.java | 120 ------------------ .../org/apache/iceberg/types/TypeUtil.java | 11 -- 5 files changed, 79 insertions(+), 181 deletions(-) delete mode 100644 api/src/main/java/org/apache/iceberg/types/AssignIds.java delete mode 100644 api/src/main/java/org/apache/iceberg/types/BaseAssignIds.java diff --git a/api/src/main/java/org/apache/iceberg/Schema.java b/api/src/main/java/org/apache/iceberg/Schema.java index e8ed1c3d109b..5e024b7c1c29 100644 --- a/api/src/main/java/org/apache/iceberg/Schema.java +++ b/api/src/main/java/org/apache/iceberg/Schema.java @@ -242,15 +242,6 @@ public Map idToName() { return lazyIdToName(); } - /** - * Returns a map for this schema between qualified field names and field id - * - * @return a map of qualified field names to field id - */ - public Map nameToId() { - return lazyNameToId(); - } - /** * Returns the underlying {@link StructType struct type} for this schema. * diff --git a/api/src/main/java/org/apache/iceberg/types/AssignFreshIds.java b/api/src/main/java/org/apache/iceberg/types/AssignFreshIds.java index d4ef83c56067..e58f76a8de56 100644 --- a/api/src/main/java/org/apache/iceberg/types/AssignFreshIds.java +++ b/api/src/main/java/org/apache/iceberg/types/AssignFreshIds.java @@ -18,13 +18,19 @@ */ 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 AssignFreshIds extends BaseAssignIds { +class AssignFreshIds extends TypeUtil.CustomOrderSchemaVisitor { + private final Schema visitingSchema; private final Schema baseSchema; private final TypeUtil.NextID nextId; AssignFreshIds(TypeUtil.NextID nextId) { + this.visitingSchema = null; this.baseSchema = null; this.nextId = nextId; } @@ -37,13 +43,12 @@ class AssignFreshIds extends BaseAssignIds { * @param nextId new id assigner */ AssignFreshIds(Schema visitingSchema, Schema baseSchema, TypeUtil.NextID nextId) { - super(visitingSchema); + this.visitingSchema = visitingSchema; this.baseSchema = baseSchema; this.nextId = nextId; } - @Override - protected int idFor(String fullName) { + private int idFor(String fullName) { if (baseSchema != null && fullName != null) { Types.NestedField field = baseSchema.findField(fullName); if (field != null) { @@ -53,4 +58,74 @@ protected int idFor(String fullName) { return nextId.get(); } + + private String name(int id) { + if (visitingSchema != null) { + return visitingSchema.findColumnName(id); + } + + return null; + } + + @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 (int i = 0; i < length; i += 1) { + newIds.add(idFor(name(fields.get(i).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(name(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(name(map.keyId())); + int newValueId = idFor(name(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/AssignIds.java b/api/src/main/java/org/apache/iceberg/types/AssignIds.java deleted file mode 100644 index 5c778ae0ac6f..000000000000 --- a/api/src/main/java/org/apache/iceberg/types/AssignIds.java +++ /dev/null @@ -1,37 +0,0 @@ -/* - * 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.Map; -import org.apache.iceberg.Schema; - -class AssignIds extends BaseAssignIds { - - private Map fieldIdMap; - - AssignIds(Schema visitingSchema, Map fieldIdMap) { - super(visitingSchema); - this.fieldIdMap = fieldIdMap; - } - - @Override - protected int idFor(String fullName) { - return fieldIdMap.get(fullName); - } -} diff --git a/api/src/main/java/org/apache/iceberg/types/BaseAssignIds.java b/api/src/main/java/org/apache/iceberg/types/BaseAssignIds.java deleted file mode 100644 index 4f33a4d6775f..000000000000 --- a/api/src/main/java/org/apache/iceberg/types/BaseAssignIds.java +++ /dev/null @@ -1,120 +0,0 @@ -/* - * 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; - -abstract class BaseAssignIds extends TypeUtil.CustomOrderSchemaVisitor { - private final Schema visitingSchema; - - BaseAssignIds() { - this.visitingSchema = null; - } - - /** - * Replaces the ids in a schema with ids from a base schema, or uses nextId to assign a fresh ids. - * - * @param visitingSchema current schema that will have ids replaced (for id to name lookup) - */ - BaseAssignIds(Schema visitingSchema) { - this.visitingSchema = visitingSchema; - } - - /** - * Return an id for qualified field name - * - * @param fullName qualified field name - * @return id to assign on field - */ - protected abstract int idFor(String fullName); - - private String name(int id) { - if (visitingSchema != null) { - return visitingSchema.findColumnName(id); - } - - return null; - } - - @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 (int i = 0; i < length; i += 1) { - newIds.add(idFor(name(fields.get(i).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(name(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(name(map.keyId())); - int newValueId = idFor(name(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 602457d20b83..7c13d6094084 100644 --- a/api/src/main/java/org/apache/iceberg/types/TypeUtil.java +++ b/api/src/main/java/org/apache/iceberg/types/TypeUtil.java @@ -251,17 +251,6 @@ public static Schema assignFreshIds(Schema schema, Schema baseSchema, NextID nex return new Schema(struct.fields(), refreshIdentifierFields(struct, schema)); } - /** - * Assigns ids from the provided field to id map for all fields in a type. - * - * @param schema a schema - * @param fieldIdMap map of qualified field names to their ids - * @return a structurally identical type with new ids assigned by the provided map - */ - public static Type assignIds(Schema schema, Map fieldIdMap) { - return TypeUtil.visit(schema.asStruct(), new AssignIds(schema, fieldIdMap)); - } - /** * Get the identifier fields in the fresh schema based on the identifier fields in the base * schema. From 2642eac4f2399276921c1e295ecfc017ddd321e5 Mon Sep 17 00:00:00 2001 From: Szehon Ho Date: Tue, 26 Mar 2024 16:20:22 -0700 Subject: [PATCH 05/15] Small fix to use cached spec instead of re-calcuating --- .../main/java/org/apache/iceberg/PositionDeletesTable.java | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/core/src/main/java/org/apache/iceberg/PositionDeletesTable.java b/core/src/main/java/org/apache/iceberg/PositionDeletesTable.java index 68e24653cda2..eb03fb89a76a 100644 --- a/core/src/main/java/org/apache/iceberg/PositionDeletesTable.java +++ b/core/src/main/java/org/apache/iceberg/PositionDeletesTable.java @@ -316,6 +316,7 @@ protected CloseableIterable doPlanFiles() { manifest, table().specs().get(manifest.partitionSpecId()), schemaString, + transformedSpecs, residualCache, specStringCache)); @@ -330,6 +331,7 @@ private CloseableIterable posDeletesScanTasks( ManifestFile manifest, PartitionSpec spec, String schemaString, + Map transformedSpecs, LoadingCache residualCache, LoadingCache specStringCache) { return new CloseableIterable() { @@ -357,8 +359,8 @@ public CloseableIterator iterator() { .scanMetrics(scanMetrics()) .liveEntries(); - // Partition Filter by metadata table filter (on transformed spec/schema) - PartitionSpec transformedSpec = transformSpec(tableSchema(), spec, fieldMap); + // Partition Filter by metadata table filter (bind on transformed spec/schema) + PartitionSpec transformedSpec = transformedSpecs.get(spec.specId()); Expression projected = Projections.inclusive(transformedSpec, isCaseSensitive()).project(filter()); Evaluator eval = From 63437ac82c487cf0065727eda307c5299047b4b6 Mon Sep 17 00:00:00 2001 From: Szehon Ho Date: Tue, 26 Mar 2024 16:42:10 -0700 Subject: [PATCH 06/15] Fix tests --- .../java/org/apache/iceberg/TestMetadataTableScans.java | 9 ++++++--- .../TestMetadataTableScansWithPartitionEvolution.java | 8 +++----- 2 files changed, 9 insertions(+), 8 deletions(-) diff --git a/core/src/test/java/org/apache/iceberg/TestMetadataTableScans.java b/core/src/test/java/org/apache/iceberg/TestMetadataTableScans.java index 4c5f1d240f57..23d684041267 100644 --- a/core/src/test/java/org/apache/iceberg/TestMetadataTableScans.java +++ b/core/src/test/java/org/apache/iceberg/TestMetadataTableScans.java @@ -1266,7 +1266,8 @@ public void testPositionDeletesWithFilter() { ScanTask task = tasks.get(0); assertThat(task).isInstanceOf(PositionDeletesScanTask.class); - Types.StructType partitionType = Partitioning.partitionType(table); + Types.StructType partitionType = + PositionDeletesTable.partitionType(table.schema(), Partitioning.partitionType(table)); PositionDeletesScanTask posDeleteTask = (PositionDeletesScanTask) task; int filePartition = posDeleteTask.file().partition().get(0, Integer.class); @@ -1333,7 +1334,8 @@ 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.partitionType(table.schema(), Partitioning.partitionType(table)); PositionDeletesScanTask posDeleteTask = (PositionDeletesScanTask) task; // base table filter should only be used to evaluate partitions @@ -1415,7 +1417,8 @@ public void testPositionDeletesWithBaseTableFilterNot() { ScanTask task = tasks.get(0); assertThat(task).isInstanceOf(PositionDeletesScanTask.class); - Types.StructType partitionType = Partitioning.partitionType(table); + Types.StructType partitionType = + PositionDeletesTable.partitionType(table.schema(), Partitioning.partitionType(table)); PositionDeletesScanTask posDeleteTask = (PositionDeletesScanTask) task; // base table filter should only be used to evaluate partitions diff --git a/core/src/test/java/org/apache/iceberg/TestMetadataTableScansWithPartitionEvolution.java b/core/src/test/java/org/apache/iceberg/TestMetadataTableScansWithPartitionEvolution.java index faccdcb3dd95..86145555a7e6 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,8 @@ public void testPositionDeletesPartitionSpecRemoval() { ScanTask task = tasks.get(0); assertThat(task).isInstanceOf(PositionDeletesScanTask.class); - Types.StructType partitionType = Partitioning.partitionType(table); + Types.StructType partitionType = + PositionDeletesTable.partitionType(table.schema(), Partitioning.partitionType(table)); PositionDeletesScanTask posDeleteTask = (PositionDeletesScanTask) task; int filePartition = posDeleteTask.file().partition().get(0, Integer.class); @@ -200,11 +200,9 @@ public void testPositionDeletesPartitionSpecRemoval() { 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(1).fieldId()); assertThat(posDeleteTask.file().specId()) .as("Expected correct partition spec id on task") .isEqualTo(table.ops().current().spec().specId()); From a23d24988e3ad4e501ee89c98f701aa6ef140b92 Mon Sep 17 00:00:00 2001 From: Szehon Ho Date: Tue, 9 Apr 2024 17:37:12 -0700 Subject: [PATCH 07/15] Make logic generic --- .../org/apache/iceberg/PartitionSpec.java | 26 +++ .../main/java/org/apache/iceberg/Schema.java | 80 ++++++++- .../org/apache/iceberg/types/AssignIds.java | 104 +++++++++++ .../org/apache/iceberg/types/TypeUtil.java | 16 ++ .../org/apache/iceberg/BaseMetadataTable.java | 22 +-- .../org/apache/iceberg/ManifestReader.java | 2 +- .../apache/iceberg/PositionDeletesTable.java | 162 +++++------------- .../iceberg/TestMetadataTableScans.java | 11 +- ...adataTableScansWithPartitionEvolution.java | 7 +- ...RewritePositionDeleteFilesSparkAction.java | 11 +- .../TestRewritePositionDeleteFilesAction.java | 67 ++++++++ ...RewritePositionDeleteFilesSparkAction.java | 12 +- .../TestRewritePositionDeleteFilesAction.java | 66 +++++++ ...RewritePositionDeleteFilesSparkAction.java | 16 +- 14 files changed, 434 insertions(+), 168 deletions(-) create mode 100644 api/src/main/java/org/apache/iceberg/types/AssignIds.java diff --git a/api/src/main/java/org/apache/iceberg/PartitionSpec.java b/api/src/main/java/org/apache/iceberg/PartitionSpec.java index 4fcb110db87c..6a4716ea7956 100644 --- a/api/src/main/java/org/apache/iceberg/PartitionSpec.java +++ b/api/src/main/java/org/apache/iceberg/PartitionSpec.java @@ -60,6 +60,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 lazyOriginalPartitionType = null; private transient volatile List fieldList = null; private final int lastAssignedFieldId; @@ -140,6 +141,31 @@ public StructType partitionType() { return lazyPartitionType; } + public StructType originalPartitionType() { + if (schema.idsToOriginal().size() == 0) { + return partitionType(); + } + if (lazyOriginalPartitionType == null) { + synchronized (this) { + if (lazyOriginalPartitionType == null) { + List structFields = Lists.newArrayListWithExpectedSize(fields.length); + + for (PartitionField field : fields) { + Type sourceType = schema.findType(field.sourceId()); + Type resultType = field.transform().getResultType(sourceType); + structFields.add( + Types.NestedField.optional( + schema.idsToOriginal().get(field.fieldId()), field.name(), resultType)); + } + + this.lazyOriginalPartitionType = Types.StructType.of(structFields); + } + } + } + + return lazyOriginalPartitionType; + } + 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..856dd906d7ca 100644 --- a/api/src/main/java/org/apache/iceberg/Schema.java +++ b/api/src/main/java/org/apache/iceberg/Schema.java @@ -26,6 +26,7 @@ import java.util.Locale; import java.util.Map; import java.util.Set; +import java.util.concurrent.atomic.AtomicInteger; import java.util.stream.Collectors; import org.apache.iceberg.relocated.com.google.common.base.Joiner; import org.apache.iceberg.relocated.com.google.common.base.Preconditions; @@ -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,25 @@ public Schema(List columns, Set identifierFieldIds) { this(DEFAULT_SCHEMA_ID, columns, identifierFieldIds); } + public Schema( + List columns, Set identifierFieldIds, Set metadataFieldIds) { + this(DEFAULT_SCHEMA_ID, columns, identifierFieldIds, metadataFieldIds); + } + 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, ImmutableSet.of()); + } + + public Schema( + int schemaId, + List columns, + Set identifierFieldIds, + Set metadataFieldIds) { + this(schemaId, columns, null, identifierFieldIds, metadataFieldIds); } public Schema( @@ -96,8 +113,22 @@ public Schema( List columns, Map aliases, Set identifierFieldIds) { + this(schemaId, columns, aliases, identifierFieldIds, ImmutableSet.of()); + } + + public Schema( + int schemaId, + List columns, + Map aliases, + Set identifierFieldIds, + Set metadataFieldIds) { this.schemaId = schemaId; - this.struct = StructType.of(columns); + + this.idsToOriginal = Maps.newHashMap(); + this.idsToReassigned = Maps.newHashMap(); + List finalColumns = reassignMetadataFieldIds(columns, metadataFieldIds); + + this.struct = StructType.of(finalColumns); this.aliasToId = aliases != null ? ImmutableBiMap.copyOf(aliases) : null; // validate IdentifierField @@ -507,4 +538,49 @@ public String toString() { .map(this::identifierFieldToString) .collect(Collectors.toList()))); } + + /** + * All ids of metadata fields are reassigned. + * + * @return map of original to reassigned field ids of metadata fields + */ + public Map idsToReassigned() { + return idsToReassigned != null ? idsToReassigned : Maps.newHashMap(); + } + + /** + * All ids of metadata fields are reassigned. + * + * @return map of reassigned to original field ids of metadata fields + */ + public Map idsToOriginal() { + return idsToOriginal != null ? idsToOriginal : Maps.newHashMap(); + } + + private List reassignMetadataFieldIds( + List columns, Set metadataFieldIds) { + Set usedIds = + Sets.newHashSet( + Sets.difference(TypeUtil.indexById(StructType.of(columns)).keySet(), metadataFieldIds)); + AtomicInteger nextId = new AtomicInteger(); + + Type res = + TypeUtil.assignIds( + StructType.of(columns), + id -> { + if (metadataFieldIds.contains(id)) { + int candidate = nextId.get(); + while (usedIds.contains(candidate)) { + candidate = nextId.incrementAndGet(); + } + usedIds.add(candidate); + idsToReassigned.put(id, candidate); + idsToOriginal.put(candidate, id); + return candidate; + } else { + return id; + } + }); + 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..15fe482ef05f --- /dev/null +++ b/api/src/main/java/org/apache/iceberg/types/AssignIds.java @@ -0,0 +1,104 @@ +/* + * 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; + + /** + * Replaces the ids in a schema with ids from a base schema, or uses nextId to assign a fresh ids. + * + * @param getID new id assigner + */ + 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..7198544a0c1d 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 nextId 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/core/src/main/java/org/apache/iceberg/BaseMetadataTable.java b/core/src/main/java/org/apache/iceberg/BaseMetadataTable.java index 9b2062ed6c6b..e1e138109f8e 100644 --- a/core/src/main/java/org/apache/iceberg/BaseMetadataTable.java +++ b/core/src/main/java/org/apache/iceberg/BaseMetadataTable.java @@ -68,24 +68,21 @@ protected BaseMetadataTable(Table table, String name) { * @return a spec used to rewrite the metadata table filters to partition filters using an * inclusive projection */ - static PartitionSpec transformSpec( - Schema metadataTableSchema, PartitionSpec spec, Map fieldMap) { + static PartitionSpec transformSpec(Schema metadataTableSchema, PartitionSpec spec) { PartitionSpec.Builder builder = PartitionSpec.builderFor(metadataTableSchema) .withSpecId(spec.specId()) .checkConflicts(false); + Map reassignedFields = metadataTableSchema.idsToReassigned(); + for (PartitionField field : spec.fields()) { - int newFieldId = fieldMap.getOrDefault(field.fieldId(), field.fieldId()); + int newFieldId = reassignedFields.getOrDefault(field.fieldId(), field.fieldId()); builder.add(newFieldId, newFieldId, field.name(), Transforms.identity()); } return builder.build(); } - static PartitionSpec transformSpec(Schema metadataTableSchema, PartitionSpec spec) { - return transformSpec(metadataTableSchema, spec, ImmutableMap.of()); - } - /** * This method transforms the given partition specs to specs that are used to rewrite the * user-provided filter expression against the given metadata table. @@ -98,19 +95,12 @@ static PartitionSpec transformSpec(Schema metadataTableSchema, PartitionSpec spe * inclusive projection */ static Map transformSpecs( - Schema metadataTableSchema, - Map specs, - Map fieldMap) { + Schema metadataTableSchema, Map specs) { return specs.values().stream() - .map(spec -> transformSpec(metadataTableSchema, spec, fieldMap)) + .map(spec -> transformSpec(metadataTableSchema, spec)) .collect(Collectors.toMap(PartitionSpec::specId, spec -> spec)); } - static Map transformSpecs( - Schema metadataTableSchema, Map specs) { - return transformSpecs(metadataTableSchema, specs, ImmutableMap.of()); - } - abstract MetadataTableType metadataTableType(); public BaseTable table() { diff --git a/core/src/main/java/org/apache/iceberg/ManifestReader.java b/core/src/main/java/org/apache/iceberg/ManifestReader.java index 4ee51aa60c31..a117ceb26d06 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.originalPartitionType()).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 eb03fb89a76a..394869c973b4 100644 --- a/core/src/main/java/org/apache/iceberg/PositionDeletesTable.java +++ b/core/src/main/java/org/apache/iceberg/PositionDeletesTable.java @@ -24,10 +24,9 @@ import java.util.Collections; import java.util.List; import java.util.Map; -import java.util.concurrent.atomic.AtomicInteger; +import java.util.Set; import java.util.function.Function; import java.util.stream.Collectors; -import org.apache.iceberg.expressions.Evaluator; import org.apache.iceberg.expressions.Expression; import org.apache.iceberg.expressions.Expressions; import org.apache.iceberg.expressions.ManifestEvaluator; @@ -35,7 +34,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.ImmutableMap; +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; @@ -55,7 +55,6 @@ public class PositionDeletesTable extends BaseMetadataTable { private final Schema schema; private final int defaultSpecId; private final Map specs; - private final Map fieldMap; PositionDeletesTable(Table table) { this(table, table.name() + ".position_deletes"); @@ -63,11 +62,9 @@ public class PositionDeletesTable extends BaseMetadataTable { PositionDeletesTable(Table table, String name) { super(table, name); - Types.StructType partitionType = Partitioning.partitionType(table()); - this.fieldMap = partitionFieldMap(table.schema(), partitionType); - this.schema = calculateSchema(partitionType, fieldMap); + this.schema = calculateSchema(); this.defaultSpecId = table.spec().specId(); - this.specs = transformSpecs(schema(), table.specs(), fieldMap); + this.specs = transformSpecs(schema(), table.specs()); } @Override @@ -83,7 +80,7 @@ public TableScan newScan() { @Override public BatchScan newBatchScan() { - return new PositionDeletesBatchScan(table(), schema(), fieldMap); + return new PositionDeletesBatchScan(table(), schema()); } @Override @@ -111,32 +108,37 @@ public Map properties() { .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue))); } - private Schema calculateSchema( - Types.StructType partitionType, Map partfieldMap) { + private Schema calculateSchema() { + Types.StructType partitionType = Partitioning.partitionType(table()); + Set metadataFieldIds = + partitionType.fields().stream().map(Types.NestedField::fieldId).collect(Collectors.toSet()); Schema result = new Schema( - MetadataColumns.DELETE_FILE_PATH, - MetadataColumns.DELETE_FILE_POS, - Types.NestedField.optional( - MetadataColumns.DELETE_FILE_ROW_FIELD_ID, - MetadataColumns.DELETE_FILE_ROW_FIELD_NAME, - table().schema().asStruct(), - MetadataColumns.DELETE_FILE_ROW_DOC), - Types.NestedField.required( - MetadataColumns.PARTITION_COLUMN_ID, - PARTITION, - partitionType(partitionType, partfieldMap), - "Partition that position delete row belongs to"), - Types.NestedField.required( - MetadataColumns.SPEC_ID_COLUMN_ID, - SPEC_ID, - Types.IntegerType.get(), - MetadataColumns.SPEC_ID_COLUMN_DOC), - Types.NestedField.required( - MetadataColumns.FILE_PATH_COLUMN_ID, - DELETE_FILE_PATH, - Types.StringType.get(), - MetadataColumns.FILE_PATH_COLUMN_DOC)); + ImmutableList.of( + MetadataColumns.DELETE_FILE_PATH, + MetadataColumns.DELETE_FILE_POS, + Types.NestedField.optional( + MetadataColumns.DELETE_FILE_ROW_FIELD_ID, + MetadataColumns.DELETE_FILE_ROW_FIELD_NAME, + table().schema().asStruct(), + MetadataColumns.DELETE_FILE_ROW_DOC), + Types.NestedField.required( + MetadataColumns.PARTITION_COLUMN_ID, + PARTITION, + partitionType, + "Partition that position delete row belongs to"), + Types.NestedField.required( + MetadataColumns.SPEC_ID_COLUMN_ID, + SPEC_ID, + Types.IntegerType.get(), + MetadataColumns.SPEC_ID_COLUMN_DOC), + Types.NestedField.required( + MetadataColumns.FILE_PATH_COLUMN_ID, + DELETE_FILE_PATH, + Types.StringType.get(), + MetadataColumns.FILE_PATH_COLUMN_DOC)), + ImmutableSet.of(), + metadataFieldIds); if (!partitionType.fields().isEmpty()) { return result; @@ -147,88 +149,25 @@ private Schema calculateSchema( } } - /** - * Handle collisions between table and partition field ids, as both need to be part of position - * deletes table - * - * @param tableSchema original table schema - * @param partitionType original table's partition type - * @return partition type with reassigned field ids - */ - public static Types.StructType partitionType(Schema tableSchema, Types.StructType partitionType) { - Map fieldMap = partitionFieldMap(tableSchema, partitionType); - return partitionType(partitionType, fieldMap); - } - - // Handle collisions between table field and partition field ids - static Map partitionFieldMap( - Schema tableSchema, Types.StructType partitionType) { - AtomicInteger nextId = new AtomicInteger(tableSchema.highestFieldId()); - return partitionType.fields().stream() - .collect(Collectors.toMap(Types.NestedField::fieldId, f -> nextId.incrementAndGet())); - } - - static Types.StructType partitionType( - Types.StructType partitionType, Map fieldMap) { - return Types.StructType.of( - partitionType.fields().stream() - .map( - f -> - Types.NestedField.of( - fieldMap.get(f.fieldId()), f.isOptional(), f.name(), f.type(), f.doc())) - .collect(Collectors.toList())); - } - public static class PositionDeletesBatchScan extends SnapshotScan> implements BatchScan { private Expression baseTableFilter = Expressions.alwaysTrue(); - private final Map fieldMap; - - protected PositionDeletesBatchScan(Table table, Schema schema, Map fieldMap) { - super(table, schema, TableScanContext.empty()); - this.fieldMap = fieldMap; - } - - protected PositionDeletesBatchScan( - Table table, - Schema schema, - TableScanContext context, - Expression baseTableFilter, - Map fieldMap) { - super(table, schema, context); - this.baseTableFilter = baseTableFilter; - this.fieldMap = fieldMap; - } - /** @deprecated since 1.5.0, will be removed in 1.6.0; use fieldMap constructor instead. */ - @Deprecated protected PositionDeletesBatchScan(Table table, Schema schema) { super(table, schema, TableScanContext.empty()); - this.fieldMap = ImmutableMap.of(); - } - - /** @deprecated since 1.5.0, will be removed in 1.6.0; use fieldMap constructor instead. */ - @Deprecated - protected PositionDeletesBatchScan(Table table, Schema schema, TableScanContext context) { - super(table, schema, context); - this.fieldMap = ImmutableMap.of(); } - /** @deprecated since 1.5.0, will be removed in 1.6.0; use fieldMap constructor instead. */ - @Deprecated protected PositionDeletesBatchScan( Table table, Schema schema, TableScanContext context, Expression baseTableFilter) { super(table, schema, context); this.baseTableFilter = baseTableFilter; - this.fieldMap = ImmutableMap.of(); } @Override protected PositionDeletesBatchScan newRefinedScan( Table newTable, Schema newSchema, TableScanContext newContext) { - return new PositionDeletesBatchScan( - newTable, newSchema, newContext, baseTableFilter, fieldMap); + return new PositionDeletesBatchScan(newTable, newSchema, newContext, baseTableFilter); } @Override @@ -265,14 +204,15 @@ protected List scanColumns() { */ public BatchScan baseTableFilter(Expression expr) { return new PositionDeletesBatchScan( - table(), schema(), context(), Expressions.and(baseTableFilter, expr), fieldMap); + table(), schema(), context(), Expressions.and(baseTableFilter, expr)); } @Override protected CloseableIterable doPlanFiles() { String schemaString = SchemaParser.toJson(tableSchema()); - Map transformedSpecs = - transformSpecs(tableSchema(), table().specs(), fieldMap); + + // prepare transformed partition specs and caches + Map transformedSpecs = transformSpecs(tableSchema(), table().specs()); LoadingCache specStringCache = partitionCacheOf(transformedSpecs, PartitionSpecParser::toJson); @@ -346,38 +286,28 @@ public void close() throws IOException { @Override public CloseableIterator iterator() { - // Partition filter by base table filter Expression partitionFilter = Projections.inclusive(spec, isCaseSensitive()).project(baseTableFilter); - // Read manifests (use original table's partition ids to de-serialize partition values) + // Filter partitions CloseableIterable> deleteFileEntries = - ManifestFiles.readDeleteManifest(manifest, table().io(), table().specs()) + ManifestFiles.readDeleteManifest(manifest, table().io(), transformedSpecs) .caseSensitive(isCaseSensitive()) .select(scanColumns()) + .filterRows(filter()) .filterPartitions(partitionFilter) .scanMetrics(scanMetrics()) .liveEntries(); - // Partition Filter by metadata table filter (bind on transformed spec/schema) - PartitionSpec transformedSpec = transformedSpecs.get(spec.specId()); - Expression projected = - Projections.inclusive(transformedSpec, isCaseSensitive()).project(filter()); - Evaluator eval = - new Evaluator(transformedSpec.partitionType(), projected, isCaseSensitive()); - deleteFileEntries = - CloseableIterable.filter( - deleteFileEntries, entry -> eval.eval(entry.file().partition())); - - // Filter by delete file type - deleteFileEntries = + // Filter delete file type + CloseableIterable> positionDeleteEntries = CloseableIterable.filter( deleteFileEntries, entry -> entry.file().content().equals(FileContent.POSITION_DELETES)); this.iterable = CloseableIterable.transform( - deleteFileEntries, + positionDeleteEntries, entry -> { int specId = entry.file().specId(); return new BasePositionDeletesScanTask( diff --git a/core/src/test/java/org/apache/iceberg/TestMetadataTableScans.java b/core/src/test/java/org/apache/iceberg/TestMetadataTableScans.java index 23d684041267..5a53f4454e13 100644 --- a/core/src/test/java/org/apache/iceberg/TestMetadataTableScans.java +++ b/core/src/test/java/org/apache/iceberg/TestMetadataTableScans.java @@ -1266,8 +1266,7 @@ public void testPositionDeletesWithFilter() { ScanTask task = tasks.get(0); assertThat(task).isInstanceOf(PositionDeletesScanTask.class); - Types.StructType partitionType = - PositionDeletesTable.partitionType(table.schema(), Partitioning.partitionType(table)); + Types.StructType partitionType = positionDeletesTable.spec().partitionType(); PositionDeletesScanTask posDeleteTask = (PositionDeletesScanTask) task; int filePartition = posDeleteTask.file().partition().get(0, Integer.class); @@ -1334,8 +1333,7 @@ private void testPositionDeletesBaseTableFilter(boolean transactional) { ScanTask task = tasks.get(0); assertThat(task).isInstanceOf(PositionDeletesScanTask.class); - Types.StructType partitionType = - PositionDeletesTable.partitionType(table.schema(), Partitioning.partitionType(table)); + Types.StructType partitionType = positionDeletesTable.spec().partitionType(); PositionDeletesScanTask posDeleteTask = (PositionDeletesScanTask) task; // base table filter should only be used to evaluate partitions @@ -1417,8 +1415,7 @@ public void testPositionDeletesWithBaseTableFilterNot() { ScanTask task = tasks.get(0); assertThat(task).isInstanceOf(PositionDeletesScanTask.class); - Types.StructType partitionType = - PositionDeletesTable.partitionType(table.schema(), Partitioning.partitionType(table)); + Types.StructType partitionType = positionDeletesTable.spec().partitionType(); PositionDeletesScanTask posDeleteTask = (PositionDeletesScanTask) task; // base table filter should only be used to evaluate partitions @@ -1429,7 +1426,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); diff --git a/core/src/test/java/org/apache/iceberg/TestMetadataTableScansWithPartitionEvolution.java b/core/src/test/java/org/apache/iceberg/TestMetadataTableScansWithPartitionEvolution.java index 86145555a7e6..a2e5386d29df 100644 --- a/core/src/test/java/org/apache/iceberg/TestMetadataTableScansWithPartitionEvolution.java +++ b/core/src/test/java/org/apache/iceberg/TestMetadataTableScansWithPartitionEvolution.java @@ -185,8 +185,7 @@ public void testPositionDeletesPartitionSpecRemoval() { ScanTask task = tasks.get(0); assertThat(task).isInstanceOf(PositionDeletesScanTask.class); - Types.StructType partitionType = - PositionDeletesTable.partitionType(table.schema(), Partitioning.partitionType(table)); + Types.StructType partitionType = positionDeletesTable.spec().partitionType(); PositionDeletesScanTask posDeleteTask = (PositionDeletesScanTask) task; int filePartition = posDeleteTask.file().partition().get(0, Integer.class); @@ -196,13 +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(partitionType.fields().get(1).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..b3236a4ca63e 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,10 +71,12 @@ 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; import org.junit.Test; +import org.junit.jupiter.api.TestTemplate; import org.junit.rules.TemporaryFolder; import org.junit.runners.Parameterized; @@ -581,6 +585,69 @@ public void testSchemaEvolution() throws Exception { assertEquals("Rows must match", expectedRecords, actualRecords); } + @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 = 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..6eaa3fe02cd4 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,10 +72,12 @@ 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; import org.junit.Test; +import org.junit.jupiter.api.TestTemplate; import org.junit.rules.TemporaryFolder; import org.junit.runners.Parameterized; @@ -618,6 +621,69 @@ public void testSchemaEvolution() throws Exception { assertEquals("Rows must match", expectedRecords, actualRecords); } + @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 = 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 7ec27063b51f..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 @@ -35,7 +35,6 @@ import org.apache.iceberg.MetadataTableUtils; import org.apache.iceberg.Partitioning; import org.apache.iceberg.PositionDeletesScanTask; -import org.apache.iceberg.PositionDeletesTable; import org.apache.iceberg.PositionDeletesTable.PositionDeletesBatchScan; import org.apache.iceberg.RewriteJobOrder; import org.apache.iceberg.StructLike; @@ -60,7 +59,6 @@ import org.apache.iceberg.relocated.com.google.common.math.IntMath; import org.apache.iceberg.relocated.com.google.common.util.concurrent.MoreExecutors; import org.apache.iceberg.relocated.com.google.common.util.concurrent.ThreadFactoryBuilder; -import org.apache.iceberg.types.Types; import org.apache.iceberg.types.Types.StructType; import org.apache.iceberg.util.PartitionUtil; import org.apache.iceberg.util.PropertyUtil; @@ -139,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); @@ -155,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(), @@ -460,7 +457,6 @@ public int totalGroupCount() { } private StructLike coercePartition(PositionDeletesScanTask task, StructType partitionType) { - Types.StructType dedupType = PositionDeletesTable.partitionType(table.schema(), partitionType); - return PartitionUtil.coercePartition(dedupType, task.spec(), task.partition()); + return PartitionUtil.coercePartition(partitionType, task.spec(), task.partition()); } } From d45ad89097ef4c60407e6093b51f45a9f02136bc Mon Sep 17 00:00:00 2001 From: Szehon Ho Date: Thu, 11 Apr 2024 15:45:40 -0700 Subject: [PATCH 08/15] Remove test template from older spark versions --- .../spark/actions/TestRewritePositionDeleteFilesAction.java | 3 +-- .../spark/actions/TestRewritePositionDeleteFilesAction.java | 3 +-- 2 files changed, 2 insertions(+), 4 deletions(-) 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 b3236a4ca63e..27a0a9d0e127 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 @@ -76,7 +76,6 @@ import org.junit.Assert; import org.junit.Rule; import org.junit.Test; -import org.junit.jupiter.api.TestTemplate; import org.junit.rules.TemporaryFolder; import org.junit.runners.Parameterized; @@ -585,7 +584,7 @@ public void testSchemaEvolution() throws Exception { assertEquals("Rows must match", expectedRecords, actualRecords); } - @TestTemplate + @Test public void testRewriteManyColumns() throws Exception { List fields = Lists.newArrayList(Types.NestedField.required(0, "id", Types.LongType.get())); 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 6eaa3fe02cd4..c1c0986bd818 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 @@ -77,7 +77,6 @@ import org.junit.Assert; import org.junit.Rule; import org.junit.Test; -import org.junit.jupiter.api.TestTemplate; import org.junit.rules.TemporaryFolder; import org.junit.runners.Parameterized; @@ -621,7 +620,7 @@ public void testSchemaEvolution() throws Exception { assertEquals("Rows must match", expectedRecords, actualRecords); } - @TestTemplate + @Test public void testRewriteManyColumns() throws Exception { List fields = Lists.newArrayList(Types.NestedField.required(0, "id", Types.LongType.get())); From 50871d29f75c8d634643796cdf9a08a1e81facb2 Mon Sep 17 00:00:00 2001 From: Szehon Ho Date: Thu, 11 Apr 2024 16:51:04 -0700 Subject: [PATCH 09/15] Fix older spark tests --- .../spark/actions/TestRewritePositionDeleteFilesAction.java | 2 +- .../spark/actions/TestRewritePositionDeleteFilesAction.java | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) 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 27a0a9d0e127..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 @@ -587,7 +587,7 @@ public void testSchemaEvolution() throws Exception { @Test public void testRewriteManyColumns() throws Exception { List fields = - Lists.newArrayList(Types.NestedField.required(0, "id", Types.LongType.get())); + 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())) 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 c1c0986bd818..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 @@ -623,7 +623,7 @@ public void testSchemaEvolution() throws Exception { @Test public void testRewriteManyColumns() throws Exception { List fields = - Lists.newArrayList(Types.NestedField.required(0, "id", Types.LongType.get())); + 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())) From 57ede3bf53c9b08218e1c749d0bbd7fe3f14e538 Mon Sep 17 00:00:00 2001 From: Szehon Ho Date: Tue, 7 May 2024 18:20:36 -0700 Subject: [PATCH 10/15] Review comments --- .palantir/revapi.yml | 6 ++++ .../org/apache/iceberg/PartitionSpec.java | 34 +++++++++---------- .../java/org/apache/iceberg/types/Types.java | 4 +++ .../org/apache/iceberg/ManifestReader.java | 2 +- 4 files changed, 28 insertions(+), 18 deletions(-) diff --git a/.palantir/revapi.yml b/.palantir/revapi.yml index a41d3ddfb8df..bd4f6fa77b0b 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: "Added new API only" 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 6a4716ea7956..100b52c022d4 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,7 +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 lazyOriginalPartitionType = null; + private transient volatile StructType lazyRawPartitionType = null; private transient volatile List fieldList = null; private final int lastAssignedFieldId; @@ -141,29 +142,28 @@ public StructType partitionType() { return lazyPartitionType; } - public StructType originalPartitionType() { - if (schema.idsToOriginal().size() == 0) { + /** + * Returns a struct with TransformID's which match the ID's used in the Table Metadata. This is + * different than the {@link #partitionType()} method which returns a struct which is guaranteed + * not to overlap with column ID's of the table by reassigning ID's. + */ + public StructType rawPartitionType() { + if (schema.idsToOriginal().isEmpty()) { return partitionType(); } - if (lazyOriginalPartitionType == null) { + if (lazyRawPartitionType == null) { synchronized (this) { - if (lazyOriginalPartitionType == null) { - List structFields = Lists.newArrayListWithExpectedSize(fields.length); - - for (PartitionField field : fields) { - Type sourceType = schema.findType(field.sourceId()); - Type resultType = field.transform().getResultType(sourceType); - structFields.add( - Types.NestedField.optional( - schema.idsToOriginal().get(field.fieldId()), field.name(), resultType)); - } - - this.lazyOriginalPartitionType = Types.StructType.of(structFields); + if (lazyRawPartitionType == null) { + this.lazyRawPartitionType = + StructType.of( + partitionType().fields().stream() + .map(f -> f.withFieldId(schema.idsToOriginal().get(f.fieldId()))) + .collect(Collectors.toList())); } } } - return lazyOriginalPartitionType; + return lazyRawPartitionType; } public Class[] javaClasses() { 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/ManifestReader.java b/core/src/main/java/org/apache/iceberg/ManifestReader.java index a117ceb26d06..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.originalPartitionType()).fields()); + this.fileSchema = new Schema(DataFile.getType(spec.rawPartitionType()).fields()); } private > PartitionSpec readPartitionSpec(InputFile inputFile) { From 394f07b6df0e304fe6681b0b41d6becc9607b6fa Mon Sep 17 00:00:00 2001 From: Szehon Ho Date: Mon, 10 Jun 2024 11:23:22 -0700 Subject: [PATCH 11/15] Fix revapi message --- .palantir/revapi.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.palantir/revapi.yml b/.palantir/revapi.yml index bd4f6fa77b0b..808a19299055 100644 --- a/.palantir/revapi.yml +++ b/.palantir/revapi.yml @@ -1023,7 +1023,7 @@ acceptedBreaks: - code: "java.class.defaultSerializationChanged" old: "class org.apache.iceberg.types.Types.NestedField" new: "class org.apache.iceberg.types.Types.NestedField" - justification: "Added new API only" + justification: "new Constructor added" apache-iceberg-0.14.0: org.apache.iceberg:iceberg-api: - code: "java.class.defaultSerializationChanged" From 6aa3993da0fada38b125d93f1380bc6625d4ced7 Mon Sep 17 00:00:00 2001 From: Szehon Ho Date: Mon, 10 Jun 2024 14:27:51 -0700 Subject: [PATCH 12/15] More review comments --- .../main/java/org/apache/iceberg/PartitionSpec.java | 10 +++++++--- api/src/main/java/org/apache/iceberg/Schema.java | 7 ++++--- 2 files changed, 11 insertions(+), 6 deletions(-) diff --git a/api/src/main/java/org/apache/iceberg/PartitionSpec.java b/api/src/main/java/org/apache/iceberg/PartitionSpec.java index 100b52c022d4..11c84d6a255f 100644 --- a/api/src/main/java/org/apache/iceberg/PartitionSpec.java +++ b/api/src/main/java/org/apache/iceberg/PartitionSpec.java @@ -143,12 +143,16 @@ public StructType partitionType() { } /** - * Returns a struct with TransformID's which match the ID's used in the Table Metadata. This is - * different than the {@link #partitionType()} method which returns a struct which is guaranteed - * not to overlap with column ID's of the table by reassigning ID's. + * While partition field Id's are based on the column they are transforming, some Schemas need to + * re-assign partition field Id's to avoid conflict with defined column field ID's. + * + * @return a struct representing the partition type, with original field ID's that match the + * column field ID's that they refer to. 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) { diff --git a/api/src/main/java/org/apache/iceberg/Schema.java b/api/src/main/java/org/apache/iceberg/Schema.java index 856dd906d7ca..bcb558c73ff3 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; @@ -540,12 +541,12 @@ public String toString() { } /** - * All ids of metadata fields are reassigned. + * Fields identified as 'Metadata Fields' will have their field ID's reassigned. * - * @return map of original to reassigned field ids of metadata fields + * @return map of original to reassigned field ids */ public Map idsToReassigned() { - return idsToReassigned != null ? idsToReassigned : Maps.newHashMap(); + return idsToReassigned != null ? idsToReassigned : Collections.emptyMap(); } /** From 1052a3367f7926f9f3871df966542d62c039c12f Mon Sep 17 00:00:00 2001 From: Szehon Ho Date: Mon, 10 Jun 2024 17:34:46 -0700 Subject: [PATCH 13/15] Review comments --- api/src/main/java/org/apache/iceberg/Schema.java | 7 +++---- api/src/main/java/org/apache/iceberg/types/AssignIds.java | 5 ----- api/src/main/java/org/apache/iceberg/types/TypeUtil.java | 2 +- 3 files changed, 4 insertions(+), 10 deletions(-) diff --git a/api/src/main/java/org/apache/iceberg/Schema.java b/api/src/main/java/org/apache/iceberg/Schema.java index bcb558c73ff3..d9c9f9ec7180 100644 --- a/api/src/main/java/org/apache/iceberg/Schema.java +++ b/api/src/main/java/org/apache/iceberg/Schema.java @@ -561,8 +561,8 @@ public Map idsToOriginal() { private List reassignMetadataFieldIds( List columns, Set metadataFieldIds) { Set usedIds = - Sets.newHashSet( - Sets.difference(TypeUtil.indexById(StructType.of(columns)).keySet(), metadataFieldIds)); + Sets.difference(TypeUtil.indexById(StructType.of(columns)).keySet(), metadataFieldIds) + .immutableCopy(); AtomicInteger nextId = new AtomicInteger(); Type res = @@ -570,11 +570,10 @@ private List reassignMetadataFieldIds( StructType.of(columns), id -> { if (metadataFieldIds.contains(id)) { - int candidate = nextId.get(); + int candidate = nextId.incrementAndGet(); while (usedIds.contains(candidate)) { candidate = nextId.incrementAndGet(); } - usedIds.add(candidate); idsToReassigned.put(id, candidate); idsToOriginal.put(candidate, id); return candidate; diff --git a/api/src/main/java/org/apache/iceberg/types/AssignIds.java b/api/src/main/java/org/apache/iceberg/types/AssignIds.java index 15fe482ef05f..68588f581adc 100644 --- a/api/src/main/java/org/apache/iceberg/types/AssignIds.java +++ b/api/src/main/java/org/apache/iceberg/types/AssignIds.java @@ -27,11 +27,6 @@ class AssignIds extends TypeUtil.CustomOrderSchemaVisitor { private final TypeUtil.GetID getID; - /** - * Replaces the ids in a schema with ids from a base schema, or uses nextId to assign a fresh ids. - * - * @param getID new id assigner - */ AssignIds(TypeUtil.GetID getID) { this.getID = getID; } 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 7198544a0c1d..07d06dcc5a89 100644 --- a/api/src/main/java/org/apache/iceberg/types/TypeUtil.java +++ b/api/src/main/java/org/apache/iceberg/types/TypeUtil.java @@ -360,7 +360,7 @@ public static Schema reassignOrRefreshIds( * * @param type a type * @param getId an id assignment function - * @return an structurally identical type with new ids assigned by the nextId 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)); From 02d46c5dfb44de1b0b39b0bd229373b7aa34e5c1 Mon Sep 17 00:00:00 2001 From: Szehon Ho Date: Tue, 11 Jun 2024 16:39:35 -0700 Subject: [PATCH 14/15] Fix problem of re-using ids in deleted columns --- .../org/apache/iceberg/PartitionSpec.java | 8 +- .../main/java/org/apache/iceberg/Schema.java | 54 ++++++------- .../apache/iceberg/PositionDeletesTable.java | 75 +++++++++++++------ 3 files changed, 75 insertions(+), 62 deletions(-) diff --git a/api/src/main/java/org/apache/iceberg/PartitionSpec.java b/api/src/main/java/org/apache/iceberg/PartitionSpec.java index 11c84d6a255f..8f1df794030a 100644 --- a/api/src/main/java/org/apache/iceberg/PartitionSpec.java +++ b/api/src/main/java/org/apache/iceberg/PartitionSpec.java @@ -143,12 +143,8 @@ public StructType partitionType() { } /** - * While partition field Id's are based on the column they are transforming, some Schemas need to - * re-assign partition field Id's to avoid conflict with defined column field ID's. - * - * @return a struct representing the partition type, with original field ID's that match the - * column field ID's that they refer to. See {@link #partitionType()} for a struct with field - * ID's potentially re-assigned to avoid conflict. + * 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()) { diff --git a/api/src/main/java/org/apache/iceberg/Schema.java b/api/src/main/java/org/apache/iceberg/Schema.java index d9c9f9ec7180..72886005c6e1 100644 --- a/api/src/main/java/org/apache/iceberg/Schema.java +++ b/api/src/main/java/org/apache/iceberg/Schema.java @@ -27,7 +27,6 @@ import java.util.Locale; import java.util.Map; import java.util.Set; -import java.util.concurrent.atomic.AtomicInteger; import java.util.stream.Collectors; import org.apache.iceberg.relocated.com.google.common.base.Joiner; import org.apache.iceberg.relocated.com.google.common.base.Preconditions; @@ -88,9 +87,8 @@ public Schema(List columns, Set identifierFieldIds) { this(DEFAULT_SCHEMA_ID, columns, identifierFieldIds); } - public Schema( - List columns, Set identifierFieldIds, Set metadataFieldIds) { - this(DEFAULT_SCHEMA_ID, columns, identifierFieldIds, metadataFieldIds); + public Schema(List columns, Set identifierFieldIds, TypeUtil.GetID getId) { + this(DEFAULT_SCHEMA_ID, columns, identifierFieldIds); } public Schema(int schemaId, List columns) { @@ -98,15 +96,15 @@ public Schema(int schemaId, List columns) { } public Schema(int schemaId, List columns, Set identifierFieldIds) { - this(schemaId, columns, null, identifierFieldIds, ImmutableSet.of()); + this(schemaId, columns, null, identifierFieldIds, null); } public Schema( int schemaId, List columns, Set identifierFieldIds, - Set metadataFieldIds) { - this(schemaId, columns, null, identifierFieldIds, metadataFieldIds); + TypeUtil.GetID getId) { + this(schemaId, columns, null, identifierFieldIds, getId); } public Schema( @@ -114,7 +112,7 @@ public Schema( List columns, Map aliases, Set identifierFieldIds) { - this(schemaId, columns, aliases, identifierFieldIds, ImmutableSet.of()); + this(schemaId, columns, aliases, identifierFieldIds, null); } public Schema( @@ -122,12 +120,12 @@ public Schema( List columns, Map aliases, Set identifierFieldIds, - Set metadataFieldIds) { + TypeUtil.GetID getID) { this.schemaId = schemaId; this.idsToOriginal = Maps.newHashMap(); this.idsToReassigned = Maps.newHashMap(); - List finalColumns = reassignMetadataFieldIds(columns, metadataFieldIds); + List finalColumns = reassignIds(columns, getID); this.struct = StructType.of(finalColumns); this.aliasToId = aliases != null ? ImmutableBiMap.copyOf(aliases) : null; @@ -541,7 +539,7 @@ public String toString() { } /** - * Fields identified as 'Metadata Fields' will have their field ID's reassigned. + * 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 */ @@ -550,36 +548,28 @@ public Map idsToReassigned() { } /** - * All ids of metadata fields are reassigned. + * 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 of metadata fields + * @return map of reassigned to original field ids */ public Map idsToOriginal() { - return idsToOriginal != null ? idsToOriginal : Maps.newHashMap(); + return idsToOriginal != null ? idsToOriginal : Collections.emptyMap(); } - private List reassignMetadataFieldIds( - List columns, Set metadataFieldIds) { - Set usedIds = - Sets.difference(TypeUtil.indexById(StructType.of(columns)).keySet(), metadataFieldIds) - .immutableCopy(); - AtomicInteger nextId = new AtomicInteger(); - + private List reassignIds(List columns, TypeUtil.GetID getID) { + if (getID == null) { + return columns; + } Type res = TypeUtil.assignIds( StructType.of(columns), - id -> { - if (metadataFieldIds.contains(id)) { - int candidate = nextId.incrementAndGet(); - while (usedIds.contains(candidate)) { - candidate = nextId.incrementAndGet(); - } - idsToReassigned.put(id, candidate); - idsToOriginal.put(candidate, id); - return candidate; - } else { - return id; + 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/core/src/main/java/org/apache/iceberg/PositionDeletesTable.java b/core/src/main/java/org/apache/iceberg/PositionDeletesTable.java index 394869c973b4..e901e6ee2e31 100644 --- a/core/src/main/java/org/apache/iceberg/PositionDeletesTable.java +++ b/core/src/main/java/org/apache/iceberg/PositionDeletesTable.java @@ -25,6 +25,7 @@ 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; @@ -112,33 +113,59 @@ private Schema calculateSchema() { Types.StructType partitionType = Partitioning.partitionType(table()); Set metadataFieldIds = partitionType.fields().stream().map(Types.NestedField::fieldId).collect(Collectors.toSet()); + List columns = + ImmutableList.of( + MetadataColumns.DELETE_FILE_PATH, + MetadataColumns.DELETE_FILE_POS, + Types.NestedField.optional( + MetadataColumns.DELETE_FILE_ROW_FIELD_ID, + MetadataColumns.DELETE_FILE_ROW_FIELD_NAME, + table().schema().asStruct(), + MetadataColumns.DELETE_FILE_ROW_DOC), + Types.NestedField.required( + MetadataColumns.PARTITION_COLUMN_ID, + PARTITION, + partitionType, + "Partition that position delete row belongs to"), + Types.NestedField.required( + MetadataColumns.SPEC_ID_COLUMN_ID, + SPEC_ID, + Types.IntegerType.get(), + MetadataColumns.SPEC_ID_COLUMN_DOC), + Types.NestedField.required( + MetadataColumns.FILE_PATH_COLUMN_ID, + DELETE_FILE_PATH, + 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 usedIds = + 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( - ImmutableList.of( - MetadataColumns.DELETE_FILE_PATH, - MetadataColumns.DELETE_FILE_POS, - Types.NestedField.optional( - MetadataColumns.DELETE_FILE_ROW_FIELD_ID, - MetadataColumns.DELETE_FILE_ROW_FIELD_NAME, - table().schema().asStruct(), - MetadataColumns.DELETE_FILE_ROW_DOC), - Types.NestedField.required( - MetadataColumns.PARTITION_COLUMN_ID, - PARTITION, - partitionType, - "Partition that position delete row belongs to"), - Types.NestedField.required( - MetadataColumns.SPEC_ID_COLUMN_ID, - SPEC_ID, - Types.IntegerType.get(), - MetadataColumns.SPEC_ID_COLUMN_DOC), - Types.NestedField.required( - MetadataColumns.FILE_PATH_COLUMN_ID, - DELETE_FILE_PATH, - Types.StringType.get(), - MetadataColumns.FILE_PATH_COLUMN_DOC)), + columns, ImmutableSet.of(), - metadataFieldIds); + oldId -> { + if (!idsToReassign.contains(oldId)) { + return oldId; + } + int candidate = nextId.incrementAndGet(); + while (usedIds.contains(candidate)) { + candidate = nextId.incrementAndGet(); + } + return candidate; + }); if (!partitionType.fields().isEmpty()) { return result; From 31dc0e09f8c180ae3186d6432389242ffd311296 Mon Sep 17 00:00:00 2001 From: Szehon Ho Date: Wed, 12 Jun 2024 15:38:30 -0700 Subject: [PATCH 15/15] Review comments, add test --- .../main/java/org/apache/iceberg/Schema.java | 2 +- .../apache/iceberg/PositionDeletesTable.java | 6 +- .../iceberg/TestMetadataTableScans.java | 63 +++++++++++++++++++ 3 files changed, 66 insertions(+), 5 deletions(-) diff --git a/api/src/main/java/org/apache/iceberg/Schema.java b/api/src/main/java/org/apache/iceberg/Schema.java index 72886005c6e1..d5ec3f250982 100644 --- a/api/src/main/java/org/apache/iceberg/Schema.java +++ b/api/src/main/java/org/apache/iceberg/Schema.java @@ -88,7 +88,7 @@ public Schema(List columns, Set identifierFieldIds) { } public Schema(List columns, Set identifierFieldIds, TypeUtil.GetID getId) { - this(DEFAULT_SCHEMA_ID, columns, identifierFieldIds); + this(DEFAULT_SCHEMA_ID, columns, identifierFieldIds, getId); } public Schema(int schemaId, List columns) { diff --git a/core/src/main/java/org/apache/iceberg/PositionDeletesTable.java b/core/src/main/java/org/apache/iceberg/PositionDeletesTable.java index e901e6ee2e31..382ad663a8d1 100644 --- a/core/src/main/java/org/apache/iceberg/PositionDeletesTable.java +++ b/core/src/main/java/org/apache/iceberg/PositionDeletesTable.java @@ -111,8 +111,6 @@ public Map properties() { private Schema calculateSchema() { Types.StructType partitionType = Partitioning.partitionType(table()); - Set metadataFieldIds = - partitionType.fields().stream().map(Types.NestedField::fieldId).collect(Collectors.toSet()); List columns = ImmutableList.of( MetadataColumns.DELETE_FILE_PATH, @@ -141,7 +139,7 @@ private Schema calculateSchema() { // Calculate used ids (for de-conflict) Set currentlyUsedIds = Collections.unmodifiableSet(TypeUtil.indexById(Types.StructType.of(columns)).keySet()); - Set usedIds = + Set allUsedIds = table().schemas().values().stream() .map(currSchema -> TypeUtil.indexById(currSchema.asStruct()).keySet()) .reduce(currentlyUsedIds, Sets::union); @@ -161,7 +159,7 @@ private Schema calculateSchema() { return oldId; } int candidate = nextId.incrementAndGet(); - while (usedIds.contains(candidate)) { + while (allUsedIds.contains(candidate)) { candidate = nextId.incrementAndGet(); } return candidate; diff --git a/core/src/test/java/org/apache/iceberg/TestMetadataTableScans.java b/core/src/test/java/org/apache/iceberg/TestMetadataTableScans.java index 5a53f4454e13..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; @@ -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"); + } }