Merge branch 'main' into er/fix/rub_cap

Cargo.lock

@@ -1855,7 +1855,7 @@ version = "0.1.0"
 dependencies = [
  "flate2",
  "hex",
- "integer-encoding",
+ "integer-encoding 3.0.2",
  "observability_deps",
  "rand 0.8.4",
  "snafu",
@@ -1878,6 +1878,12 @@ version = "1.1.7"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "48dc51180a9b377fd75814d0cc02199c20f8e99433d6762f650d39cdbbd3b56f"
 
+[[package]]
+name = "integer-encoding"
+version = "3.0.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "90c11140ffea82edce8dcd74137ce9324ec24b3cf0175fc9d7e29164da9915b8"
+
 [[package]]
 name = "internal_types"
 version = "0.1.0"
@@ -4552,7 +4558,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "0c6d965454947cc7266d22716ebfd07b18d84ebaf35eec558586bbb2a8cb6b5b"
 dependencies = [
  "byteorder",
- "integer-encoding",
+ "integer-encoding 1.1.7",
  "log",
  "ordered-float 1.1.1",
  "threadpool",

influxdb_tsm/Cargo.toml

@@ -5,7 +5,7 @@ authors = ["Edd Robinson <me@edd.io>"]
 edition = "2018"
 
 [dependencies] # In alphabetical order
-integer-encoding = "1.0.7"
+integer-encoding = "3.0.2"
 snafu = "0.6.2"
 snap = "1.0.0"
 observability_deps = { path = "../observability_deps" }

influxdb_tsm/src/encoders/boolean.rs

@@ -66,10 +66,7 @@ pub fn decode(src: &[u8], dst: &mut Vec<bool>) -> Result<(), Box<dyn Error>> {
     assert_eq!(src[0], BOOLEAN_COMPRESSED_BIT_PACKED << 4);
     let src = &src[HEADER_LEN..];
 
-    let (count, num_bytes_read) = u64::decode_var(src);
-    if num_bytes_read == 0 {
-        return Err("boolean decoder: invalid count".into());
-    }
+    let (count, num_bytes_read) = u64::decode_var(src).ok_or("boolean decoder: invalid count")?;
 
     let mut count: usize = count.try_into()?;
     let src = &src[num_bytes_read..];

influxdb_tsm/src/encoders/integer.rs

@@ -162,16 +162,11 @@ fn decode_rle(src: &[u8], dst: &mut Vec<i64>) -> Result<(), Box<dyn Error>> {
     }
 
     let mut i = 8; // Skip first value
-    let (delta, n) = u64::decode_var(&src[i..]);
-    if n == 0 {
-        return Err(From::from("unable to decode delta"));
-    }
+    let (delta, n) = u64::decode_var(&src[i..]).ok_or("unable to decode delta")?;
+
     i += n;
 
-    let (count, n) = usize::decode_var(&src[i..]);
-    if n == 0 {
-        return Err(From::from("unable to decode count"));
-    }
+    let (count, _n) = usize::decode_var(&src[i..]).ok_or("unable to decode count")?;
 
     if dst.capacity() < count {
         dst.reserve_exact(count - dst.capacity());

influxdb_tsm/src/encoders/string.rs

@@ -90,11 +90,9 @@ pub fn decode(src: &[u8], dst: &mut Vec<Vec<u8>>) -> Result<(), Box<dyn Error>>
     let mut i = 0;
 
     while i < num_decoded_bytes {
-        let (length, num_bytes_read) = u64::decode_var(&decoded_bytes[i..]);
+        let (length, num_bytes_read) =
+            u64::decode_var(&decoded_bytes[i..]).ok_or("invalid encoded string length")?;
         let length: usize = length.try_into()?;
-        if num_bytes_read == 0 {
-            return Err("invalid encoded string length".into());
-        }
 
         let lower = i + num_bytes_read;
         let upper = lower + length;

influxdb_tsm/src/encoders/timestamp.rs

@@ -195,17 +195,11 @@ fn decode_rle(src: &[u8], dst: &mut Vec<i64>) -> Result<(), Box<dyn Error>> {
     let mut a: [u8; 8] = [0; 8];
     a.copy_from_slice(&src[i..i + 8]);
     i += 8;
-    let (mut delta, n) = u64::decode_var(&src[i..]);
-    if n == 0 {
-        return Err(From::from("unable to decode delta"));
-    }
+    let (mut delta, n) = u64::decode_var(&src[i..]).ok_or("unable to decode delta")?;
     i += n;
     delta *= scaler;
 
-    let (count, n) = usize::decode_var(&src[i..]);
-    if n == 0 {
-        return Err(From::from("unable to decode count"));
-    }
+    let (count, _n) = usize::decode_var(&src[i..]).ok_or("unable to decode count")?;
 
     if dst.capacity() < count {
         dst.reserve_exact(count - dst.capacity());

influxdb_tsm/src/reader.rs

@@ -599,7 +599,11 @@ where
 
                 // first decode the timestamp block.
                 let mut ts = Vec::with_capacity(MAX_BLOCK_VALUES); // 1000 is the max block size
-                let (len, n) = u64::decode_var(&data[idx..]); // size of timestamp block
+                                                                   // size of timestamp block
+                let (len, n) = u64::decode_var(&data[idx..]).ok_or_else(|| TsmError {
+                    description: "unable to decode timestamp".into(),
+                })?;
+
                 idx += n;
                 encoders::timestamp::decode(&data[idx..idx + (len as usize)], &mut ts).map_err(
                     |e| TsmError {