class Level1Merger extends Thread {

	public static void mergeSort(LevelQueue lq1, LevelQueue lq2, int ways, String dir, short shard)
			throws IOException, ClassNotFoundException {
		boolean hasLevel2MapTable = lq2.size() > 0;
		List<AbstractMapTable> tables = new ArrayList<AbstractMapTable>(ways);
		lq1.getReadLock().lock();
		try {
			Iterator<AbstractMapTable> iter = lq1.descendingIterator();
			for(int i = 0; i < ways - 1; i++) {
				tables.add(iter.next());
			}
			if (hasLevel2MapTable) {
				tables.add(lq2.get(0));
			} else {
				tables.add(iter.next());
			}
		} finally {
			lq1.getReadLock().unlock();
		}

		long expectedInsertions = 0;
		for(AbstractMapTable table : tables) {
			expectedInsertions += table.getAppendedSize();
		}
		if (expectedInsertions > Integer.MAX_VALUE) expectedInsertions = Integer.MAX_VALUE;
		// target table
		AbstractSortedMapTable sortedMapTable = new FCMapTable(dir, shard, SDB.LEVEL2, System.nanoTime(), (int)expectedInsertions);

		PriorityQueue<QueueElement> pq = new PriorityQueue<QueueElement>();
		// build initial heap
		for(AbstractMapTable table : tables) {
			QueueElement qe = new QueueElement();
			qe.sortedMapTable = table;
			qe.size = qe.sortedMapTable.getAppendedSize();
			qe.index = 0;
			qe.queue = new LinkedList<IMapEntry>();
			IMapEntry me = qe.getNextMapEntry();
			if (me != null) {
				qe.key = me.getKey();
				qe.mapEntry = me;
				qe.keyHash = me.getKeyHash();
				pq.add(qe);
			}
		}

		LinkedList<IMapEntry> targetCacheQueue = new LinkedList<IMapEntry>();
		// merge sort
		while(pq.size() > 0) {
			QueueElement qe1 = pq.poll();
			// remove old/stale entries
			while(pq.peek() != null && qe1.keyHash == pq.peek().keyHash && BytesUtil.compare(qe1.key, pq.peek().key) == 0) {
				QueueElement qe2 = pq.poll();
				IMapEntry me = qe2.getNextMapEntry();
				if (me != null) {
					qe2.key = me.getKey();
					qe2.mapEntry = me;
					qe2.keyHash = me.getKeyHash();
					pq.add(qe2);
				}
			}
			// remove deleted or expired entries in final merge sorting
			if (!qe1.mapEntry.isDeleted() && !qe1.mapEntry.isExpired()) {
				targetCacheQueue.add(qe1.mapEntry);
			}
			if (targetCacheQueue.size() >= CACHED_MAP_ENTRIES * DEFAULT_MERGE_WAYS) {
				while(targetCacheQueue.size() > 0) {
					IMapEntry mapEntry = targetCacheQueue.poll();
					byte[] value = mapEntry.getValue();
					// disk space optimization
					if (mapEntry.isExpired()) {
						continue;
					}
					sortedMapTable.appendNew(mapEntry.getKey(), mapEntry.getKeyHash(), value, mapEntry.getTimeToLive(),
							mapEntry.getCreatedTime(), mapEntry.isDeleted(), mapEntry.isCompressed());
				}
			}
			IMapEntry me = qe1.getNextMapEntry();
			if (me != null) {
				qe1.key = me.getKey();
				qe1.mapEntry = me;
				qe1.keyHash = me.getKeyHash();
				pq.add(qe1);
			}
		}

		// remaining cached entries
		while(targetCacheQueue.size() > 0) {
			IMapEntry mapEntry = targetCacheQueue.poll();
			byte[] value = mapEntry.getValue();
			// disk space optimization
			if (mapEntry.isExpired()) {
				continue;
			}
			sortedMapTable.appendNew(mapEntry.getKey(), mapEntry.getKeyHash(), value, mapEntry.getTimeToLive(),
					mapEntry.getCreatedTime(), mapEntry.isDeleted(), mapEntry.isCompressed());
		}

		// persist metadata
		sortedMapTable.reMap();
		sortedMapTable.saveMetadata();

		// switching
		lq1.getWriteLock().lock();
		lq2.getWriteLock().lock();
		try {
			for(int i = 0; i < ways - 1; i++) {
				lq1.removeLast();
			}
			if (hasLevel2MapTable) {
				lq2.removeLast();
			} else {
				lq1.removeLast();
			}
			for(AbstractMapTable table : tables) {
				table.markUsable(false);
			}

			sortedMapTable.markUsable(true);
			lq2.addFirst(sortedMapTable);
		} finally {
			lq2.getWriteLock().unlock();
			lq1.getWriteLock().unlock();
		}

		for(AbstractMapTable table : tables) {
			table.close();
			table.delete();
		}
	}

	public  Level1Merger(SDB sdb, List<LevelQueue> levelQueueList, CountDownLatch countDownLatch, short shard,
                        SDBStats stats);

}