Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

storage: support reverse scan #3033

Merged
merged 3 commits into from
May 11, 2018
Merged

storage: support reverse scan #3033

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
16a8566
storage: support reverse scan
huachaohuang May 7, 2018
7bec39b
Update kvproto and address comment
huachaohuang May 8, 2018
cf2a834
Merge branch 'master' into reverse-scan
huachaohuang May 11, 2018
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions src/server/service/kv.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -200,6 +200,7 @@ impl<T: RaftStoreRouter + 'static> tikvpb_grpc::Tikv for Service<T> {

let mut options = Options::default();
options.key_only = req.get_key_only();
options.reverse_scan = req.get_reverse();

let future = self.storage
.async_scan(
Expand Down
49 changes: 32 additions & 17 deletions src/storage/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -389,6 +389,7 @@ pub struct Options {
pub lock_ttl: u64,
pub skip_constraint_check: bool,
pub key_only: bool,
pub reverse_scan: bool,
}

impl Options {
Expand All @@ -397,8 +398,14 @@ impl Options {
lock_ttl,
skip_constraint_check,
key_only,
reverse_scan: false,
}
}

pub fn reverse_scan(mut self) -> Options {
self.reverse_scan = true;
self
}
}

#[derive(Clone)]
Expand Down Expand Up @@ -657,23 +664,31 @@ impl Storage {
ctx.get_isolation_level(),
!ctx.get_not_fill_cache(),
);
snap_store
.scanner(ScanMode::Forward, options.key_only, None, None)
.and_then(|mut scanner| {
let res = scanner.scan(start_key, limit);
let statistics = scanner.get_statistics();
thread_ctx.collect_scan_count(CMD, statistics);
thread_ctx.collect_read_flow(ctx.get_region_id(), statistics);
res
})
.map_err(Error::from)
.map(|results| {
thread_ctx.collect_key_reads(CMD, results.len() as u64);
results
.into_iter()
.map(|x| x.map_err(Error::from))
.collect()
})

let scan_mode = if options.reverse_scan {
ScanMode::Backward
} else {
ScanMode::Forward
};

let mut scanner = snap_store.scanner(scan_mode, options.key_only, None, None)?;
let res = if options.reverse_scan {
scanner.reverse_scan(start_key, limit)
} else {
scanner.scan(start_key, limit)
};

let statistics = scanner.get_statistics();
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'm still wondering this case. Previously, we have collect_scan_count and collect_read_flow in snap_store.scanner(..).and_then(..), but now there is no and_then. There should be some differences in edge cases. The rest LGTM.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

what is the difference?

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

When snap_store.scanner(..) returns Err, there will be differences. and_then only executes when it is Ok().

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The previous code did nothing when snap_store.scanner() returns Err, and now it returns directly when snap_store.scanner() returns Err, they are exactly the same.

thread_ctx.collect_scan_count(CMD, statistics);
thread_ctx.collect_read_flow(ctx.get_region_id(), statistics);

res.map_err(Error::from).map(|results| {
thread_ctx.collect_key_reads(CMD, results.len() as u64);
results
.into_iter()
.map(|x| x.map_err(Error::from))
.collect()
})
})
.then(move |r| {
_timer.observe_duration();
Expand Down
19 changes: 19 additions & 0 deletions tests/storage/assert_storage.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -281,6 +281,25 @@ impl AssertionStorage {
assert_eq!(result, expect);
}

pub fn reverse_scan_ok(
&self,
start_key: &[u8],
limit: usize,
ts: u64,
expect: Vec<Option<(&[u8], &[u8])>>,
) {
let key_address = make_key(start_key);
let result = self.store
.reverse_scan(self.ctx.clone(), key_address, limit, false, ts)
.unwrap();
let result: Vec<Option<KvPair>> = result.into_iter().map(Result::ok).collect();
let expect: Vec<Option<KvPair>> = expect
.into_iter()
.map(|x| x.map(|(k, v)| (k.to_vec(), v.to_vec())))
.collect();
assert_eq!(result, expect);
}

pub fn scan_key_only_ok(
&self,
start_key: &[u8],
Expand Down
19 changes: 19 additions & 0 deletions tests/storage/sync_storage.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -101,6 +101,25 @@ impl SyncStorage {
.wait()
}

pub fn reverse_scan(
&self,
ctx: Context,
key: Key,
limit: usize,
key_only: bool,
start_ts: u64,
) -> Result<Vec<Result<KvPair>>> {
self.store
.async_scan(
ctx,
key,
limit,
start_ts,
Options::new(0, false, key_only).reverse_scan(),
)
.wait()
}

pub fn prewrite(
&self,
ctx: Context,
Expand Down
137 changes: 137 additions & 0 deletions tests/storage_cases/test_storage.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -285,6 +285,143 @@ fn test_txn_store_scan() {
check_v40();
}

#[test]
fn test_txn_store_reverse_scan() {
let store = AssertionStorage::default();

// ver10: A(10) - B(_) - C(10) - D(_) - E(10)
store.put_ok(b"A", b"A10", 5, 10);
store.put_ok(b"C", b"C10", 5, 10);
store.put_ok(b"E", b"E10", 5, 10);

let check_v10 = || {
store.reverse_scan_ok(b"Z", 0, 10, vec![]);
store.reverse_scan_ok(b"Z", 1, 10, vec![Some((b"E", b"E10"))]);
store.reverse_scan_ok(
b"Z",
2,
10,
vec![Some((b"E", b"E10")), Some((b"C", b"C10"))],
);
store.reverse_scan_ok(
b"Z",
3,
10,
vec![
Some((b"E", b"E10")),
Some((b"C", b"C10")),
Some((b"A", b"A10")),
],
);
store.reverse_scan_ok(
b"Z",
4,
10,
vec![
Some((b"E", b"E10")),
Some((b"C", b"C10")),
Some((b"A", b"A10")),
],
);
store.reverse_scan_ok(
b"E\x00",
3,
10,
vec![
Some((b"E", b"E10")),
Some((b"C", b"C10")),
Some((b"A", b"A10")),
],
);
store.reverse_scan_ok(
b"E",
3,
10,
vec![Some((b"C", b"C10")), Some((b"A", b"A10"))],
);
store.reverse_scan_ok(b"", 1, 10, vec![]);
};
check_v10();

// ver20: A(10) - B(20) - C(10) - D(20) - E(10)
store.put_ok(b"B", b"B20", 15, 20);
store.put_ok(b"D", b"D20", 15, 20);

let check_v20 = || {
store.reverse_scan_ok(
b"Z",
5,
20,
vec![
Some((b"E", b"E10")),
Some((b"D", b"D20")),
Some((b"C", b"C10")),
Some((b"B", b"B20")),
Some((b"A", b"A10")),
],
);
store.reverse_scan_ok(
b"C",
5,
20,
vec![Some((b"B", b"B20")), Some((b"A", b"A10"))],
);
store.reverse_scan_ok(b"D\x00", 1, 20, vec![Some((b"D", b"D20"))]);
};
check_v10();
check_v20();

// ver30: A(_) - B(20) - C(10) - D(_) - E(10)
store.delete_ok(b"A", 25, 30);
store.delete_ok(b"D", 25, 30);

let check_v30 = || {
store.reverse_scan_ok(
b"Z",
5,
30,
vec![
Some((b"E", b"E10")),
Some((b"C", b"C10")),
Some((b"B", b"B20")),
],
);
store.reverse_scan_ok(b"E", 1, 30, vec![Some((b"C", b"C10"))]);
store.reverse_scan_ok(b"B\x00", 5, 30, vec![Some((b"B", b"B20"))]);
};
check_v10();
check_v20();
check_v30();

// ver40: A(_) - B(_) - C(40) - D(40) - E(10)
store.delete_ok(b"B", 35, 40);
store.put_ok(b"C", b"C40", 35, 40);
store.put_ok(b"D", b"D40", 35, 40);

let check_v40 = || {
store.reverse_scan_ok(
b"Z",
5,
40,
vec![
Some((b"E", b"E10")),
Some((b"D", b"D40")),
Some((b"C", b"C40")),
],
);
store.reverse_scan_ok(
b"E",
5,
100,
vec![Some((b"D", b"D40")), Some((b"C", b"C40"))],
);
};
check_v10();
check_v20();
check_v30();
check_v40();
}

#[test]
fn test_txn_store_scan_key_only() {
let store = AssertionStorage::default();
Expand Down