922 lines
31 KiB
Python
922 lines
31 KiB
Python
import os
|
|
import sys
|
|
import shutil
|
|
import gc
|
|
import unittest
|
|
import rocksdb
|
|
from itertools import takewhile
|
|
import struct
|
|
import tempfile
|
|
from rocksdb.merge_operators import UintAddOperator, StringAppendOperator
|
|
|
|
|
|
def int_to_bytes(ob):
|
|
return str(ob).encode('ascii')
|
|
|
|
|
|
class TestHelper(unittest.TestCase):
|
|
|
|
def setUp(self):
|
|
self.db_loc = tempfile.mkdtemp()
|
|
self.addCleanup(self._close_db)
|
|
|
|
def _close_db(self):
|
|
del self.db
|
|
gc.collect()
|
|
if os.path.exists(self.db_loc):
|
|
shutil.rmtree(self.db_loc)
|
|
|
|
|
|
class TestDB(TestHelper):
|
|
def setUp(self):
|
|
TestHelper.setUp(self)
|
|
opts = rocksdb.Options(create_if_missing=True)
|
|
self.db = rocksdb.DB(os.path.join(self.db_loc, "test"), opts)
|
|
|
|
def test_options_used_twice(self):
|
|
if sys.version_info[0] == 3:
|
|
assertRaisesRegex = self.assertRaisesRegex
|
|
else:
|
|
assertRaisesRegex = self.assertRaisesRegexp
|
|
expected = "Options object is already used by another DB"
|
|
with assertRaisesRegex(Exception, expected):
|
|
rocksdb.DB(os.path.join(self.db_loc, "test2"), self.db.options)
|
|
|
|
def test_unicode_path(self):
|
|
name = os.path.join(self.db_loc, b'M\xc3\xbcnchen'.decode('utf8'))
|
|
rocksdb.DB(name, rocksdb.Options(create_if_missing=True))
|
|
self.addCleanup(shutil.rmtree, name)
|
|
self.assertTrue(os.path.isdir(name))
|
|
|
|
def test_get_none(self):
|
|
self.assertIsNone(self.db.get(b'xxx'))
|
|
|
|
def test_put_get(self):
|
|
self.db.put(b"a", b"b")
|
|
self.assertEqual(b"b", self.db.get(b"a"))
|
|
|
|
def test_put_then_get_from_secondary(self):
|
|
secondary_location = os.path.join(self.db_loc, "secondary")
|
|
secondary = rocksdb.DB(
|
|
os.path.join(self.db_loc, "test"),
|
|
rocksdb.Options(create_if_missing=True, max_open_files=-1),
|
|
secondary_name=secondary_location
|
|
)
|
|
self.addCleanup(secondary.close)
|
|
|
|
self.assertIsNone(secondary.get(b"a"))
|
|
self.db.put(b"a", b"b")
|
|
self.assertEqual(b"b", self.db.get(b"a"))
|
|
self.assertIsNone(secondary.get(b"a"))
|
|
secondary.try_catch_up_with_primary()
|
|
self.assertEqual(b"b", secondary.get(b"a"))
|
|
|
|
secondary2_location = os.path.join(self.db_loc, "secondary2")
|
|
secondary2 = rocksdb.DB(
|
|
os.path.join(self.db_loc, "test"),
|
|
rocksdb.Options(create_if_missing=True, max_open_files=-1),
|
|
secondary_name=secondary2_location
|
|
)
|
|
self.addCleanup(secondary2.close)
|
|
|
|
self.assertEqual(b"b", secondary2.get(b"a"))
|
|
self.db.put(b"a", b"c")
|
|
self.assertEqual(b"b", secondary.get(b"a"))
|
|
self.assertEqual(b"b", secondary2.get(b"a"))
|
|
self.assertEqual(b"c", self.db.get(b"a"))
|
|
secondary.try_catch_up_with_primary()
|
|
secondary2.try_catch_up_with_primary()
|
|
self.assertEqual(b"c", secondary.get(b"a"))
|
|
self.assertEqual(b"c", secondary2.get(b"a"))
|
|
|
|
def test_multi_get(self):
|
|
self.db.put(b"a", b"1")
|
|
self.db.put(b"b", b"2")
|
|
self.db.put(b"c", b"3")
|
|
|
|
ret = self.db.multi_get([b'a', b'b', b'c'])
|
|
ref = {b'a': b'1', b'c': b'3', b'b': b'2'}
|
|
self.assertEqual(ref, ret)
|
|
|
|
def test_delete(self):
|
|
self.db.put(b"a", b"b")
|
|
self.assertEqual(b"b", self.db.get(b"a"))
|
|
self.db.delete(b"a")
|
|
self.assertIsNone(self.db.get(b"a"))
|
|
|
|
def test_write_batch(self):
|
|
batch = rocksdb.WriteBatch()
|
|
batch.put(b"key", b"v1")
|
|
batch.delete(b"key")
|
|
batch.put(b"key", b"v2")
|
|
batch.put(b"key", b"v3")
|
|
batch.put(b"a", b"b")
|
|
|
|
self.db.write(batch)
|
|
ref = {b'a': b'b', b'key': b'v3'}
|
|
ret = self.db.multi_get([b'key', b'a'])
|
|
self.assertEqual(ref, ret)
|
|
|
|
def test_write_batch_context(self):
|
|
with self.db.write_batch() as batch:
|
|
batch.put(b"key", b"v1")
|
|
batch.delete(b"key")
|
|
batch.put(b"key", b"v2")
|
|
batch.put(b"key", b"v3")
|
|
batch.put(b"a", b"b")
|
|
|
|
ref = {b'a': b'b', b'key': b'v3'}
|
|
ret = self.db.multi_get([b'key', b'a'])
|
|
self.assertEqual(ref, ret)
|
|
|
|
def test_write_batch_iter(self):
|
|
batch = rocksdb.WriteBatch()
|
|
self.assertEqual([], list(batch))
|
|
|
|
batch.put(b"key1", b"v1")
|
|
batch.put(b"key2", b"v2")
|
|
batch.put(b"key3", b"v3")
|
|
batch.delete(b'a')
|
|
batch.delete(b'key1')
|
|
batch.merge(b'xxx', b'value')
|
|
|
|
it = iter(batch)
|
|
del batch
|
|
ref = [
|
|
('Put', b'key1', b'v1'),
|
|
('Put', b'key2', b'v2'),
|
|
('Put', b'key3', b'v3'),
|
|
('Delete', b'a', b''),
|
|
('Delete', b'key1', b''),
|
|
('Merge', b'xxx', b'value')
|
|
]
|
|
self.assertEqual(ref, list(it))
|
|
|
|
def test_key_may_exists(self):
|
|
self.db.put(b"a", b'1')
|
|
|
|
self.assertEqual((False, None), self.db.key_may_exist(b"x"))
|
|
self.assertEqual((False, None), self.db.key_may_exist(b'x', True))
|
|
self.assertEqual((True, None), self.db.key_may_exist(b'a'))
|
|
self.assertEqual((True, b'1'), self.db.key_may_exist(b'a', True))
|
|
|
|
def test_seek_for_prev(self):
|
|
self.db.put(b'a1', b'a1_value')
|
|
self.db.put(b'a3', b'a3_value')
|
|
self.db.put(b'b1', b'b1_value')
|
|
self.db.put(b'b2', b'b2_value')
|
|
self.db.put(b'c2', b'c2_value')
|
|
self.db.put(b'c4', b'c4_value')
|
|
|
|
self.assertEqual(self.db.get(b'a1'), b'a1_value')
|
|
|
|
it = self.db.iterkeys()
|
|
|
|
it.seek(b'a1')
|
|
self.assertEqual(it.get(), b'a1')
|
|
it.seek(b'a3')
|
|
self.assertEqual(it.get(), b'a3')
|
|
it.seek_for_prev(b'c4')
|
|
self.assertEqual(it.get(), b'c4')
|
|
it.seek_for_prev(b'c3')
|
|
self.assertEqual(it.get(), b'c2')
|
|
|
|
it = self.db.itervalues()
|
|
it.seek(b'a1')
|
|
self.assertEqual(it.get(), b'a1_value')
|
|
it.seek(b'a3')
|
|
self.assertEqual(it.get(), b'a3_value')
|
|
it.seek_for_prev(b'c4')
|
|
self.assertEqual(it.get(), b'c4_value')
|
|
it.seek_for_prev(b'c3')
|
|
self.assertEqual(it.get(), b'c2_value')
|
|
|
|
it = self.db.iteritems()
|
|
it.seek(b'a1')
|
|
self.assertEqual(it.get(), (b'a1', b'a1_value'))
|
|
it.seek(b'a3')
|
|
self.assertEqual(it.get(), (b'a3', b'a3_value'))
|
|
it.seek_for_prev(b'c4')
|
|
self.assertEqual(it.get(), (b'c4', b'c4_value'))
|
|
it.seek_for_prev(b'c3')
|
|
self.assertEqual(it.get(), (b'c2', b'c2_value'))
|
|
|
|
reverse_it = reversed(it)
|
|
it.seek_for_prev(b'c3')
|
|
self.assertEqual(it.get(), (b'c2', b'c2_value'))
|
|
|
|
def test_iter_keys(self):
|
|
for x in range(300):
|
|
self.db.put(int_to_bytes(x), int_to_bytes(x))
|
|
|
|
it = self.db.iterkeys()
|
|
|
|
self.assertEqual([], list(it))
|
|
|
|
it.seek_to_last()
|
|
self.assertEqual([b'99'], list(it))
|
|
|
|
ref = sorted([int_to_bytes(x) for x in range(300)])
|
|
it.seek_to_first()
|
|
self.assertEqual(ref, list(it))
|
|
|
|
it.seek(b'90')
|
|
ref = [
|
|
b'90',
|
|
b'91',
|
|
b'92',
|
|
b'93',
|
|
b'94',
|
|
b'95',
|
|
b'96',
|
|
b'97',
|
|
b'98',
|
|
b'99'
|
|
]
|
|
self.assertEqual(ref, list(it))
|
|
|
|
def test_iter_values(self):
|
|
for x in range(300):
|
|
self.db.put(int_to_bytes(x), int_to_bytes(x * 1000))
|
|
|
|
it = self.db.itervalues()
|
|
|
|
self.assertEqual([], list(it))
|
|
|
|
it.seek_to_last()
|
|
self.assertEqual([b'99000'], list(it))
|
|
|
|
ref = sorted([int_to_bytes(x) for x in range(300)])
|
|
ref = [int_to_bytes(int(x) * 1000) for x in ref]
|
|
it.seek_to_first()
|
|
self.assertEqual(ref, list(it))
|
|
|
|
it.seek(b'90')
|
|
ref = [int_to_bytes(x * 1000) for x in range(90, 100)]
|
|
self.assertEqual(ref, list(it))
|
|
|
|
def test_iter_items(self):
|
|
for x in range(300):
|
|
self.db.put(int_to_bytes(x), int_to_bytes(x * 1000))
|
|
|
|
it = self.db.iteritems()
|
|
|
|
self.assertEqual([], list(it))
|
|
|
|
it.seek_to_last()
|
|
self.assertEqual([(b'99', b'99000')], list(it))
|
|
|
|
ref = sorted([int_to_bytes(x) for x in range(300)])
|
|
ref = [(x, int_to_bytes(int(x) * 1000)) for x in ref]
|
|
it.seek_to_first()
|
|
self.assertEqual(ref, list(it))
|
|
|
|
it.seek(b'90')
|
|
ref = [(int_to_bytes(x), int_to_bytes(x * 1000)) for x in range(90, 100)]
|
|
self.assertEqual(ref, list(it))
|
|
|
|
def test_reverse_iter(self):
|
|
for x in range(100):
|
|
self.db.put(int_to_bytes(x), int_to_bytes(x * 1000))
|
|
|
|
it = self.db.iteritems()
|
|
it.seek_to_last()
|
|
|
|
ref = reversed(sorted([int_to_bytes(x) for x in range(100)]))
|
|
ref = [(x, int_to_bytes(int(x) * 1000)) for x in ref]
|
|
|
|
self.assertEqual(ref, list(reversed(it)))
|
|
|
|
def test_snapshot(self):
|
|
self.db.put(b"a", b"1")
|
|
self.db.put(b"b", b"2")
|
|
|
|
snapshot = self.db.snapshot()
|
|
self.db.put(b"a", b"2")
|
|
self.db.delete(b"b")
|
|
|
|
it = self.db.iteritems()
|
|
it.seek_to_first()
|
|
self.assertEqual({b'a': b'2'}, dict(it))
|
|
|
|
it = self.db.iteritems(snapshot=snapshot)
|
|
it.seek_to_first()
|
|
self.assertEqual({b'a': b'1', b'b': b'2'}, dict(it))
|
|
|
|
def test_get_property(self):
|
|
for x in range(300):
|
|
x = int_to_bytes(x)
|
|
self.db.put(x, x)
|
|
|
|
self.assertIsNotNone(self.db.get_property(b'rocksdb.stats'))
|
|
self.assertIsNotNone(self.db.get_property(b'rocksdb.sstables'))
|
|
self.assertIsNotNone(self.db.get_property(b'rocksdb.num-files-at-level0'))
|
|
self.assertIsNone(self.db.get_property(b'does not exsits'))
|
|
|
|
def test_compact_range(self):
|
|
for x in range(10000):
|
|
x = int_to_bytes(x)
|
|
self.db.put(x, x)
|
|
|
|
self.db.compact_range()
|
|
|
|
|
|
class AssocCounter(rocksdb.interfaces.AssociativeMergeOperator):
|
|
def merge(self, key, existing_value, value):
|
|
if existing_value:
|
|
return (True, int_to_bytes(int(existing_value) + int(value)))
|
|
return (True, value)
|
|
|
|
def name(self):
|
|
return b'AssocCounter'
|
|
|
|
|
|
class TestUint64Merge(TestHelper):
|
|
def setUp(self):
|
|
TestHelper.setUp(self)
|
|
opts = rocksdb.Options()
|
|
opts.create_if_missing = True
|
|
opts.merge_operator = UintAddOperator()
|
|
self.db = rocksdb.DB(os.path.join(self.db_loc, 'test'), opts)
|
|
|
|
def test_merge(self):
|
|
self.db.put(b'a', struct.pack('Q', 5566))
|
|
for x in range(1000):
|
|
self.db.merge(b"a", struct.pack('Q', x))
|
|
self.assertEqual(5566 + sum(range(1000)), struct.unpack('Q', self.db.get(b'a'))[0])
|
|
|
|
|
|
# class TestPutMerge(TestHelper):
|
|
# def setUp(self):
|
|
# TestHelper.setUp(self)
|
|
# opts = rocksdb.Options()
|
|
# opts.create_if_missing = True
|
|
# opts.merge_operator = "put"
|
|
# self.db = rocksdb.DB(os.path.join(self.db_loc, 'test'), opts)
|
|
|
|
# def test_merge(self):
|
|
# self.db.put(b'a', b'ccc')
|
|
# self.db.merge(b'a', b'ddd')
|
|
# self.assertEqual(self.db.get(b'a'), 'ddd')
|
|
|
|
# class TestPutV1Merge(TestHelper):
|
|
# def setUp(self):
|
|
# TestHelper.setUp(self)
|
|
# opts = rocksdb.Options()
|
|
# opts.create_if_missing = True
|
|
# opts.merge_operator = "put_v1"
|
|
# self.db = rocksdb.DB(os.path.join(self.db_loc, 'test'), opts)
|
|
|
|
# def test_merge(self):
|
|
# self.db.put(b'a', b'ccc')
|
|
# self.db.merge(b'a', b'ddd')
|
|
# self.assertEqual(self.db.get(b'a'), 'ddd')
|
|
|
|
class TestStringAppendOperatorMerge(TestHelper):
|
|
def setUp(self):
|
|
TestHelper.setUp(self)
|
|
opts = rocksdb.Options()
|
|
opts.create_if_missing = True
|
|
opts.merge_operator = StringAppendOperator()
|
|
self.db = rocksdb.DB(os.path.join(self.db_loc, 'test'), opts)
|
|
|
|
# NOTE(sileht): Raise "Corruption: Error: Could not perform merge." on PY3
|
|
#@unittest.skipIf(sys.version_info[0] == 3,
|
|
# "Unexpected behavior on PY3")
|
|
def test_merge(self):
|
|
self.db.put(b'a', b'ccc')
|
|
self.db.merge(b'a', b'ddd')
|
|
self.assertEqual(self.db.get(b'a'), b'ccc,ddd')
|
|
|
|
# class TestStringMaxOperatorMerge(TestHelper):
|
|
# def setUp(self):
|
|
# TestHelper.setUp(self)
|
|
# opts = rocksdb.Options()
|
|
# opts.create_if_missing = True
|
|
# opts.merge_operator = "max"
|
|
# self.db = rocksdb.DB(os.path.join(self.db_loc, 'test'), opts)
|
|
|
|
# def test_merge(self):
|
|
# self.db.put(b'a', int_to_bytes(55))
|
|
# self.db.merge(b'a', int_to_bytes(56))
|
|
# self.assertEqual(int(self.db.get(b'a')), 56)
|
|
|
|
|
|
class TestAssocMerge(TestHelper):
|
|
def setUp(self):
|
|
TestHelper.setUp(self)
|
|
opts = rocksdb.Options()
|
|
opts.create_if_missing = True
|
|
opts.merge_operator = AssocCounter()
|
|
self.db = rocksdb.DB(os.path.join(self.db_loc, 'test'), opts)
|
|
|
|
def test_merge(self):
|
|
for x in range(1000):
|
|
self.db.merge(b"a", int_to_bytes(x))
|
|
self.assertEqual(sum(range(1000)), int(self.db.get(b'a')))
|
|
|
|
|
|
class FullCounter(rocksdb.interfaces.MergeOperator):
|
|
def name(self):
|
|
return b'fullcounter'
|
|
|
|
def full_merge(self, key, existing_value, operand_list):
|
|
ret = sum([int(x) for x in operand_list])
|
|
if existing_value:
|
|
ret += int(existing_value)
|
|
|
|
return (True, int_to_bytes(ret))
|
|
|
|
def partial_merge(self, key, left, right):
|
|
return (True, int_to_bytes(int(left) + int(right)))
|
|
|
|
|
|
class TestFullMerge(TestHelper):
|
|
def setUp(self):
|
|
TestHelper.setUp(self)
|
|
opts = rocksdb.Options()
|
|
opts.create_if_missing = True
|
|
opts.merge_operator = FullCounter()
|
|
self.db = rocksdb.DB(os.path.join(self.db_loc, 'test'), opts)
|
|
|
|
def test_merge(self):
|
|
for x in range(1000):
|
|
self.db.merge(b"a", int_to_bytes(x))
|
|
self.assertEqual(sum(range(1000)), int(self.db.get(b'a')))
|
|
|
|
|
|
class SimpleComparator(rocksdb.interfaces.Comparator):
|
|
def name(self):
|
|
return b'mycompare'
|
|
|
|
def compare(self, a, b):
|
|
a = int(a)
|
|
b = int(b)
|
|
if a < b:
|
|
return -1
|
|
if a == b:
|
|
return 0
|
|
if a > b:
|
|
return 1
|
|
|
|
|
|
class TestComparator(TestHelper):
|
|
def setUp(self):
|
|
TestHelper.setUp(self)
|
|
opts = rocksdb.Options()
|
|
opts.create_if_missing = True
|
|
opts.comparator = SimpleComparator()
|
|
self.db = rocksdb.DB(os.path.join(self.db_loc, 'test'), opts)
|
|
|
|
def test_compare(self):
|
|
for x in range(1000):
|
|
self.db.put(int_to_bytes(x), int_to_bytes(x))
|
|
|
|
self.assertEqual(b'300', self.db.get(b'300'))
|
|
|
|
class StaticPrefix(rocksdb.interfaces.SliceTransform):
|
|
def name(self):
|
|
return b'static'
|
|
|
|
def transform(self, src):
|
|
return (0, 5)
|
|
|
|
def in_domain(self, src):
|
|
return len(src) >= 5
|
|
|
|
def in_range(self, dst):
|
|
return len(dst) == 5
|
|
|
|
|
|
class TestPrefixExtractor(TestHelper):
|
|
def setUp(self):
|
|
TestHelper.setUp(self)
|
|
opts = rocksdb.Options(create_if_missing=True)
|
|
opts.prefix_extractor = StaticPrefix()
|
|
self.db = rocksdb.DB(os.path.join(self.db_loc, 'test'), opts)
|
|
|
|
def _fill_db(self):
|
|
for x in range(3000):
|
|
keyx = hex(x)[2:].zfill(5).encode('utf8') + b'.x'
|
|
keyy = hex(x)[2:].zfill(5).encode('utf8') + b'.y'
|
|
keyz = hex(x)[2:].zfill(5).encode('utf8') + b'.z'
|
|
self.db.put(keyx, b'x')
|
|
self.db.put(keyy, b'y')
|
|
self.db.put(keyz, b'z')
|
|
|
|
|
|
def test_prefix_iterkeys(self):
|
|
self._fill_db()
|
|
self.assertEqual(b'x', self.db.get(b'00001.x'))
|
|
self.assertEqual(b'y', self.db.get(b'00001.y'))
|
|
self.assertEqual(b'z', self.db.get(b'00001.z'))
|
|
|
|
it = self.db.iterkeys()
|
|
it.seek(b'00002')
|
|
|
|
ref = [b'00002.x', b'00002.y', b'00002.z']
|
|
ret = takewhile(lambda key: key.startswith(b'00002'), it)
|
|
self.assertEqual(ref, list(ret))
|
|
|
|
def test_prefix_iteritems(self):
|
|
self._fill_db()
|
|
|
|
it = self.db.iteritems()
|
|
it.seek(b'00002')
|
|
|
|
ref = {b'00002.z': b'z', b'00002.y': b'y', b'00002.x': b'x'}
|
|
ret = takewhile(lambda item: item[0].startswith(b'00002'), it)
|
|
self.assertEqual(ref, dict(ret))
|
|
|
|
class TestDBColumnFamilies(TestHelper):
|
|
def setUp(self):
|
|
TestHelper.setUp(self)
|
|
opts = rocksdb.Options(create_if_missing=True)
|
|
self.db = rocksdb.DB(
|
|
os.path.join(self.db_loc, 'test'),
|
|
opts,
|
|
)
|
|
|
|
self.cf_a = self.db.create_column_family(b'A', rocksdb.ColumnFamilyOptions())
|
|
self.cf_b = self.db.create_column_family(b'B', rocksdb.ColumnFamilyOptions())
|
|
|
|
def test_column_families(self):
|
|
families = self.db.column_families
|
|
names = [handle.name for handle in families]
|
|
self.assertEqual([b'default', b'A', b'B'], names)
|
|
for name in names:
|
|
self.assertIn(self.db.get_column_family(name), families)
|
|
|
|
self.assertEqual(
|
|
names,
|
|
rocksdb.list_column_families(
|
|
os.path.join(self.db_loc, 'test'),
|
|
rocksdb.Options(),
|
|
)
|
|
)
|
|
|
|
def test_get_none(self):
|
|
self.assertIsNone(self.db.get(b'k'))
|
|
self.assertIsNone(self.db.get((self.cf_a, b'k')))
|
|
self.assertIsNone(self.db.get((self.cf_b, b'k')))
|
|
|
|
def test_put_get(self):
|
|
key = (self.cf_a, b'k')
|
|
self.db.put(key, b"v")
|
|
self.assertEqual(b"v", self.db.get(key))
|
|
self.assertIsNone(self.db.get(b"k"))
|
|
self.assertIsNone(self.db.get((self.cf_b, b"k")))
|
|
|
|
def test_multi_get(self):
|
|
data = [
|
|
(b'a', b'1default'),
|
|
(b'b', b'2default'),
|
|
(b'c', b'3default'),
|
|
((self.cf_a, b'a'), b'1a'),
|
|
((self.cf_a, b'b'), b'2a'),
|
|
((self.cf_a, b'c'), b'3a'),
|
|
((self.cf_b, b'a'), b'1b'),
|
|
((self.cf_b, b'b'), b'2b'),
|
|
((self.cf_b, b'c'), b'3b'),
|
|
]
|
|
for value in data:
|
|
self.db.put(*value)
|
|
|
|
multi_get_lookup = [value[0] for value in data]
|
|
|
|
ret = self.db.multi_get(multi_get_lookup)
|
|
ref = {value[0]: value[1] for value in data}
|
|
self.assertEqual(ref, ret)
|
|
|
|
def test_delete(self):
|
|
self.db.put((self.cf_a, b"a"), b"b")
|
|
self.assertEqual(b"b", self.db.get((self.cf_a, b"a")))
|
|
self.db.delete((self.cf_a, b"a"))
|
|
self.assertIsNone(self.db.get((self.cf_a, b"a")))
|
|
|
|
def test_write_batch(self):
|
|
cfa = self.db.get_column_family(b"A")
|
|
batch = rocksdb.WriteBatch()
|
|
batch.put((cfa, b"key"), b"v1")
|
|
batch.delete((self.cf_a, b"key"))
|
|
batch.put((cfa, b"key"), b"v2")
|
|
batch.put((cfa, b"key"), b"v3")
|
|
batch.put((cfa, b"a"), b"1")
|
|
batch.put((cfa, b"b"), b"2")
|
|
|
|
self.db.write(batch)
|
|
query = [(cfa, b"key"), (cfa, b"a"), (cfa, b"b")]
|
|
ret = self.db.multi_get(query)
|
|
|
|
self.assertEqual(b"v3", ret[query[0]])
|
|
self.assertEqual(b"1", ret[query[1]])
|
|
self.assertEqual(b"2", ret[query[2]])
|
|
|
|
def test_key_may_exists(self):
|
|
self.db.put((self.cf_a, b"a"), b'1')
|
|
|
|
self.assertEqual(
|
|
(False, None),
|
|
self.db.key_may_exist((self.cf_a, b"x"))
|
|
)
|
|
self.assertEqual(
|
|
(False, None),
|
|
self.db.key_may_exist((self.cf_a, b'x'), fetch=True)
|
|
)
|
|
self.assertEqual(
|
|
(True, None),
|
|
self.db.key_may_exist((self.cf_a, b'a'))
|
|
)
|
|
self.assertEqual(
|
|
(True, b'1'),
|
|
self.db.key_may_exist((self.cf_a, b'a'), fetch=True)
|
|
)
|
|
|
|
def test_iter_keys(self):
|
|
for x in range(300):
|
|
self.db.put((self.cf_a, int_to_bytes(x)), int_to_bytes(x))
|
|
|
|
it = self.db.iterkeys(self.cf_a)
|
|
self.assertEqual([], list(it))
|
|
|
|
it.seek_to_last()
|
|
self.assertEqual([(self.cf_a, b'99')], list(it))
|
|
|
|
ref = sorted([(self.cf_a, int_to_bytes(x)) for x in range(300)])
|
|
it.seek_to_first()
|
|
self.assertEqual(ref, list(it))
|
|
|
|
it.seek(b'90')
|
|
ref = sorted([(self.cf_a, int_to_bytes(x)) for x in range(90, 100)])
|
|
self.assertEqual(ref, list(it))
|
|
|
|
def test_iter_values(self):
|
|
for x in range(300):
|
|
self.db.put((self.cf_b, int_to_bytes(x)), int_to_bytes(x * 1000))
|
|
|
|
it = self.db.itervalues(self.cf_b)
|
|
self.assertEqual([], list(it))
|
|
|
|
it.seek_to_last()
|
|
self.assertEqual([b'99000'], list(it))
|
|
|
|
ref = sorted([int_to_bytes(x) for x in range(300)])
|
|
ref = [int_to_bytes(int(x) * 1000) for x in ref]
|
|
it.seek_to_first()
|
|
self.assertEqual(ref, list(it))
|
|
|
|
it.seek(b'90')
|
|
ref = [int_to_bytes(x * 1000) for x in range(90, 100)]
|
|
self.assertEqual(ref, list(it))
|
|
|
|
def test_iter_items(self):
|
|
for x in range(300):
|
|
self.db.put((self.cf_b, int_to_bytes(x)), int_to_bytes(x * 1000))
|
|
|
|
it = self.db.iteritems(self.cf_b)
|
|
self.assertEqual([], list(it))
|
|
|
|
it.seek_to_last()
|
|
self.assertEqual([((self.cf_b, b'99'), b'99000')], list(it))
|
|
|
|
ref = sorted([int_to_bytes(x) for x in range(300)])
|
|
ref = [((self.cf_b, x), int_to_bytes(int(x) * 1000)) for x in ref]
|
|
it.seek_to_first()
|
|
self.assertEqual(ref, list(it))
|
|
|
|
it.seek(b'90')
|
|
ref = [((self.cf_b, int_to_bytes(x)), int_to_bytes(x * 1000)) for x in range(90, 100)]
|
|
self.assertEqual(ref, list(it))
|
|
|
|
def test_reverse_iter(self):
|
|
for x in range(100):
|
|
self.db.put((self.cf_a, int_to_bytes(x)), int_to_bytes(x * 1000))
|
|
|
|
it = self.db.iteritems(self.cf_a)
|
|
it.seek_to_last()
|
|
|
|
ref = reversed(sorted([(self.cf_a, int_to_bytes(x)) for x in range(100)]))
|
|
ref = [(x, int_to_bytes(int(x[1]) * 1000)) for x in ref]
|
|
|
|
self.assertEqual(ref, list(reversed(it)))
|
|
|
|
def test_snapshot(self):
|
|
cfa = self.db.get_column_family(b'A')
|
|
self.db.put((cfa, b"a"), b"1")
|
|
self.db.put((cfa, b"b"), b"2")
|
|
|
|
snapshot = self.db.snapshot()
|
|
self.db.put((cfa, b"a"), b"2")
|
|
self.db.delete((cfa, b"b"))
|
|
|
|
it = self.db.iteritems(cfa)
|
|
it.seek_to_first()
|
|
self.assertEqual({(cfa, b'a'): b'2'}, dict(it))
|
|
|
|
it = self.db.iteritems(cfa, snapshot=snapshot)
|
|
it.seek_to_first()
|
|
self.assertEqual({(cfa, b'a'): b'1', (cfa, b'b'): b'2'}, dict(it))
|
|
|
|
def test_get_property(self):
|
|
secondary_location = os.path.join(self.db_loc, "secondary")
|
|
cf = {
|
|
b'A': rocksdb.ColumnFamilyOptions(),
|
|
b'B': rocksdb.ColumnFamilyOptions()
|
|
}
|
|
secondary = rocksdb.DB(
|
|
os.path.join(self.db_loc, "test"),
|
|
rocksdb.Options(create_if_missing=True, max_open_files=-1),
|
|
secondary_name=secondary_location, column_families=cf
|
|
)
|
|
self.addCleanup(secondary.close)
|
|
|
|
for x in range(300):
|
|
x = int_to_bytes(x)
|
|
self.db.put((self.cf_a, x), x)
|
|
|
|
self.assertIsNone(self.db.get_property(b'does not exsits', self.cf_a))
|
|
self.assertEqual(b"0", secondary.get_property(b'rocksdb.estimate-num-keys', secondary.get_column_family(b'A')))
|
|
self.assertEqual(b"300", self.db.get_property(b'rocksdb.estimate-num-keys', self.cf_a))
|
|
|
|
secondary.try_catch_up_with_primary()
|
|
|
|
self.assertEqual(b"300", secondary.get_property(b'rocksdb.estimate-num-keys', secondary.get_column_family(b'A')))
|
|
self.assertEqual(b"300", self.db.get_property(b'rocksdb.estimate-num-keys', self.cf_a))
|
|
|
|
def test_compact_range(self):
|
|
for x in range(10000):
|
|
x = int_to_bytes(x)
|
|
self.db.put((self.cf_b, x), x)
|
|
|
|
self.db.compact_range(column_family=self.cf_b)
|
|
|
|
|
|
class OneCharacterPrefix(rocksdb.interfaces.SliceTransform):
|
|
def name(self):
|
|
return b'test prefix'
|
|
|
|
def transform(self, src):
|
|
return (0, 1)
|
|
|
|
def in_domain(self, src):
|
|
return len(src) >= 1
|
|
|
|
def in_range(self, dst):
|
|
return len(dst) == 1
|
|
|
|
|
|
class TestPrefixIterator(TestHelper):
|
|
def setUp(self):
|
|
TestHelper.setUp(self)
|
|
opts = rocksdb.Options(create_if_missing=True)
|
|
self.db = rocksdb.DB(os.path.join(self.db_loc, 'test'), opts)
|
|
|
|
def test_iterator(self):
|
|
self.db.put(b'a0', b'a0_value')
|
|
self.db.put(b'a1', b'a1_value')
|
|
self.db.put(b'a1b', b'a1b_value')
|
|
self.db.put(b'a2b', b'a2b_value')
|
|
self.db.put(b'a3', b'a3_value')
|
|
self.db.put(b'a4', b'a4_value')
|
|
self.db.put(b'b0', b'b0_value')
|
|
self.assertListEqual(
|
|
[(b'a0', b'a0_value'), (b'a1', b'a1_value'), (b'a1b', b'a1b_value'), (b'a2b', b'a2b_value'),
|
|
(b'a3', b'a3_value'), (b'a4', b'a4_value')],
|
|
list(self.db.iterator(start=b'a', iterate_upper_bound=b'b'))
|
|
)
|
|
self.assertListEqual(
|
|
[b'a0', b'a1', b'a1b', b'a2b', b'a3', b'a4'],
|
|
list(self.db.iterator(start=b'a', iterate_upper_bound=b'b', include_value=False))
|
|
)
|
|
self.assertListEqual(
|
|
[b'a0', b'a1', b'a1b', b'a2b', b'a3', b'a4'],
|
|
list(self.db.iterator(start=b'a0', iterate_upper_bound=b'a5', include_value=False))
|
|
)
|
|
self.assertListEqual(
|
|
[b'a4', b'a3', b'a2b', b'a1b', b'a1', b'a0'],
|
|
list(reversed(self.db.iterator(start=b'a0', iterate_upper_bound=b'a5', include_value=False)))
|
|
)
|
|
self.assertListEqual(
|
|
[b'a0', b'a1', b'a1b', b'a2b', b'a3'],
|
|
list(self.db.iterator(start=b'a0', iterate_upper_bound=b'a4', include_value=False))
|
|
)
|
|
self.assertListEqual(
|
|
[b'a0', b'a1', b'a1b'],
|
|
list(self.db.iterator(start=b'a0', iterate_upper_bound=b'a2', include_value=False))
|
|
)
|
|
self.assertListEqual(
|
|
[b'a0', b'a1', b'a1b'],
|
|
list(self.db.iterator(start=b'a0', iterate_upper_bound=b'a2', include_value=False))
|
|
)
|
|
self.assertListEqual(
|
|
[b'a1b', b'a1', b'a0'],
|
|
list(reversed(self.db.iterator(start=b'a0', iterate_upper_bound=b'a2', include_value=False)))
|
|
)
|
|
self.assertListEqual(
|
|
[b'a0', b'a1', b'a1b', b'a2b', b'a3', b'a4'],
|
|
list(self.db.iterator(start=b'a', iterate_upper_bound=b'b0', include_value=False))
|
|
)
|
|
|
|
|
|
class TestPrefixIteratorWithExtractor(TestHelper):
|
|
def setUp(self):
|
|
TestHelper.setUp(self)
|
|
opts = rocksdb.Options(create_if_missing=True)
|
|
opts.prefix_extractor = OneCharacterPrefix()
|
|
self.db = rocksdb.DB(os.path.join(self.db_loc, 'test'), opts)
|
|
|
|
def test_iterator(self):
|
|
self.db.put(b'a0', b'a0_value')
|
|
self.db.put(b'a1', b'a1_value')
|
|
self.db.put(b'a1b', b'a1b_value')
|
|
self.db.put(b'a2b', b'a2b_value')
|
|
self.db.put(b'a3', b'a3_value')
|
|
self.db.put(b'a4', b'a4_value')
|
|
self.db.put(b'b0', b'b0_value')
|
|
self.assertListEqual(
|
|
[(b'a0', b'a0_value'), (b'a1', b'a1_value'), (b'a1b', b'a1b_value'), (b'a2b', b'a2b_value'),
|
|
(b'a3', b'a3_value'), (b'a4', b'a4_value')],
|
|
list(self.db.iterator(start=b'a', prefix_same_as_start=True))
|
|
)
|
|
self.assertListEqual(
|
|
[b'a0', b'a1', b'a1b', b'a2b', b'a3', b'a4'],
|
|
list(self.db.iterator(start=b'a', include_value=False, prefix_same_as_start=True))
|
|
)
|
|
self.assertListEqual(
|
|
[b'a0', b'a1', b'a1b', b'a2b', b'a3', b'a4'],
|
|
list(self.db.iterator(start=b'a0', iterate_upper_bound=b'a5', include_value=False))
|
|
)
|
|
self.assertListEqual(
|
|
[b'a4', b'a3', b'a2b', b'a1b', b'a1', b'a0'],
|
|
list(reversed(self.db.iterator(start=b'a0', iterate_upper_bound=b'a5', include_value=False)))
|
|
)
|
|
self.assertListEqual(
|
|
[b'a0', b'a1', b'a1b', b'a2b', b'a3'],
|
|
list(self.db.iterator(start=b'a0', iterate_upper_bound=b'a4', include_value=False))
|
|
)
|
|
self.assertListEqual(
|
|
[b'a0', b'a1', b'a1b'],
|
|
list(self.db.iterator(start=b'a0', iterate_upper_bound=b'a2', include_value=False))
|
|
)
|
|
self.assertListEqual(
|
|
[b'a0', b'a1', b'a1b'],
|
|
list(self.db.iterator(start=b'a0', iterate_upper_bound=b'a2', include_value=False))
|
|
)
|
|
self.assertListEqual(
|
|
[b'a1b', b'a1', b'a0'],
|
|
list(reversed(self.db.iterator(start=b'a0', iterate_upper_bound=b'a2', include_value=False)))
|
|
)
|
|
self.assertListEqual(
|
|
[b'a0', b'a1', b'a1b', b'a2b', b'a3', b'a4'],
|
|
list(self.db.iterator(start=b'a', iterate_upper_bound=b'b0', include_value=False))
|
|
)
|
|
|
|
def test_column_family_iterator(self):
|
|
cf_a = self.db.create_column_family(b'first', rocksdb.ColumnFamilyOptions())
|
|
cf_b = self.db.create_column_family(b'second', rocksdb.ColumnFamilyOptions())
|
|
|
|
self.db.put((cf_a, b'a0'), b'a0_value')
|
|
self.db.put((cf_a, b'a1'), b'a1_value')
|
|
self.db.put((cf_a, b'a1b'), b'a1b_value')
|
|
self.db.put((cf_a, b'a2b'), b'a2b_value')
|
|
self.db.put((cf_a, b'a3'), b'a3_value')
|
|
self.db.put((cf_a, b'a4'), b'a4_value')
|
|
self.db.put((cf_b, b'b0'), b'b0_value')
|
|
|
|
self.assertListEqual(
|
|
[(b'a0', b'a0_value'), (b'a1', b'a1_value'), (b'a1b', b'a1b_value'), (b'a2b', b'a2b_value'),
|
|
(b'a3', b'a3_value'), (b'a4', b'a4_value')],
|
|
list(map(lambda x: (x[0][-1], x[1]), self.db.iterator(column_family=cf_a, start=b'a', prefix_same_as_start=True)))
|
|
)
|
|
self.assertListEqual(
|
|
[b'a0', b'a1', b'a1b', b'a2b', b'a3', b'a4'],
|
|
list(map(lambda x: x[-1], self.db.iterator(column_family=cf_a, start=b'a', include_value=False, prefix_same_as_start=True)))
|
|
)
|
|
self.assertListEqual(
|
|
[b'a0', b'a1', b'a1b', b'a2b', b'a3', b'a4'],
|
|
list(map(lambda x: x[-1], self.db.iterator(column_family=cf_a, start=b'a0', iterate_upper_bound=b'a5', include_value=False)))
|
|
)
|
|
self.assertListEqual(
|
|
[b'a4', b'a3', b'a2b', b'a1b', b'a1', b'a0'],
|
|
list(map(lambda x: x[-1],
|
|
reversed(self.db.iterator(
|
|
column_family=cf_a, start=b'a0', iterate_upper_bound=b'a5', include_value=False
|
|
))))
|
|
)
|
|
self.assertListEqual(
|
|
[b'a0', b'a1', b'a1b', b'a2b', b'a3'],
|
|
list(map(lambda x: x[-1], self.db.iterator(column_family=cf_a, start=b'a0', iterate_upper_bound=b'a4', include_value=False)))
|
|
)
|
|
self.assertListEqual(
|
|
[b'a0', b'a1', b'a1b'],
|
|
list(map(lambda x: x[-1], self.db.iterator(column_family=cf_a, start=b'a0', iterate_upper_bound=b'a2', include_value=False)))
|
|
)
|
|
self.assertListEqual(
|
|
[b'a1b', b'a1', b'a0'],
|
|
list(map(lambda x: x[-1], reversed(
|
|
self.db.iterator(column_family=cf_a, start=b'a0', iterate_upper_bound=b'a2', include_value=False))))
|
|
)
|
|
self.assertListEqual(
|
|
[b'b0'],
|
|
list(map(lambda x: x[-1], self.db.iterator(column_family=cf_b, start=b'b', include_value=False)))
|
|
)
|