koodin9 commented on code in PR #18003: URL: https://github.com/apache/iceberg/pull/18003#discussion_r4357527446
########## kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/data/EqualityDeleteConverter.java: ########## @@ -0,0 +1,336 @@ +/* + * 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.connect.data; + +import java.io.IOException; +import java.io.UncheckedIOException; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Set; +import java.util.UUID; +import java.util.concurrent.atomic.AtomicInteger; +import org.apache.iceberg.DeleteFile; +import org.apache.iceberg.FileContent; +import org.apache.iceberg.FileFormat; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.Schema; +import org.apache.iceberg.Snapshot; +import org.apache.iceberg.StructLike; +import org.apache.iceberg.Table; +import org.apache.iceberg.data.BaseDeleteLoader; +import org.apache.iceberg.data.DeleteLoader; +import org.apache.iceberg.deletes.BaseDVFileWriter; +import org.apache.iceberg.deletes.PositionDeleteIndex; +import org.apache.iceberg.expressions.Expression; +import org.apache.iceberg.expressions.Expressions; +import org.apache.iceberg.io.DeleteWriteResult; +import org.apache.iceberg.io.OutputFileFactory; +import org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting; +import org.apache.iceberg.relocated.com.google.common.base.Preconditions; +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.Lists; +import org.apache.iceberg.relocated.com.google.common.collect.Maps; +import org.apache.iceberg.types.TypeUtil; +import org.apache.iceberg.util.StructLikeSet; +import org.apache.iceberg.util.StructLikeWrapper; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Resolves equality delete files against the rows of a table and writes the matching row positions + * as deletion vectors, so the equality delete files never have to be committed. + * + * <p>An equality delete written by the sink deletes every row of an earlier commit whose identifier + * fields equal one of the deleted keys. The converter reads the keys, groups them by equality + * fields and partition, and asks a {@link KeyPositionResolver} for the rows that hold them. The + * default resolver, {@link ScanKeyPositionResolver}, scans the data files that the column + * statistics cannot rule out; an index-backed resolver can be plugged in instead. A data file that + * already has a deletion vector gets a merged one; the replaced vector is reported in {@link + * Result#rewrittenDvFiles()} and must be removed by the commit. + * + * <p>The positions are only valid for the snapshot they were resolved against, see {@link + * Result#baseSnapshotId()}. The commit that adds the deletion vectors has to validate that no + * conflicting data or delete files were added since then and re-run the conversion otherwise. + * {@link Result#conflictFilter()} narrows that validation to the deleted keys. + * + * <p>Requires format version 3. + */ +public class EqualityDeleteConverter { + + private static final Logger LOG = LoggerFactory.getLogger(EqualityDeleteConverter.class); + + private final Table table; + private final String branch; + private final KeyPositionResolver resolver; + private final DeleteLoader deleteLoader; + private final AtomicInteger scannedDataFiles = new AtomicInteger(); + + /** + * Creates a converter that finds the rows by scanning data files. + * + * @param table the table to resolve the deletes against + * @param branch the branch that is written, or null for the main branch + */ + public EqualityDeleteConverter(Table table, String branch) { + this(table, branch, new ScanKeyPositionResolver(table)); + } + + EqualityDeleteConverter(Table table, String branch, KeyPositionResolver resolver) { + this.table = table; + this.branch = branch; + this.resolver = resolver; + this.deleteLoader = new BaseDeleteLoader(deleteFile -> table.io().newInputFile(deleteFile)); + } + + /** + * Converts the given equality delete files into deletion vectors on the current snapshot of the + * branch. The table is expected to be refreshed by the caller. + */ + public Result convert(List<DeleteFile> eqDeleteFiles) { + Snapshot base = branch == null ? table.currentSnapshot() : table.snapshot(branch); + if (base == null) { + // there are no rows an equality delete could remove + LOG.info( + "Table {} has no snapshot on branch {}, {} equality delete file(s) match no rows", + table.name(), + branch == null ? "main" : branch, + eqDeleteFiles.size()); + return Result.empty(); Review Comment: Fixed in 92ee01a75. The keys are loaded and the conflict filter is built before the snapshot check, so a result without a base snapshot still carries the filter. `Result.baseSnapshotId()` now documents that the commit must then validate added data files against `conflictFilter()` without a starting snapshot. `testConflictFilterCoversKeysWithoutSnapshot` shows that a concurrent first commit with a deleted key fails that validation. A related case: a branch that doesn't exist yet was also resolved as empty, but a commit to it builds on the current snapshot of main, so rows of main that match the deletes survived in the new branch. Such a branch is now resolved against main, covered by `testConvertOnBranchThatDoesNotExistYet`. In part 2, the coordinator will always set the conflict filter and the conflict checks, and call `validateFromSnapshot` only when there is a base snapshot. ########## kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/data/TestEqualityDeleteConverter.java: ########## @@ -0,0 +1,720 @@ +/* + * 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.connect.data; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +import java.io.IOException; +import java.io.UncheckedIOException; +import java.util.Arrays; +import java.util.Collection; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.function.Function; +import java.util.function.LongConsumer; +import org.apache.iceberg.DataFile; +import org.apache.iceberg.DeleteFile; +import org.apache.iceberg.FileFormat; +import org.apache.iceberg.FileMetadata; +import org.apache.iceberg.PartitionKey; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.RowDelta; +import org.apache.iceberg.Schema; +import org.apache.iceberg.StructLike; +import org.apache.iceberg.Table; +import org.apache.iceberg.catalog.Namespace; +import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.iceberg.data.GenericFileWriterFactory; +import org.apache.iceberg.data.GenericRecord; +import org.apache.iceberg.data.IcebergGenerics; +import org.apache.iceberg.data.InternalRecordWrapper; +import org.apache.iceberg.data.Record; +import org.apache.iceberg.deletes.EqualityDeleteWriter; +import org.apache.iceberg.deletes.PositionDeleteIndex; +import org.apache.iceberg.expressions.Expressions; +import org.apache.iceberg.inmemory.InMemoryCatalog; +import org.apache.iceberg.io.CloseableIterable; +import org.apache.iceberg.io.DataWriter; +import org.apache.iceberg.io.FileWriterFactory; +import org.apache.iceberg.io.OutputFileFactory; +import org.apache.iceberg.io.WriteResult; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList; +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.transforms.Transforms; +import org.apache.iceberg.types.TypeUtil; +import org.apache.iceberg.types.Types; +import org.apache.iceberg.util.ContentFileUtil; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +public class TestEqualityDeleteConverter { + + private static final Namespace NAMESPACE = Namespace.of("db"); + private static final TableIdentifier TABLE_IDENTIFIER = TableIdentifier.of(NAMESPACE, "tbl"); + private static final Schema SCHEMA = + new Schema( + ImmutableList.of( + Types.NestedField.required(1, "id", Types.LongType.get()), + Types.NestedField.required(2, "category", Types.StringType.get()), + Types.NestedField.optional(3, "data", Types.StringType.get())), + ImmutableSet.of(1)); + + /** The change of one row, as a CDC writer would produce it. */ + private enum Op { + INSERT, + UPDATE, + DELETE + } + + private InMemoryCatalog catalog; + private List<Integer> equalityFieldIds; + + @BeforeEach + public void before() { + catalog = new InMemoryCatalog(); + catalog.initialize("test_catalog", ImmutableMap.of()); + catalog.createNamespace(NAMESPACE); + equalityFieldIds = ImmutableList.copyOf(SCHEMA.identifierFieldIds()); + } + + @AfterEach + public void after() throws IOException { + catalog.close(); + } + + @Test + public void testConvertsDeletesOfEarlierCommits() throws IOException { + Table table = createTable(PartitionSpec.unpartitioned()); + WriteResult first = + write( + table, + insert(1L, "a", "v1"), + insert(2L, "a", "v1"), + insert(3L, "a", "v1"), + insert(4L, "a", "v1"), + insert(5L, "a", "v1")); + commit(table, first.dataFiles(), first.deleteFiles(), ImmutableList.of()); + DataFile firstDataFile = first.dataFiles()[0]; + long firstSnapshotId = table.currentSnapshot().snapshotId(); + + WriteResult second = write(table, delete(2L, "a"), update(4L, "a", "v2")); + assertThat(second.deleteFiles()).hasSize(1); + + EqualityDeleteConverter converter = new EqualityDeleteConverter(table, null); + EqualityDeleteConverter.Result result = converter.convert(Arrays.asList(second.deleteFiles())); + + assertThat(result.baseSnapshotId()).isEqualTo(firstSnapshotId); + assertThat(result.rewrittenDvFiles()).isEmpty(); + assertThat(result.dvFiles()).hasSize(1); + DeleteFile dv = result.dvFiles().get(0); + assertThat(ContentFileUtil.isDV(dv)).isTrue(); + assertThat(dv.referencedDataFile()).isEqualTo(firstDataFile.location()); + assertThat(dv.recordCount()).isEqualTo(2L); + assertThat(converter.scannedDataFiles()).isEqualTo(1); + + commit(table, second.dataFiles(), result.dvFiles(), result.rewrittenDvFiles()); + assertThat(readRows(table)).containsExactlyInAnyOrder("1|a|v1", "3|a|v1", "4|a|v2", "5|a|v1"); + + // a second conversion for the same data file merges the existing deletion vector + WriteResult third = write(table, delete(1L, "a")); + result = converter.convert(Arrays.asList(third.deleteFiles())); + + assertThat(result.dvFiles()).hasSize(1); + assertThat(result.dvFiles().get(0).referencedDataFile()).isEqualTo(firstDataFile.location()); + assertThat(result.dvFiles().get(0).recordCount()).isEqualTo(3L); + assertThat(result.rewrittenDvFiles()).hasSize(1); + assertThat(result.rewrittenDvFiles().get(0).location()).isEqualTo(dv.location()); + assertThat(result.rewrittenDvFiles().get(0).contentOffset()).isEqualTo(dv.contentOffset()); + // the data file written by the second batch only holds id 4 and is pruned by its column bounds + assertThat(converter.scannedDataFiles()).isEqualTo(1); + + commit(table, third.dataFiles(), result.dvFiles(), result.rewrittenDvFiles()); + assertThat(readRows(table)).containsExactlyInAnyOrder("3|a|v1", "4|a|v2", "5|a|v1"); + } + + @Test + public void testPartitionPruning() throws IOException { + Table table = createTable(PartitionSpec.builderFor(SCHEMA).identity("category").build()); + // the partition column has to be part of the key for the delete to land in its partition + equalityFieldIds = ImmutableList.of(1, 2); + + WriteResult first = + write( + table, + insert(1L, "a", "v1"), + insert(2L, "a", "v1"), + insert(3L, "b", "v1"), + insert(4L, "b", "v1")); + assertThat(first.dataFiles()).hasSize(2); + commit(table, first.dataFiles(), first.deleteFiles(), ImmutableList.of()); + + WriteResult second = write(table, delete(1L, "a")); + EqualityDeleteConverter converter = new EqualityDeleteConverter(table, null); + EqualityDeleteConverter.Result result = converter.convert(Arrays.asList(second.deleteFiles())); + + // only the data file of partition a is a candidate + assertThat(converter.scannedDataFiles()).isEqualTo(1); + assertThat(result.dvFiles()).hasSize(1); + assertThat(result.dvFiles().get(0).recordCount()).isEqualTo(1L); + + commit(table, second.dataFiles(), result.dvFiles(), result.rewrittenDvFiles()); + assertThat(readRows(table)).containsExactlyInAnyOrder("2|a|v1", "3|b|v1", "4|b|v1"); + } + + @Test + public void testPartitionedDeletesSkipDataFilesOfOtherSpecs() throws IOException { + Table table = createTable(PartitionSpec.builderFor(SCHEMA).identity("category").build()); + equalityFieldIds = ImmutableList.of(1, 2); + + WriteResult first = write(table, insert(1L, "a", "v1"), insert(2L, "a", "v1")); + commit(table, first.dataFiles(), first.deleteFiles(), ImmutableList.of()); + + table.updateSpec().addField(Expressions.bucket("id", 4)).commit(); + WriteResult second = write(table, insert(3L, "a", "v1")); + commit(table, second.dataFiles(), second.deleteFiles(), ImmutableList.of()); + long baseSnapshotId = table.currentSnapshot().snapshotId(); + + WriteResult third = write(table, delete(1L, "a"), delete(3L, "a")); + + // the equality deletes committed as they are define the expected rows: a delete stored with + // the new spec does not apply to the data file written with the old spec + table.manageSnapshots().createBranch("expected", baseSnapshotId).commit(); + RowDelta expected = table.newRowDelta().toBranch("expected"); + Arrays.stream(third.deleteFiles()).forEach(expected::addDeletes); + expected.commit(); + Set<String> expectedRows = readRows(table, table.snapshot("expected").snapshotId()); + assertThat(expectedRows).containsExactlyInAnyOrder("1|a|v1", "2|a|v1"); + + EqualityDeleteConverter converter = new EqualityDeleteConverter(table, null); + EqualityDeleteConverter.Result result = converter.convert(Arrays.asList(third.deleteFiles())); + commit(table, third.dataFiles(), result.dvFiles(), result.rewrittenDvFiles()); + + assertThat(readRows(table)).isEqualTo(expectedRows); + } + + @Test + public void testManyKeys() throws IOException { + Table table = createTable(PartitionSpec.unpartitioned()); + + List<Object[]> inserts = Lists.newArrayList(); + for (long id = 1; id <= 300; id++) { + inserts.add(insert(id, "a", "v1")); + } + WriteResult first = write(table, inserts.toArray(new Object[0][])); + commit(table, first.dataFiles(), first.deleteFiles(), ImmutableList.of()); + + List<Object[]> deletes = Lists.newArrayList(); + for (long id = 1; id <= 250; id++) { + deletes.add(delete(id, "a")); + } + WriteResult second = write(table, deletes.toArray(new Object[0][])); + + EqualityDeleteConverter converter = new EqualityDeleteConverter(table, null); + EqualityDeleteConverter.Result result = converter.convert(Arrays.asList(second.deleteFiles())); + + assertThat(result.dvFiles()).hasSize(1); + assertThat(result.dvFiles().get(0).recordCount()).isEqualTo(250L); + + commit(table, second.dataFiles(), result.dvFiles(), result.rewrittenDvFiles()); + Set<String> rows = readRows(table); + assertThat(rows).hasSize(50); + assertThat(rows).contains("251|a|v1", "300|a|v1"); + assertThat(rows).doesNotContain("1|a|v1", "250|a|v1"); + } + + @Test + public void testClusteredKeysPruneToTheirFiles() throws IOException { + Table table = createTable(PartitionSpec.unpartitioned()); + + // 200 data files with 50 consecutive ids each + int fileCount = 200; + int rowsPerFile = 50; + List<DataFile> dataFiles = Lists.newArrayList(); + for (int file = 0; file < fileCount; file++) { + List<Object[]> inserts = Lists.newArrayList(); + for (int row = 0; row < rowsPerFile; row++) { + inserts.add(insert((long) file * rowsPerFile + row, "a", "v1")); + } + WriteResult result = write(table, inserts.toArray(new Object[0][])); + assertThat(result.dataFiles()).hasSize(1); + dataFiles.add(result.dataFiles()[0]); + } + commit(table, dataFiles.toArray(new DataFile[0]), new DeleteFile[0], ImmutableList.of()); + + // 295 hot keys in the last six files and five keys scattered over the table + List<Object[]> deletes = Lists.newArrayList(); + for (long id = 9700; id < 9995; id++) { + deletes.add(delete(id, "a")); + } + for (long id : new long[] {5, 1005, 3005, 6005, 8005}) { + deletes.add(delete(id, "a")); + } + WriteResult second = write(table, deletes.toArray(new Object[0][])); + + EqualityDeleteConverter converter = new EqualityDeleteConverter(table, null); + EqualityDeleteConverter.Result result = converter.convert(Arrays.asList(second.deleteFiles())); + + // more keys than one IN list holds, so the keys are matched with several lists: the hot + // range covers six files and every scattered key stays in its own file + assertThat(deletes).hasSizeGreaterThan(KeyFilters.IN_PREDICATE_LIMIT); + assertThat(converter.scannedDataFiles()).isEqualTo(11); + assertThat(result.dvFiles()).hasSize(11); + assertThat(result.dvFiles().stream().mapToLong(DeleteFile::recordCount).sum()).isEqualTo(300L); + + commit(table, second.dataFiles(), result.dvFiles(), result.rewrittenDvFiles()); + assertThat(readRows(table)).hasSize(fileCount * rowsPerFile - 300); + } + + @Test + public void testPartitionIsPinnedWithManyKeys() throws IOException { + Table table = createTable(PartitionSpec.builderFor(SCHEMA).identity("category").build()); + // the partition column has to be part of the key for the delete to land in its partition + equalityFieldIds = ImmutableList.of(1, 2); + + // three files of 100 ids in each of the partitions a and b, ids overlap across partitions + List<DataFile> dataFiles = Lists.newArrayList(); + for (String category : new String[] {"a", "b"}) { + for (int file = 0; file < 3; file++) { + List<Object[]> inserts = Lists.newArrayList(); + for (int row = 0; row < 100; row++) { + inserts.add(insert((long) file * 100 + row, category, "v1")); + } + WriteResult result = write(table, inserts.toArray(new Object[0][])); + dataFiles.add(result.dataFiles()[0]); + } + } + commit(table, dataFiles.toArray(new DataFile[0]), new DeleteFile[0], ImmutableList.of()); + + List<Object[]> deletes = Lists.newArrayList(); + for (long id = 0; id < 250; id++) { + deletes.add(delete(id, "a")); + } + WriteResult second = write(table, deletes.toArray(new Object[0][])); + + EqualityDeleteConverter converter = new EqualityDeleteConverter(table, null); + EqualityDeleteConverter.Result result = converter.convert(Arrays.asList(second.deleteFiles())); + + // the id filter alone covers files of both partitions, the partition pins it to a Review Comment: Done in 3cecc5f9c. The resolver now counts the files the plan returns before the spec and partition check, and the count is renamed to `plannedDataFiles` since not all of those files are read. I also changed both tests so that only the partition filter can prune: the identity test keys on `id` alone, and the bucket test deletes 1,200 keys, so the ids are matched with a range that `bucket(4, id)` can't project. With `partitionFilter` returning `alwaysTrue`, the two tests now see 6 and 12 planned files instead of 3. ########## kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/data/TestEqualityDeleteConverter.java: ########## @@ -0,0 +1,720 @@ +/* + * 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.connect.data; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +import java.io.IOException; +import java.io.UncheckedIOException; +import java.util.Arrays; +import java.util.Collection; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.function.Function; +import java.util.function.LongConsumer; +import org.apache.iceberg.DataFile; +import org.apache.iceberg.DeleteFile; +import org.apache.iceberg.FileFormat; +import org.apache.iceberg.FileMetadata; +import org.apache.iceberg.PartitionKey; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.RowDelta; +import org.apache.iceberg.Schema; +import org.apache.iceberg.StructLike; +import org.apache.iceberg.Table; +import org.apache.iceberg.catalog.Namespace; +import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.iceberg.data.GenericFileWriterFactory; +import org.apache.iceberg.data.GenericRecord; +import org.apache.iceberg.data.IcebergGenerics; +import org.apache.iceberg.data.InternalRecordWrapper; +import org.apache.iceberg.data.Record; +import org.apache.iceberg.deletes.EqualityDeleteWriter; +import org.apache.iceberg.deletes.PositionDeleteIndex; +import org.apache.iceberg.expressions.Expressions; +import org.apache.iceberg.inmemory.InMemoryCatalog; +import org.apache.iceberg.io.CloseableIterable; +import org.apache.iceberg.io.DataWriter; +import org.apache.iceberg.io.FileWriterFactory; +import org.apache.iceberg.io.OutputFileFactory; +import org.apache.iceberg.io.WriteResult; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList; +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.transforms.Transforms; +import org.apache.iceberg.types.TypeUtil; +import org.apache.iceberg.types.Types; +import org.apache.iceberg.util.ContentFileUtil; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +public class TestEqualityDeleteConverter { + + private static final Namespace NAMESPACE = Namespace.of("db"); + private static final TableIdentifier TABLE_IDENTIFIER = TableIdentifier.of(NAMESPACE, "tbl"); + private static final Schema SCHEMA = + new Schema( + ImmutableList.of( + Types.NestedField.required(1, "id", Types.LongType.get()), + Types.NestedField.required(2, "category", Types.StringType.get()), + Types.NestedField.optional(3, "data", Types.StringType.get())), + ImmutableSet.of(1)); + + /** The change of one row, as a CDC writer would produce it. */ + private enum Op { + INSERT, + UPDATE, + DELETE + } + + private InMemoryCatalog catalog; + private List<Integer> equalityFieldIds; + + @BeforeEach + public void before() { + catalog = new InMemoryCatalog(); + catalog.initialize("test_catalog", ImmutableMap.of()); + catalog.createNamespace(NAMESPACE); + equalityFieldIds = ImmutableList.copyOf(SCHEMA.identifierFieldIds()); + } + + @AfterEach + public void after() throws IOException { + catalog.close(); + } + + @Test + public void testConvertsDeletesOfEarlierCommits() throws IOException { + Table table = createTable(PartitionSpec.unpartitioned()); + WriteResult first = + write( + table, + insert(1L, "a", "v1"), + insert(2L, "a", "v1"), + insert(3L, "a", "v1"), + insert(4L, "a", "v1"), + insert(5L, "a", "v1")); + commit(table, first.dataFiles(), first.deleteFiles(), ImmutableList.of()); + DataFile firstDataFile = first.dataFiles()[0]; + long firstSnapshotId = table.currentSnapshot().snapshotId(); + + WriteResult second = write(table, delete(2L, "a"), update(4L, "a", "v2")); + assertThat(second.deleteFiles()).hasSize(1); + + EqualityDeleteConverter converter = new EqualityDeleteConverter(table, null); + EqualityDeleteConverter.Result result = converter.convert(Arrays.asList(second.deleteFiles())); + + assertThat(result.baseSnapshotId()).isEqualTo(firstSnapshotId); + assertThat(result.rewrittenDvFiles()).isEmpty(); + assertThat(result.dvFiles()).hasSize(1); + DeleteFile dv = result.dvFiles().get(0); + assertThat(ContentFileUtil.isDV(dv)).isTrue(); + assertThat(dv.referencedDataFile()).isEqualTo(firstDataFile.location()); + assertThat(dv.recordCount()).isEqualTo(2L); + assertThat(converter.scannedDataFiles()).isEqualTo(1); + + commit(table, second.dataFiles(), result.dvFiles(), result.rewrittenDvFiles()); + assertThat(readRows(table)).containsExactlyInAnyOrder("1|a|v1", "3|a|v1", "4|a|v2", "5|a|v1"); + + // a second conversion for the same data file merges the existing deletion vector + WriteResult third = write(table, delete(1L, "a")); + result = converter.convert(Arrays.asList(third.deleteFiles())); + + assertThat(result.dvFiles()).hasSize(1); + assertThat(result.dvFiles().get(0).referencedDataFile()).isEqualTo(firstDataFile.location()); + assertThat(result.dvFiles().get(0).recordCount()).isEqualTo(3L); + assertThat(result.rewrittenDvFiles()).hasSize(1); + assertThat(result.rewrittenDvFiles().get(0).location()).isEqualTo(dv.location()); + assertThat(result.rewrittenDvFiles().get(0).contentOffset()).isEqualTo(dv.contentOffset()); + // the data file written by the second batch only holds id 4 and is pruned by its column bounds + assertThat(converter.scannedDataFiles()).isEqualTo(1); + + commit(table, third.dataFiles(), result.dvFiles(), result.rewrittenDvFiles()); + assertThat(readRows(table)).containsExactlyInAnyOrder("3|a|v1", "4|a|v2", "5|a|v1"); + } + + @Test + public void testPartitionPruning() throws IOException { + Table table = createTable(PartitionSpec.builderFor(SCHEMA).identity("category").build()); + // the partition column has to be part of the key for the delete to land in its partition + equalityFieldIds = ImmutableList.of(1, 2); + + WriteResult first = + write( + table, + insert(1L, "a", "v1"), + insert(2L, "a", "v1"), + insert(3L, "b", "v1"), + insert(4L, "b", "v1")); + assertThat(first.dataFiles()).hasSize(2); + commit(table, first.dataFiles(), first.deleteFiles(), ImmutableList.of()); + + WriteResult second = write(table, delete(1L, "a")); + EqualityDeleteConverter converter = new EqualityDeleteConverter(table, null); + EqualityDeleteConverter.Result result = converter.convert(Arrays.asList(second.deleteFiles())); + + // only the data file of partition a is a candidate + assertThat(converter.scannedDataFiles()).isEqualTo(1); + assertThat(result.dvFiles()).hasSize(1); + assertThat(result.dvFiles().get(0).recordCount()).isEqualTo(1L); + + commit(table, second.dataFiles(), result.dvFiles(), result.rewrittenDvFiles()); + assertThat(readRows(table)).containsExactlyInAnyOrder("2|a|v1", "3|b|v1", "4|b|v1"); + } + + @Test + public void testPartitionedDeletesSkipDataFilesOfOtherSpecs() throws IOException { + Table table = createTable(PartitionSpec.builderFor(SCHEMA).identity("category").build()); + equalityFieldIds = ImmutableList.of(1, 2); + + WriteResult first = write(table, insert(1L, "a", "v1"), insert(2L, "a", "v1")); + commit(table, first.dataFiles(), first.deleteFiles(), ImmutableList.of()); + + table.updateSpec().addField(Expressions.bucket("id", 4)).commit(); + WriteResult second = write(table, insert(3L, "a", "v1")); + commit(table, second.dataFiles(), second.deleteFiles(), ImmutableList.of()); + long baseSnapshotId = table.currentSnapshot().snapshotId(); + + WriteResult third = write(table, delete(1L, "a"), delete(3L, "a")); + + // the equality deletes committed as they are define the expected rows: a delete stored with + // the new spec does not apply to the data file written with the old spec + table.manageSnapshots().createBranch("expected", baseSnapshotId).commit(); + RowDelta expected = table.newRowDelta().toBranch("expected"); + Arrays.stream(third.deleteFiles()).forEach(expected::addDeletes); + expected.commit(); + Set<String> expectedRows = readRows(table, table.snapshot("expected").snapshotId()); + assertThat(expectedRows).containsExactlyInAnyOrder("1|a|v1", "2|a|v1"); + + EqualityDeleteConverter converter = new EqualityDeleteConverter(table, null); + EqualityDeleteConverter.Result result = converter.convert(Arrays.asList(third.deleteFiles())); + commit(table, third.dataFiles(), result.dvFiles(), result.rewrittenDvFiles()); + + assertThat(readRows(table)).isEqualTo(expectedRows); + } + + @Test + public void testManyKeys() throws IOException { + Table table = createTable(PartitionSpec.unpartitioned()); + + List<Object[]> inserts = Lists.newArrayList(); + for (long id = 1; id <= 300; id++) { + inserts.add(insert(id, "a", "v1")); + } + WriteResult first = write(table, inserts.toArray(new Object[0][])); + commit(table, first.dataFiles(), first.deleteFiles(), ImmutableList.of()); + + List<Object[]> deletes = Lists.newArrayList(); + for (long id = 1; id <= 250; id++) { + deletes.add(delete(id, "a")); + } + WriteResult second = write(table, deletes.toArray(new Object[0][])); + + EqualityDeleteConverter converter = new EqualityDeleteConverter(table, null); + EqualityDeleteConverter.Result result = converter.convert(Arrays.asList(second.deleteFiles())); + + assertThat(result.dvFiles()).hasSize(1); + assertThat(result.dvFiles().get(0).recordCount()).isEqualTo(250L); + + commit(table, second.dataFiles(), result.dvFiles(), result.rewrittenDvFiles()); + Set<String> rows = readRows(table); + assertThat(rows).hasSize(50); + assertThat(rows).contains("251|a|v1", "300|a|v1"); + assertThat(rows).doesNotContain("1|a|v1", "250|a|v1"); + } + + @Test + public void testClusteredKeysPruneToTheirFiles() throws IOException { + Table table = createTable(PartitionSpec.unpartitioned()); + + // 200 data files with 50 consecutive ids each + int fileCount = 200; + int rowsPerFile = 50; + List<DataFile> dataFiles = Lists.newArrayList(); + for (int file = 0; file < fileCount; file++) { + List<Object[]> inserts = Lists.newArrayList(); + for (int row = 0; row < rowsPerFile; row++) { + inserts.add(insert((long) file * rowsPerFile + row, "a", "v1")); + } + WriteResult result = write(table, inserts.toArray(new Object[0][])); + assertThat(result.dataFiles()).hasSize(1); + dataFiles.add(result.dataFiles()[0]); + } + commit(table, dataFiles.toArray(new DataFile[0]), new DeleteFile[0], ImmutableList.of()); + + // 295 hot keys in the last six files and five keys scattered over the table + List<Object[]> deletes = Lists.newArrayList(); + for (long id = 9700; id < 9995; id++) { + deletes.add(delete(id, "a")); + } + for (long id : new long[] {5, 1005, 3005, 6005, 8005}) { + deletes.add(delete(id, "a")); + } + WriteResult second = write(table, deletes.toArray(new Object[0][])); + + EqualityDeleteConverter converter = new EqualityDeleteConverter(table, null); + EqualityDeleteConverter.Result result = converter.convert(Arrays.asList(second.deleteFiles())); + + // more keys than one IN list holds, so the keys are matched with several lists: the hot Review Comment: Added in b0e5bc8cb: 30 files of consecutive ids and 1,200 deleted keys in two clusters at both ends, checking the planned files and the exact surviving rows. Since 286bc46e5 replaced the ranges with bounds, the same test (now `testKeysAboveTheListLimitAreMatchedWithBounds`) goes through the bounds path: all 30 files are planned, and the row check also verifies that the rows inside the bounds that weren't deleted survive. ########## kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/data/TestEqualityDeleteConverter.java: ########## @@ -0,0 +1,720 @@ +/* + * 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.connect.data; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +import java.io.IOException; +import java.io.UncheckedIOException; +import java.util.Arrays; +import java.util.Collection; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.function.Function; +import java.util.function.LongConsumer; +import org.apache.iceberg.DataFile; +import org.apache.iceberg.DeleteFile; +import org.apache.iceberg.FileFormat; +import org.apache.iceberg.FileMetadata; +import org.apache.iceberg.PartitionKey; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.RowDelta; +import org.apache.iceberg.Schema; +import org.apache.iceberg.StructLike; +import org.apache.iceberg.Table; +import org.apache.iceberg.catalog.Namespace; +import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.iceberg.data.GenericFileWriterFactory; +import org.apache.iceberg.data.GenericRecord; +import org.apache.iceberg.data.IcebergGenerics; +import org.apache.iceberg.data.InternalRecordWrapper; +import org.apache.iceberg.data.Record; +import org.apache.iceberg.deletes.EqualityDeleteWriter; +import org.apache.iceberg.deletes.PositionDeleteIndex; +import org.apache.iceberg.expressions.Expressions; +import org.apache.iceberg.inmemory.InMemoryCatalog; +import org.apache.iceberg.io.CloseableIterable; +import org.apache.iceberg.io.DataWriter; +import org.apache.iceberg.io.FileWriterFactory; +import org.apache.iceberg.io.OutputFileFactory; +import org.apache.iceberg.io.WriteResult; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList; +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.transforms.Transforms; +import org.apache.iceberg.types.TypeUtil; +import org.apache.iceberg.types.Types; +import org.apache.iceberg.util.ContentFileUtil; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +public class TestEqualityDeleteConverter { + + private static final Namespace NAMESPACE = Namespace.of("db"); + private static final TableIdentifier TABLE_IDENTIFIER = TableIdentifier.of(NAMESPACE, "tbl"); + private static final Schema SCHEMA = + new Schema( + ImmutableList.of( + Types.NestedField.required(1, "id", Types.LongType.get()), + Types.NestedField.required(2, "category", Types.StringType.get()), + Types.NestedField.optional(3, "data", Types.StringType.get())), + ImmutableSet.of(1)); + + /** The change of one row, as a CDC writer would produce it. */ + private enum Op { + INSERT, + UPDATE, + DELETE + } + + private InMemoryCatalog catalog; + private List<Integer> equalityFieldIds; + + @BeforeEach + public void before() { + catalog = new InMemoryCatalog(); + catalog.initialize("test_catalog", ImmutableMap.of()); + catalog.createNamespace(NAMESPACE); + equalityFieldIds = ImmutableList.copyOf(SCHEMA.identifierFieldIds()); + } + + @AfterEach + public void after() throws IOException { + catalog.close(); + } + + @Test + public void testConvertsDeletesOfEarlierCommits() throws IOException { + Table table = createTable(PartitionSpec.unpartitioned()); + WriteResult first = + write( + table, + insert(1L, "a", "v1"), + insert(2L, "a", "v1"), + insert(3L, "a", "v1"), + insert(4L, "a", "v1"), + insert(5L, "a", "v1")); + commit(table, first.dataFiles(), first.deleteFiles(), ImmutableList.of()); + DataFile firstDataFile = first.dataFiles()[0]; + long firstSnapshotId = table.currentSnapshot().snapshotId(); + + WriteResult second = write(table, delete(2L, "a"), update(4L, "a", "v2")); + assertThat(second.deleteFiles()).hasSize(1); + + EqualityDeleteConverter converter = new EqualityDeleteConverter(table, null); + EqualityDeleteConverter.Result result = converter.convert(Arrays.asList(second.deleteFiles())); + + assertThat(result.baseSnapshotId()).isEqualTo(firstSnapshotId); + assertThat(result.rewrittenDvFiles()).isEmpty(); + assertThat(result.dvFiles()).hasSize(1); + DeleteFile dv = result.dvFiles().get(0); + assertThat(ContentFileUtil.isDV(dv)).isTrue(); + assertThat(dv.referencedDataFile()).isEqualTo(firstDataFile.location()); + assertThat(dv.recordCount()).isEqualTo(2L); + assertThat(converter.scannedDataFiles()).isEqualTo(1); + + commit(table, second.dataFiles(), result.dvFiles(), result.rewrittenDvFiles()); + assertThat(readRows(table)).containsExactlyInAnyOrder("1|a|v1", "3|a|v1", "4|a|v2", "5|a|v1"); + + // a second conversion for the same data file merges the existing deletion vector + WriteResult third = write(table, delete(1L, "a")); + result = converter.convert(Arrays.asList(third.deleteFiles())); + + assertThat(result.dvFiles()).hasSize(1); + assertThat(result.dvFiles().get(0).referencedDataFile()).isEqualTo(firstDataFile.location()); + assertThat(result.dvFiles().get(0).recordCount()).isEqualTo(3L); + assertThat(result.rewrittenDvFiles()).hasSize(1); + assertThat(result.rewrittenDvFiles().get(0).location()).isEqualTo(dv.location()); + assertThat(result.rewrittenDvFiles().get(0).contentOffset()).isEqualTo(dv.contentOffset()); + // the data file written by the second batch only holds id 4 and is pruned by its column bounds + assertThat(converter.scannedDataFiles()).isEqualTo(1); + + commit(table, third.dataFiles(), result.dvFiles(), result.rewrittenDvFiles()); + assertThat(readRows(table)).containsExactlyInAnyOrder("3|a|v1", "4|a|v2", "5|a|v1"); + } + + @Test + public void testPartitionPruning() throws IOException { + Table table = createTable(PartitionSpec.builderFor(SCHEMA).identity("category").build()); + // the partition column has to be part of the key for the delete to land in its partition + equalityFieldIds = ImmutableList.of(1, 2); + + WriteResult first = + write( + table, + insert(1L, "a", "v1"), + insert(2L, "a", "v1"), + insert(3L, "b", "v1"), + insert(4L, "b", "v1")); + assertThat(first.dataFiles()).hasSize(2); + commit(table, first.dataFiles(), first.deleteFiles(), ImmutableList.of()); + + WriteResult second = write(table, delete(1L, "a")); + EqualityDeleteConverter converter = new EqualityDeleteConverter(table, null); + EqualityDeleteConverter.Result result = converter.convert(Arrays.asList(second.deleteFiles())); + + // only the data file of partition a is a candidate + assertThat(converter.scannedDataFiles()).isEqualTo(1); + assertThat(result.dvFiles()).hasSize(1); + assertThat(result.dvFiles().get(0).recordCount()).isEqualTo(1L); + + commit(table, second.dataFiles(), result.dvFiles(), result.rewrittenDvFiles()); + assertThat(readRows(table)).containsExactlyInAnyOrder("2|a|v1", "3|b|v1", "4|b|v1"); + } + + @Test + public void testPartitionedDeletesSkipDataFilesOfOtherSpecs() throws IOException { + Table table = createTable(PartitionSpec.builderFor(SCHEMA).identity("category").build()); + equalityFieldIds = ImmutableList.of(1, 2); + + WriteResult first = write(table, insert(1L, "a", "v1"), insert(2L, "a", "v1")); + commit(table, first.dataFiles(), first.deleteFiles(), ImmutableList.of()); + + table.updateSpec().addField(Expressions.bucket("id", 4)).commit(); + WriteResult second = write(table, insert(3L, "a", "v1")); + commit(table, second.dataFiles(), second.deleteFiles(), ImmutableList.of()); + long baseSnapshotId = table.currentSnapshot().snapshotId(); + + WriteResult third = write(table, delete(1L, "a"), delete(3L, "a")); + + // the equality deletes committed as they are define the expected rows: a delete stored with + // the new spec does not apply to the data file written with the old spec + table.manageSnapshots().createBranch("expected", baseSnapshotId).commit(); + RowDelta expected = table.newRowDelta().toBranch("expected"); + Arrays.stream(third.deleteFiles()).forEach(expected::addDeletes); + expected.commit(); + Set<String> expectedRows = readRows(table, table.snapshot("expected").snapshotId()); + assertThat(expectedRows).containsExactlyInAnyOrder("1|a|v1", "2|a|v1"); + + EqualityDeleteConverter converter = new EqualityDeleteConverter(table, null); + EqualityDeleteConverter.Result result = converter.convert(Arrays.asList(third.deleteFiles())); + commit(table, third.dataFiles(), result.dvFiles(), result.rewrittenDvFiles()); + + assertThat(readRows(table)).isEqualTo(expectedRows); + } + + @Test + public void testManyKeys() throws IOException { + Table table = createTable(PartitionSpec.unpartitioned()); + + List<Object[]> inserts = Lists.newArrayList(); + for (long id = 1; id <= 300; id++) { + inserts.add(insert(id, "a", "v1")); + } + WriteResult first = write(table, inserts.toArray(new Object[0][])); + commit(table, first.dataFiles(), first.deleteFiles(), ImmutableList.of()); + + List<Object[]> deletes = Lists.newArrayList(); + for (long id = 1; id <= 250; id++) { + deletes.add(delete(id, "a")); + } + WriteResult second = write(table, deletes.toArray(new Object[0][])); + + EqualityDeleteConverter converter = new EqualityDeleteConverter(table, null); + EqualityDeleteConverter.Result result = converter.convert(Arrays.asList(second.deleteFiles())); + + assertThat(result.dvFiles()).hasSize(1); + assertThat(result.dvFiles().get(0).recordCount()).isEqualTo(250L); + + commit(table, second.dataFiles(), result.dvFiles(), result.rewrittenDvFiles()); + Set<String> rows = readRows(table); + assertThat(rows).hasSize(50); + assertThat(rows).contains("251|a|v1", "300|a|v1"); + assertThat(rows).doesNotContain("1|a|v1", "250|a|v1"); + } + + @Test + public void testClusteredKeysPruneToTheirFiles() throws IOException { + Table table = createTable(PartitionSpec.unpartitioned()); + + // 200 data files with 50 consecutive ids each + int fileCount = 200; + int rowsPerFile = 50; + List<DataFile> dataFiles = Lists.newArrayList(); + for (int file = 0; file < fileCount; file++) { + List<Object[]> inserts = Lists.newArrayList(); + for (int row = 0; row < rowsPerFile; row++) { + inserts.add(insert((long) file * rowsPerFile + row, "a", "v1")); + } + WriteResult result = write(table, inserts.toArray(new Object[0][])); + assertThat(result.dataFiles()).hasSize(1); + dataFiles.add(result.dataFiles()[0]); + } + commit(table, dataFiles.toArray(new DataFile[0]), new DeleteFile[0], ImmutableList.of()); + + // 295 hot keys in the last six files and five keys scattered over the table + List<Object[]> deletes = Lists.newArrayList(); + for (long id = 9700; id < 9995; id++) { + deletes.add(delete(id, "a")); + } + for (long id : new long[] {5, 1005, 3005, 6005, 8005}) { + deletes.add(delete(id, "a")); + } + WriteResult second = write(table, deletes.toArray(new Object[0][])); + + EqualityDeleteConverter converter = new EqualityDeleteConverter(table, null); + EqualityDeleteConverter.Result result = converter.convert(Arrays.asList(second.deleteFiles())); + + // more keys than one IN list holds, so the keys are matched with several lists: the hot + // range covers six files and every scattered key stays in its own file + assertThat(deletes).hasSizeGreaterThan(KeyFilters.IN_PREDICATE_LIMIT); + assertThat(converter.scannedDataFiles()).isEqualTo(11); + assertThat(result.dvFiles()).hasSize(11); + assertThat(result.dvFiles().stream().mapToLong(DeleteFile::recordCount).sum()).isEqualTo(300L); + + commit(table, second.dataFiles(), result.dvFiles(), result.rewrittenDvFiles()); + assertThat(readRows(table)).hasSize(fileCount * rowsPerFile - 300); + } + + @Test + public void testPartitionIsPinnedWithManyKeys() throws IOException { + Table table = createTable(PartitionSpec.builderFor(SCHEMA).identity("category").build()); + // the partition column has to be part of the key for the delete to land in its partition + equalityFieldIds = ImmutableList.of(1, 2); + + // three files of 100 ids in each of the partitions a and b, ids overlap across partitions + List<DataFile> dataFiles = Lists.newArrayList(); + for (String category : new String[] {"a", "b"}) { + for (int file = 0; file < 3; file++) { + List<Object[]> inserts = Lists.newArrayList(); + for (int row = 0; row < 100; row++) { + inserts.add(insert((long) file * 100 + row, category, "v1")); + } + WriteResult result = write(table, inserts.toArray(new Object[0][])); + dataFiles.add(result.dataFiles()[0]); + } + } + commit(table, dataFiles.toArray(new DataFile[0]), new DeleteFile[0], ImmutableList.of()); + + List<Object[]> deletes = Lists.newArrayList(); + for (long id = 0; id < 250; id++) { + deletes.add(delete(id, "a")); + } + WriteResult second = write(table, deletes.toArray(new Object[0][])); + + EqualityDeleteConverter converter = new EqualityDeleteConverter(table, null); + EqualityDeleteConverter.Result result = converter.convert(Arrays.asList(second.deleteFiles())); + + // the id filter alone covers files of both partitions, the partition pins it to a + assertThat(converter.scannedDataFiles()).isEqualTo(3); + assertThat(result.dvFiles()).hasSize(3); + + commit(table, second.dataFiles(), result.dvFiles(), result.rewrittenDvFiles()); + Set<String> rows = readRows(table); + assertThat(rows).hasSize(350); + assertThat(rows).contains("0|b|v1", "250|a|v1").doesNotContain("0|a|v1", "249|a|v1"); + } + + @Test + public void testTransformPartitionIsPinnedWithManyKeys() throws IOException { + Table table = createTable(PartitionSpec.builderFor(SCHEMA).bucket("id", 4).build()); + + // three batches, each writes one file per bucket + List<DataFile> dataFiles = Lists.newArrayList(); + for (int batch = 0; batch < 3; batch++) { + List<Object[]> inserts = Lists.newArrayList(); + for (long id = batch * 1000L; id < (batch + 1) * 1000L; id++) { + inserts.add(insert(id, "a", "v1")); + } + WriteResult result = write(table, inserts.toArray(new Object[0][])); + assertThat(result.dataFiles()).hasSize(4); + dataFiles.addAll(Arrays.asList(result.dataFiles())); + } + commit(table, dataFiles.toArray(new DataFile[0]), new DeleteFile[0], ImmutableList.of()); + + // more keys than the IN limit, all of them in bucket 0 and spread over all three batches + Function<Object, Integer> bucket = Transforms.bucket(4).bind(Types.LongType.get()); + List<Long> bucketZeroIds = Lists.newArrayList(); + for (long id = 0; id < 3000; id++) { + if (bucket.apply(id) == 0) { + bucketZeroIds.add(id); + } + } + List<Object[]> deletes = Lists.newArrayList(); + for (int i = 0; i < 250; i++) { + deletes.add(delete(bucketZeroIds.get(i * bucketZeroIds.size() / 250), "a")); + } + WriteResult second = write(table, deletes.toArray(new Object[0][])); + assertThat(second.deleteFiles()).hasSize(1); + + EqualityDeleteConverter converter = new EqualityDeleteConverter(table, null); + EqualityDeleteConverter.Result result = converter.convert(Arrays.asList(second.deleteFiles())); + + // the pinned bucket keeps only the three files of bucket 0 + assertThat(converter.scannedDataFiles()).isEqualTo(3); + assertThat(result.dvFiles()).hasSize(3); + assertThat(result.dvFiles().stream().mapToLong(DeleteFile::recordCount).sum()).isEqualTo(250L); + + commit(table, second.dataFiles(), result.dvFiles(), result.rewrittenDvFiles()); + assertThat(readRows(table)).hasSize(3000 - 250); + } + + @Test + public void testScatteredKeysAreMatchedWithLists() throws IOException { + Table table = createTable(PartitionSpec.unpartitioned()); + + // 200 data files with 50 consecutive ids each + int fileCount = 200; + int rowsPerFile = 50; + List<DataFile> dataFiles = Lists.newArrayList(); + for (int file = 0; file < fileCount; file++) { + List<Object[]> inserts = Lists.newArrayList(); + for (int row = 0; row < rowsPerFile; row++) { + inserts.add(insert((long) file * rowsPerFile + row, "a", "v1")); + } + dataFiles.add(write(table, inserts.toArray(new Object[0][])).dataFiles()[0]); + } + commit(table, dataFiles.toArray(new DataFile[0]), new DeleteFile[0], ImmutableList.of()); + + // 300 hot keys in the last six files plus one key in each of the first 150 files: more + // scattered keys than value ranges could isolate, so the keys are matched with IN lists + List<Object[]> deletes = Lists.newArrayList(); + for (long id = 9700; id < 10000; id++) { + deletes.add(delete(id, "a")); + } + for (int file = 0; file < 150; file++) { + deletes.add(delete((long) file * rowsPerFile + 7, "a")); + } + WriteResult second = write(table, deletes.toArray(new Object[0][])); + + EqualityDeleteConverter converter = new EqualityDeleteConverter(table, null); + EqualityDeleteConverter.Result result = converter.convert(Arrays.asList(second.deleteFiles())); + + // exactly the files that hold a deleted key are read: 6 hot files and 150 scattered files + assertThat(converter.scannedDataFiles()).isEqualTo(156); + assertThat(result.dvFiles()).hasSize(156); + assertThat(result.dvFiles().stream().mapToLong(DeleteFile::recordCount).sum()).isEqualTo(450L); + + commit(table, second.dataFiles(), result.dvFiles(), result.rewrittenDvFiles()); + assertThat(readRows(table)).hasSize(fileCount * rowsPerFile - 450); + } + + @Test + public void testCustomResolver() throws IOException { + Table table = createTable(PartitionSpec.unpartitioned()); + WriteResult first = write(table, insert(1L, "a", "v1"), insert(2L, "a", "v1")); + commit(table, first.dataFiles(), first.deleteFiles(), ImmutableList.of()); + DataFile dataFile = first.dataFiles()[0]; + + WriteResult second = write(table, delete(2L, "a")); + + // a resolver that knows the position of the row without reading the data file + KeyPositionResolver resolver = + (base, keySchema, keys, deleteSpec, deletePartition) -> { + assertThat(keys).hasSize(1); + KeyPositionResolver.FileMatches matches = + new KeyPositionResolver.FileMatches() { + @Override + public String path() { + return dataFile.location(); + } + + @Override + public PartitionSpec spec() { + return table.spec(); + } + + @Override + public StructLike partition() { + return dataFile.partition(); + } + + @Override + public PositionDeleteIndex existingDeletes() { + return null; + } + + @Override + public void forEachPosition(LongConsumer consumer) { + consumer.accept(1L); + } + }; + return new KeyPositionResolver.Resolution() { + @Override + public Collection<KeyPositionResolver.FileMatches> matches() { + return ImmutableList.of(matches); + } + + @Override + public int scannedDataFiles() { + return 0; + } + }; + }; + + EqualityDeleteConverter converter = new EqualityDeleteConverter(table, null, resolver); + EqualityDeleteConverter.Result result = converter.convert(Arrays.asList(second.deleteFiles())); + + assertThat(converter.scannedDataFiles()).isEqualTo(0); + assertThat(result.dvFiles()).hasSize(1); + assertThat(result.dvFiles().get(0).referencedDataFile()).isEqualTo(dataFile.location()); + assertThat(result.dvFiles().get(0).recordCount()).isEqualTo(1L); + + commit(table, second.dataFiles(), result.dvFiles(), result.rewrittenDvFiles()); + assertThat(readRows(table)).containsExactly("1|a|v1"); + } + + @Test + public void testSplitIntoRanges() { Review Comment: Moved to `TestKeyFilters` as one test per case in 8ec2676a3. 286bc46e5 then removes `splitIntoRanges` itself, so these tests are not in the final diff. ########## kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/data/KeyFilters.java: ########## @@ -0,0 +1,342 @@ +/* + * 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.connect.data; + +import java.math.BigDecimal; +import java.util.Comparator; +import java.util.List; +import java.util.Set; +import java.util.function.ToDoubleFunction; +import org.apache.iceberg.PartitionField; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.Schema; +import org.apache.iceberg.Snapshot; +import org.apache.iceberg.SnapshotSummary; +import org.apache.iceberg.StructLike; +import org.apache.iceberg.expressions.Expression; +import org.apache.iceberg.expressions.Expressions; +import org.apache.iceberg.expressions.UnboundTerm; +import org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting; +import org.apache.iceberg.relocated.com.google.common.collect.Lists; +import org.apache.iceberg.relocated.com.google.common.collect.Sets; +import org.apache.iceberg.transforms.Transform; +import org.apache.iceberg.transforms.UnknownTransform; +import org.apache.iceberg.types.Comparators; +import org.apache.iceberg.types.Type; +import org.apache.iceberg.types.Types; +import org.apache.iceberg.util.StructLikeSet; + +/** + * Row filters built from the keys of equality delete files. + * + * <p>The filters are inclusive: every row with a deleted key satisfies them, and rows without a + * deleted key may satisfy them too. {@link ScanKeyPositionResolver} uses them to prune the data + * files it has to read; {@link EqualityDeleteConverter} uses them to detect files that another + * writer added concurrently for the deleted keys. + */ +final class KeyFilters { + + // metrics evaluators stop pruning with an IN list above this many values, so keys are matched + // with several lists of at most this size + @VisibleForTesting static final int IN_PREDICATE_LIMIT = 200; + + // The filters are evaluated once per data file entry during planning, so their size is chosen + // from the table's data file count: a budget of comparisons for the whole plan, divided by the + // number of data files. The budget is what a fixed cap of 100 value ranges (200 comparisons per + // file) would spend on a table with one million data files, so such a table keeps that cost + // and smaller tables get a finer filter. The lower bound is that same cap and applies when the + // count is missing; the upper bound limits what the filter costs when every file is read + // anyway, e.g. keys spread over a small table, where a finer filter only adds evaluation time. + @VisibleForTesting static final long COMPARISON_BUDGET = 200_000_000L; Review Comment: Agreed, done in 286bc46e5. Up to 1,000 keys are matched with IN lists of at most 200 values, and more keys with the min and max of the leading key column; the row-level match keeps the result exact either way. This removes the budget constants, the range splitting and their tests, which is most of the deletions in that commit. I'll propose adaptive sizing again, with a benchmark, once the coordinator calls the converter. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
