leaves12138 commented on code in PR #9177: URL: https://github.com/apache/paimon/pull/9177#discussion_r3763322935
########## 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: The selected ranges are derived only from DV anchor files, but the resulting scan can include a dedicated file whose physical range extends beyond those anchors without including all normal files needed to cover that dedicated file. This creates an incomplete logical read group. A local reproduction is: 1. normal ranges: `[0,4]`, `[5,9]`, `[10,14]`; 2. write a partial BLOB update as one file covering `[5,14]`; 3. create a DV only for the anchor in `[5,9]`; 4. run materialization. This scan selects `[5,9]`, includes the BLOB file `[5,14]` because it intersects, but excludes the normal file `[10,14]`. `DataEvolutionMaterializeDeletionCompactTask` then fails in `DataEvolutionSplitRead` with `The merged rowCount 10 of blob file bunch should be aligned with normal files 5.` Please expand the batch to the complete normal-file coverage required by every included dedicated file (and repeat to closure if necessary), or otherwise ensure a materialization task never contains a partial dedicated-file read group. A regression with a BLOB/vector file spanning adjacent normal ranges would be useful. -- 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]
