JingsongLi commented on code in PR #9177: URL: https://github.com/apache/paimon/pull/9177#discussion_r3763818114
########## paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionDeletionVectorMaterializeCoordinator.java: ########## @@ -0,0 +1,522 @@ +/* + * 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.paimon.append.dataevolution; + +import org.apache.paimon.CoreOptions; +import org.apache.paimon.Snapshot; +import org.apache.paimon.annotation.VisibleForTesting; +import org.apache.paimon.data.BinaryRow; +import org.apache.paimon.index.IndexFileHandler; +import org.apache.paimon.io.DataFileMeta; +import org.apache.paimon.manifest.IndexManifestEntry; +import org.apache.paimon.manifest.ManifestEntry; +import org.apache.paimon.manifest.ManifestFileMeta; +import org.apache.paimon.operation.FileStoreScan; +import org.apache.paimon.partition.PartitionPredicate; +import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.table.source.DeletionFile; +import org.apache.paimon.table.source.EndOfScanException; +import org.apache.paimon.table.source.ScanMode; +import org.apache.paimon.table.source.snapshot.SnapshotReader; +import org.apache.paimon.utils.Range; +import org.apache.paimon.utils.RangeHelper; + +import javax.annotation.Nullable; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashSet; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.stream.Collectors; + +import static org.apache.paimon.deletionvectors.DeletionVectorsIndexFile.DELETION_VECTORS_INDEX; +import static org.apache.paimon.format.blob.BlobFileFormat.isBlobFile; +import static org.apache.paimon.manifest.ManifestFileMeta.allContainsRowId; +import static org.apache.paimon.table.BucketMode.UNAWARE_BUCKET; +import static org.apache.paimon.types.VectorType.isVectorStoreFile; +import static org.apache.paimon.utils.DataEvolutionUtils.retrieveAnchorFile; +import static org.apache.paimon.utils.Preconditions.checkArgument; +import static org.apache.paimon.utils.Preconditions.checkState; + +/** Plans tasks which physically apply deletion vectors and assign new row IDs. */ +public class DataEvolutionDeletionVectorMaterializeCoordinator { + + // Soft target. One overlapping row-id component can exceed it. + private static final int DELETION_FILES_PER_BATCH = 100_000; + + private final MaterializeScanner scanner; + private final MaterializePlanner planner; + + public DataEvolutionDeletionVectorMaterializeCoordinator( + FileStoreTable table, + @Nullable PartitionPredicate partitionPredicate, + Snapshot snapshot) { + this(table, partitionPredicate, snapshot, DELETION_FILES_PER_BATCH); + } + + @VisibleForTesting + public DataEvolutionDeletionVectorMaterializeCoordinator( + FileStoreTable table, + @Nullable PartitionPredicate partitionPredicate, + Snapshot snapshot, + int deletionFilesPerBatch) { + CoreOptions options = table.coreOptions(); + checkArgument( + options.dataEvolutionEnabled(), + "Materializing deletion vectors requires a data evolution table."); + checkArgument( + options.deletionVectorsEnabled(), + "Materializing deletion vectors requires deletion vectors to be enabled."); + + this.scanner = + new MaterializeScanner(table, partitionPredicate, snapshot, deletionFilesPerBatch); + this.planner = + new MaterializePlanner(options.targetFileSize(false), options.splitOpenFileCost()); + } + + public List<DataEvolutionCompactTask> plan() { + return planner.plan(scanner.scan()); + } + + public Snapshot snapshot() { + return scanner.snapshot; + } + + private static class MaterializeScanner { + + @Nullable private final FileStoreScan anchorScan; + @Nullable private final FileStoreScan rangeScan; + private final Snapshot snapshot; + private final List<ManifestFileMeta> manifests; + private final LinkedHashMap<BinaryRow, LinkedHashMap<String, DeletionFile>> + remainingDeletionFiles; + private final int deletionFilesPerBatch; + + private MaterializeScanner( + FileStoreTable table, + @Nullable PartitionPredicate partitionPredicate, + Snapshot snapshot, + int deletionFilesPerBatch) { + this.snapshot = snapshot; + checkArgument(deletionFilesPerBatch > 0, "Deletion files per batch must be positive."); + this.deletionFilesPerBatch = deletionFilesPerBatch; + this.remainingDeletionFiles = + scanDeletionFiles( + table.store().newIndexFileHandler(), snapshot, partitionPredicate); + + if (remainingDeletionFiles.isEmpty()) { + this.anchorScan = null; + this.rangeScan = null; + this.manifests = Collections.emptyList(); + } else { + SnapshotReader snapshotReader = + table.newSnapshotReader().withPartitionFilter(partitionPredicate); + this.manifests = + snapshotReader + .manifestsReader() + .read(snapshot, ScanMode.ALL) + .filteredManifests; + this.anchorScan = + table.store().newScan().withPartitionFilter(partitionPredicate).dropStats(); + this.rangeScan = + table.store().newScan().withPartitionFilter(partitionPredicate).dropStats(); + } + } + + private MaterializeScanBatch scan() { + if (remainingDeletionFiles.isEmpty()) { + throw new EndOfScanException(); + } + + Map<BinaryRow, Map<String, DeletionFile>> candidates = takeCandidates(); + Set<String> candidateFileNames = new HashSet<>(); + candidates.values().forEach(files -> candidateFileNames.addAll(files.keySet())); + + checkState(anchorScan != null && rangeScan != null); + anchorScan.withDataFileNameFilter(candidateFileNames::contains); + List<ManifestEntry> anchors = new ArrayList<>(candidateFileNames.size()); + anchorScan.readFileIterator(manifests).forEachRemaining(anchors::add); + + Map<BinaryRow, Set<String>> missing = new LinkedHashMap<>(); + for (Map.Entry<BinaryRow, Map<String, DeletionFile>> candidate : + candidates.entrySet()) { + missing.put(candidate.getKey(), new HashSet<>(candidate.getValue().keySet())); + } + List<Range> ranges = new ArrayList<>(anchors.size()); + for (ManifestEntry anchor : anchors) { + Set<String> partitionMissing = missing.get(anchor.partition()); + if (partitionMissing == null + || !partitionMissing.remove(anchor.file().fileName())) { + continue; + } + checkArgument( + !isBlobFile(anchor.file().fileName()) + && !isVectorStoreFile(anchor.file().fileName()), + "Deletion vector anchor '%s' must be a normal data file.", + anchor.file().fileName()); + ranges.add(anchor.file().nonNullRowIdRange()); + } + List<String> missingFiles = + missing.values().stream().flatMap(Set::stream).collect(Collectors.toList()); + checkState( + missingFiles.isEmpty(), + "Cannot find live data files for deletion vectors: %s", + missingFiles); + + ranges = Range.sortAndMergeOverlap(ranges); + rangeScan.withRowRanges(ranges); Review Comment: Thanks for the precise reproduction. This is fixed in `c33177648d`. The materialization planner now expands the DV-derived range to the complete overlapping-range closure before creating a task. It accumulates the normal and dedicated files overlapping the current frontier; when a BLOB/vector file extends the physical coverage, the newly required normal range becomes the next frontier, and this repeats until the component is stable. A connected component is kept in one task even when it exceeds the soft batch target, so a dedicated-file read group is never split. I added `testMaterializeRangeCoveredBySpanningBlobFile`, covering normal ranges `[0,4]`, `[5,9]`, `[10,14]`, a BLOB file spanning `[5,14]`, and a DV anchored in `[5,9]`. Materialization now includes `[10,14]`, completes successfully, and preserves the expected rows. The relevant Core materialization/coordinator tests pass. -- 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]
