NextGraph
/
rust-rocksdb
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Fix bad merge.

Browse Source
master
Tyler Neely 9 years ago
parent 0fc2441a2e b262535559
commit 3dc7aa9e3d
7 changed files with 38 additions and 241 deletions
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Show Stats Download Patch File Download Diff File
  1. 3
      src/lib.rs
  2. 9
      src/main.rs
  3. 18
      test/test_column_family.rs
  4. 141
      test/test_column_family.rs.orig
  5. 34
      test/test_iterator.rs
  6. 18
      test/test_multithreaded.rs
  7. 56
      test/test_multithreaded.rs.orig

3
src/lib.rs
Unescape View File

@ -13,9 +13,6 @@
See the License for the specific language governing permissions and See the License for the specific language governing permissions and
limitations under the License. limitations under the License.
*/ */
#![crate_id = "rocksdb"]
#![crate_type = "lib"]
pub use ffi as rocksdb_ffi; pub use ffi as rocksdb_ffi;
pub use ffi::{new_bloom_filter, DBCompactionStyle, DBComparator}; pub use ffi::{new_bloom_filter, DBCompactionStyle, DBComparator};
pub use rocksdb::{DB, DBVector, WriteBatch, Writable, Direction}; pub use rocksdb::{DB, DBVector, WriteBatch, Writable, Direction};

9
src/main.rs
Unescape View File

@ -14,8 +14,7 @@
limitations under the License. limitations under the License.
*/ */
extern crate rocksdb; extern crate rocksdb;
use rocksdb::{Options, DB, MergeOperands, new_bloom_filter, Writable, }; use rocksdb::{Options, DB, MergeOperands, Writable, };
use rocksdb::DBCompactionStyle::DBUniversalCompaction;
//fn snapshot_test() { //fn snapshot_test() {
// let path = "_rust_rocksdb_iteratortest"; // let path = "_rust_rocksdb_iteratortest";
@ -47,7 +46,7 @@ use rocksdb::DBCompactionStyle::DBUniversalCompaction;
#[cfg(not(feature = "valgrind"))] #[cfg(not(feature = "valgrind"))]
fn main() { fn main() {
let path = "/tmp/rust-rocksdb"; let path = "/tmp/rust-rocksdb";
let mut db = DB::open_default(path).unwrap(); let db = DB::open_default(path).unwrap();
assert!(db.put(b"my key", b"my value").is_ok()); assert!(db.put(b"my key", b"my value").is_ok());
match db.get(b"my key") { match db.get(b"my key") {
Ok(Some(value)) => { Ok(Some(value)) => {
@ -67,8 +66,8 @@ fn main() {
custom_merge(); custom_merge();
} }
fn concat_merge(new_key: &[u8], existing_val: Option<&[u8]>, fn concat_merge(_: &[u8], existing_val: Option<&[u8]>,
mut operands: &mut MergeOperands) -> Vec<u8> { operands: &mut MergeOperands) -> Vec<u8> {
let mut result: Vec<u8> = Vec::with_capacity(operands.size_hint().0); let mut result: Vec<u8> = Vec::with_capacity(operands.size_hint().0);
match existing_val { match existing_val {
Some(v) => for e in v { Some(v) => for e in v {

18
test/test_column_family.rs
Unescape View File

@ -13,7 +13,7 @@
See the License for the specific language governing permissions and See the License for the specific language governing permissions and
limitations under the License. limitations under the License.
*/ */
use rocksdb::{Options, DB, Writable, Direction, MergeOperands}; use rocksdb::{Options, DB, Writable, MergeOperands};
#[test] #[test]
pub fn test_column_family() { pub fn test_column_family() {
@ -58,7 +58,7 @@ pub fn test_column_family() {
{ {
let mut opts = Options::new(); let mut opts = Options::new();
opts.add_merge_operator("test operator", test_provided_merge); opts.add_merge_operator("test operator", test_provided_merge);
let mut db = match DB::open_cf(&opts, path, &["cf1"]) { let db = match DB::open_cf(&opts, path, &["cf1"]) {
Ok(db) => { Ok(db) => {
println!("successfully opened db with column family"); println!("successfully opened db with column family");
db db
@ -70,10 +70,10 @@ pub fn test_column_family() {
assert!(db.get_cf(cf1, b"k1").unwrap().unwrap().to_utf8().unwrap() == "v1"); assert!(db.get_cf(cf1, b"k1").unwrap().unwrap().to_utf8().unwrap() == "v1");
let p = db.put_cf(cf1, b"k1", b"a"); let p = db.put_cf(cf1, b"k1", b"a");
assert!(p.is_ok()); assert!(p.is_ok());
db.merge_cf(cf1, b"k1", b"b"); db.merge_cf(cf1, b"k1", b"b").unwrap();
db.merge_cf(cf1, b"k1", b"c"); db.merge_cf(cf1, b"k1", b"c").unwrap();
db.merge_cf(cf1, b"k1", b"d"); db.merge_cf(cf1, b"k1", b"d").unwrap();
db.merge_cf(cf1, b"k1", b"efg"); db.merge_cf(cf1, b"k1", b"efg").unwrap();
let m = db.merge_cf(cf1, b"k1", b"h"); let m = db.merge_cf(cf1, b"k1", b"h");
println!("m is {:?}", m); println!("m is {:?}", m);
// TODO assert!(m.is_ok()); // TODO assert!(m.is_ok());
@ -90,7 +90,7 @@ pub fn test_column_family() {
_ => panic!("value not present!"), _ => panic!("value not present!"),
} }
let r = db.get_cf(cf1, b"k1"); let _ = db.get_cf(cf1, b"k1");
// TODO assert!(r.unwrap().to_utf8().unwrap() == "abcdefgh"); // TODO assert!(r.unwrap().to_utf8().unwrap() == "abcdefgh");
assert!(db.delete(b"k1").is_ok()); assert!(db.delete(b"k1").is_ok());
assert!(db.get(b"k1").unwrap().is_none()); assert!(db.get(b"k1").unwrap().is_none());
@ -113,9 +113,9 @@ pub fn test_column_family() {
assert!(DB::destroy(&Options::new(), path).is_ok()); assert!(DB::destroy(&Options::new(), path).is_ok());
} }
fn test_provided_merge(new_key: &[u8], fn test_provided_merge(_: &[u8],
existing_val: Option<&[u8]>, existing_val: Option<&[u8]>,
mut operands: &mut MergeOperands) operands: &mut MergeOperands)
-> Vec<u8> { -> Vec<u8> {
let nops = operands.size_hint().0; let nops = operands.size_hint().0;
let mut result: Vec<u8> = Vec::with_capacity(nops); let mut result: Vec<u8> = Vec::with_capacity(nops);

141
test/test_column_family.rs.orig
Unescape View File

@ -1,141 +0,0 @@
/*
Copyright 2014 Tyler Neely
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
use rocksdb::{Options, DB, DBResult, Writable, Direction, MergeOperands};
#[test]
pub fn test_column_family() {
let path = "_rust_rocksdb_cftest";
// should be able to create column families
{
let mut opts = Options::new();
opts.create_if_missing(true);
opts.add_merge_operator("test operator", test_provided_merge);
let mut db = DB::open(&opts, path).unwrap();
let opts = Options::new();
match db.create_cf("cf1", &opts) {
Ok(_) => println!("cf1 created successfully"),
Err(e) => {
panic!("could not create column family: {}", e);
},
}
}
// should fail to open db without specifying same column families
{
let mut opts = Options::new();
opts.add_merge_operator("test operator", test_provided_merge);
match DB::open(&opts, path) {
Ok(_) => panic!("should not have opened DB successfully without specifying column
families"),
Err(e) => assert!(e.starts_with("Invalid argument: You have to open all column families.")),
}
}
// should properly open db when specyfing all column families
{
let mut opts = Options::new();
opts.add_merge_operator("test operator", test_provided_merge);
match DB::open_cf(&opts, path, &["cf1"]) {
Ok(_) => println!("successfully opened db with column family"),
Err(e) => panic!("failed to open db with column family: {}", e),
}
}
// TODO should be able to write, read, merge, batch, and iterate over a cf
{
let mut opts = Options::new();
opts.add_merge_operator("test operator", test_provided_merge);
let mut db = match DB::open_cf(&opts, path, &["cf1"]) {
Ok(db) => {
println!("successfully opened db with column family");
db
},
Err(e) => panic!("failed to open db with column family: {}", e),
};
let cf1 = *db.cf_handle("cf1").unwrap();
assert!(db.put_cf(cf1, b"k1", b"v1").is_ok());
assert!(db.get_cf(cf1, b"k1").unwrap().unwrap().to_utf8().unwrap() == "v1");
let p = db.put_cf(cf1, b"k1", b"a");
assert!(p.is_ok());
db.merge_cf(cf1, b"k1", b"b");
db.merge_cf(cf1, b"k1", b"c");
db.merge_cf(cf1, b"k1", b"d");
db.merge_cf(cf1, b"k1", b"efg");
let m = db.merge_cf(cf1, b"k1", b"h");
println!("m is {:?}", m);
// TODO assert!(m.is_ok());
match db.get(b"k1") {
Ok(Some(value)) => {
match value.to_utf8() {
Some(v) =>
println!("retrieved utf8 value: {}", v),
None =>
println!("did not read valid utf-8 out of the db"),
}
<<<<<<< Updated upstream
}).on_absent( || { println!("value not present!") })
.on_error( |e| { println!("error reading value")}); //: {", e) });
=======
},
Err(e) => println!("error reading value"),
_ => panic!("value not present!"),
}
>>>>>>> Stashed changes
let r = db.get_cf(cf1, b"k1");
// TODO assert!(r.unwrap().to_utf8().unwrap() == "abcdefgh");
assert!(db.delete(b"k1").is_ok());
assert!(db.get(b"k1").unwrap().is_none());
}
// TODO should be able to use writebatch ops with a cf
{
}
// TODO should be able to iterate over a cf
{
}
// should b able to drop a cf
{
let mut db = DB::open_cf(&Options::new(), path, &["cf1"]).unwrap();
match db.drop_cf("cf1") {
Ok(_) => println!("cf1 successfully dropped."),
Err(e) => panic!("failed to drop column family: {}", e),
}
}
assert!(DB::destroy(&Options::new(), path).is_ok());
}
fn test_provided_merge(new_key: &[u8],
existing_val: Option<&[u8]>,
mut operands: &mut MergeOperands)
-> Vec<u8> {
let nops = operands.size_hint().0;
let mut result: Vec<u8> = Vec::with_capacity(nops);
match existing_val {
Some(v) => {
for e in v {
result.push(*e);
}
},
None => (),
}
for op in operands {
for e in op {
result.push(*e);
}
}
result
}

34
test/test_iterator.rs
Unescape View File

@ -1,5 +1,4 @@
use rocksdb::{Options, DB, Writable, Direction}; use rocksdb::{Options, DB, Writable, Direction};
use std;
fn cba(input: &Box<[u8]>) -> Box<[u8]> { fn cba(input: &Box<[u8]>) -> Box<[u8]> {
input.iter().cloned().collect::<Vec<_>>().into_boxed_slice() input.iter().cloned().collect::<Vec<_>>().into_boxed_slice()
@ -17,7 +16,7 @@ pub fn test_iterator() {
let v2:Box<[u8]> = b"v2222".to_vec().into_boxed_slice(); let v2:Box<[u8]> = b"v2222".to_vec().into_boxed_slice();
let v3:Box<[u8]> = b"v3333".to_vec().into_boxed_slice(); let v3:Box<[u8]> = b"v3333".to_vec().into_boxed_slice();
let v4:Box<[u8]> = b"v4444".to_vec().into_boxed_slice(); let v4:Box<[u8]> = b"v4444".to_vec().into_boxed_slice();
let mut db = DB::open_default(path).unwrap(); let db = DB::open_default(path).unwrap();
let p = db.put(&*k1, &*v1); let p = db.put(&*k1, &*v1);
assert!(p.is_ok()); assert!(p.is_ok());
let p = db.put(&*k2, &*v2); let p = db.put(&*k2, &*v2);
@ -27,83 +26,82 @@ pub fn test_iterator() {
let mut view1 = db.iterator(); let mut view1 = db.iterator();
let expected = vec![(cba(&k1), cba(&v1)), (cba(&k2), cba(&v2)), (cba(&k3), cba(&v3))]; let expected = vec![(cba(&k1), cba(&v1)), (cba(&k2), cba(&v2)), (cba(&k3), cba(&v3))];
{ {
let mut iterator1 = view1.from_start(); let iterator1 = view1.from_start();
assert_eq!(iterator1.collect::<Vec<_>>(), expected); assert_eq!(iterator1.collect::<Vec<_>>(), expected);
} }
// Test that it's reusable a few times // Test that it's reusable a few times
{ {
let mut iterator1 = view1.from_start(); let iterator1 = view1.from_start();
assert_eq!(iterator1.collect::<Vec<_>>(), expected); assert_eq!(iterator1.collect::<Vec<_>>(), expected);
} }
{ {
let mut iterator1 = view1.from_start(); let iterator1 = view1.from_start();
assert_eq!(iterator1.collect::<Vec<_>>(), expected); assert_eq!(iterator1.collect::<Vec<_>>(), expected);
} }
{ {
let mut iterator1 = view1.from_start(); let iterator1 = view1.from_start();
assert_eq!(iterator1.collect::<Vec<_>>(), expected); assert_eq!(iterator1.collect::<Vec<_>>(), expected);
} }
// Test it in reverse a few times // Test it in reverse a few times
{ {
let mut iterator1 = view1.from_end(); let iterator1 = view1.from_end();
let mut tmp_vec = iterator1.collect::<Vec<_>>(); let mut tmp_vec = iterator1.collect::<Vec<_>>();
tmp_vec.reverse(); tmp_vec.reverse();
assert_eq!(tmp_vec, expected); assert_eq!(tmp_vec, expected);
} }
{ {
let mut iterator1 = view1.from_end(); let iterator1 = view1.from_end();
let mut tmp_vec = iterator1.collect::<Vec<_>>(); let mut tmp_vec = iterator1.collect::<Vec<_>>();
tmp_vec.reverse(); tmp_vec.reverse();
assert_eq!(tmp_vec, expected); assert_eq!(tmp_vec, expected);
} }
{ {
let mut iterator1 = view1.from_end(); let iterator1 = view1.from_end();
let mut tmp_vec = iterator1.collect::<Vec<_>>(); let mut tmp_vec = iterator1.collect::<Vec<_>>();
tmp_vec.reverse(); tmp_vec.reverse();
assert_eq!(tmp_vec, expected); assert_eq!(tmp_vec, expected);
} }
{ {
let mut iterator1 = view1.from_end(); let iterator1 = view1.from_end();
let mut tmp_vec = iterator1.collect::<Vec<_>>(); let mut tmp_vec = iterator1.collect::<Vec<_>>();
tmp_vec.reverse(); tmp_vec.reverse();
assert_eq!(tmp_vec, expected); assert_eq!(tmp_vec, expected);
} }
{ {
let mut iterator1 = view1.from_end(); let iterator1 = view1.from_end();
let mut tmp_vec = iterator1.collect::<Vec<_>>(); let mut tmp_vec = iterator1.collect::<Vec<_>>();
tmp_vec.reverse(); tmp_vec.reverse();
assert_eq!(tmp_vec, expected); assert_eq!(tmp_vec, expected);
} }
// Try it forward again // Try it forward again
{ {
let mut iterator1 = view1.from_start(); let iterator1 = view1.from_start();
assert_eq!(iterator1.collect::<Vec<_>>(), expected); assert_eq!(iterator1.collect::<Vec<_>>(), expected);
} }
{ {
let mut iterator1 = view1.from_start(); let iterator1 = view1.from_start();
assert_eq!(iterator1.collect::<Vec<_>>(), expected); assert_eq!(iterator1.collect::<Vec<_>>(), expected);
} }
let mut view2 = db.iterator();
let p = db.put(&*k4, &*v4); let p = db.put(&*k4, &*v4);
assert!(p.is_ok()); assert!(p.is_ok());
let mut view3 = db.iterator(); let mut view3 = db.iterator();
let expected2 = vec![(cba(&k1), cba(&v1)), (cba(&k2), cba(&v2)), (cba(&k3), cba(&v3)), (cba(&k4), cba(&v4))]; let expected2 = vec![(cba(&k1), cba(&v1)), (cba(&k2), cba(&v2)), (cba(&k3), cba(&v3)), (cba(&k4), cba(&v4))];
{ {
let mut iterator1 = view1.from_start(); let iterator1 = view1.from_start();
assert_eq!(iterator1.collect::<Vec<_>>(), expected); assert_eq!(iterator1.collect::<Vec<_>>(), expected);
} }
{ {
let mut iterator1 = view3.from_start(); let iterator1 = view3.from_start();
assert_eq!(iterator1.collect::<Vec<_>>(), expected2); assert_eq!(iterator1.collect::<Vec<_>>(), expected2);
} }
{ {
let mut iterator1 = view3.from(b"k2", Direction::forward); let iterator1 = view3.from(b"k2", Direction::forward);
let expected = vec![(cba(&k2), cba(&v2)), (cba(&k3), cba(&v3)), (cba(&k4), cba(&v4))]; let expected = vec![(cba(&k2), cba(&v2)), (cba(&k3), cba(&v3)), (cba(&k4), cba(&v4))];
assert_eq!(iterator1.collect::<Vec<_>>(), expected); assert_eq!(iterator1.collect::<Vec<_>>(), expected);
} }
{ {
let mut iterator1 = view3.from(b"k2", Direction::reverse); let iterator1 = view3.from(b"k2", Direction::reverse);
let expected = vec![(cba(&k2), cba(&v2)), (cba(&k1), cba(&v1))]; let expected = vec![(cba(&k2), cba(&v2)), (cba(&k1), cba(&v1))];
assert_eq!(iterator1.collect::<Vec<_>>(), expected); assert_eq!(iterator1.collect::<Vec<_>>(), expected);
} }

18
test/test_multithreaded.rs
Unescape View File

@ -11,25 +11,25 @@ pub fn test_multithreaded() {
let db = DB::open_default(path).unwrap(); let db = DB::open_default(path).unwrap();
let db = Arc::new(db); let db = Arc::new(db);
db.put(b"key", b"value1"); db.put(b"key", b"value1").unwrap();
let db1 = db.clone(); let db1 = db.clone();
let j1 = thread::spawn(move|| { let j1 = thread::spawn(move|| {
for i in 1..N { for _ in 1..N {
db1.put(b"key", b"value1"); db1.put(b"key", b"value1").unwrap();
} }
}); });
let db2 = db.clone(); let db2 = db.clone();
let j2 = thread::spawn(move|| { let j2 = thread::spawn(move|| {
for i in 1..N { for _ in 1..N {
db2.put(b"key", b"value2"); db2.put(b"key", b"value2").unwrap();
} }
}); });
let db3 = db.clone(); let db3 = db.clone();
let j3 = thread::spawn(move|| { let j3 = thread::spawn(move|| {
for i in 1..N { for _ in 1..N {
match db3.get(b"key") { match db3.get(b"key") {
Ok(Some(v)) => { Ok(Some(v)) => {
if &v[..] != b"value1" && &v[..] != b"value2" { if &v[..] != b"value1" && &v[..] != b"value2" {
@ -43,9 +43,9 @@ pub fn test_multithreaded() {
} }
}); });
j1.join(); j1.join().unwrap();
j2.join(); j2.join().unwrap();
j3.join(); j3.join().unwrap();
} }
assert!(DB::destroy(&Options::new(), path).is_ok()); assert!(DB::destroy(&Options::new(), path).is_ok());
} }

56
test/test_multithreaded.rs.orig
Unescape View File

@ -1,56 +0,0 @@
<<<<<<< Updated upstream
use rocksdb::{Options, DB, Writable, Direction, DBResult};
use std::thread::{self, Builder};
=======
use rocksdb::{Options, DB, Writable};
use std::thread;
>>>>>>> Stashed changes
use std::sync::Arc;
const N: usize = 100_000;
#[test]
pub fn test_multithreaded() {
let path = "_rust_rocksdb_multithreadtest";
{
let db = DB::open_default(path).unwrap();
let db = Arc::new(db);
db.put(b"key", b"value1");
let db1 = db.clone();
let j1 = thread::spawn(move|| {
for i in 1..N {
db1.put(b"key", b"value1");
}
});
let db2 = db.clone();
let j2 = thread::spawn(move|| {
for i in 1..N {
db2.put(b"key", b"value2");
}
});
let db3 = db.clone();
let j3 = thread::spawn(move|| {
for i in 1..N {
match db3.get(b"key") {
Ok(Some(v)) => {
if &v[..] != b"value1" && &v[..] != b"value2" {
assert!(false);
}
}
_ => {
assert!(false);
}
}
}
});
j1.join();
j2.join();
j3.join();
}
assert!(DB::destroy(&Options::new(), path).is_ok());
}
Write Preview
Loading…
Cancel
Save