laskoviymishka commented on code in PR #18003: URL: https://github.com/apache/iceberg/pull/18003#discussion_r4371597433
########## kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/data/EqualityDeleteConverter.java: ########## @@ -0,0 +1,357 @@ +/* + * 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 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.TableUtil; +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.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 finds those rows by reading the data files + * that their partition values and column statistics cannot rule out. 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; + + /** + * 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 + * @throws IllegalArgumentException if the table's format version is lower than 3 + */ + public EqualityDeleteConverter(Table table, String branch) { + this(table, branch, new ScanKeyPositionResolver(table)); + } + + EqualityDeleteConverter(Table table, String branch, KeyPositionResolver resolver) { + int formatVersion = TableUtil.formatVersion(table); + Preconditions.checkArgument( + formatVersion >= 3, + "Cannot convert equality deletes of table %s: deletion vectors require format version 3, " + + "found %s", + table.name(), + formatVersion); + 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. A branch that does not exist yet is resolved against the current snapshot of main, + * which a commit creating the branch builds on. The table is expected to be refreshed by the + * caller. + */ + public Result convert(List<DeleteFile> eqDeleteFiles) { + Snapshot base = + branch == null || !table.refs().containsKey(branch) + ? table.currentSnapshot() + : table.snapshot(branch); + long start = System.currentTimeMillis(); + + // the conflict filter covers the keys even when the branch has no snapshot, so that the commit + // detects rows another writer commits to the branch first + Map<DeleteGroup, StructLikeSet> keysByGroup = Maps.newLinkedHashMap(); + Expression conflictFilter = Expressions.alwaysFalse(); + long deletedKeys = 0L; + for (Map.Entry<DeleteGroup, List<DeleteFile>> entry : + groupDeleteFiles(eqDeleteFiles).entrySet()) { + DeleteGroup group = entry.getKey(); + Schema keySchema = keySchema(group.equalityFieldIds()); + StructLikeSet keys = deleteLoader.loadEqualityDeletes(entry.getValue(), keySchema); + if (keys.isEmpty()) { + continue; + } + + keysByGroup.put(group, keys); + deletedKeys += keys.size(); + conflictFilter = + Expressions.or( + conflictFilter, + Expressions.and( + KeyFilters.partitionFilter(table.schema(), group.spec(), group.partition()), + KeyFilters.keyFilter(keySchema, keys))); + } + + if (base == null) { + LOG.info( + "Table {} has no snapshot on branch {}, {} equality delete file(s) with {} key(s) " + + "match no rows", + table.name(), + branch == null ? "main" : branch, + eqDeleteFiles.size(), + deletedKeys); + return new Result(null, ImmutableList.of(), ImmutableList.of(), conflictFilter, 0); + } + + OutputFileFactory fileFactory = + OutputFileFactory.builderFor(table, 1, System.currentTimeMillis()) + .defaultSpec(table.spec()) + .operationId(UUID.randomUUID().toString()) + .format(FileFormat.PUFFIN) + .build(); + // position deletes a matched data file already has, merged into its new deletion vector + Map<String, PositionDeleteIndex> previousDeletes = Maps.newHashMap(); + BaseDVFileWriter dvWriter = new BaseDVFileWriter(fileFactory, previousDeletes::get); + + long matchedRows = 0L; + int plannedDataFiles = 0; + try { + for (Map.Entry<DeleteGroup, StructLikeSet> entry : keysByGroup.entrySet()) { + DeleteGroup group = entry.getKey(); + KeyPositionResolver.Resolution resolution = + resolver.resolve( + base, + keySchema(group.equalityFieldIds()), + entry.getValue(), + group.spec(), + group.partition()); + plannedDataFiles += resolution.plannedDataFiles(); + + for (KeyPositionResolver.FileMatches matches : resolution.matches()) { + if (matches.existingDeletes() != null) { + previousDeletes.put(matches.path(), matches.existingDeletes()); + } + + long[] count = new long[1]; + matches.forEachPosition( + pos -> { + dvWriter.delete(matches.path(), pos, matches.spec(), matches.partition()); + count[0]++; + }); + matchedRows += count[0]; + } + } + + dvWriter.close(); + } catch (IOException e) { + throw new UncheckedIOException(e); + } + + DeleteWriteResult writeResult = dvWriter.result(); + LOG.info( + "Converted {} equality delete file(s) with {} key(s) into {} deletion vector(s) covering {} row(s) " + + "of table {} at snapshot {}, planned {} data file(s) in {} ms", + eqDeleteFiles.size(), + deletedKeys, + writeResult.deleteFiles().size(), + matchedRows, + table.name(), + base.snapshotId(), + plannedDataFiles, + System.currentTimeMillis() - start); + + return new Result( + base.snapshotId(), + writeResult.deleteFiles(), + writeResult.rewrittenDeleteFiles(), + conflictFilter, + plannedDataFiles); + } + + /** + * Groups the delete files by equality field set and partition. A partitioned equality delete only + * applies to its own partition, and a smaller key set per group prunes better. + */ + private Map<DeleteGroup, List<DeleteFile>> groupDeleteFiles(List<DeleteFile> eqDeleteFiles) { + Map<DeleteGroup, List<DeleteFile>> groups = Maps.newLinkedHashMap(); + for (DeleteFile deleteFile : eqDeleteFiles) { + Preconditions.checkArgument( + deleteFile.content() == FileContent.EQUALITY_DELETES, + "Not an equality delete file: %s", + deleteFile.location()); + PartitionSpec spec = table.specs().get(deleteFile.specId()); + Preconditions.checkArgument( + spec != null, + "Unknown partition spec %s for %s", + deleteFile.specId(), + deleteFile.location()); + DeleteGroup group = + new DeleteGroup( + ImmutableSet.copyOf(deleteFile.equalityFieldIds()), spec, deleteFile.partition()); + groups.computeIfAbsent(group, key -> Lists.newArrayList()).add(deleteFile); + } + + return groups; + } + + private Schema keySchema(Set<Integer> equalityFieldIds) { + Schema keySchema = TypeUtil.select(table.schema(), equalityFieldIds); + Preconditions.checkArgument( + TypeUtil.getProjectedIds(keySchema).containsAll(equalityFieldIds), + "Equality field IDs %s are not all present in table schema %s", + equalityFieldIds, + table.schema()); + return keySchema; + } + + /** Outcome of a conversion, to be committed together with the data files of the batch. */ + public static class Result { + private final Long baseSnapshotId; + private final List<DeleteFile> dvFiles; + private final List<DeleteFile> rewrittenDvFiles; + private final Expression conflictFilter; + private final int plannedDataFiles; + + Result( + Long baseSnapshotId, + List<DeleteFile> dvFiles, + List<DeleteFile> rewrittenDvFiles, + Expression conflictFilter, + int plannedDataFiles) { + this.baseSnapshotId = baseSnapshotId; + this.dvFiles = ImmutableList.copyOf(dvFiles); + this.rewrittenDvFiles = ImmutableList.copyOf(rewrittenDvFiles); + this.conflictFilter = conflictFilter; + this.plannedDataFiles = plannedDataFiles; + } + + /** + * Snapshot the positions were resolved against, or null when there was no snapshot to resolve + * against. A result without a snapshot has no deletion vectors, but the commit must still + * validate that no data files matching {@link #conflictFilter()} were added, without a starting + * snapshot, so that rows another writer commits to the branch first are detected. + */ + public Long baseSnapshotId() { + return baseSnapshotId; + } + + /** + * Deletion vectors to add. The data files they reference must still exist when the vectors are + * committed, so the commit should pass their referenced data files to {@link + * org.apache.iceberg.RowDelta#validateDataFilesExist(Iterable)}. + */ + public List<DeleteFile> dvFiles() { Review Comment: Could the tests commit a `Result` through a small helper, covering a concurrent compaction of a DV's data file and a concurrent in-key append on a non-null base? Conflict validation [only covers append and overwrite](https://github.com/apache/iceberg/blob/286bc46e511948e9320c8535e465d5a728130df4/core/src/main/java/org/apache/iceberg/MergingSnapshotProducer.java#L73-L77), so only `validateDataFilesExist` catches the compaction, and the [`commit()` helper](https://github.com/apache/iceberg/blob/286bc46e511948e9320c8535e465d5a728130df4/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/data/TestEqualityDeleteConverter.java#L834-L848) never calls it. ########## kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/data/KeyPositionResolver.java: ########## @@ -0,0 +1,85 @@ +/* + * 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. */ +interface KeyPositionResolver { + + /** + * Resolves the keys of one group of equality delete files. + * + * @param base snapshot whose rows are resolved + * @param keySchema the identifier columns, a projection of the table schema + * @param keys deleted keys in the internal representation of {@code keySchema}, as produced by + * {@link org.apache.iceberg.data.InternalRecordWrapper} + * @param deleteSpec partition spec of the equality delete files + * @param deletePartition partition of the equality delete files. If {@code deleteSpec} is + * partitioned, only rows of data files with the same spec ID and partition values are + * returned; otherwise rows of every data file are candidates + * @return the matched rows, grouped by data file + */ + Resolution resolve( + Snapshot base, + Schema keySchema, + StructLikeSet keys, + PartitionSpec deleteSpec, + StructLike deletePartition) + throws IOException; + + /** Rows matched by one resolution. */ + interface Resolution { + Collection<FileMatches> matches(); + + /** + * Number of data files planned as candidates, before the spec and partition of each candidate + * are checked against the deletes. + */ + int plannedDataFiles(); + } + + /** Matched rows of one data file. */ + interface FileMatches { + String path(); + + PartitionSpec spec(); + + StructLike partition(); + + /** + * Position deletes the data file already has in the base snapshot: its deletion vector, or + * position delete files left from format version 2. Returns null only if the file has none. The + * converter merges them into the new deletion vector, which replaces them for readers, so + * returning null for a file that has some leaves it with two vectors or brings back rows the + * older deletes removed. + */ + PositionDeleteIndex existingDeletes(); Review Comment: @pvary, does this obligation fit the index in #16961? Every resolver has to return the file's existing DV and v2 position deletes, and once the new DV exists, a null for v2 files [silently brings their deleted rows back](https://github.com/apache/iceberg/blob/286bc46e511948e9320c8535e465d5a728130df4/core/src/main/java/org/apache/iceberg/DeleteFileIndex.java#L158-L175). The draft spec [doesn't define whether an entry points back to its row](https://github.com/pvary/iceberg/blob/701330c795e8aaae7bb6e6fca338e21795e8bfea/format/index-spec.md?plain=1#L367-L377), so an index-backed resolver may have no positions to give. ########## kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/data/KeyFilters.java: ########## @@ -0,0 +1,203 @@ +/* + * 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.util.Comparator; +import java.util.List; +import java.util.Set; +import org.apache.iceberg.PartitionField; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.Schema; +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.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; above this many keys the + // lists are replaced by the bounds of the leading key column, two comparisons per entry + @VisibleForTesting static final int MAX_LISTED_KEYS = 1_000; + + private KeyFilters() {} + + /** + * Pins the partition of an equality delete file. Every partition field is expressed through its + * transform, e.g. {@code bucket(16, id) = 3}, so the scan planner prunes to the partition even + * when the key filter is a range that a transform like bucket cannot project. Identity fields + * reference the column directly. Void fields carry no information and unknown transforms cannot + * be evaluated, so both are skipped. + */ + @SuppressWarnings("unchecked") + static Expression partitionFilter(Schema tableSchema, PartitionSpec spec, StructLike partition) { + Expression filter = Expressions.alwaysTrue(); + List<PartitionField> fields = spec.fields(); + for (int pos = 0; pos < fields.size(); pos++) { + PartitionField field = fields.get(pos); + Transform<?, ?> transform = field.transform(); + if (transform.isVoid() || transform instanceof UnknownTransform) { + continue; + } + + String column = tableSchema.findColumnName(field.sourceId()); + UnboundTerm<Object> term = + transform.isIdentity() + ? Expressions.ref(column) + : Expressions.transform(column, (Transform<?, Object>) transform); + Object value = partition.get(pos, Object.class); + filter = + Expressions.and( + filter, value == null ? Expressions.isNull(term) : Expressions.equal(term, value)); + } + + return filter; + } + + /** + * Filter for the deleted keys. Up to {@link #MAX_LISTED_KEYS} keys are matched with {@code IN} + * lists of at most {@link #IN_PREDICATE_LIMIT} keys each, one list per key column. More keys are + * matched with the minimum and maximum of the leading key column, which may also cover files that + * hold no deleted key. + */ + static Expression keyFilter(Schema keySchema, StructLikeSet keys) { + if (keys.size() <= MAX_LISTED_KEYS) { Review Comment: Above 1,000 keys this falls back to one min/max range on the first key column. For scattered keys (UUIDs, hashes, updates across history) that means every commit reads the key columns of every file in the partition, or the whole table if unpartitioned, with no cap on planned files. The [1,200-key test](https://github.com/apache/iceberg/blob/286bc46e511948e9320c8535e465d5a728130df4/kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/data/TestEqualityDeleteConverter.java#L454-L455) already plans all 30 of 30 files. The same range is the conflict filter, so a retry repeats the scan. What envelope (keys per commit, files per partition, key distribution) should the part-2 benchmark cover, and what happens when `plannedDataFiles()` goes past it? -- 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]
