tikv
/
client-rust
mirror of https://github.com/tikv/client-rust.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #269 from nrc/docs 

Improve documentation
This commit is contained in:
Nick Cameron 2021-04-22 16:25:36 +12:00 committed by GitHub
parent 20318abcd8 8a128ac7ff
commit da26bb4927
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
13 changed files with 407 additions and 301 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

155
README.md
View File

@ -2,127 +2,138 @@
[![Build Status](https://travis-ci.org/tikv/client-rust.svg?branch=master)](https://travis-ci.org/tikv/client-rust)
> Currently this crate is experimental and some portions (e.g. the Transactional API) are still in active development. You're encouraged to use this library for testing and to help us find problems!
[Docs](https://www.tikv.dev/doc/rust-client/tikv_client/)
This crate provides a clean, ready to use client for [TiKV](https://github.com/tikv/tikv), a
distributed transactional Key-Value database written in Rust.
This crate provides an easy-to-use client for [TiKV](https://github.com/tikv/tikv), a distributed, transactional key-value database written in Rust.
With this crate you can easily connect to any TiKV deployment, interact with it, and mutate the data it contains. It uses async/await internally and exposes some `async fn` APIs as well.
This crate lets you connect to a TiKV cluster and use either a transactional or raw (simple get/put style without transactional consistency guarantees) API to access and update your data.
This is an open source (Apache 2) project hosted by the Cloud Native Computing Foundation (CNCF) and maintained by the TiKV Authors. *We'd love it if you joined us in improving this project.*
This is an open source (Apache 2) project maintained by the TiKV Authors. We welcome community contributions, see below for more info.
## Getting started
The TiKV client is a Rust library (crate). To use this crate in your project, add following dependencies in your `Cargo.toml`:
The TiKV client is a Rust library (crate). To use this crate in your project, add the following dependency to your `Cargo.toml`:
```toml
[dependencies]
tikv-client = { git = "https://github.com/tikv/client-rust.git" }
tikv-client = 0.1
```
The client requires a Git dependency until we can [publish it](https://github.com/tikv/client-rust/issues/32).
Note, that you will need `tikv-client = { git = "https://github.com/tikv/client-rust.git" }` until we publish the crate (should be any day now).
The client provides two modes to interact with TiKV: raw and transactional.
In the current version (0.0.0), the transactional API supports optimistic transactions. Pessimistic transactions are implemented but not well tested.
The minimum supported version of Rust is 1.40. The minimum supported version of TiKV is 5.0.
Important note: It is **not recommended or supported** to use both the raw and transactional APIs on the same database.
The general flow of using the client crate is to create either a raw or transaction client object (which can be configured) then send commands using the client object, or use it to create transactions objects. In the latter case, the transaction is built up using various commands and then committed (or rolled back).
### Code examples
### Examples
Raw mode:
```rust
let config = Config::new(vec!["127.0.0.1:2379"]);
let client = RawClient::new(config).await?;
client.put("key".to_owned(), "value".to_owned()).await;
let value = client.get("key".to_owned()).await;
use tikv_client::RawClient;
let client = RawClient::new(vec!["127.0.0.1:2379"]).await?;
client.put("key".to_owned(), "value".to_owned()).await?;
let value = client.get("key".to_owned()).await?;
```
Transactional mode:
```rust
let config = Config::new(vec!["127.0.0.1:2379"]);
let txn_client = TransactionClient::new(config).await?;
let mut txn = txn_client.begin().await?;
use tikv_client::TransactionClient;
let txn_client = TransactionClient::new(vec!["127.0.0.1:2379"]).await?;
let mut txn = txn_client.begin_optimistic().await?;
txn.put("key".to_owned(), "value".to_owned()).await?;
let value = txn.get("key".to_owned()).await;
let value = txn.get("key".to_owned()).await?;
txn.commit().await?;
```
There are some [examples](examples) which show how to use the client in a Rust program.
## API summary
### API
The TiKV Rust client supports several levels of abstraction. The most convenient way to use the client is via `RawClient` and `TransactionClient`. This gives a very high-level API which mostly abstracts over the distributed nature of the store and has sensible defaults for all protocols. This interface can be configured, primarily when creating the client or transaction objects via the `Config` and `TransactionOptions` structs. Using some options, you can take over parts of the protocols (such as retrying failed messages) yourself.
#### Raw requests
The lowest level of abstraction is to create and send gRPC messages directly to TiKV (and PD) nodes. The `tikv-client-store` and `tikv-client-pd` crates make this easier than using the protobuf definitions and a gRPC library directly, but give you the same level of control.
| Request | Main parameter type | Successful result type | Noteworthy Behavior |
| -------------- | ------------------- | ---------------------- | --------------------------------------------- |
| `put` | `KvPair` | `()` | |
| `get` | `Key` | `Option<Value>` | |
| `delete` | `Key` | `()` | |
| `scan` | `BoundRange` | `Vec<KvPair>` | |
| `batch_put` | `Iter<KvPair>` | `()` | |
| `batch_get` | `Iter<Key>` | `Vec<KvPair>` | Skip non-existent keys; Does not retain order |
| `batch_delete` | `Iter<Key>` | `()` | |
| `delete_range` | `BoundRange` | `()` | |
In between these levels of abstraction, you can send and receive individual messages to the TiKV cluster, but take advantage of library code for common operations such as resolving data to regions and thus nodes in the cluster, or retrying failed messages. This can be useful for testing a TiKV cluster or for some advanced use cases. See the `client_rust::request` module for this API, and `client_rust::raw::lowering` and `client_rust::transaction::lowering` for convenience methods for creating request objects.
#### Transactional requests
The rest of this document describes only the `RawClient`/`TransactionClient` APIs.
| Request | Main parameter type | Successful result type | Noteworthy Behavior |
| ----------- | ------------------- | ---------------------- | ------------------------------------------------------------------ |
| `put` | `KvPair` | `()` | |
| `get` | `Key` | `Option<value>` | |
| `key_exists` | `Key` | `bool` | |
| `delete` | `Key` | `()` | |
| `scan` | `BoundRange` | `Iter<KvPair>` | |
| `scan_keys` | `BoundRange` | `Iter<Key>` | |
| `batch_get` | `Iter<Key>` | `Iter<KvPair>` | Skip non-existent keys; Does not retain order |
| `lock_keys` | `Iter<Key>` | `()` | |
| `gc` | `Timestamp` | `bool` | It returns whether the latest safepoint in PD equals the parameter |
Important note: It is **not recommended or supported** to use both the raw and transactional APIs on the same database.
For detailed behavior of each request, please refer to the [doc](#Access-the-documentation).
### Types
#### Experimental raw requests
`Key`: a key in the store. `String` and `Vec<u8>` implement `Into<Key>`, so you can pass them directly into client functions.
You must be careful if you want to use the following request(s). Read the description for reasons.
`Value`: a value in the store; just an alias of `Vec<u8>`.
| Request | Main parameter type | Successful result type |
| ------------ | ------------------- | ---------------------- |
| `batch_scan` | `Iter<BoundRange>` | `Vec<KvPair>` |
`KvPair`: a pair of a `Key` and a `Value`. It provides convenience methods for conversion to and from other types.
The `each_limit` parameter does not work as expected. It does not limit the number of results returned of each range, instead it limits the number of results in each region of each range. As a result, you may get **more than** `each_limit` key-value pairs for each range. But you should not miss any entries.
`BoundRange`: used for range related requests like `scan`. It implements `From` for Rust ranges so you can pass a Rust range of keys to the request, e.g., `client.delete_range(vec![]..)`.
The results of `batch_scan` are flattened. The order of ranges is retained.
### Raw requests
### Useful types
| Request | Main parameter type | Result type | Noteworthy Behavior |
| ------------------ | ------------------- | ---------------- | ---------------------------------------------- |
| `put` | `KvPair` | | |
| `get` | `Key` | `Option<Value>` | |
| `delete` | `Key` | | |
| `delete_range` | `BoundRange` | | |
| `scan` | `BoundRange` | `Vec<KvPair>` | |
| `batch_put` | `Iter<KvPair>` | | |
| `batch_get` | `Iter<Key>` | `Vec<KvPair>` | Skips non-existent keys; does not retain order |
| `batch_delete` | `Iter<Key>` | | |
| `batch_scan` | `Iter<BoundRange>` | `Vec<KvPair>` | See docs for `each_limit` parameter behavior. The order of ranges is retained. |
| `batch_scan_keys` | `Iter<BoundRange>` | `Vec<Key>` | See docs for `each_limit` parameter behavior. The order of ranges is retained. |
| `compare_and_swap` | `Key` + 2x `Value` | `(Option<Value>, bool)` | |
To use the client, there are 4 types you will need.
### Transactional requests
`Key` is simply a vector of bytes(`Vec<u8>`). `String` and `Vec<u8>` implements `Into<Key>`, so you can directly pass them to clients.
| Request | Main parameter type | Result type | Noteworthy Behavior |
| ---------------------- | ------------------- | --------------- | ------------------------------------------------------------------ |
| `put` | `KvPair` | | |
| `get` | `Key` | `Option<value>` | |
| `get_for_update` | `Key` | `Option<value>` | |
| `key_exists` | `Key` | `bool` | |
| `delete` | `Key` | | |
| `scan` | `BoundRange` | `Iter<KvPair>` | |
| `scan_keys` | `BoundRange` | `Iter<Key>` | |
| `batch_get` | `Iter<Key>` | `Iter<KvPair>` | Skips non-existent keys; does not retain order |
| `batch_get_for_update` | `Iter<Key>` | `Iter<KvPair>` | Skips non-existent keys; does not retain order |
| `lock_keys` | `Iter<Key>` | | |
| `send_heart_beat` | | `u64` (TTL) | |
| `gc` | `Timestamp` | `bool` | Returns true if the latest safepoint in PD equals the parameter |
`Value` is just an alias of `Vec<u8>`.
# Development and contributing
`KvPair` is a tuple consisting of a `Key` and a `Value`. It also provides some convenience methods for conversion to and from other types.
We welcome your contributions! Contributing code is great, we also appreciate filing [issues](https://github.com/tikv/client-rust/issues/new) to identify bugs and provide feedback, adding tests or examples, and improvements to documentation.
`BoundRange` is used for range related requests like `scan`. It implements `From` for usual ranges so you can just create a range and pass them to the request.For instance, `client.scan("k2".to_owned()..="k5".to_owned(), 5)` or `client.delete_range(vec![]..)`.
## Building and testing
## Access the documentation
We use the standard Cargo workflows, e.g., `cargo build` to build and `cargo test` to run unit tests. You will need to use a nightly Rust toolchain to build and run tests.
We've done our best to include ample, tested, and understandable examples.
Running integration tests or manually testing the client with a TiKV cluster is a little bit more involved. The easiest way is to use [TiUp](https://github.com/pingcap/tiup) to initialise a cluster on your local machine:
We recommend using the officially maintained documentation [here](https://www.tikv.dev/doc/rust-client/tikv_client/).
You can also access the documentation on your machine by running the following in any project that depends on `tikv-client`.
```bash
cargo doc --package tikv-client --open
# If it didn't work, browse file URL it tried to open with your browser.
```
tiup playground nightly --db 0 --tiflash 0 --monitor false
```
## Minimal Rust version
Then if you want to run integration tests:
This crate supports Rust 1.40 and above.
```
PD_ADDRS="127.0.0.1:2379" cargo test --package tikv-client --test integration_tests --features integration-tests
```
For development, a nightly Rust compiler is needed to compile the tests.
## Creating a PR
We use a standard GitHub PR workflow. We run CI on every PR and require all PRs to build without warnings (including clippy and Rustfmt warnings), pass tests, have a DCO sign-off (use `-s` when you commit, the DCO bot will guide you through completing the DCO agreement for your first PR), and have at least one review. If any of this is difficult for you, don't worry about it and ask on the PR.
To run CI-like tests locally, we recommend you run `cargo clippy`, `cargo test`, and `cargo fmt` before submitting your PR. See above for running integration tests, but you probably won't need to worry about this for your first few PRs.
Please follow PingCAP's [Rust style guide](https://pingcap.github.io/style-guide/rust/). All code PRs should include new tests or test cases.
## Getting help
If you need help, either to find something to work on, or with any technical problem, the easiest way to get it is via Slack. We monitor the client-rust (better for general client questions) and sig-transaction (better for technical questions about TiKV's transaction protocol) channels on the [tikv-wg slack](https://tikv.org/chat).
You can also get help on GitHub issues or PRs directly. You can just ask a question; if you don't get a response, you should ping @nrc or @ekexium.

43
src/config.rs
View File

@ -3,22 +3,11 @@
use serde_derive::{Deserialize, Serialize};
use std::{path::PathBuf, time::Duration};
/// The configuration for either a `raw::Client` or a `transaction::Client`.
/// The configuration for either a [`RawClient`](crate::RawClient) or a
/// [`TransactionClient`](crate::TransactionClient).
///
/// Because TiKV is managed by a [PD](https://github.com/pingcap/pd/) cluster, the endpoints for PD
/// must be provided, **not** the TiKV nodes.
///
/// It's important to **include more than one PD endpoint** (include all, if possible!)
/// This helps avoid having a *single point of failure*.
///
/// By default, this client will use an insecure connection over instead of one protected by
/// Transport Layer Security (TLS). Your deployment may have chosen to rely on security measures
/// such as a private network, or a VPN layer to provide secure transmission.
///
/// To use a TLS secured connection, use the `with_security` function to set the required
/// parameters.
///
/// TiKV does not currently offer encrypted storage (or encryption-at-rest).
/// See also [`TransactionOptions`](crate::TransactionOptions) which provides more ways to configure
/// requests.
#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
#[serde(default)]
#[serde(rename_all = "kebab-case")]
@ -43,11 +32,16 @@ impl Default for Config {
}
impl Config {
/// Set the certificate authority, certificate, and key locations for the
/// [`Config`](Config).
/// Set the certificate authority, certificate, and key locations for clients.
///
/// By default, TiKV connections do not utilize transport layer security. Enable it by setting
/// these values.
/// By default, this client will use an insecure connection over instead of one protected by
/// Transport Layer Security (TLS). Your deployment may have chosen to rely on security measures
/// such as a private network, or a VPN layer to provide secure transmission.
///
/// To use a TLS secured connection, use the `with_security` function to set the required
/// parameters.
///
/// TiKV does not currently offer encrypted storage (or encryption-at-rest).
///
/// # Examples
/// ```rust
@ -66,16 +60,21 @@ impl Config {
self
}
/// Set the timeout for the [`Config`](Config).
/// Set the timeout for clients.
///
/// The timeout is used for all requests when using or connecting to a TiKV cluster (including
/// PD nodes). If the request does not complete within timeout, the request is cancelled and
/// an error returned to the user.
///
/// The default timeout is two seconds.
///
/// # Examples
/// ```rust
/// # use tikv_client::Config;
/// # use std::time::Duration;
/// let config = Config::default().timeout(Duration::from_secs(10));
/// let config = Config::default().with_timeout(Duration::from_secs(10));
/// ```
pub fn timeout(mut self, timeout: Duration) -> Self {
pub fn with_timeout(mut self, timeout: Duration) -> Self {
self.timeout = timeout;
self
}

126
src/lib.rs
View File

@ -1,91 +1,108 @@
// Copyright 2018 TiKV Project Authors. Licensed under Apache-2.0.
// Long and nested future chains can quickly result in large generic types.
#![type_length_limit = "16777216"]
#![allow(clippy::redundant_closure)]
#![allow(clippy::type_complexity)]
#![allow(incomplete_features)]
//! This crate provides a clean, ready to use client for [TiKV](https://github.com/tikv/tikv), a
//! distributed transactional Key-Value database written in Rust.
//! This crate provides an easy-to-use client for [TiKV](https://github.com/tikv/tikv), a
//! distributed, transactional key-value database written in Rust.
//!
//! With this crate you can easily connect to any TiKV deployment, interact with it, and mutate the
//! data it contains.
//! This crate lets you connect to a TiKV cluster and use either a transactional or raw (simple
//! get/put style without transactional consistency guarantees) API to access and update your data.
//!
//! The TiKV Rust client supports several levels of abstraction. The most convenient way to use the
//! client is via [`RawClient`] and [`TransactionClient`]. This gives a very high-level API which
//! mostly abstracts over the distributed nature of the store and has sensible defaults for all
//! protocols. This interface can be configured, primarily when creating the client or transaction
//! objects via the [`Config`] and [`TransactionOptions`] structs. Using some options, you can take
//! over parts of the protocols (such as retrying failed messages) yourself.
//!
//! The lowest level of abstraction is to create and send gRPC messages directly to TiKV (and PD)
//! nodes. The `tikv-client-store` and `tikv-client-pd` crates make this easier than using the
//! protobuf definitions and a gRPC library directly, but give you the same level of control.
//!
//! In between these levels of abstraction, you can send and receive individual messages to the TiKV
//! cluster, but take advantage of library code for common operations such as resolving data to
//! regions and thus nodes in the cluster, or retrying failed messages. This can be useful for
//! testing a TiKV cluster or for some advanced use cases. See the [`request`] module for
//! this API, and [`raw::lowering`] and [`transaction::lowering`] for
//! convenience methods for creating request objects.
//!
//! ## Choosing an API
//!
//! This crate offers both [**raw**](raw/index.html) and
//! [**transactional**](transaction/index.html) APIs. You should choose just one for your system.
//! This crate offers both [raw](RawClient) and
//! [transactional](Transaction) APIs. You should choose just one for your system.
//!
//! The *consequence* of supporting transactions is increased overhead of coordination with the
//! placement driver for timestamp acquisition. This is approximately 1 RTT.
//! The consequence of supporting transactions is increased overhead of coordination with the
//! placement driver and TiKV, and additional code complexity.
//!
//! *While it is possible to use both APIs at the same time, doing so is unsafe and unsupported.*
//!
//! Choose the one that suites your needs as described below, then add the import statement to your
//! file where you need to use the library.
//!
//! ### Transactional
//!
//! The [transactional](transaction/index.html) API supports **transactions** via Multi-Version
//! Concurrency Control (MVCC).
//! The [transactional](Transaction) API supports **transactions** via multi-version
//! concurrency control (MVCC).
//!
//! **Best when you mostly do** complex sets of actions, actions which may require a rollback,
//! operations affecting multiple keys or values, or operations that depend on strong ordering.
//! Best when you mostly do complex sets of actions, actions which may require a rollback,
//! operations affecting multiple keys or values, or operations that depend on strong consistency.
//!
//! ```rust
//! use tikv_client::*;
//! ```
//!
//! ### Raw
//!
//! The [raw](raw/index.html) API has **reduced coordination overhead**, but lacks any
//! The [raw](RawClient) API has reduced coordination overhead, but lacks any
//! transactional abilities.
//!
//! **Best when you mostly do** single row changes, and have very limited cross-row (eg. foreign
//! key) requirements. You will not be able to use transactions with this API.
//! Best when you mostly do single value changes, and have very limited cross-value
//! requirements. You will not be able to use transactions with this API.
//!
//! ```rust
//! use tikv_client::*;
//! ```
//! ## Usage
//!
//! ## Connect
//! The general flow of using the client crate is to create either a raw or transaction client
//! object (which can be configured) then send commands using the client object, or use it to create
//! transactions objects. In the latter case, the transaction is built up using various commands and
//! then committed (or rolled back).
//!
//! Regardless of which API you choose, you'll need to connect your client
//! ([raw](raw::Client), [transactional](transaction::Client)).
//! ### Examples
//!
//! Raw mode:
//!
//! ```rust,no_run
//! # use tikv_client::*;
//! # use tikv_client::{RawClient, Result};
//! # use futures::prelude::*;
//!
//! # fn main() -> Result<()> {
//! # futures::executor::block_on(async {
//! // Configure endpoints and optional TLS.
//! let config = Config::default().with_security("root.ca", "internal.cert", "internal.key");
//!
//! // Get a transactional client.
//! let client = TransactionClient::new_with_config(
//! vec![
//! // A list of PD endpoints.
//! "192.168.0.100:2379",
//! "192.168.0.101:2379",
//! ], config).await.unwrap();
//! # });
//! let client = RawClient::new(vec!["127.0.0.1:2379"]).await?;
//! client.put("key".to_owned(), "value".to_owned()).await?;
//! let value = client.get("key".to_owned()).await?;
//! # Ok(())
//! # })}
//! ```
//!
//! At this point, you should seek the documentation in the related API modules.
//! Transactional mode:
//!
//! ```rust,no_run
//! # use tikv_client::{TransactionClient, Result};
//! # use futures::prelude::*;
//! # fn main() -> Result<()> {
//! # futures::executor::block_on(async {
//! let txn_client = TransactionClient::new(vec!["127.0.0.1:2379"]).await?;
//! let mut txn = txn_client.begin_optimistic().await?;
//! txn.put("key".to_owned(), "value".to_owned()).await?;
//! let value = txn.get("key".to_owned()).await?;
//! txn.commit().await?;
//! # Ok(())
//! # })}
//! ```
#[macro_use]
mod request;
pub mod request;
#[macro_use]
mod transaction;
#[doc(hidden)]
pub mod transaction;
mod backoff;
mod compat;
mod config;
mod kv;
mod pd;
mod raw;
#[doc(hidden)]
pub mod raw;
mod region;
mod stats;
mod store;
@ -105,18 +122,17 @@ pub use crate::backoff::Backoff;
#[doc(inline)]
pub use crate::kv::{BoundRange, IntoOwnedRange, Key, KvPair, Value};
#[doc(inline)]
pub use crate::raw::{lowering::*, Client as RawClient, ColumnFamily};
pub use crate::raw::{lowering as raw_lowering, Client as RawClient, ColumnFamily};
#[doc(inline)]
pub use crate::request::RetryOptions;
#[doc(inline)]
pub use crate::timestamp::{Timestamp, TimestampExt};
#[doc(inline)]
pub use crate::transaction::{
lowering::*, CheckLevel, Client as TransactionClient, Snapshot, Transaction, TransactionOptions,
lowering as transaction_lowering, CheckLevel, Client as TransactionClient, Snapshot,
Transaction, TransactionOptions,
};
#[doc(inline)]
pub use config::Config;
#[doc(inline)]
pub use region::{Region, RegionId, RegionVerId, StoreId};
#[doc(inline)]
pub use tikv_client_common::{security::SecurityManager, Error, Result};

3
src/mock.rs
View File

@ -7,8 +7,9 @@
use crate::{
pd::{PdClient, PdRpcClient, RetryClient},
region::{Region, RegionId},
store::Store,
Config, Error, Key, Region, RegionId, Result, Timestamp,
Config, Error, Key, Result, Timestamp,
};
use async_trait::async_trait;
use derive_new::new;

8
src/pd/client.rs
View File

@ -1,8 +1,12 @@
// Copyright 2018 TiKV Project Authors. Licensed under Apache-2.0.
use crate::{
compat::stream_fn, kv::codec, pd::RetryClient, store::Store, BoundRange, Config, Key, Region,
RegionId, Result, SecurityManager, Timestamp,
compat::stream_fn,
kv::codec,
pd::RetryClient,
region::{Region, RegionId},
store::Store,
BoundRange, Config, Key, Result, SecurityManager, Timestamp,
};
use async_trait::async_trait;
use futures::{prelude::*, stream::BoxStream};

6
src/pd/retry.rs
View File

@ -2,7 +2,11 @@
//! A utility module for managing and retrying PD requests.
use crate::{stats::pd_stats, Error, Region, RegionId, Result, SecurityManager, StoreId};
use crate::{
region::{Region, RegionId, StoreId},
stats::pd_stats,
Error, Result, SecurityManager,
};
use async_trait::async_trait;
use futures_timer::Delay;
use grpcio::Environment;

54
src/raw/client.rs
View File

@ -19,7 +19,8 @@ const MAX_RAW_KV_SCAN_LIMIT: u32 = 10240;
/// Raw requests don't need a wrapping transaction.
/// Each request is immediately processed once executed.
///
/// The returned results of raw requests are [`Future`](std::future::Future)s that must be awaited to execute.
/// The returned results of raw request methods are [`Future`](std::future::Future)s that must be
/// awaited to execute.
#[derive(Clone)]
pub struct Client {
rpc: Arc<PdRpcClient>,
@ -29,14 +30,16 @@ pub struct Client {
}
impl Client {
/// Create a raw [`Client`](Client).
/// Create a raw [`Client`] and connect to the TiKV cluster.
///
/// It's important to **include more than one PD endpoint** (include all, if possible!)
/// This helps avoid having a *single point of failure*.
/// Because TiKV is managed by a [PD](https://github.com/pingcap/pd/) cluster, the endpoints for
/// PD must be provided, not the TiKV nodes. It's important to include more than one PD endpoint
/// (include all endpoints, if possible), this helps avoid having a single point of failure.
///
/// # Examples
///
/// ```rust,no_run
/// # use tikv_client::{Config, RawClient};
/// # use tikv_client::RawClient;
/// # use futures::prelude::*;
/// # futures::executor::block_on(async {
/// let client = RawClient::new(vec!["192.168.0.100"]).await.unwrap();
@ -46,17 +49,23 @@ impl Client {
Self::new_with_config(pd_endpoints, Config::default()).await
}
/// Create a raw [`Client`](Client).
/// Create a raw [`Client`] with a custom configuration, and connect to the TiKV cluster.
///
/// It's important to **include more than one PD endpoint** (include all, if possible!)
/// This helps avoid having a *single point of failure*.
/// Because TiKV is managed by a [PD](https://github.com/pingcap/pd/) cluster, the endpoints for
/// PD must be provided, not the TiKV nodes. It's important to include more than one PD endpoint
/// (include all endpoints, if possible), this helps avoid having a single point of failure.
///
/// # Examples
///
/// ```rust,no_run
/// # use tikv_client::{Config, RawClient};
/// # use futures::prelude::*;
/// # use std::time::Duration;
/// # futures::executor::block_on(async {
/// let client = RawClient::new(vec!["192.168.0.100"]).await.unwrap();
/// let client = RawClient::new_with_config(
/// vec!["192.168.0.100"],
/// Config::default().with_timeout(Duration::from_secs(60)),
/// ).await.unwrap();
/// # });
/// ```
pub async fn new_with_config<S: Into<String>>(
@ -72,22 +81,27 @@ impl Client {
})
}
/// Set the column family of requests.
/// Create a new client which is a clone of `self`, but which uses an explicit column family for
/// all requests.
///
/// This function returns a new `Client`, requests created with it will have the
/// supplied column family constraint. The original `Client` can still be used.
/// This function returns a new `Client`; requests created with the new client will use the
/// supplied column family. The original `Client` can still be used (without the new
/// column family).
///
/// By default, raw client uses the `Default` column family.
///
/// For normal users of the raw API, you don't need to use other column families.
/// By default, raw clients use the `Default` column family.
///
/// # Examples
///
/// ```rust,no_run
/// # use tikv_client::{Config, RawClient, ColumnFamily};
/// # use futures::prelude::*;
/// # use std::convert::TryInto;
/// # futures::executor::block_on(async {
/// let client = RawClient::new(vec!["192.168.0.100"]).await.unwrap().with_cf(ColumnFamily::Write);
/// let client = RawClient::new(vec!["192.168.0.100"])
/// .await
/// .unwrap()
/// .with_cf(ColumnFamily::Write);
/// // Fetch a value at "foo" from the Write CF.
/// let get_request = client.get("foo".to_owned());
/// # });
/// ```
@ -411,11 +425,11 @@ impl Client {
///
/// Once resolved this request will result in a set of scanners over the given keys.
///
/// **Warning**: This method is experimental. The `each_limit` parameter does not work as expected.
/// It does not limit the number of results returned of each range,
/// **Warning**: This method is experimental.
/// The `each_limit` parameter does not limit the number of results returned of each range,
/// instead it limits the number of results in each region of each range.
/// As a result, you may get **more than** `each_limit` key-value pairs for each range.
/// But you should not miss any entries.
/// As a result, you may get **more than** `each_limit` key-value pairs for each range,
/// but you should not miss any entries.
///
/// # Examples
/// ```rust,no_run

9
src/store.rs
View File

@ -1,6 +1,6 @@
// Copyright 2019 TiKV Project Authors. Licensed under Apache-2.0.
use crate::{pd::PdClient, BoundRange, Key, Region, Result};
use crate::{pd::PdClient, region::Region, BoundRange, Key, Result};
use derive_new::new;
use futures::{prelude::*, stream::BoxStream};
use std::{
@ -48,6 +48,7 @@ where
.boxed()
}
#[allow(clippy::type_complexity)]
pub fn store_stream_for_range<PdC: PdClient>(
range: (Vec<u8>, Vec<u8>),
pd_client: Arc<PdC>,
@ -67,16 +68,16 @@ pub fn store_stream_for_range<PdC: PdClient>(
}
pub fn store_stream_for_range_by_start_key<PdC: PdClient>(
start_key: Vec<u8>,
start_key: Key,
pd_client: Arc<PdC>,
) -> BoxStream<'static, Result<(Vec<u8>, Store)>> {
let bnd_range = BoundRange::range_from(start_key.clone().into());
let bnd_range = BoundRange::range_from(start_key.clone());
pd_client
.stores_for_range(bnd_range)
.map_ok(move |store| {
let region_range = store.region.range();
(
range_intersection(region_range, (start_key.clone().into(), vec![].into()))
range_intersection(region_range, (start_key.clone(), vec![].into()))
.0
.into(),
store,

5
src/transaction/buffer.rs
View File

@ -75,10 +75,7 @@ impl Buffer {
let (cached_results, undetermined_keys) = {
// Partition the keys into those we have buffered and those we have to
// get from the store.
let (undetermined_keys, cached_results): (
Vec<(Key, MutationValue)>,
Vec<(Key, MutationValue)>,
) = keys
let (undetermined_keys, cached_results): (Vec<_>, Vec<_>) = keys
.map(|key| {
let value = self
.entry_map

119
src/transaction/client.rs
View File

@ -13,39 +13,41 @@ use crate::{
use std::{mem, sync::Arc};
use tikv_client_proto::{kvrpcpb, pdpb::Timestamp};
const SCAN_LOCK_BATCH_SIZE: u32 = 1024; // FIXME: cargo-culted value
// FIXME: cargo-culted value
const SCAN_LOCK_BATCH_SIZE: u32 = 1024;
/// The TiKV transactional `Client` is used to interact with TiKV using transactional (MVCC) requests.
/// The TiKV transactional `Client` is used to interact with TiKV using transactional requests.
///
/// A [`Transaction`](crate::transaction::Transaction) provides a SQL-like interface.
/// It begins with a [`begin_optimistic`](Client::begin_optimistic) or [`begin_pessimistic`](Client::begin_pessimistic) request
/// and ends with a `rollback` or `commit` request.
/// If a `Transaction` is dropped before it's rolled back or committed, it is automatically rolled back.
/// Transactions support optimistic and pessimistic modes. For more details see the SIG-transaction
/// [docs](https://github.com/tikv/sig-transaction/tree/master/doc/tikv#optimistic-and-pessimistic-transactions).
///
/// Transaction supports optimistic and pessimistic modes, for mroe deatils, check our
/// [SIG-transaction](https://github.com/tikv/sig-transaction/tree/master/doc/tikv#optimistic-and-pessimistic-transactions).
/// Begin a [`Transaction`] by calling [`begin_optimistic`](Client::begin_optimistic) or
/// [`begin_pessimistic`](Client::begin_pessimistic). A transaction must be rolled back or committed.
///
/// Besides transaction, the client provides some utility methods:
/// - `gc`: execute a GC process which clear stale data. It is not stablized yet.
/// - `current_timestamp`: get the current `Timestamp`.
/// - `snapshot`: get the [`Snapshot`](crate::transaction::Snapshot) of the database at a certain timestamp.
/// Besides transactions, the client provides some further functionality:
/// - `gc`: trigger a GC process which clears stale data in the cluster.
/// - `current_timestamp`: get the current `Timestamp` from PD.
/// - `snapshot`: get a [`Snapshot`] of the database at a specified timestamp.
/// A `Snapshot` is a read-only transaction.
///
/// The returned results of transactional requests are [`Future`](std::future::Future)s that must be awaited to execute.
/// The returned results of transactional requests are [`Future`](std::future::Future)s that must be
/// awaited to execute.
pub struct Client {
pd: Arc<PdRpcClient>,
}
impl Client {
/// Creates a transactional [`Client`](Client).
/// Create a transactional [`Client`] and connect to the TiKV cluster.
///
/// It's important to **include more than one PD endpoint** (include all, if possible!)
/// This helps avoid having a *single point of failure*.
/// Because TiKV is managed by a [PD](https://github.com/pingcap/pd/) cluster, the endpoints for
/// PD must be provided, not the TiKV nodes. It's important to include more than one PD endpoint
/// (include all endpoints, if possible), this helps avoid having a single point of failure.
///
/// # Examples
///
/// ```rust,no_run
/// use tikv_client::{Config, TransactionClient};
/// use futures::prelude::*;
/// # use tikv_client::{Config, TransactionClient};
/// # use futures::prelude::*;
/// # futures::executor::block_on(async {
/// let client = TransactionClient::new(vec!["192.168.0.100"]).await.unwrap();
/// # });
@ -54,17 +56,23 @@ impl Client {
Self::new_with_config(pd_endpoints, Config::default()).await
}
/// Creates a transactional [`Client`](Client).
/// Create a transactional [`Client`] with a custom configuration, and connect to the TiKV cluster.
///
/// It's important to **include more than one PD endpoint** (include all, if possible!)
/// This helps avoid having a *single point of failure*.
/// Because TiKV is managed by a [PD](https://github.com/pingcap/pd/) cluster, the endpoints for
/// PD must be provided, not the TiKV nodes. It's important to include more than one PD endpoint
/// (include all endpoints, if possible), this helps avoid having a single point of failure.
///
/// # Examples
///
/// ```rust,no_run
/// use tikv_client::{Config, TransactionClient};
/// use futures::prelude::*;
/// # use tikv_client::{Config, TransactionClient};
/// # use futures::prelude::*;
/// # use std::time::Duration;
/// # futures::executor::block_on(async {
/// let client = TransactionClient::new(vec!["192.168.0.100"]).await.unwrap();
/// let client = TransactionClient::new_with_config(
/// vec!["192.168.0.100"],
/// Config::default().with_timeout(Duration::from_secs(60)),
/// ).await.unwrap();
/// # });
/// ```
pub async fn new_with_config<S: Into<String>>(
@ -76,24 +84,24 @@ impl Client {
Ok(Client { pd })
}
/// Creates a new [`Transaction`](Transaction) in optimistic mode.
/// Creates a new optimistic [`Transaction`].
///
/// Using the transaction you can issue commands like [`get`](Transaction::get) or [`put`](Transaction::put).
/// Use the transaction to issue requests like [`get`](Transaction::get) or
/// [`put`](Transaction::put).
///
/// Write operations do not lock data in TiKV, thus commit request may fail due to write conflict.
///
/// For details, check our [SIG-transaction](https://github.com/tikv/sig-transaction/tree/master/doc/tikv#optimistic-and-pessimistic-transactions).
/// Write operations do not lock data in TiKV, thus the commit request may fail due to a write
/// conflict.
///
/// # Examples
///
/// ```rust,no_run
/// use tikv_client::{Config, TransactionClient};
/// use futures::prelude::*;
/// # use tikv_client::{Config, TransactionClient};
/// # use futures::prelude::*;
/// # futures::executor::block_on(async {
/// let client = TransactionClient::new(vec!["192.168.0.100"]).await.unwrap();
/// let mut transaction = client.begin_optimistic().await.unwrap();
/// // ... Issue some commands.
/// let commit = transaction.commit();
/// let result = commit.await.unwrap();
/// transaction.commit().await.unwrap();
/// # });
/// ```
pub async fn begin_optimistic(&self) -> Result<Transaction> {
@ -101,21 +109,21 @@ impl Client {
Ok(self.new_transaction(timestamp, TransactionOptions::new_optimistic()))
}
/// Creates a new [`Transaction`](Transaction) in pessimistic mode.
/// Creates a new pessimistic [`Transaction`].
///
/// Write operations will lock the data until commit, thus commit requests should not suffer from write conflict.
/// For details, check our [SIG-transaction](https://github.com/tikv/sig-transaction/tree/master/doc/tikv#optimistic-and-pessimistic-transactions).
/// Write operations will lock the data until committed, thus commit requests should not suffer
/// from write conflicts.
///
/// # Examples
///
/// ```rust,no_run
/// use tikv_client::{Config, TransactionClient};
/// use futures::prelude::*;
/// # use tikv_client::{Config, TransactionClient};
/// # use futures::prelude::*;
/// # futures::executor::block_on(async {
/// let client = TransactionClient::new(vec!["192.168.0.100"]).await.unwrap();
/// let mut transaction = client.begin_pessimistic().await.unwrap();
/// // ... Issue some commands.
/// let commit = transaction.commit();
/// let result = commit.await.unwrap();
/// transaction.commit().await.unwrap();
/// # });
/// ```
pub async fn begin_pessimistic(&self) -> Result<Transaction> {
@ -123,12 +131,13 @@ impl Client {
Ok(self.new_transaction(timestamp, TransactionOptions::new_pessimistic()))
}
/// Creates a new customized [`Transaction`](Transaction).
/// Create a new customized [`Transaction`].
///
/// # Examples
///
/// ```rust,no_run
/// use tikv_client::{Config, TransactionClient, TransactionOptions};
/// use futures::prelude::*;
/// # use tikv_client::{Config, TransactionClient, TransactionOptions};
/// # use futures::prelude::*;
/// # futures::executor::block_on(async {
/// let client = TransactionClient::new(vec!["192.168.0.100"]).await.unwrap();
/// let mut transaction = client
@ -136,8 +145,7 @@ impl Client {
/// .await
/// .unwrap();
/// // ... Issue some commands.
/// let commit = transaction.commit();
/// let result = commit.await.unwrap();
/// transaction.commit().await.unwrap();
/// # });
/// ```
pub async fn begin_with_options(&self, options: TransactionOptions) -> Result<Transaction> {
@ -145,17 +153,18 @@ impl Client {
Ok(self.new_transaction(timestamp, options))
}
/// Creates a new [`Snapshot`](Snapshot) at the given [`Timestamp`](Timestamp).
/// Create a new [`Snapshot`](Snapshot) at the given [`Timestamp`](Timestamp).
pub fn snapshot(&self, timestamp: Timestamp, options: TransactionOptions) -> Snapshot {
Snapshot::new(self.new_transaction(timestamp, options.read_only()))
}
/// Retrieves the current [`Timestamp`](Timestamp).
/// Retrieve the current [`Timestamp`].
///
/// # Examples
///
/// ```rust,no_run
/// use tikv_client::{Config, TransactionClient};
/// use futures::prelude::*;
/// # use tikv_client::{Config, TransactionClient};
/// # use futures::prelude::*;
/// # futures::executor::block_on(async {
/// let client = TransactionClient::new(vec!["192.168.0.100"]).await.unwrap();
/// let timestamp = client.current_timestamp().await.unwrap();
@ -165,16 +174,18 @@ impl Client {
self.pd.clone().get_timestamp().await
}
/// Cleans MVCC records whose timestamp is lower than the given `timestamp` in TiKV.
/// Request garbage collection (GC) of the TiKV cluster.
///
/// GC deletes MVCC records whose timestamp is lower than the given `safepoint`.
///
/// For each key, the last mutation record (unless it's a deletion) before `safepoint` is retained.
///
/// It is done by:
/// 1. resolve all locks with ts <= `safepoint`
/// 2. update safepoint to PD
/// GC is performed by:
/// 1. resolving all locks with timestamp <= `safepoint`
/// 2. updating PD's known safepoint
///
/// This is a simplified version of [GC in TiDB](https://docs.pingcap.com/tidb/stable/garbage-collection-overview).
/// We omit the second step "delete ranges" which is an optimization for TiDB.
/// We skip the second step "delete ranges" which is an optimization for TiDB.
pub async fn gc(&self, safepoint: Timestamp) -> Result<bool> {
// scan all locks with ts <= safepoint
let mut locks: Vec<kvrpcpb::LockInfo> = vec![];

3
src/transaction/lock.rs
View File

@ -3,10 +3,11 @@
use crate::{
backoff::{Backoff, DEFAULT_REGION_BACKOFF, OPTIMISTIC_BACKOFF},
pd::PdClient,
region::RegionVerId,
request::Plan,
timestamp::TimestampExt,
transaction::requests,
Error, RegionVerId, Result,
Error, Result,
};
use std::{
collections::{HashMap, HashSet},

6
src/transaction/requests.rs
View File

@ -278,11 +278,11 @@ impl KvRequest for kvrpcpb::CommitRequest {
shardable_keys!(kvrpcpb::CommitRequest);
pub fn new_batch_rollback_request(
keys: Vec<Key>,
keys: Vec<Vec<u8>>,
start_version: u64,
) -> kvrpcpb::BatchRollbackRequest {
let mut req = kvrpcpb::BatchRollbackRequest::default();
req.set_keys(keys.into_iter().map(Into::into).collect());
req.set_keys(keys);
req.set_start_version(start_version);
req
@ -429,7 +429,7 @@ impl Shardable for kvrpcpb::ScanLockRequest {
&self,
pd_client: &Arc<impl PdClient>,
) -> BoxStream<'static, Result<(Self::Shard, Store)>> {
store_stream_for_range_by_start_key(self.start_key.clone(), pd_client.clone())
store_stream_for_range_by_start_key(self.start_key.clone().into(), pd_client.clone())
}
fn apply_shard(&mut self, shard: Self::Shard, store: &Store) -> Result<()> {

171
src/transaction/transaction.rs
View File

@ -17,30 +17,40 @@ use tokio::{sync::RwLock, time::Duration};
/// An undo-able set of actions on the dataset.
///
/// Using a transaction you can prepare a set of actions (such as `get`, or `put`) on data at a
/// particular timestamp called `start_ts` obtained from the placement driver.
/// Once a transaction is commited, a new timestamp called `commit_ts` is obtained from the placement driver.
/// Create a transaction using a [`TransactionClient`](crate::TransactionClient), then run actions
/// (such as `get`, or `put`) on the transaction. Reads are executed immediately, writes are
/// buffered locally. Once complete, `commit` the transaction. Behind the scenes, the client will
/// perform a two phase commit and return success as soon as the writes are guaranteed to be
/// committed (some finalisation may continue in the background after the return, but no data can be
/// lost).
///
/// The snapshot isolation in TiKV ensures that a transaction behaves as if it operates on the snapshot taken at
/// `start_ts` and its mutations take effect at `commit_ts`.
/// In other words, the transaction can read mutations with `commit_ts` <= its `start_ts`,
/// and its mutations are readable for transactions with `start_ts` >= its `commit_ts`.
/// TiKV transactions use multi-version concurrency control. All reads logically happen at the start
/// of the transaction (at the start timestamp, `start_ts`). Once a transaction is commited, a
/// its writes atomically become visible to other transactions at (logically) the commit timestamp.
///
/// Mutations, or write operations made in a transaction are buffered locally and sent at the time of commit,
/// except for pessimistic locking.
/// In pessimistic mode, all write operations or `xxx_for_update` operations will first acquire pessimistic locks in TiKV.
/// A lock exists until the transaction is committed (in the first phase of 2PC) or rolled back, or it exceeds its Time To Live (TTL).
/// In other words, a transaction can read data that was committed at `commit_ts` < its `start_ts`,
/// and its writes are readable by transactions with `start_ts` >= its `commit_ts`.
///
/// Mutations are buffered locally and sent to the TiKV cluster at the time of commit.
/// In a pessimistic transaction, all write operations and `xxx_for_update` operations will immediately
/// acquire locks from TiKV. Such a lock blocks other transactions from writing to that key.
/// A lock exists until the transaction is committed or rolled back, or the lock reaches its time to
/// live (TTL).
///
/// For details, the [SIG-Transaction](https://github.com/tikv/sig-transaction)
/// provides materials explaining designs and implementations of multiple features in TiKV transactions.
/// provides materials explaining designs and implementations of TiKV transactions.
///
/// # Examples
///
/// ```rust,no_run
/// use tikv_client::{Config, TransactionClient};
/// use futures::prelude::*;
/// # use tikv_client::{Config, TransactionClient};
/// # use futures::prelude::*;
/// # futures::executor::block_on(async {
/// let client = TransactionClient::new(vec!["192.168.0.100"]).await.unwrap();
/// let txn = client.begin_optimistic().await.unwrap();
/// let mut txn = client.begin_optimistic().await.unwrap();
/// let foo = txn.get("foo".to_owned()).await.unwrap().unwrap();
/// txn.put("bar".to_owned(), foo).await.unwrap();
/// txn.commit().await.unwrap();
/// # });
/// ```
pub struct Transaction<PdC: PdClient = PdRpcClient> {
@ -89,8 +99,6 @@ impl<PdC: PdClient> Transaction<PdC> {
/// let mut txn = client.begin_optimistic().await.unwrap();
/// let key = "TiKV".to_owned();
/// let result: Option<Value> = txn.get(key).await.unwrap();
/// // Finish the transaction...
/// txn.commit().await.unwrap();
/// # });
/// ```
pub async fn get(&mut self, key: impl Into<Key>) -> Result<Option<Value>> {
@ -115,7 +123,7 @@ impl<PdC: PdClient> Transaction<PdC> {
.await
}
/// Create a `get for udpate` request.
/// Create a `get for update` request.
///
/// The request reads and "locks" a key. It is similar to `SELECT ... FOR
/// UPDATE` in TiDB, and has different behavior in optimistic and
@ -135,13 +143,15 @@ impl<PdC: PdClient> Transaction<PdC> {
/// A lock will be acquired immediately with this request, which prevents
/// concurrent transactions from mutating the keys.
///
/// The "current timestamp" (also called `for_update_ts` of the request) is
/// fetched immediately from the timestamp oracle.
/// The "current timestamp" (also called `for_update_ts` of the request) is fetched from PD.
///
/// Note: The behavior of the request under pessimistic transaction does not
/// follow snapshot isolation.
/// Note: The behavior of this command under pessimistic transaction does not follow snapshot.
/// It reads the latest value (using current timestamp), and the value is not cached in the
/// local buffer. So normal `get`-like commands after `get_for_update` will not be influenced,
/// they still read values at the transaction's `start_ts`.
///
/// # Examples
///
/// ```rust,no_run
/// # use tikv_client::{Value, Config, TransactionClient};
/// # use futures::prelude::*;
@ -171,9 +181,10 @@ impl<PdC: PdClient> Transaction<PdC> {
}
}
/// Check whether the key exists.
/// Check whether a key exists.
///
/// # Examples
///
/// ```rust,no_run
/// # use tikv_client::{Value, Config, TransactionClient};
/// # use futures::prelude::*;
@ -194,9 +205,11 @@ impl<PdC: PdClient> Transaction<PdC> {
/// Once resolved this request will result in the fetching of the values associated with the
/// given keys.
///
/// Non-existent entries will not appear in the result. The order of the keys is not retained in the result.
/// Non-existent entries will not appear in the result. The order of the keys is not retained in
/// the result.
///
/// # Examples
///
/// ```rust,no_run
/// # use tikv_client::{Key, Value, Config, TransactionClient};
/// # use futures::prelude::*;
@ -242,13 +255,14 @@ impl<PdC: PdClient> Transaction<PdC> {
/// Create a new 'batch get for update' request.
///
/// Similar [`get_for_update`](Transaction::get_for_update), but it works
/// Similar to [`get_for_update`](Transaction::get_for_update), but it works
/// for a batch of keys.
///
/// Non-existent entries will not appear in the result. The order of the
/// keys is not retained in the result.
///
/// # Examples
///
/// ```rust,no_run
/// # use tikv_client::{Key, Value, Config, TransactionClient, KvPair};
/// # use futures::prelude::*;
@ -256,12 +270,12 @@ impl<PdC: PdClient> Transaction<PdC> {
/// # futures::executor::block_on(async {
/// # let client = TransactionClient::new(vec!["192.168.0.100", "192.168.0.101"]).await.unwrap();
/// let mut txn = client.begin_pessimistic().await.unwrap();
/// let keys = vec!["TiKV".to_owned(), "TiDB".to_owned()];
/// let keys = vec!["foo".to_owned(), "bar".to_owned()];
/// let result: Vec<KvPair> = txn
/// .batch_get_for_update(keys)
/// .await
/// .unwrap();
/// // now "TiKV" and "TiDB" are both locked
/// // now "foo" and "bar" are both locked
/// // Finish the transaction...
/// txn.commit().await.unwrap();
/// # });
@ -282,12 +296,14 @@ impl<PdC: PdClient> Transaction<PdC> {
/// Create a new 'scan' request.
///
/// Once resolved this request will result in a `Vec` of key-value pairs that lies in the specified range.
/// Once resolved this request will result in a `Vec` of all key-value pairs that lie in the
/// specified range.
///
/// If the number of eligible key-value pairs are greater than `limit`,
/// only the first `limit` pairs are returned, ordered by the key.
/// only the first `limit` pairs are returned, ordered by key.
///
/// # Examples
///
/// ```rust,no_run
/// # use tikv_client::{Key, KvPair, Value, Config, TransactionClient};
/// # use futures::prelude::*;
@ -295,8 +311,8 @@ impl<PdC: PdClient> Transaction<PdC> {
/// # futures::executor::block_on(async {
/// # let client = TransactionClient::new(vec!["192.168.0.100", "192.168.0.101"]).await.unwrap();
/// let mut txn = client.begin_optimistic().await.unwrap();
/// let key1: Key = b"TiKV".to_vec().into();
/// let key2: Key = b"TiDB".to_vec().into();
/// let key1: Key = b"foo".to_vec().into();
/// let key2: Key = b"bar".to_vec().into();
/// let result: Vec<KvPair> = txn
/// .scan(key1..key2, 10)
/// .await
@ -319,9 +335,10 @@ impl<PdC: PdClient> Transaction<PdC> {
/// Once resolved this request will result in a `Vec` of keys that lies in the specified range.
///
/// If the number of eligible keys are greater than `limit`,
/// only the first `limit` keys are returned, ordered by the key.
/// only the first `limit` keys are returned, ordered by key.
///
/// # Examples
///
/// ```rust,no_run
/// # use tikv_client::{Key, KvPair, Value, Config, TransactionClient};
/// # use futures::prelude::*;
@ -329,8 +346,8 @@ impl<PdC: PdClient> Transaction<PdC> {
/// # futures::executor::block_on(async {
/// # let client = TransactionClient::new(vec!["192.168.0.100", "192.168.0.101"]).await.unwrap();
/// let mut txn = client.begin_optimistic().await.unwrap();
/// let key1: Key = b"TiKV".to_vec().into();
/// let key2: Key = b"TiDB".to_vec().into();
/// let key1: Key = b"foo".to_vec().into();
/// let key2: Key = b"bar".to_vec().into();
/// let result: Vec<Key> = txn
/// .scan_keys(key1..key2, 10)
/// .await
@ -353,7 +370,7 @@ impl<PdC: PdClient> Transaction<PdC> {
/// Create a 'scan_reverse' request.
///
/// Similar to [`scan`](Transaction::scan), but in the reverse direction.
/// Similar to [`scan`](Transaction::scan), but scans in the reverse direction.
pub(crate) fn scan_reverse(&self, _range: impl RangeBounds<Key>) -> BoxStream<Result<KvPair>> {
unimplemented!()
}
@ -361,16 +378,16 @@ impl<PdC: PdClient> Transaction<PdC> {
/// Sets the value associated with the given key.
///
/// # Examples
///
/// ```rust,no_run
/// # use tikv_client::{Key, Value, Config, TransactionClient};
/// # use futures::prelude::*;
/// # futures::executor::block_on(async {
/// # let client = TransactionClient::new(vec!["192.168.0.100", "192.168.0.101"]).await.unwrap();
/// let mut txn = client.begin_optimistic().await.unwrap();
/// let key = "TiKV".to_owned();
/// let val = "TiKV".to_owned();
/// let key = "foo".to_owned();
/// let val = "FOO".to_owned();
/// txn.put(key, val);
/// // Finish the transaction...
/// txn.commit().await.unwrap();
/// # });
/// ```
@ -386,19 +403,21 @@ impl<PdC: PdClient> Transaction<PdC> {
}
/// Inserts the value associated with the given key.
/// It has a constraint that key should not exist before.
///
/// Similar to [`put'], but it has an additional constraint that the key should not exist
/// before this operation.
///
/// # Examples
///
/// ```rust,no_run
/// # use tikv_client::{Key, Value, Config, TransactionClient};
/// # use futures::prelude::*;
/// # futures::executor::block_on(async {
/// # let client = TransactionClient::new(vec!["192.168.0.100", "192.168.0.101"]).await.unwrap();
/// let mut txn = client.begin_optimistic().await.unwrap();
/// let key = "TiKV".to_owned();
/// let val = "TiKV".to_owned();
/// let key = "foo".to_owned();
/// let val = "FOO".to_owned();
/// txn.insert(key, val);
/// // Finish the transaction...
/// txn.commit().await.unwrap();
/// # });
/// ```
@ -419,20 +438,20 @@ impl<PdC: PdClient> Transaction<PdC> {
Ok(())
}
/// Deletes the given key.
/// Deletes the given key and its value from the database.
///
/// Deleting a non-existent key will not result in an error.
///
/// # Examples
///
/// ```rust,no_run
/// # use tikv_client::{Key, Config, TransactionClient};
/// # use futures::prelude::*;
/// # futures::executor::block_on(async {
/// # let client = TransactionClient::new(vec!["192.168.0.100", "192.168.0.101"]).await.unwrap();
/// let mut txn = client.begin_optimistic().await.unwrap();
/// let key = "TiKV".to_owned();
/// let key = "foo".to_owned();
/// txn.delete(key);
/// // Finish the transaction...
/// txn.commit().await.unwrap();
/// # });
/// ```
@ -447,17 +466,18 @@ impl<PdC: PdClient> Transaction<PdC> {
Ok(())
}
/// Lock the given keys without mutating value (at the time of commit).
/// Lock the given keys without mutating their values.
///
/// In optimistic mode, write conflicts are not checked until commit.
/// So use this command to indicate that
/// "I do not want to commit if the value associated with this key has been modified".
/// It's useful to avoid *write skew* anomaly.
/// It's useful to avoid the *write skew* anomaly.
///
/// In pessimistic mode, it is similar to [`batch_get_for_update`](Transaction::batch_get_for_update),
/// except that it does not read values.
///
/// # Examples
///
/// ```rust,no_run
/// # use tikv_client::{Config, TransactionClient};
/// # use futures::prelude::*;
@ -488,10 +508,11 @@ impl<PdC: PdClient> Transaction<PdC> {
Ok(())
}
/// Commits the actions of the transaction. On success, we return the commit timestamp (or None
/// if there was nothing to commit).
/// Commits the actions of the transaction. On success, we return the commit timestamp (or
/// `None` if there was nothing to commit).
///
/// # Examples
///
/// ```rust,no_run
/// # use tikv_client::{Config, Timestamp, TransactionClient};
/// # use futures::prelude::*;
@ -499,8 +520,7 @@ impl<PdC: PdClient> Transaction<PdC> {
/// # let client = TransactionClient::new(vec!["192.168.0.100"]).await.unwrap();
/// let mut txn = client.begin_optimistic().await.unwrap();
/// // ... Do some actions.
/// let req = txn.commit();
/// let result: Timestamp = req.await.unwrap().unwrap();
/// let result: Timestamp = txn.commit().await.unwrap().unwrap();
/// # });
/// ```
pub async fn commit(&mut self) -> Result<Option<Timestamp>> {
@ -543,7 +563,20 @@ impl<PdC: PdClient> Transaction<PdC> {
/// Rollback the transaction.
///
/// If it succeeds, all mutations made by this transaciton will not take effect.
/// If it succeeds, all mutations made by this transaction will be discarded.
///
/// # Examples
///
/// ```rust,no_run
/// # use tikv_client::{Config, Timestamp, TransactionClient};
/// # use futures::prelude::*;
/// # futures::executor::block_on(async {
/// # let client = TransactionClient::new(vec!["192.168.0.100"]).await.unwrap();
/// let mut txn = client.begin_optimistic().await.unwrap();
/// // ... Do some actions.
/// txn.rollback().await.unwrap();
/// # });
/// ```
pub async fn rollback(&mut self) -> Result<()> {
{
let status = self.status.read().await;
@ -583,7 +616,7 @@ impl<PdC: PdClient> Transaction<PdC> {
/// Send a heart beat message to keep the transaction alive on the server and update its TTL.
///
/// Returns the TTL set on the lock by the server.
/// Returns the TTL set on the transaction's locks by TiKV.
pub async fn send_heart_beat(&mut self) -> Result<u64> {
self.check_allow_operation().await?;
let primary_key = match self.buffer.get_primary_key().await {
@ -792,11 +825,13 @@ impl<PdC: PdClient> Drop for Transaction<PdC> {
#[derive(Clone, PartialEq, Debug)]
pub enum TransactionKind {
Optimistic,
/// Argument is for_update_ts
/// Argument is the transaction's for_update_ts
Pessimistic(Timestamp),
}
/// Options for configuring a transaction.
///
/// `TransactionOptions` has a builder-style API.
#[derive(Clone, PartialEq, Debug)]
pub struct TransactionOptions {
/// Optimistic or pessimistic (default) transaction.
@ -815,13 +850,6 @@ pub struct TransactionOptions {
auto_heartbeat: bool,
}
#[derive(Clone, Eq, PartialEq, Debug)]
pub enum CheckLevel {
Panic,
Warn,
None,
}
impl Default for TransactionOptions {
fn default() -> TransactionOptions {
Self::new_pessimistic()
@ -909,11 +937,14 @@ impl TransactionOptions {
}
}
/// The transaction should not automatically send heartbeat messages to TiKV to keep itself
// alive.
pub fn no_auto_hearbeat(mut self) -> TransactionOptions {
self.auto_heartbeat = false;
self
}
// Returns true if these options describe a pessimistic transaction.
pub fn is_pessimistic(&self) -> bool {
match self.kind {
TransactionKind::Pessimistic(_) => true,
@ -922,6 +953,22 @@ impl TransactionOptions {
}
}
/// Determines what happens when a transaction is dropped without being rolled back or committed.
///
/// The default is to panic.
#[derive(Clone, Eq, PartialEq, Debug)]
pub enum CheckLevel {
/// The program will panic.
///
/// Note that if the thread is already panicking, then we will not double-panic and abort, but
/// just ignore the issue.
Panic,
/// Log a warning.
Warn,
/// Do nothing
None,
}
/// The default TTL of a lock in milliseconds.
const DEFAULT_LOCK_TTL: u64 = 3000;
/// The default heartbeat interval.
@ -1035,7 +1082,7 @@ impl<PdC: PdClient> Committer<PdC> {
r.min_commit_ts
})
.max()
.map(|ts| Timestamp::from_version(ts));
.map(Timestamp::from_version);
Ok(min_commit_ts)
}