diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java index 1eacfa71f78d..e8a8dadfdf90 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java @@ -204,7 +204,15 @@ public FileStoreCommitImpl( this.manifestList = manifestListFactory.create(); this.indexManifestFile = indexManifestFileFactory.create(); this.rollback = rollback; - this.scanner = new CommitScanner(scanSupplier, snapshotManager, indexManifestFile, options); + this.scanner = + new CommitScanner( + scanSupplier, + snapshotManager, + indexManifestFile, + manifestFile, + manifestList, + partitionType, + options); this.commitPreCallbacks = commitPreCallbacks; this.commitCallbacks = commitCallbacks; this.retryWaiter = @@ -962,67 +970,82 @@ CommitResult tryCommitOnce( boolean discardDuplicate = options.commitDiscardDuplicateFiles() && commitKind == CommitKind.APPEND; boolean checkConflicts = latestSnapshot != null && (discardDuplicate || detectConflicts); + boolean checkGlobalIndexOnly = + checkConflicts + && commitKind == CommitKind.APPEND + && deltaFiles.isEmpty() + && commitPreCallbacks.isEmpty() + && !conflictDetection.hasRowIdCheckFromSnapshot() + && conflictDetection.canCheckGlobalIndexOnly(indexFiles); // By default, if checkConflicts is required, we do not have to do the extra check bucket // here. if (!checkConflicts && shouldCheckSameBucket(commitKind)) { checkSameBucketFromSnapshot(deltaFiles, latestSnapshot); } if (checkConflicts) { - // latestSnapshotId is different from the snapshot id we've checked for conflicts, - // so we have to check again - if (changedPartitions == null) { - changedPartitions = changedPartitions(deltaFiles, indexFiles); - } - CommitFailRetryResult commitFailRetry = - retryResult instanceof CommitFailRetryResult - ? (CommitFailRetryResult) retryResult - : null; - // An overwrite may replace the base manifest list without recording the replacements - // in its delta manifest, so the cached base cannot always be refreshed incrementally. - if (commitFailRetry != null - && commitFailRetry.latestSnapshot != null - && commitFailRetry.baseDataFiles != null - && !hasOverwriteSinceLastAttempt) { - baseDataFiles = new ArrayList<>(commitFailRetry.baseDataFiles); - List incremental = - scanner.readIncrementalChanges( - commitFailRetry.latestSnapshot, latestSnapshot, changedPartitions); - if (!incremental.isEmpty()) { - baseDataFiles.addAll(incremental); - baseDataFiles = new ArrayList<>(FileEntry.mergeEntries(baseDataFiles)); - } + Optional exception; + if (checkGlobalIndexOnly) { + exception = conflictDetection.checkGlobalIndexOnly(latestSnapshot, indexFiles); } else { - baseDataFiles = - scanner.readAllEntriesFromChangedPartitions( - latestSnapshot, changedPartitions); - } - if (discardDuplicate) { - Set baseIdentifiers = - baseDataFiles.stream() - .map(FileEntry::identifier) - .collect(Collectors.toSet()); - deltaFiles = - deltaFiles.stream() - .filter(entry -> !baseIdentifiers.contains(entry.identifier())) - .collect(Collectors.toList()); - } - RowIdColumnConflictChecker rowIdColumnConflictChecker = null; - if (conflictDetection.hasRowIdCheckFromSnapshot()) { - rowIdColumnConflictChecker = - RowIdColumnConflictChecker.fromDataFiles( - schemaManager, - deltaFiles.stream() - .map(ManifestEntry::file) - .collect(Collectors.toList())); + // latestSnapshotId is different from the snapshot id we've checked for conflicts, + // so we have to check again + if (changedPartitions == null) { + changedPartitions = changedPartitions(deltaFiles, indexFiles); + } + CommitFailRetryResult commitFailRetry = + retryResult instanceof CommitFailRetryResult + ? (CommitFailRetryResult) retryResult + : null; + // An overwrite may replace the base manifest list without recording the + // replacements in its delta manifest, so the cached base cannot always be + // refreshed incrementally. + if (commitFailRetry != null + && commitFailRetry.latestSnapshot != null + && commitFailRetry.baseDataFiles != null + && !hasOverwriteSinceLastAttempt) { + baseDataFiles = new ArrayList<>(commitFailRetry.baseDataFiles); + List incremental = + scanner.readIncrementalChanges( + commitFailRetry.latestSnapshot, + latestSnapshot, + changedPartitions); + if (!incremental.isEmpty()) { + baseDataFiles.addAll(incremental); + baseDataFiles = new ArrayList<>(FileEntry.mergeEntries(baseDataFiles)); + } + } else { + baseDataFiles = + scanner.readAllEntriesFromChangedPartitions( + latestSnapshot, changedPartitions); + } + if (discardDuplicate) { + Set baseIdentifiers = + baseDataFiles.stream() + .map(FileEntry::identifier) + .collect(Collectors.toSet()); + deltaFiles = + deltaFiles.stream() + .filter(entry -> !baseIdentifiers.contains(entry.identifier())) + .collect(Collectors.toList()); + } + RowIdColumnConflictChecker rowIdColumnConflictChecker = null; + if (conflictDetection.hasRowIdCheckFromSnapshot()) { + rowIdColumnConflictChecker = + RowIdColumnConflictChecker.fromDataFiles( + schemaManager, + deltaFiles.stream() + .map(ManifestEntry::file) + .collect(Collectors.toList())); + } + exception = + conflictDetection.checkConflicts( + latestSnapshot, + baseDataFiles, + SimpleFileEntry.from(deltaFiles), + indexFiles, + rowIdColumnConflictChecker, + commitKind); } - Optional exception = - conflictDetection.checkConflicts( - latestSnapshot, - baseDataFiles, - SimpleFileEntry.from(deltaFiles), - indexFiles, - rowIdColumnConflictChecker, - commitKind); if (exception.isPresent()) { if (allowRollback && rollback != null) { if (rollback.tryToRollback(latestSnapshot)) { @@ -1033,6 +1056,8 @@ CommitResult tryCommitOnce( } } + boolean indexOnlyCommit = + deltaFiles.isEmpty() && changelogFiles.isEmpty() && !indexFiles.isEmpty(); Snapshot newSnapshot; Pair baseManifestList = null; Pair deltaManifestList = null; @@ -1066,6 +1091,11 @@ CommitResult tryCommitOnce( mergeBeforeManifests = emptyList(); mergeAfterManifests = emptyList(); oldIndexManifest = null; + } else if (indexOnlyCommit) { + // An index-only commit cannot change table data. Preserve the existing data + // manifests instead of opportunistically sorting or compacting them. + mergeAfterManifests = mergeBeforeManifests; + skipManifestMergeOnRetry = true; } else { ManifestMergeReuse manifestMergeReuse = tryReuseManifestMergeResult(retryResult, mergeBeforeManifests); diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/commit/CommitScanner.java b/paimon-core/src/main/java/org/apache/paimon/operation/commit/CommitScanner.java index e63e76590d78..154b77aef8a4 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/commit/CommitScanner.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/commit/CommitScanner.java @@ -21,33 +21,68 @@ import org.apache.paimon.CoreOptions; import org.apache.paimon.Snapshot; import org.apache.paimon.data.BinaryRow; +import org.apache.paimon.index.GlobalIndexMeta; +import org.apache.paimon.io.BinaryDataFileMeta; +import org.apache.paimon.io.DataFileMeta; +import org.apache.paimon.manifest.BinaryManifestEntry; +import org.apache.paimon.manifest.BinaryManifestEntry.Projection; +import org.apache.paimon.manifest.BinaryManifestEntry.ReusableIdentifier; +import org.apache.paimon.manifest.DeletedIdentifierSet; import org.apache.paimon.manifest.IndexManifestEntry; import org.apache.paimon.manifest.IndexManifestFile; import org.apache.paimon.manifest.ManifestEntry; +import org.apache.paimon.manifest.ManifestFile; +import org.apache.paimon.manifest.ManifestFileMeta; +import org.apache.paimon.manifest.ManifestList; import org.apache.paimon.manifest.SimpleFileEntry; import org.apache.paimon.operation.FileStoreScan; import org.apache.paimon.partition.PartitionPredicate; import org.apache.paimon.table.source.ScanMode; +import org.apache.paimon.types.DataField; +import org.apache.paimon.types.RowType; +import org.apache.paimon.utils.ByteArrayKey; +import org.apache.paimon.utils.ByteArrayLookupKey; +import org.apache.paimon.utils.CloseableIterator; +import org.apache.paimon.utils.PrimitiveRowRanges; +import org.apache.paimon.utils.Range; +import org.apache.paimon.utils.RowRangeIndex; import org.apache.paimon.utils.SnapshotManager; import javax.annotation.Nullable; import java.util.ArrayList; +import java.util.Arrays; import java.util.HashMap; import java.util.HashSet; import java.util.Iterator; import java.util.List; import java.util.Map; +import java.util.Optional; import java.util.Set; import java.util.function.Supplier; +import static org.apache.paimon.utils.SerializationUtils.serializeBinaryRow; + /** Manifest entries scanner for commit. */ public class CommitScanner { + private static final Projection GLOBAL_INDEX_ROW_ID_PROJECTION = + manifestProjection( + DataFileMeta.FILE_NAME, + DataFileMeta.ROW_COUNT, + DataFileMeta.LEVEL, + DataFileMeta.EXTRA_FILES, + DataFileMeta.EMBEDDED_FILE_INDEX, + DataFileMeta.EXTERNAL_PATH, + DataFileMeta.FIRST_ROW_ID); + private final FileStoreScan scan; private final Supplier scanSupplier; private final SnapshotManager snapshotManager; private final IndexManifestFile indexManifestFile; + private final @Nullable ManifestFile manifestFile; + private final @Nullable ManifestList manifestList; + private final @Nullable RowType partitionType; private final boolean dropStats; public CommitScanner( @@ -55,10 +90,24 @@ public CommitScanner( SnapshotManager snapshotManager, IndexManifestFile indexManifestFile, CoreOptions options) { + this(scanSupplier, snapshotManager, indexManifestFile, null, null, null, options); + } + + public CommitScanner( + Supplier scanSupplier, + SnapshotManager snapshotManager, + IndexManifestFile indexManifestFile, + @Nullable ManifestFile manifestFile, + @Nullable ManifestList manifestList, + @Nullable RowType partitionType, + CoreOptions options) { this.scanSupplier = scanSupplier; this.scan = scanSupplier.get(); this.snapshotManager = snapshotManager; this.indexManifestFile = indexManifestFile; + this.manifestFile = manifestFile; + this.manifestList = manifestList; + this.partitionType = partitionType; // Stats in DELETE Manifest Entries is useless this.dropStats = options.manifestDeleteFileDropStats(); if (dropStats) { @@ -101,6 +150,263 @@ public List readAllEntriesFromChangedPartitions( } } + /** + * Checks global-index row ranges by streaming projected binary manifest entries. + * + *

Only live data ranges which intersect a requested index range are retained. This avoids + * materializing all data files in the changed partitions as {@link SimpleFileEntry}s. + * + * @return the first index entry whose row range is not covered by live data files + */ + public Optional firstGlobalIndexWithMissingRowIds( + Snapshot snapshot, List indexesToCheck) { + if (indexesToCheck.isEmpty()) { + return Optional.empty(); + } + if (manifestFile == null || manifestList == null || partitionType == null) { + throw new IllegalStateException( + "Projected manifest scanning is not configured for this commit scanner."); + } + + try { + GlobalIndexCoveragePlan coveragePlan = createGlobalIndexCoverage(indexesToCheck); + RowRangeIndex requestedRanges = requestedRanges(indexesToCheck); + List candidateManifests = + candidateManifests(snapshot, coveragePlan.targetPartitions(), requestedRanges); + + DeletedIdentifierSet deletedIdentifiers = new DeletedIdentifierSet(); + try { + collectDeletedIdentifiers( + candidateManifests, coveragePlan.byPartitionBytes, deletedIdentifiers); + collectLiveRowIdRanges( + candidateManifests, coveragePlan.byPartitionBytes, deletedIdentifiers); + } finally { + deletedIdentifiers.release(); + } + + for (IndexManifestEntry indexEntry : indexesToCheck) { + GlobalIndexMeta index = indexEntry.indexFile().globalIndexMeta(); + GlobalIndexCoverage coverage = coveragePlan.forIndex(indexEntry); + if (!coverage.liveDataRanges.covers(index.rowRangeStart(), index.rowRangeEnd())) { + return Optional.of(indexEntry); + } + } + return Optional.empty(); + } catch (Throwable e) { + throw new RuntimeException("Cannot check global-index row IDs from data manifests.", e); + } + } + + private static GlobalIndexCoveragePlan createGlobalIndexCoverage( + List indexesToCheck) { + Map> byPartitionRows = new HashMap<>(); + int nextCoverageId = 0; + for (IndexManifestEntry indexEntry : indexesToCheck) { + GlobalIndexMeta index = indexEntry.indexFile().globalIndexMeta(); + BinaryRow partition = indexEntry.partition(); + Map byBucket = byPartitionRows.get(partition); + if (byBucket == null) { + byBucket = new HashMap<>(); + byPartitionRows.put(partition.copy(), byBucket); + } + GlobalIndexCoverage coverage = byBucket.get(indexEntry.bucket()); + if (coverage == null) { + coverage = new GlobalIndexCoverage(nextCoverageId++); + byBucket.put(indexEntry.bucket(), coverage); + } + coverage.requestedRanges.add(index.rowRange()); + } + Map> byPartitionBytes = new HashMap<>(); + for (Map.Entry> partition : + byPartitionRows.entrySet()) { + byPartitionBytes.put( + new ByteArrayKey(serializeBinaryRow(partition.getKey())), partition.getValue()); + } + for (Map byBucket : byPartitionRows.values()) { + for (GlobalIndexCoverage coverage : byBucket.values()) { + coverage.prepare(); + } + } + return new GlobalIndexCoveragePlan(byPartitionRows, byPartitionBytes); + } + + private static RowRangeIndex requestedRanges(List indexesToCheck) { + List ranges = new ArrayList<>(indexesToCheck.size()); + for (IndexManifestEntry entry : indexesToCheck) { + ranges.add(entry.indexFile().globalIndexMeta().rowRange()); + } + return RowRangeIndex.create(ranges); + } + + private List candidateManifests( + Snapshot snapshot, Set targetPartitions, RowRangeIndex requestedRanges) { + PartitionPredicate partitionPredicate = + PartitionPredicate.fromMultiple(partitionType, targetPartitions); + List result = new ArrayList<>(); + for (ManifestFileMeta meta : manifestList.readDataManifests(snapshot)) { + Long minRowId = meta.minRowId(); + Long maxRowId = meta.maxRowId(); + if (minRowId != null + && maxRowId != null + && !requestedRanges.intersects(minRowId, maxRowId)) { + continue; + } + if (partitionPredicate != null + && !partitionPredicate.test( + meta.numAddedFiles() + meta.numDeletedFiles(), + meta.partitionStats().minValues(), + meta.partitionStats().maxValues(), + meta.partitionStats().nullCounts())) { + continue; + } + result.add(meta); + } + return result; + } + + private void collectDeletedIdentifiers( + List manifests, + Map> coverageByPartition, + DeletedIdentifierSet deletedIdentifiers) + throws Exception { + ReusableIdentifier identifier = new ReusableIdentifier(); + ByteArrayLookupKey partitionLookup = new ByteArrayLookupKey(); + try { + for (ManifestFileMeta meta : manifests) { + if (meta.numDeletedFiles() == 0) { + continue; + } + try (CloseableIterator entries = + manifestFile.scan( + meta.fileName(), + meta.fileSize(), + BinaryManifestEntry.DELETE_ENTRY_PROJECTION)) { + while (entries.hasNext()) { + BinaryManifestEntry entry = entries.next(); + if (!entry.isDelete()) { + continue; + } + GlobalIndexCoverage coverage = + requestedBucket(entry, coverageByPartition, partitionLookup); + if (coverage != null) { + deletedIdentifiers.add(coverage.id, identifier.replace(entry)); + } + } + } + } + } finally { + identifier.release(); + partitionLookup.clear(); + } + } + + private void collectLiveRowIdRanges( + List manifests, + Map> coverageByPartition, + DeletedIdentifierSet deletedIdentifiers) + throws Exception { + ReusableIdentifier identifier = new ReusableIdentifier(); + ByteArrayLookupKey partitionLookup = new ByteArrayLookupKey(); + try { + for (ManifestFileMeta meta : manifests) { + if (meta.numAddedFiles() == 0) { + continue; + } + try (CloseableIterator entries = + manifestFile.scan( + meta.fileName(), meta.fileSize(), GLOBAL_INDEX_ROW_ID_PROJECTION)) { + while (entries.hasNext()) { + BinaryManifestEntry entry = entries.next(); + if (!entry.isAdd()) { + continue; + } + GlobalIndexCoverage coverage = + requestedBucket(entry, coverageByPartition, partitionLookup); + if (coverage == null + || (!deletedIdentifiers.isEmpty() + && deletedIdentifiers.contains( + coverage.id, identifier.replace(entry)))) { + continue; + } + BinaryDataFileMeta file = entry.file(); + if (!file.hasFirstRowId()) { + continue; + } + long start = file.firstRowId(); + long end = Math.addExact(start, file.rowCount() - 1L); + if (coverage.requestedRangeIndex.intersects(start, end)) { + coverage.liveDataRanges.add(start, end); + } + } + } + } + } finally { + identifier.release(); + partitionLookup.clear(); + } + } + + @Nullable + private static GlobalIndexCoverage requestedBucket( + BinaryManifestEntry entry, + Map> coverageByPartition, + ByteArrayLookupKey partitionLookup) { + partitionLookup.reset(entry.partitionBytes()); + Map byBucket = coverageByPartition.get(partitionLookup); + return byBucket == null ? null : byBucket.get(entry.bucket()); + } + + private static Projection manifestProjection(String... projectedFileFields) { + List fields = + new ArrayList<>( + Arrays.asList( + ManifestEntry.MANIFEST_ROW_TYPE.getField(ManifestEntry.KIND), + ManifestEntry.MANIFEST_ROW_TYPE.getField(ManifestEntry.PARTITION), + ManifestEntry.MANIFEST_ROW_TYPE.getField(ManifestEntry.BUCKET), + ManifestEntry.MANIFEST_ROW_TYPE + .getField(ManifestEntry.FILE) + .newType( + DataFileMeta.SCHEMA.project(projectedFileFields)))); + return Projection.create(new RowType(false, fields)); + } + + private static class GlobalIndexCoverage { + + private final int id; + private final List requestedRanges = new ArrayList<>(); + private final PrimitiveRowRanges liveDataRanges = new PrimitiveRowRanges(16); + private RowRangeIndex requestedRangeIndex; + + private GlobalIndexCoverage(int id) { + this.id = id; + } + + private void prepare() { + requestedRangeIndex = RowRangeIndex.create(requestedRanges); + } + } + + private static class GlobalIndexCoveragePlan { + + private final Map> byPartitionRows; + private final Map> byPartitionBytes; + + private GlobalIndexCoveragePlan( + Map> byPartitionRows, + Map> byPartitionBytes) { + this.byPartitionRows = byPartitionRows; + this.byPartitionBytes = byPartitionBytes; + } + + private Set targetPartitions() { + return byPartitionRows.keySet(); + } + + private GlobalIndexCoverage forIndex(IndexManifestEntry entry) { + return byPartitionRows.get(entry.partition()).get(entry.bucket()); + } + } + public Map readTotalBuckets( Snapshot snapshot, List changedPartitions) { try { diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/commit/ConflictDetection.java b/paimon-core/src/main/java/org/apache/paimon/operation/commit/ConflictDetection.java index 34c5a29a4612..bdb021fc3f97 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/commit/ConflictDetection.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/commit/ConflictDetection.java @@ -135,6 +135,36 @@ public boolean hasRowIdCheckFromSnapshot() { return rowIdCheckFromSnapshot != null; } + /** + * Returns whether an index-only commit can use the projected manifest conflict check. + * + *

Deletion-vector changes still require the complete base-file conflict path. + */ + public boolean canCheckGlobalIndexOnly(List deltaIndexEntries) { + if (!dataEvolutionEnabled) { + return false; + } + + boolean hasGlobalIndexAddition = false; + for (IndexManifestEntry entry : deltaIndexEntries) { + if (entry.indexFile().indexType().equals(DELETION_VECTORS_INDEX)) { + return false; + } + hasGlobalIndexAddition |= + entry.kind() == FileKind.ADD && entry.indexFile().globalIndexMeta() != null; + } + return hasGlobalIndexAddition; + } + + /** Checks global-index row IDs without materializing the table's data manifest entries. */ + public Optional checkGlobalIndexOnly( + Snapshot latestSnapshot, List deltaIndexEntries) { + List indexesToCheck = globalIndexFileAdditions(deltaIndexEntries); + return commitScanner + .firstGlobalIndexWithMissingRowIds(latestSnapshot, indexesToCheck) + .map(this::globalIndexRowIdExistenceConflict); + } + @Nullable public Comparator keyComparator() { return keyComparator; @@ -660,20 +690,24 @@ private Optional checkGlobalIndexRowIdExistence( RowRangeIndex rowRangeIndex = rowRangeIndexes.get(Pair.of(indexEntry.partition(), indexEntry.bucket())); if (rowRangeIndex == null || !rowRangeIndex.contains(indexRange)) { - return Optional.of( - new RuntimeException( - String.format( - "Global index row ID existence conflict: index file '%s' " - + "references row range %s, but this range " - + "is not fully covered by current data " - + "files. The referenced row IDs may have been " - + "reassigned or removed by a concurrent commit.", - indexEntry.indexFile().fileName(), indexRange))); + return Optional.of(globalIndexRowIdExistenceConflict(indexEntry)); } } return Optional.empty(); } + private RuntimeException globalIndexRowIdExistenceConflict(IndexManifestEntry indexEntry) { + return new RuntimeException( + String.format( + "Global index row ID existence conflict: index file '%s' " + + "references row range %s, but this range " + + "is not fully covered by current data " + + "files. The referenced row IDs may have been " + + "reassigned or removed by a concurrent commit.", + indexEntry.indexFile().fileName(), + indexEntry.indexFile().globalIndexMeta().rowRange())); + } + private List globalIndexFileAdditions( List indexFileChanges) { List result = new ArrayList<>(); diff --git a/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java b/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java index 97f4a5b80426..13bfce61d714 100644 --- a/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java @@ -1326,6 +1326,8 @@ public void testGlobalIndexCommitChecksExistingRowIds() throws Exception { BinaryRow partition = gen.getPartition(keyValues.get(0)); Snapshot dataSnapshot = store.commitData(keyValues, s -> partition, kv -> 0).get(0); assertThat(dataSnapshot.nextRowId()).isEqualTo(1L); + List dataManifests = + store.manifestListFactory().create().readDataManifests(dataSnapshot); try (FileStoreCommitImpl commit = store.newCommit()) { commit.commit(indexCommittable(partition, "existing-index", 0, 0), false); @@ -1333,6 +1335,27 @@ public void testGlobalIndexCommitChecksExistingRowIds() throws Exception { Snapshot latest = checkNotNull(store.snapshotManager().latestSnapshot()); assertThat(latest.indexManifest()).isNotNull(); + assertThat(store.manifestListFactory().create().readDataManifests(latest)) + .containsExactlyElementsOf(dataManifests); + } + + @Test + public void testGlobalIndexCommitChecksAdjacentDataRanges() throws Exception { + TestFileStore store = createRowTrackingDataEvolutionStore(); + + KeyValue first = generateDataList(1).get(0); + BinaryRow partition = gen.getPartition(first); + store.commitData(Collections.singletonList(first), s -> partition, kv -> 0); + Snapshot secondSnapshot = + store.commitData(generateDataList(1), s -> partition, kv -> 0).get(0); + assertThat(secondSnapshot.nextRowId()).isEqualTo(2L); + + try (FileStoreCommitImpl commit = store.newCommit()) { + commit.commit(indexCommittable(partition, "adjacent-index", 0, 1), false); + } + + assertThat(checkNotNull(store.snapshotManager().latestSnapshot()).indexManifest()) + .isNotNull(); } @Test @@ -1360,6 +1383,75 @@ public void testGlobalIndexCommitFailsForMissingRowIds() throws Exception { } } + @Test + public void testGlobalIndexCommitDoesNotUseDeletedDataFileRowIds() throws Exception { + TestFileStore store = createRowTrackingDataEvolutionStore(); + + KeyValue original = generateDataList(1).get(0); + BinaryRow partition = gen.getPartition(original); + store.commitData(Collections.singletonList(original), s -> partition, kv -> 0); + Snapshot overwrite = + store.overwriteData( + generateDataList(1), + s -> partition, + kv -> 0, + Collections.emptyMap()) + .get(0); + assertThat(overwrite.nextRowId()).isEqualTo(2L); + + try (FileStoreCommitImpl commit = store.newCommit()) { + assertThatThrownBy( + () -> + commit.commit( + indexCommittable( + partition, "deleted-row-id-index", 0, 0), + false)) + .hasMessageContaining("Global index row ID existence conflict") + .hasMessageContaining("deleted-row-id-index"); + } + + try (FileStoreCommitImpl commit = store.newCommit()) { + commit.commit(indexCommittable(partition, "replacement-row-id-index", 1, 1), false); + } + } + + @Test + public void testGlobalIndexCommitChecksPartitionAndBucket() throws Exception { + TestFileStore store = createRowTrackingDataEvolutionStore(); + + KeyValue data = gen.nextInsert("20201110", 10, 1L, new int[] {1, 1}, "data"); + KeyValue otherPartition = gen.nextInsert("20201111", 11, 2L, new int[] {2, 2}, "other"); + BinaryRow dataPartition = gen.getPartition(data); + BinaryRow missingPartition = gen.getPartition(otherPartition); + assertThat(missingPartition).isNotEqualTo(dataPartition); + store.commitData(Collections.singletonList(data), s -> dataPartition, kv -> 0); + + try (FileStoreCommitImpl commit = store.newCommit()) { + assertThatThrownBy( + () -> + commit.commit( + indexCommittable( + missingPartition, + "wrong-partition-index", + 0, + 0), + false)) + .hasMessageContaining("Global index row ID existence conflict") + .hasMessageContaining("wrong-partition-index"); + } + + try (FileStoreCommitImpl commit = store.newCommit()) { + assertThatThrownBy( + () -> + commit.commit( + indexCommittable( + dataPartition, 1, "wrong-bucket-index", 0, 0), + false)) + .hasMessageContaining("Global index row ID existence conflict") + .hasMessageContaining("wrong-bucket-index"); + } + } + @Test public void testCommitTwiceWithDifferentKind() throws Exception { TestFileStore store = createStore(false); @@ -1898,11 +1990,20 @@ private FileStoreCommitImpl newCommitWithSnapshotCommit( private ManifestCommittable indexCommittable( BinaryRow partition, String fileName, long rowRangeStart, long rowRangeEnd) { + return indexCommittable(partition, 0, fileName, rowRangeStart, rowRangeEnd); + } + + private ManifestCommittable indexCommittable( + BinaryRow partition, + int bucket, + String fileName, + long rowRangeStart, + long rowRangeEnd) { ManifestCommittable committable = new ManifestCommittable(0); committable.addFileCommittable( new CommitMessageImpl( partition, - 0, + bucket, null, DataIncrement.indexIncrement( Collections.singletonList(