@Test public void testPendingCompactionWithDuplicateFileIdsAcrossPartitions() throws Exception { String partitionPath1 = "2016/05/01"; String partitionPath2 = "2016/05/02"; String partitionPath3 = "2016/05/03"; String fullPartitionPath1 = basePath + "/" + partitionPath1 + "/"; new File(fullPartitionPath1).mkdirs(); String fullPartitionPath2 = basePath + "/" + partitionPath2 + "/"; new File(fullPartitionPath2).mkdirs(); String fullPartitionPath3 = basePath + "/" + partitionPath3 + "/"; new File(fullPartitionPath3).mkdirs(); String instantTime1 = "1"; String deltaInstantTime1 = "2"; String deltaInstantTime2 = "3"; String fileId = UUID.randomUUID().toString(); String dataFileName = FSUtils.makeDataFileName(instantTime1, TEST_WRITE_TOKEN, fileId); new File(fullPartitionPath1 + dataFileName).createNewFile(); String fileName1 = FSUtils.makeLogFileName(fileId, HoodieLogFile.DELTA_EXTENSION, instantTime1, 0, TEST_WRITE_TOKEN); new File(fullPartitionPath1 + fileName1).createNewFile(); new File(fullPartitionPath2 + FSUtils.makeDataFileName(instantTime1, TEST_WRITE_TOKEN, fileId)).createNewFile(); new File(fullPartitionPath2 + fileName1).createNewFile(); new File(fullPartitionPath3 + FSUtils.makeDataFileName(instantTime1, TEST_WRITE_TOKEN, fileId)).createNewFile(); new File(fullPartitionPath3 + fileName1).createNewFile(); HoodieActiveTimeline commitTimeline = metaClient.getActiveTimeline(); HoodieInstant instant1 = new HoodieInstant(true, HoodieTimeline.COMMIT_ACTION, instantTime1); HoodieInstant deltaInstant2 = new HoodieInstant(true, HoodieTimeline.DELTA_COMMIT_ACTION, deltaInstantTime1); HoodieInstant deltaInstant3 = new HoodieInstant(true, HoodieTimeline.DELTA_COMMIT_ACTION, deltaInstantTime2); saveAsComplete(commitTimeline, instant1, Option.empty()); saveAsComplete(commitTimeline, deltaInstant2, Option.empty()); saveAsComplete(commitTimeline, deltaInstant3, Option.empty()); FileStatus[] statuses = metaClient.getFs().listStatus( new Path[] {new Path(fullPartitionPath1), new Path(fullPartitionPath2), new Path(fullPartitionPath3)}); assertEquals(6, statuses.length); refreshFsView(); Arrays.asList(partitionPath1, partitionPath2, partitionPath3).forEach(p -> fsView.getAllFileGroups(p).count()); List<HoodieFileGroup> groups = Stream.of(partitionPath1, partitionPath2, partitionPath3) .flatMap(p -> fsView.getAllFileGroups(p)).collect(Collectors.toList()); assertEquals(3, groups.size(), "Expected number of file-groups"); assertEquals(3, groups.stream().map(HoodieFileGroup::getPartitionPath).collect(Collectors.toSet()).size(), "Partitions must be different for file-groups"); Set<String> fileIds = groups.stream().map(HoodieFileGroup::getFileGroupId).map(HoodieFileGroupId::getFileId) .collect(Collectors.toSet()); assertEquals(1, fileIds.size(), "File Id must be same"); assertTrue(fileIds.contains(fileId), "Expected FileId"); List<Pair<String, FileSlice>> partitionFileSlicesPairs = new ArrayList<>(); List<FileSlice> fileSlices = rtView.getLatestFileSlices(partitionPath1).collect(Collectors.toList()); partitionFileSlicesPairs.add(Pair.of(partitionPath1, fileSlices.get(0))); fileSlices = rtView.getLatestFileSlices(partitionPath2).collect(Collectors.toList()); partitionFileSlicesPairs.add(Pair.of(partitionPath2, fileSlices.get(0))); fileSlices = rtView.getLatestFileSlices(partitionPath3).collect(Collectors.toList()); partitionFileSlicesPairs.add(Pair.of(partitionPath3, fileSlices.get(0))); String compactionRequestedTime = "2"; String compactDataFileName = FSUtils.makeDataFileName(compactionRequestedTime, TEST_WRITE_TOKEN, fileId); HoodieCompactionPlan compactionPlan = CompactionUtils.buildFromFileSlices(partitionFileSlicesPairs, Option.empty(), Option.empty()); new File(basePath + "/" + partitionPath1 + "/" + compactDataFileName).createNewFile(); new File(basePath + "/" + partitionPath2 + "/" + compactDataFileName).createNewFile(); HoodieInstant compactionInstant = new HoodieInstant(State.INFLIGHT, HoodieTimeline.COMPACTION_ACTION, compactionRequestedTime); HoodieInstant requested = HoodieTimeline.getCompactionRequestedInstant(compactionInstant.getTimestamp()); metaClient.getActiveTimeline().saveToCompactionRequested(requested, TimelineMetadataUtils.serializeCompactionPlan(compactionPlan)); metaClient.getActiveTimeline().transitionCompactionRequestedToInflight(requested); String deltaInstantTime4 = "4"; String deltaInstantTime5 = "6"; String fileName3 = FSUtils.makeLogFileName(fileId, HoodieLogFile.DELTA_EXTENSION, compactionRequestedTime, 0, TEST_WRITE_TOKEN); String fileName4 = FSUtils.makeLogFileName(fileId, HoodieLogFile.DELTA_EXTENSION, compactionRequestedTime, 1, TEST_WRITE_TOKEN); new File(basePath + "/" + partitionPath1 + "/" + fileName3).createNewFile(); new File(basePath + "/" + partitionPath1 + "/" + fileName4).createNewFile(); new File(basePath + "/" + partitionPath2 + "/" + fileName3).createNewFile(); new File(basePath + "/" + partitionPath2 + "/" + fileName4).createNewFile(); new File(basePath + "/" + partitionPath3 + "/" + fileName3).createNewFile(); new File(basePath + "/" + partitionPath3 + "/" + fileName4).createNewFile(); HoodieInstant deltaInstant4 = new HoodieInstant(true, HoodieTimeline.DELTA_COMMIT_ACTION, deltaInstantTime4); HoodieInstant deltaInstant5 = new HoodieInstant(true, HoodieTimeline.DELTA_COMMIT_ACTION, deltaInstantTime5); saveAsComplete(commitTimeline, deltaInstant4, Option.empty()); saveAsComplete(commitTimeline, deltaInstant5, Option.empty()); refreshFsView(); List<HoodieBaseFile> dataFiles = roView.getAllBaseFiles(partitionPath1).collect(Collectors.toList()); assertEquals(1, dataFiles.size(), "One data-file is expected as there is only one file-group"); assertEquals("1", dataFiles.get(0).getCommitTime(), "Expect only valid commit"); dataFiles = roView.getAllBaseFiles(partitionPath2).collect(Collectors.toList()); assertEquals(1, dataFiles.size(), "One data-file is expected as there is only one file-group"); assertEquals("1", dataFiles.get(0).getCommitTime(), "Expect only valid commit"); Arrays.asList(partitionPath1, partitionPath2, partitionPath3).forEach(partitionPath -> { List<FileSlice> fileSliceList = rtView.getLatestMergedFileSlicesBeforeOrOn(partitionPath, deltaInstantTime5).collect(Collectors.toList()); assertEquals(1, fileSliceList.size(), "Expect file-slice to be merged"); FileSlice fileSlice = fileSliceList.get(0); assertEquals(fileId, fileSlice.getFileId()); assertEquals(dataFileName, fileSlice.getBaseFile().get().getFileName(), "Data file must be present"); assertEquals(instantTime1, fileSlice.getBaseInstantTime(), "Base Instant of penultimate file-slice must be base instant"); List<HoodieLogFile> logFiles = fileSlice.getLogFiles().collect(Collectors.toList()); assertEquals(3, logFiles.size(), "Log files must include those after compaction request"); assertEquals(fileName4, logFiles.get(0).getFileName(), "Log File Order check"); assertEquals(fileName3, logFiles.get(1).getFileName(), "Log File Order check"); assertEquals(fileName1, logFiles.get(2).getFileName(), "Log File Order check"); fileSliceList = rtView.getLatestFileSlicesBeforeOrOn(partitionPath, deltaInstantTime5, true).collect(Collectors.toList()); assertEquals(1, fileSliceList.size(), "Expect only one file-id"); fileSlice = fileSliceList.get(0); assertEquals(fileId, fileSlice.getFileId()); assertFalse(fileSlice.getBaseFile().isPresent(), "No data-file expected in latest file-slice"); assertEquals(compactionRequestedTime, fileSlice.getBaseInstantTime(), "Compaction requested instant must be base instant"); logFiles = fileSlice.getLogFiles().collect(Collectors.toList()); assertEquals(2, logFiles.size(), "Log files must include only those after compaction request"); assertEquals(fileName4, logFiles.get(0).getFileName(), "Log File Order check"); assertEquals(fileName3, logFiles.get(1).getFileName(), "Log File Order check"); fileSliceList = rtView.getLatestFileSlicesBeforeOrOn(partitionPath, deltaInstantTime5, false).collect(Collectors.toList()); assertEquals(0, fileSliceList.size(), "Expect empty list as file-id is in pending compaction"); }); assertEquals(3, fsView.getPendingCompactionOperations().count()); Set<String> partitionsInCompaction = fsView.getPendingCompactionOperations().map(Pair::getValue) .map(CompactionOperation::getPartitionPath).collect(Collectors.toSet()); assertEquals(3, partitionsInCompaction.size()); assertTrue(partitionsInCompaction.contains(partitionPath1)); assertTrue(partitionsInCompaction.contains(partitionPath2)); assertTrue(partitionsInCompaction.contains(partitionPath3)); Set<String> fileIdsInCompaction = fsView.getPendingCompactionOperations().map(Pair::getValue) .map(CompactionOperation::getFileId).collect(Collectors.toSet()); assertEquals(1, fileIdsInCompaction.size()); assertTrue(fileIdsInCompaction.contains(fileId)); }
public Stream<HoodieFileGroup> getAllFileGroups() { return fetchAllStoredFileGroups(); }
HoodieTableFileSystemView extends IncrementalTimelineSyncFileSystemView { public Stream<HoodieFileGroup> getAllFileGroups() { return fetchAllStoredFileGroups(); } }
HoodieTableFileSystemView extends IncrementalTimelineSyncFileSystemView { public Stream<HoodieFileGroup> getAllFileGroups() { return fetchAllStoredFileGroups(); } HoodieTableFileSystemView(boolean enableIncrementalTimelineSync);  HoodieTableFileSystemView(HoodieTableMetaClient metaClient, HoodieTimeline visibleActiveTimeline);  HoodieTableFileSystemView(HoodieTableMetaClient metaClient, HoodieTimeline visibleActiveTimeline,
      boolean enableIncrementalTimelineSync);  HoodieTableFileSystemView(HoodieTableMetaClient metaClient, HoodieTimeline visibleActiveTimeline,
      FileStatus[] fileStatuses); }
HoodieTableFileSystemView extends IncrementalTimelineSyncFileSystemView { public Stream<HoodieFileGroup> getAllFileGroups() { return fetchAllStoredFileGroups(); } HoodieTableFileSystemView(boolean enableIncrementalTimelineSync);  HoodieTableFileSystemView(HoodieTableMetaClient metaClient, HoodieTimeline visibleActiveTimeline);  HoodieTableFileSystemView(HoodieTableMetaClient metaClient, HoodieTimeline visibleActiveTimeline,
      boolean enableIncrementalTimelineSync);  HoodieTableFileSystemView(HoodieTableMetaClient metaClient, HoodieTimeline visibleActiveTimeline,
      FileStatus[] fileStatuses); @Override void init(HoodieTableMetaClient metaClient, HoodieTimeline visibleActiveTimeline); void init(HoodieTableMetaClient metaClient, HoodieTimeline visibleActiveTimeline,
      FileStatus[] fileStatuses); Stream<HoodieFileGroup> getAllFileGroups(); @Override Stream<HoodieFileGroup> fetchAllStoredFileGroups(); @Override void close(); @Override boolean isClosed(); }
HoodieTableFileSystemView extends IncrementalTimelineSyncFileSystemView { public Stream<HoodieFileGroup> getAllFileGroups() { return fetchAllStoredFileGroups(); } HoodieTableFileSystemView(boolean enableIncrementalTimelineSync);  HoodieTableFileSystemView(HoodieTableMetaClient metaClient, HoodieTimeline visibleActiveTimeline);  HoodieTableFileSystemView(HoodieTableMetaClient metaClient, HoodieTimeline visibleActiveTimeline,
      boolean enableIncrementalTimelineSync);  HoodieTableFileSystemView(HoodieTableMetaClient metaClient, HoodieTimeline visibleActiveTimeline,
      FileStatus[] fileStatuses); @Override void init(HoodieTableMetaClient metaClient, HoodieTimeline visibleActiveTimeline); void init(HoodieTableMetaClient metaClient, HoodieTimeline visibleActiveTimeline,
      FileStatus[] fileStatuses); Stream<HoodieFileGroup> getAllFileGroups(); @Override Stream<HoodieFileGroup> fetchAllStoredFileGroups(); @Override void close(); @Override boolean isClosed();  }