laskoviymishka commented on code in PR #18003: URL: https://github.com/apache/iceberg/pull/18003#discussion_r4345793639
########## 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: When the branch has no snapshot this drops the keys and returns `alwaysFalse` as the conflict filter, so part 2's conflict validation passes no matter what was appended concurrently. If another writer makes the first commit to the branch before ours, its rows sit at a lower sequence number and the equality delete would have removed them; here they survive. Could we still build the key filter on this path, and document what a null `baseSnapshotId` means for the commit? ########## kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/data/KeyPositionResolver.java: ########## @@ -0,0 +1,87 @@ +/* + * 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.util.Collection; +import java.util.function.LongConsumer; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.Schema; +import org.apache.iceberg.Snapshot; +import org.apache.iceberg.StructLike; +import org.apache.iceberg.deletes.PositionDeleteIndex; +import org.apache.iceberg.util.StructLikeSet; + +/** + * Finds the rows of a table snapshot whose identifier values are in a set of deleted keys. + * + * <p>{@link EqualityDeleteConverter} asks a resolver for the position of every row an equality + * delete would remove and writes those positions as deletion vectors. Implementations differ in how + * they locate the rows. The default {@link ScanKeyPositionResolver} reads the data files that the + * column statistics cannot rule out; an implementation backed by an index can look the keys up + * without touching data files. + */ +public interface KeyPositionResolver { Review Comment: Does this need to be public before part 2 has a caller? The only constructor that takes a resolver is package-private, so the "can be plugged in" Javadoc describes something code outside the package can't do yet. ########## 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: The budget clamps to 1000 for these table sizes, and every converter test stays under it (250, 300 and 450 keys), so the range path only runs against metrics in `TestKeyFilters`, never through planning and reading. Could we add one converter test that forces ranges and checks the surviving rows exactly? ########## kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/data/ScanKeyPositionResolver.java: ########## @@ -0,0 +1,319 @@ +/* + * 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.util.Collection; +import java.util.List; +import java.util.Map; +import java.util.function.LongConsumer; +import org.apache.iceberg.Accessor; +import org.apache.iceberg.DataFile; +import org.apache.iceberg.DeleteFile; +import org.apache.iceberg.FileContent; +import org.apache.iceberg.FileScanTask; +import org.apache.iceberg.MetadataColumns; +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.data.InternalRecordWrapper; +import org.apache.iceberg.data.Record; +import org.apache.iceberg.deletes.PositionDeleteIndex; +import org.apache.iceberg.expressions.Expression; +import org.apache.iceberg.expressions.Expressions; +import org.apache.iceberg.formats.FormatModelRegistry; +import org.apache.iceberg.formats.ReadBuilder; +import org.apache.iceberg.io.CloseableIterable; +import org.apache.iceberg.io.InputFile; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList; +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.Types; +import org.apache.iceberg.util.ContentFileUtil; +import org.apache.iceberg.util.StructLikeSet; +import org.apache.iceberg.util.StructLikeWrapper; +import org.apache.iceberg.util.StructProjection; +import org.apache.iceberg.util.Tasks; +import org.apache.iceberg.util.ThreadPools; + +/** + * Resolves deleted keys by scanning data files. No state is kept between calls. + * + * <p>The candidate files are planned with the filters of {@link KeyFilters}: the partition of the + * equality delete files is pinned through its transform and the keys are matched with {@code IN} + * lists while they fit the comparison budget the table's data file count allows, with value ranges + * split at the largest gaps of the leading key column above that. Only the key columns and the row + * position of the candidate files are read, in parallel, and a row is matched by comparing its key + * with the deleted set. A cluster of hot keys and a few scattered keys therefore cost the files + * that hold them; keys spread evenly over a large table still match most files of the partition. + * + * <p>The filters only prune. A candidate is resolved only if the deletes apply to it under the scan + * planning rules: an unpartitioned delete applies to every data file, a partitioned delete only to + * data files with the same spec ID and partition values. + * + * <p>Data files that carry position delete files instead of deletion vectors are rejected, because + * a deletion vector replaces all position deletes of its data file. + */ +class ScanKeyPositionResolver implements KeyPositionResolver { + + private final Table table; + private final DeleteLoader deleteLoader; + + ScanKeyPositionResolver(Table table) { + this.table = table; + this.deleteLoader = new BaseDeleteLoader(deleteFile -> table.io().newInputFile(deleteFile)); + } + + @Override + public Resolution resolve( + Snapshot base, + Schema keySchema, + StructLikeSet keys, + PartitionSpec deleteSpec, + StructLike deletePartition) + throws IOException { + // the plan filter carries the partition pin and the key filter; only the key filter is passed + // to the file readers, whose row group filters evaluate column references, + // not partition transforms + Expression keyFilter = + KeyFilters.keyFilter(keySchema, keys, KeyFilters.comparisonsPerFile(base)); + Expression planFilter = + Expressions.and( + KeyFilters.partitionFilter(table.schema(), deleteSpec, deletePartition), keyFilter); + + // the partition filter only prunes; it also keeps files of other specs whose rows satisfy the + // transforms, so the spec and partition of every candidate are checked as well + PartitionMatcher partitionMatcher = new PartitionMatcher(deleteSpec, deletePartition); + List<FileScanTask> tasks = Lists.newArrayList(); + try (CloseableIterable<FileScanTask> planned = + table + .newScan() + .useSnapshot(base.snapshotId()) + .filter(planFilter) + .ignoreResiduals() + .planFiles()) { + for (FileScanTask task : planned) { + if (partitionMatcher.matches(task.spec(), task.file().partition())) { + tasks.add(task); + } + } + } + + Map<String, FileMatches> matches = Maps.newConcurrentMap(); + Tasks.foreach(tasks) + .executeWith(ThreadPools.getWorkerPool()) + .stopOnFailure() + .throwFailureWhenFinished() + .run(task -> match(task, keySchema, keys, keyFilter, matches), IOException.class); + + return new ScanResolution(matches.values(), tasks.size()); + } + + /** Reads the key columns and row positions of one data file and records the matching rows. */ + private void match( + FileScanTask task, + Schema keySchema, + StructLikeSet keys, + Expression keyFilter, + Map<String, FileMatches> matches) + throws IOException { + DataFile file = task.file(); + + List<DeleteFile> dvs = Lists.newArrayList(); + for (DeleteFile delete : task.deletes()) { + if (ContentFileUtil.isDV(delete)) { + dvs.add(delete); + } else if (delete.content() == FileContent.POSITION_DELETES) { + throw new IllegalStateException( Review Comment: The description says these are rejected because a data file can have only one DV, but as far as I can tell `BaseDVFileWriter` already folds file-scoped position deletes into the new DV and reports them as rewritten. As written this throws for any candidate file, before a key has matched, so one legacy delete file blocks every commit touching that partition. Could we reject only partition-scoped ones, and only on files that actually match? ########## 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) { Review Comment: The class doc says this requires format version 3, but nothing checks it. On a v2 table we'd scan and write the Puffin files, and only `addDeletes` at commit time would reject them, leaving the files orphaned. A `Preconditions.checkArgument(TableUtil.formatVersion(table) >= 3, ...)` at the top, with a test, would fail before any of that work. ########## 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: `scannedDataFiles` is counted after `PartitionMatcher`, so this test and the bucket one would still pass if `partitionFilter` returned `alwaysTrue`. Here the key filter already includes `category IN ('a')`, and for the bucket case `id IN (...)` projects through `bucket(4, id)` on its own. Counting planned files before the matcher would let these tests see the pin. ########## 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(); + } + + Map<DeleteGroup, List<DeleteFile>> groups = groupDeleteFiles(eqDeleteFiles); + + OutputFileFactory fileFactory = + OutputFileFactory.builderFor(table, 1, System.currentTimeMillis()) + .defaultSpec(table.spec()) + .operationId(UUID.randomUUID().toString()) + .format(FileFormat.PUFFIN) + .build(); + // deletion vectors already attached to a matched data file, folded into the new vector + Map<String, PositionDeleteIndex> previousDeletes = Maps.newHashMap(); + BaseDVFileWriter dvWriter = new BaseDVFileWriter(fileFactory, previousDeletes::get); + + Expression conflictFilter = Expressions.alwaysFalse(); + int comparisonsPerFile = KeyFilters.comparisonsPerFile(base); + long matchedRows = 0L; + long deletedKeys = 0L; + scannedDataFiles.set(0); + + long start = System.currentTimeMillis(); + try { + for (Map.Entry<DeleteGroup, List<DeleteFile>> entry : groups.entrySet()) { + DeleteGroup group = entry.getKey(); + Schema keySchema = keySchema(group.equalityFieldIds()); + StructLikeSet keys = deleteLoader.loadEqualityDeletes(entry.getValue(), keySchema); + if (keys.isEmpty()) { + continue; + } + + deletedKeys += keys.size(); + conflictFilter = + Expressions.or( + conflictFilter, + Expressions.and( + KeyFilters.partitionFilter(table.schema(), group.spec(), group.partition()), + KeyFilters.keyFilter(keySchema, keys, comparisonsPerFile))); + + KeyPositionResolver.Resolution resolution = + resolver.resolve(base, keySchema, keys, group.spec(), group.partition()); + scannedDataFiles.addAndGet(resolution.scannedDataFiles()); + + for (KeyPositionResolver.FileMatches matches : resolution.matches()) { + if (matches.existingDeletes() != null) { Review Comment: This only merges the old DV when the resolver returns `existingDeletes`. If a resolver returns null for a file that has a DV, the old one is neither merged nor removed, so the file ends up with two DVs and reads fail with "Can't index multiple DVs". The scan resolver handles this, but an index-backed one would have to know. Could `FileMatches.existingDeletes()` state that null means no DV exists, or the converter look the DVs up itself? ########## 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: For part 1, would a capped IN list plus the exact matcher be enough, with the adaptive budget following once part 2 gives it a caller and there's a benchmark in the repo? The constants are hard to review without numbers we can rerun. wdyt? ########## 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: This tests `KeyFilters.splitIntoRanges` directly, so it fits better in `TestKeyFilters`, as one test per case. ########## 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(); Review Comment: This is reset on each `convert()`, so two calls on one instance overwrite each other's count and the log line can print the wrong number. Returning the count on `Result` would drop the field and the `@VisibleForTesting` accessor. -- 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]
