*: support split table

etc/config-template.toml

@@ -134,6 +134,9 @@
 # consistency-check-interval = 0
 
 [coprocessor]
+# When it is true, it will try to split a region with table prefix if
+# that region crosses tables.
+# split-region-on-table = false
 # When the region's size exceeds region-max-size, we will split the region
 # into two which the left region's size will be region-split-size or a little
 # bit smaller.

src/coprocessor/codec/table.rs

@@ -35,6 +35,7 @@ pub const RECORD_PREFIX_SEP: &'static [u8] = b"_r";
 pub const INDEX_PREFIX_SEP: &'static [u8] = b"_i";
 pub const SEP_LEN: usize = 2;
 pub const TABLE_PREFIX_LEN: usize = 1;
+pub const TABLE_PREFIX_KEY_LEN: usize = TABLE_PREFIX_LEN + ID_LEN;
 
 
 trait TableEncoder: NumberEncoder {
@@ -53,6 +54,19 @@ trait TableEncoder: NumberEncoder {
 
 impl<T: Write> TableEncoder for T {}
 
+/// Extract table prefix from table record or index.
+// It is useful in tests.
+pub fn extract_table_prefix(key: &[u8]) -> Result<&[u8]> {
+    if !key.starts_with(TABLE_PREFIX) || key.len() < TABLE_PREFIX_KEY_LEN {
+        Err(invalid_type!(
+            "record key or index key expected, but got {:?}",
+            key
+        ))
+    } else {
+        Ok(&key[..TABLE_PREFIX_KEY_LEN])
+    }
+}
+
 pub fn flatten(data: Datum) -> Result<Datum> {
     match data {
         Datum::Dur(d) => Ok(Datum::I64(d.to_nanos())),
@@ -570,4 +584,15 @@ mod test {
         assert!(res.0.is_empty());
         assert!(res.1.is_none());
     }
+
+    #[test]
+    fn test_extract_table_prefix() {
+        extract_table_prefix(&[]).unwrap_err();
+        extract_table_prefix(b"a\x80\x00\x00\x00\x00\x00\x00\x01").unwrap_err();
+        extract_table_prefix(b"t\x80\x00\x00\x00\x00\x00\x01").unwrap_err();
+        assert_eq!(
+            extract_table_prefix(b"t\x80\x00\x00\x00\x00\x00\x00\x01").unwrap(),
+            extract_table_prefix(b"t\x80\x00\x00\x00\x00\x00\x00\x01_r\xff\xff").unwrap()
+        );
+    }
 }

src/raftstore/coprocessor/config.rs

@@ -18,6 +18,10 @@ use super::Result;
 #[serde(default)]
 #[serde(rename_all = "kebab-case")]
 pub struct Config {
+    /// When it is true, it will try to split a region with table prefix if
+    /// that region crosses tables.
+    pub split_region_on_table: bool,
+
     /// When region [a, b) size meets region_max_size, it will be split
     /// into two region into [a, c), [c, b). And the size of [a, c) will
     /// be region_split_size (or a little bit smaller).
@@ -32,6 +36,7 @@ impl Default for Config {
     fn default() -> Config {
         let split_size = ReadableSize::mb(SPLIT_SIZE_MB);
         Config {
+            split_region_on_table: false,
             region_split_size: split_size,
             region_max_size: split_size / 2 * 3,
         }

src/raftstore/coprocessor/dispatcher.rs

@@ -63,6 +63,12 @@ impl CoprocessorHost {
             SIZE_CHECK_OBSERVER_PRIORITY,
             Box::new(split_size_check_observer),
         );
+        if cfg.split_region_on_table {
+            registry.register_observer(
+                TABLE_CHECK_OBSERVER_PRIORITY,
+                Box::new(TableCheckObserver::default()),
+            );
+        }
         CoprocessorHost { registry: registry }
     }
 

src/raftstore/coprocessor/mod.rs

@@ -28,8 +28,8 @@ pub use self::config::Config;
 pub use self::region_snapshot::{RegionIterator, RegionSnapshot};
 pub use self::dispatcher::{CoprocessorHost, Registry};
 pub use self::error::{Error, Result};
-pub use self::split_check::{SizeCheckObserver, Status as SplitCheckStatus,
-                            SIZE_CHECK_OBSERVER_PRIORITY};
+pub use self::split_check::{SizeCheckObserver, Status as SplitCheckStatus, TableCheckObserver,
+                            SIZE_CHECK_OBSERVER_PRIORITY, TABLE_CHECK_OBSERVER_PRIORITY};
 
 /// Coprocessor is used to provide a convient way to inject code to
 /// KV processing.

src/raftstore/coprocessor/split_check/mod.rs

@@ -11,21 +11,29 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+mod table;
 mod size;
 
 use self::size::SizeStatus;
+use self::table::TableStatus;
 
 pub use self::size::SizeCheckObserver;
 pub const SIZE_CHECK_OBSERVER_PRIORITY: u32 = 200;
+pub use self::table::TableCheckObserver;
+// TableCheckObserver has higher priority than TableCheckObserver.
+// Note that higher means less.
+pub const TABLE_CHECK_OBSERVER_PRIORITY: u32 = SIZE_CHECK_OBSERVER_PRIORITY - 1;
 
 #[derive(Default)]
 pub struct Status {
+    // For TableCheckObserver
+    table: Option<TableStatus>,
     // For SizeCheckObserver
     size: Option<SizeStatus>,
 }
 
 impl Status {
     pub fn skip(&self) -> bool {
-        self.size.is_none()
+        self.table.is_none() && self.size.is_none()
     }
 }