@Test public void testUnpartitionedDeletes() { DeleteFileIndex index = new DeleteFileIndex( ImmutableMap.of( PartitionSpec.unpartitioned().specId(), PartitionSpec.unpartitioned(), 1, SPEC), new long[] { 3, 5, 5, 6 }, DELETE_FILES, ImmutableMap.of()); Assert.assertArrayEquals("All deletes should apply to seq 0", DELETE_FILES, index.forDataFile(0, UNPARTITIONED_FILE)); Assert.assertArrayEquals("All deletes should apply to seq 3", DELETE_FILES, index.forDataFile(3, UNPARTITIONED_FILE)); Assert.assertArrayEquals("Last 3 deletes should apply to seq 4", Arrays.copyOfRange(DELETE_FILES, 1, 4), index.forDataFile(4, UNPARTITIONED_FILE)); Assert.assertArrayEquals("Last 3 deletes should apply to seq 5", Arrays.copyOfRange(DELETE_FILES, 1, 4), index.forDataFile(5, UNPARTITIONED_FILE)); Assert.assertArrayEquals("Last delete should apply to seq 6", Arrays.copyOfRange(DELETE_FILES, 3, 4), index.forDataFile(6, UNPARTITIONED_FILE)); Assert.assertArrayEquals("No deletes should apply to seq 7", new DataFile[0], index.forDataFile(7, UNPARTITIONED_FILE)); Assert.assertArrayEquals("No deletes should apply to seq 10", new DataFile[0], index.forDataFile(10, UNPARTITIONED_FILE)); DataFile partitionedFileA = FILE_A.copy(); ((BaseFile<?>) partitionedFileA).setSpecId(1); Assert.assertArrayEquals("All global deletes should apply to a partitioned file", DELETE_FILES, index.forDataFile(0, partitionedFileA)); }
DeleteFile[] forDataFile(long sequenceNumber, DataFile file) { Pair<Integer, StructLikeWrapper> partition = partition(file.specId(), file.partition()); Pair<long[], DeleteFile[]> partitionDeletes = sortedDeletesByPartition.get(partition); if (partitionDeletes == null) { return limitBySequenceNumber(sequenceNumber, globalSeqs, globalDeletes); } else if (globalDeletes == null) { return limitBySequenceNumber(sequenceNumber, partitionDeletes.first(), partitionDeletes.second()); } else { return Stream.concat( Stream.of(limitBySequenceNumber(sequenceNumber, globalSeqs, globalDeletes)), Stream.of(limitBySequenceNumber(sequenceNumber, partitionDeletes.first(), partitionDeletes.second())) ).toArray(DeleteFile[]::new); } }
DeleteFileIndex { DeleteFile[] forDataFile(long sequenceNumber, DataFile file) { Pair<Integer, StructLikeWrapper> partition = partition(file.specId(), file.partition()); Pair<long[], DeleteFile[]> partitionDeletes = sortedDeletesByPartition.get(partition); if (partitionDeletes == null) { return limitBySequenceNumber(sequenceNumber, globalSeqs, globalDeletes); } else if (globalDeletes == null) { return limitBySequenceNumber(sequenceNumber, partitionDeletes.first(), partitionDeletes.second()); } else { return Stream.concat( Stream.of(limitBySequenceNumber(sequenceNumber, globalSeqs, globalDeletes)), Stream.of(limitBySequenceNumber(sequenceNumber, partitionDeletes.first(), partitionDeletes.second())) ).toArray(DeleteFile[]::new); } } }
DeleteFileIndex { DeleteFile[] forDataFile(long sequenceNumber, DataFile file) { Pair<Integer, StructLikeWrapper> partition = partition(file.specId(), file.partition()); Pair<long[], DeleteFile[]> partitionDeletes = sortedDeletesByPartition.get(partition); if (partitionDeletes == null) { return limitBySequenceNumber(sequenceNumber, globalSeqs, globalDeletes); } else if (globalDeletes == null) { return limitBySequenceNumber(sequenceNumber, partitionDeletes.first(), partitionDeletes.second()); } else { return Stream.concat( Stream.of(limitBySequenceNumber(sequenceNumber, globalSeqs, globalDeletes)), Stream.of(limitBySequenceNumber(sequenceNumber, partitionDeletes.first(), partitionDeletes.second())) ).toArray(DeleteFile[]::new); } } DeleteFileIndex(Map<Integer, PartitionSpec> specsById, long[] globalSeqs, DeleteFile[] globalDeletes,
                  Map<Pair<Integer, StructLikeWrapper>, Pair<long[], DeleteFile[]>> sortedDeletesByPartition); }
DeleteFileIndex { DeleteFile[] forDataFile(long sequenceNumber, DataFile file) { Pair<Integer, StructLikeWrapper> partition = partition(file.specId(), file.partition()); Pair<long[], DeleteFile[]> partitionDeletes = sortedDeletesByPartition.get(partition); if (partitionDeletes == null) { return limitBySequenceNumber(sequenceNumber, globalSeqs, globalDeletes); } else if (globalDeletes == null) { return limitBySequenceNumber(sequenceNumber, partitionDeletes.first(), partitionDeletes.second()); } else { return Stream.concat( Stream.of(limitBySequenceNumber(sequenceNumber, globalSeqs, globalDeletes)), Stream.of(limitBySequenceNumber(sequenceNumber, partitionDeletes.first(), partitionDeletes.second())) ).toArray(DeleteFile[]::new); } } DeleteFileIndex(Map<Integer, PartitionSpec> specsById, long[] globalSeqs, DeleteFile[] globalDeletes,
                  Map<Pair<Integer, StructLikeWrapper>, Pair<long[], DeleteFile[]>> sortedDeletesByPartition);  }
DeleteFileIndex { DeleteFile[] forDataFile(long sequenceNumber, DataFile file) { Pair<Integer, StructLikeWrapper> partition = partition(file.specId(), file.partition()); Pair<long[], DeleteFile[]> partitionDeletes = sortedDeletesByPartition.get(partition); if (partitionDeletes == null) { return limitBySequenceNumber(sequenceNumber, globalSeqs, globalDeletes); } else if (globalDeletes == null) { return limitBySequenceNumber(sequenceNumber, partitionDeletes.first(), partitionDeletes.second()); } else { return Stream.concat( Stream.of(limitBySequenceNumber(sequenceNumber, globalSeqs, globalDeletes)), Stream.of(limitBySequenceNumber(sequenceNumber, partitionDeletes.first(), partitionDeletes.second())) ).toArray(DeleteFile[]::new); } } DeleteFileIndex(Map<Integer, PartitionSpec> specsById, long[] globalSeqs, DeleteFile[] globalDeletes,
                  Map<Pair<Integer, StructLikeWrapper>, Pair<long[], DeleteFile[]>> sortedDeletesByPartition);   }