@Test public void pop() throws KitDBException { String head = "pop0"; RSet set = db.getSet(); int num = 10 * 10000; try { for (int i = 0; i < num; i++) { set.add(head, ("hello world" + i).getBytes()); } Set<String> strings = new HashSet<>(); int n = num / 1000; for (int i = 0; i < n; i++) { List<byte[]> list = set.pop(head, 1000); Assert.assertEquals(num - 1000 * (i + 1), set.size(head)); for (byte[] bytes : list) { strings.add(new String(bytes)); } } Assert.assertEquals(num, strings.size()); for (int i = 0; i < num; i++) { Assert.assertTrue(strings.contains("hello world" + i)); } } finally { set.delete(head); } }
public List<byte[]> pop(String key, int num) throws KitDBException { checkTxStart(); try (CloseLock ignored = checkClose()) { LockEntity lockEntity = lock(key); try { List<byte[]> values = new ArrayList<>(); byte[] key_b = getKey(key); MetaV metaV = getMeta(key_b); if (metaV == null) { checkTxCommit(); return values; } try (final RocksIterator iterator = newIterator(SstColumnFamily.DEFAULT)) { start(); List<byte[]> dels = new ArrayList<>(); SData sData = new SData(key_b.length, key_b, metaV.getVersion(), ArrayKits.intToBytes(0)); byte[] head = sData.getHead(); iterator.seek(head); int count = 0; while (iterator.isValid() && count++ < num) { byte[] key_bs = iterator.key(); if (!BytesUtil.checkHead(head, key_bs)) return values; SDataD sDataD = SDataD.build(key_bs); values.add(sDataD.getValue()); dels.add(key_bs); iterator.next(); } removeDo(metaV, dels); putDB(key_b, metaV.convertMetaBytes().toBytes(), SstColumnFamily.META); commit(); checkTxCommit(); return values; } } finally { unlock(lockEntity); release(); } } catch (KitDBException e) { checkTxRollBack(); throw e; } }
RSet extends RCollection { public List<byte[]> pop(String key, int num) throws KitDBException { checkTxStart(); try (CloseLock ignored = checkClose()) { LockEntity lockEntity = lock(key); try { List<byte[]> values = new ArrayList<>(); byte[] key_b = getKey(key); MetaV metaV = getMeta(key_b); if (metaV == null) { checkTxCommit(); return values; } try (final RocksIterator iterator = newIterator(SstColumnFamily.DEFAULT)) { start(); List<byte[]> dels = new ArrayList<>(); SData sData = new SData(key_b.length, key_b, metaV.getVersion(), ArrayKits.intToBytes(0)); byte[] head = sData.getHead(); iterator.seek(head); int count = 0; while (iterator.isValid() && count++ < num) { byte[] key_bs = iterator.key(); if (!BytesUtil.checkHead(head, key_bs)) return values; SDataD sDataD = SDataD.build(key_bs); values.add(sDataD.getValue()); dels.add(key_bs); iterator.next(); } removeDo(metaV, dels); putDB(key_b, metaV.convertMetaBytes().toBytes(), SstColumnFamily.META); commit(); checkTxCommit(); return values; } } finally { unlock(lockEntity); release(); } } catch (KitDBException e) { checkTxRollBack(); throw e; } } }
RSet extends RCollection { public List<byte[]> pop(String key, int num) throws KitDBException { checkTxStart(); try (CloseLock ignored = checkClose()) { LockEntity lockEntity = lock(key); try { List<byte[]> values = new ArrayList<>(); byte[] key_b = getKey(key); MetaV metaV = getMeta(key_b); if (metaV == null) { checkTxCommit(); return values; } try (final RocksIterator iterator = newIterator(SstColumnFamily.DEFAULT)) { start(); List<byte[]> dels = new ArrayList<>(); SData sData = new SData(key_b.length, key_b, metaV.getVersion(), ArrayKits.intToBytes(0)); byte[] head = sData.getHead(); iterator.seek(head); int count = 0; while (iterator.isValid() && count++ < num) { byte[] key_bs = iterator.key(); if (!BytesUtil.checkHead(head, key_bs)) return values; SDataD sDataD = SDataD.build(key_bs); values.add(sDataD.getValue()); dels.add(key_bs); iterator.next(); } removeDo(metaV, dels); putDB(key_b, metaV.convertMetaBytes().toBytes(), SstColumnFamily.META); commit(); checkTxCommit(); return values; } } finally { unlock(lockEntity); release(); } } catch (KitDBException e) { checkTxRollBack(); throw e; } } protected  RSet(DB db); }
RSet extends RCollection { public List<byte[]> pop(String key, int num) throws KitDBException { checkTxStart(); try (CloseLock ignored = checkClose()) { LockEntity lockEntity = lock(key); try { List<byte[]> values = new ArrayList<>(); byte[] key_b = getKey(key); MetaV metaV = getMeta(key_b); if (metaV == null) { checkTxCommit(); return values; } try (final RocksIterator iterator = newIterator(SstColumnFamily.DEFAULT)) { start(); List<byte[]> dels = new ArrayList<>(); SData sData = new SData(key_b.length, key_b, metaV.getVersion(), ArrayKits.intToBytes(0)); byte[] head = sData.getHead(); iterator.seek(head); int count = 0; while (iterator.isValid() && count++ < num) { byte[] key_bs = iterator.key(); if (!BytesUtil.checkHead(head, key_bs)) return values; SDataD sDataD = SDataD.build(key_bs); values.add(sDataD.getValue()); dels.add(key_bs); iterator.next(); } removeDo(metaV, dels); putDB(key_b, metaV.convertMetaBytes().toBytes(), SstColumnFamily.META); commit(); checkTxCommit(); return values; } } finally { unlock(lockEntity); release(); } } catch (KitDBException e) { checkTxRollBack(); throw e; } } protected  RSet(DB db); List<byte[]> pop(String key, int num); void remove(String key, byte[]... values); boolean contains(String key, byte[] value); void addMayTTL(String key, int ttl, byte[]... values); void add(String key, byte[]... values); @Override void delete(String key); @Override KeyIterator getKeyIterator(); void deleteFast(String key); @Override @SuppressWarnings("unchecked") RIterator<RSet> iterator(String key); @Override int getTtl(String key); @Override void delTtl(String key); @Override void ttl(String key, int ttl); @Override boolean isExist(String key); @Override int size(String key); @Override Entry getEntry(RocksIterator iterator); }
RSet extends RCollection { public List<byte[]> pop(String key, int num) throws KitDBException { checkTxStart(); try (CloseLock ignored = checkClose()) { LockEntity lockEntity = lock(key); try { List<byte[]> values = new ArrayList<>(); byte[] key_b = getKey(key); MetaV metaV = getMeta(key_b); if (metaV == null) { checkTxCommit(); return values; } try (final RocksIterator iterator = newIterator(SstColumnFamily.DEFAULT)) { start(); List<byte[]> dels = new ArrayList<>(); SData sData = new SData(key_b.length, key_b, metaV.getVersion(), ArrayKits.intToBytes(0)); byte[] head = sData.getHead(); iterator.seek(head); int count = 0; while (iterator.isValid() && count++ < num) { byte[] key_bs = iterator.key(); if (!BytesUtil.checkHead(head, key_bs)) return values; SDataD sDataD = SDataD.build(key_bs); values.add(sDataD.getValue()); dels.add(key_bs); iterator.next(); } removeDo(metaV, dels); putDB(key_b, metaV.convertMetaBytes().toBytes(), SstColumnFamily.META); commit(); checkTxCommit(); return values; } } finally { unlock(lockEntity); release(); } } catch (KitDBException e) { checkTxRollBack(); throw e; } } protected  RSet(DB db); List<byte[]> pop(String key, int num); void remove(String key, byte[]... values); boolean contains(String key, byte[] value); void addMayTTL(String key, int ttl, byte[]... values); void add(String key, byte[]... values); @Override void delete(String key); @Override KeyIterator getKeyIterator(); void deleteFast(String key); @Override @SuppressWarnings("unchecked") RIterator<RSet> iterator(String key); @Override int getTtl(String key); @Override void delTtl(String key); @Override void ttl(String key, int ttl); @Override boolean isExist(String key); @Override int size(String key); @Override Entry getEntry(RocksIterator iterator); final static String HEAD; final static byte[] HEAD_B; final static byte[] HEAD_V_B; }