Level0Merger extends Thread { public static void mergeSort(LevelQueue source, LevelQueue target, int ways, String dir, short shard) throws IOException, ClassNotFoundException { List<HashMapTable> tables = new ArrayList<HashMapTable>(ways); source.getReadLock().lock(); try { Iterator<AbstractMapTable> iter = source.descendingIterator(); for(int i = 0; i < ways; i++) { tables.add((HashMapTable) iter.next()); } } finally { source.getReadLock().unlock(); } int expectedInsertions = 0; for(HashMapTable table : tables) { expectedInsertions += table.getRealSize(); } MMFMapTable sortedMapTable = new MMFMapTable(dir, shard, SDB.LEVEL1, System.nanoTime(), expectedInsertions, ways); PriorityQueue<QueueElement> pq = new PriorityQueue<QueueElement>(); for(HashMapTable table : tables) { QueueElement qe = new QueueElement(); final HashMapTable hmTable = table; qe.hashMapTable = hmTable; List<Map.Entry<ByteArrayWrapper, InMemIndex>> list = new ArrayList<Map.Entry<ByteArrayWrapper, InMemIndex>>(qe.hashMapTable.getEntrySet()); Collections.sort(list, new Comparator<Map.Entry<ByteArrayWrapper, InMemIndex>>() { @Override public int compare( Entry<ByteArrayWrapper, InMemIndex> o1, Entry<ByteArrayWrapper, InMemIndex> o2) { IMapEntry mapEntry1 = hmTable.getMapEntry(o1.getValue().getIndex()); IMapEntry mapEntry2 = hmTable.getMapEntry(o2.getValue().getIndex()); try { int hash1 = mapEntry1.getKeyHash(); int hash2 = mapEntry2.getKeyHash(); if (hash1 < hash2) return -1; else if (hash1 > hash2) return 1; else { return o1.getKey().compareTo(o2.getKey()); } } catch (IOException e) { throw new RuntimeException("Fail to get hash code in map entry", e); } } }); qe.iterator = list.iterator(); if (qe.iterator.hasNext()) { Map.Entry<ByteArrayWrapper, InMemIndex> me = qe.iterator.next(); qe.key = me.getKey().getData(); qe.inMemIndex = me.getValue(); IMapEntry mapEntry = table.getMapEntry(qe.inMemIndex.getIndex()); qe.keyHash = mapEntry.getKeyHash(); pq.add(qe); } } while(pq.size() > 0) { QueueElement qe1 = pq.poll(); while(pq.peek() != null && qe1.keyHash == pq.peek().keyHash && BytesUtil.compare(qe1.key, pq.peek().key) == 0) { QueueElement qe2 = pq.poll(); if (qe2.iterator.hasNext()) { Map.Entry<ByteArrayWrapper, InMemIndex> me = qe2.iterator.next(); qe2.key = me.getKey().getData(); qe2.inMemIndex = me.getValue(); IMapEntry mapEntry = qe2.hashMapTable.getMapEntry(qe2.inMemIndex.getIndex()); qe2.keyHash = mapEntry.getKeyHash(); pq.add(qe2); } } IMapEntry mapEntry = qe1.hashMapTable.getMapEntry(qe1.inMemIndex.getIndex()); byte[] value = mapEntry.getValue(); if (mapEntry.isDeleted() || mapEntry.isExpired()) { value = new byte[] {0}; } sortedMapTable.appendNew(mapEntry.getKey(), mapEntry.getKeyHash(), value, mapEntry.getTimeToLive(), mapEntry.getCreatedTime(), mapEntry.isDeleted(), mapEntry.isCompressed()); if (qe1.iterator.hasNext()) { Map.Entry<ByteArrayWrapper, InMemIndex> me = qe1.iterator.next(); qe1.key = me.getKey().getData(); qe1.inMemIndex = me.getValue(); IMapEntry mEntry = qe1.hashMapTable.getMapEntry(qe1.inMemIndex.getIndex()); qe1.keyHash = mEntry.getKeyHash(); pq.add(qe1); } } sortedMapTable.reMap(); sortedMapTable.saveMetadata(); source.getWriteLock().lock(); target.getWriteLock().lock(); try { for(int i = 0; i < ways; i++) { source.removeLast(); } for(HashMapTable table : tables) { table.markUsable(false); } sortedMapTable.markUsable(true); target.addFirst(sortedMapTable); } finally { target.getWriteLock().unlock(); source.getWriteLock().unlock(); } for(HashMapTable table : tables) { table.close(); table.delete(); } } Level0Merger(SDB sdb, List<LevelQueue> levelQueueList, CountDownLatch countDownLatch, short shard,
                        SDBStats stats); @Override void run(); static void mergeSort(LevelQueue source, LevelQueue target, int ways, String dir, short shard); void setStop(); static final int DEFAULT_MERGE_WAYS; }
@Test public void testCase02() throws IOException, ClassNotFoundException { HashMapTable[] sourceTables = new HashMapTable[4]; LevelQueue lq0 = new LevelQueue(); for(int i = 0; i < 4; i++) { sourceTables[i] = new HashMapTable(testDir, SDB.LEVEL0, System.nanoTime() + i); sourceTables[i].setCompressionEnabled(true); lq0.addFirst(sourceTables[i]); assertTrue(sourceTables[i].isImmutable()); } int max = AbstractMapTable.INIT_INDEX_ITEMS_PER_TABLE; HashMapTable table4 = sourceTables[3]; int start = 0; while(start < max) { table4.delete(String.valueOf(start).getBytes()); start = start + 4; } HashMapTable table3 = sourceTables[2]; start = 1; while(start < max) { table3.put(String.valueOf(start).getBytes(), String.valueOf(start).getBytes(), 200, System.currentTimeMillis()); start = start + 4; } try { Thread.sleep(2000); } catch (InterruptedException e) { } HashMapTable table2 = sourceTables[1]; start = 2; while(start < max) { table2.put(String.valueOf(start).getBytes(), String.valueOf(start).getBytes(), 60 * 1000, System.currentTimeMillis()); start = start + 4; } HashMapTable table1 = sourceTables[0]; start = 0; while(start < max) { table1.put(String.valueOf(start).getBytes(), String.valueOf(start).getBytes(), 120 * 1000, System.currentTimeMillis()); start = start + 1; } LevelQueue lq1 = new LevelQueue(); Level0Merger.mergeSort(lq0, lq1, 4, testDir, (short)0); for(int i = 0; i < 4; i++) { assertTrue(sourceTables[i].isImmutable()); } assertTrue(lq1.size() == 1); MMFMapTable targetTable = (MMFMapTable) lq1.poll(); assertTrue(targetTable.getLevel() == SDB.LEVEL1); assertTrue(targetTable.getAppendedSize() == max); start = 0; while(start < max) { GetResult result = targetTable.get(String.valueOf(start).getBytes()); assertTrue(result.isFound()); assertTrue(result.isDeleted()); start += 4; } start = 1; while(start < max) { GetResult result = targetTable.get(String.valueOf(start).getBytes()); assertTrue(result.isFound()); assertTrue(result.isExpired()); start += 4; } start = 2; while(start < max) { GetResult result = targetTable.get(String.valueOf(start).getBytes()); assertTrue(result.isFound()); assertTrue(result.getTimeToLive() == 60 * 1000); start += 4; } start = 3; while(start < max) { GetResult result = targetTable.get(String.valueOf(start).getBytes()); assertTrue(result.isFound()); assertTrue(result.getTimeToLive() == 120 * 1000); start += 4; } List<String> keyList = new ArrayList<String>(); for(long i = 0; i < max; i++) { keyList.add(String.valueOf(i)); } Collections.sort(keyList, new Comparator<String>() { @Override public int compare(String arg0, String arg1) { int hash0 = Arrays.hashCode(arg0.getBytes()); int hash1 = Arrays.hashCode(arg1.getBytes()); if (hash0 < hash1) return -1; else if (hash0 > hash1) return 1; else return 0; } }); for(int i = 0; i < max; i++) { IMapEntry mapEntry = targetTable.getMapEntry(i); assertTrue(mapEntry.getIndex() == i); assertTrue(new String(mapEntry.getKey()).equals(keyList.get(i))); int intKey = Integer.parseInt(new String(mapEntry.getKey())); if (intKey % 4 == 0) assertTrue(Arrays.equals(new byte[] {0}, mapEntry.getValue())); else if (intKey % 4 == 1) assertTrue(Arrays.equals(new byte[] {0}, mapEntry.getValue())); else { assertTrue(new String(Snappy.uncompress(mapEntry.getValue())).equals(keyList.get(i))); } } Random random = new Random(); for(int i = 0; i < 1024; i++) { long key = random.nextInt(max); GetResult result = targetTable.get(String.valueOf(key).getBytes()); assertTrue(result.isFound()); } targetTable.close(); targetTable.delete(); }
@Test public void testCase01() throws IOException, ClassNotFoundException { String value = TestUtil.randomString(128); HashMapTable[] sourceTables = new HashMapTable[4]; LevelQueue lq0 = new LevelQueue(); for(int i = 0; i < 4; i++) { sourceTables[i] = new HashMapTable(testDir, SDB.LEVEL0, System.nanoTime() + i); sourceTables[i].setCompressionEnabled(true); lq0.addFirst(sourceTables[i]); assertTrue(sourceTables[i].isImmutable()); } int totalCount = 0; int max = 0; for(int i = 0; i < 4; i++) { int start = i; HashMapTable table = sourceTables[i]; while(table.put(String.valueOf(start).getBytes(), value.getBytes(), AbstractMapTable.NO_TIMEOUT, System.currentTimeMillis(), false)) { totalCount++; if (start > max) max = start; start = start + 4; } } LevelQueue lq1 = new LevelQueue(); long start = System.currentTimeMillis(); Level0Merger.mergeSort(lq0, lq1, 4, testDir, (short)1); long end = System.currentTimeMillis(); System.out.println("Time spent to merge " + totalCount + " items in 4 ways is " + (end - start) / 1000 + "s"); for(int i = 0; i < 4; i++) { assertTrue(sourceTables[i].isImmutable()); } assertTrue(lq1.size() == 1); MMFMapTable targetTable = (MMFMapTable) lq1.poll(); assertTrue(targetTable.getLevel() == SDB.LEVEL1); assertTrue(targetTable.getAppendedSize() == totalCount); List<String> keyList = new ArrayList<String>(); for(long i = 0; i < totalCount; i++) { keyList.add(String.valueOf(i)); } Collections.sort(keyList, new Comparator<String>() { @Override public int compare(String arg0, String arg1) { int hash0 = Arrays.hashCode(arg0.getBytes()); int hash1 = Arrays.hashCode(arg1.getBytes()); if (hash0 < hash1) return -1; else if (hash0 > hash1) return 1; else return 0; } }); for(int i = 0; i < totalCount; i++) { IMapEntry mapEntry = targetTable.getMapEntry(i); assertTrue(mapEntry.getIndex() == i); assertTrue(new String(mapEntry.getKey()).equals(keyList.get(i))); assertTrue(new String(Snappy.uncompress(mapEntry.getValue())).equals(value)); } start = System.currentTimeMillis(); Random random = new Random(); for(int i = 0; i < 1024; i++) { long key = random.nextInt(totalCount); GetResult result = targetTable.get(String.valueOf(key).getBytes()); assertTrue(result.isFound()); } end = System.currentTimeMillis(); System.out.println("Time to lookup 1024 random key in the target table is " + (end - start) + "ms"); targetTable.close(); targetTable.delete(); }