@Test public void testMergeOverlappingSingleSplit() { SortedMap<KeyExtent,Files> mappings = new TreeMap<>(); addMapping(mappings, null, "m", "f0"); addMapping(mappings, "m", "s", "f1", "f2"); addMapping(mappings, "p", "s", "f3"); addMapping(mappings, "m", "p", "f4"); addMapping(mappings, "s", null, "f5"); var actual = BulkImport.mergeOverlapping(mappings); SortedMap<KeyExtent,Files> expected = new TreeMap<>(); addMapping(expected, null, "m", "f0"); addMapping(expected, "m", "s", "f1", "f2", "f3", "f4"); addMapping(expected, "s", null, "f5"); assertEquals(expected, actual); }
static SortedMap<KeyExtent,Bulk.Files> mergeOverlapping(SortedMap<KeyExtent,Bulk.Files> mappings) { List<KeyExtent> extents = new ArrayList<>(mappings.keySet()); for (KeyExtent ke : extents) { Set<KeyExtent> overlapping = KeyExtent.findOverlapping(ke, mappings); for (KeyExtent oke : overlapping) { if (ke.equals(oke)) { continue; } if (ke.contains(oke)) { mappings.get(ke).merge(mappings.remove(oke)); } else if (!oke.contains(ke)) { throw new RuntimeException("Error during bulk import: Unable to merge overlapping " + "tablets where neither tablet contains the other. This may be caused by " + "a concurrent merge. Key extents " + oke + " and " + ke + " overlap, but " + "neither contains the other."); } } } return mappings; }
BulkImport implements ImportDestinationArguments, ImportMappingOptions { static SortedMap<KeyExtent,Bulk.Files> mergeOverlapping(SortedMap<KeyExtent,Bulk.Files> mappings) { List<KeyExtent> extents = new ArrayList<>(mappings.keySet()); for (KeyExtent ke : extents) { Set<KeyExtent> overlapping = KeyExtent.findOverlapping(ke, mappings); for (KeyExtent oke : overlapping) { if (ke.equals(oke)) { continue; } if (ke.contains(oke)) { mappings.get(ke).merge(mappings.remove(oke)); } else if (!oke.contains(ke)) { throw new RuntimeException("Error during bulk import: Unable to merge overlapping " + "tablets where neither tablet contains the other. This may be caused by " + "a concurrent merge. Key extents " + oke + " and " + ke + " overlap, but " + "neither contains the other."); } } } return mappings; } }
BulkImport implements ImportDestinationArguments, ImportMappingOptions { static SortedMap<KeyExtent,Bulk.Files> mergeOverlapping(SortedMap<KeyExtent,Bulk.Files> mappings) { List<KeyExtent> extents = new ArrayList<>(mappings.keySet()); for (KeyExtent ke : extents) { Set<KeyExtent> overlapping = KeyExtent.findOverlapping(ke, mappings); for (KeyExtent oke : overlapping) { if (ke.equals(oke)) { continue; } if (ke.contains(oke)) { mappings.get(ke).merge(mappings.remove(oke)); } else if (!oke.contains(ke)) { throw new RuntimeException("Error during bulk import: Unable to merge overlapping " + "tablets where neither tablet contains the other. This may be caused by " + "a concurrent merge. Key extents " + oke + " and " + ke + " overlap, but " + "neither contains the other."); } } } return mappings; } BulkImport(String directory, ClientContext context); }
BulkImport implements ImportDestinationArguments, ImportMappingOptions { static SortedMap<KeyExtent,Bulk.Files> mergeOverlapping(SortedMap<KeyExtent,Bulk.Files> mappings) { List<KeyExtent> extents = new ArrayList<>(mappings.keySet()); for (KeyExtent ke : extents) { Set<KeyExtent> overlapping = KeyExtent.findOverlapping(ke, mappings); for (KeyExtent oke : overlapping) { if (ke.equals(oke)) { continue; } if (ke.contains(oke)) { mappings.get(ke).merge(mappings.remove(oke)); } else if (!oke.contains(ke)) { throw new RuntimeException("Error during bulk import: Unable to merge overlapping " + "tablets where neither tablet contains the other. This may be caused by " + "a concurrent merge. Key extents " + oke + " and " + ke + " overlap, but " + "neither contains the other."); } } } return mappings; } BulkImport(String directory, ClientContext context); @Override ImportMappingOptions tableTime(boolean value); @Override void load(); @Override ImportMappingOptions executor(Executor service); @Override ImportMappingOptions threads(int numThreads); @Override ImportMappingOptions plan(LoadPlan plan); @Override ImportMappingOptions to(String tableName); static Map<KeyExtent,Long> estimateSizes(AccumuloConfiguration acuConf, Path mapFile,
      long fileSize, Collection<KeyExtent> extents, FileSystem ns, Cache<String,Long> fileLenCache,
      CryptoService cs); static List<KeyExtent> findOverlappingTablets(KeyExtentCache extentCache,
      FileSKVIterator reader); static List<KeyExtent> findOverlappingTablets(ClientContext context,
      KeyExtentCache extentCache, Path file, FileSystem fs, Cache<String,Long> fileLenCache,
      CryptoService cs); static List<FileStatus> filterInvalid(FileStatus[] files); SortedMap<KeyExtent,Bulk.Files> computeFileToTabletMappings(FileSystem fs, TableId tableId,
      Path dirPath, Executor executor, ClientContext context, int maxTablets); }
BulkImport implements ImportDestinationArguments, ImportMappingOptions { static SortedMap<KeyExtent,Bulk.Files> mergeOverlapping(SortedMap<KeyExtent,Bulk.Files> mappings) { List<KeyExtent> extents = new ArrayList<>(mappings.keySet()); for (KeyExtent ke : extents) { Set<KeyExtent> overlapping = KeyExtent.findOverlapping(ke, mappings); for (KeyExtent oke : overlapping) { if (ke.equals(oke)) { continue; } if (ke.contains(oke)) { mappings.get(ke).merge(mappings.remove(oke)); } else if (!oke.contains(ke)) { throw new RuntimeException("Error during bulk import: Unable to merge overlapping " + "tablets where neither tablet contains the other. This may be caused by " + "a concurrent merge. Key extents " + oke + " and " + ke + " overlap, but " + "neither contains the other."); } } } return mappings; } BulkImport(String directory, ClientContext context); @Override ImportMappingOptions tableTime(boolean value); @Override void load(); @Override ImportMappingOptions executor(Executor service); @Override ImportMappingOptions threads(int numThreads); @Override ImportMappingOptions plan(LoadPlan plan); @Override ImportMappingOptions to(String tableName); static Map<KeyExtent,Long> estimateSizes(AccumuloConfiguration acuConf, Path mapFile,
      long fileSize, Collection<KeyExtent> extents, FileSystem ns, Cache<String,Long> fileLenCache,
      CryptoService cs); static List<KeyExtent> findOverlappingTablets(KeyExtentCache extentCache,
      FileSKVIterator reader); static List<KeyExtent> findOverlappingTablets(ClientContext context,
      KeyExtentCache extentCache, Path file, FileSystem fs, Cache<String,Long> fileLenCache,
      CryptoService cs); static List<FileStatus> filterInvalid(FileStatus[] files); SortedMap<KeyExtent,Bulk.Files> computeFileToTabletMappings(FileSystem fs, TableId tableId,
      Path dirPath, Executor executor, ClientContext context, int maxTablets);  }