Fix out of bounds crash in RleValueDecoder (#7441)

Author: apilloud

parquet/src/encodings/decoding.rs

@@ -454,7 +454,14 @@ impl<T: DataType> Decoder<T> for RleValueDecoder<T> {
 
         // We still need to remove prefix of i32 from the stream.
         const I32_SIZE: usize = mem::size_of::<i32>();
+        if data.len() < I32_SIZE {
+            return Err(eof_err!("Not enough bytes to decode"));
+        }
         let data_size = bit_util::read_num_bytes::<i32>(I32_SIZE, data.as_ref()) as usize;
+        if data.len() - I32_SIZE < data_size {
+            return Err(eof_err!("Not enough bytes to decode"));
+        }
+
         self.decoder = RleDecoder::new(1);
         self.decoder
             .set_data(data.slice(I32_SIZE..I32_SIZE + data_size));
@@ -1451,6 +1458,18 @@ mod tests {
         decoder.set_data(Bytes::from(vec![5, 0, 0, 0]), 1).unwrap();
     }
 
+    #[test]
+    fn test_rle_value_decode_missing_size() {
+        let mut decoder = RleValueDecoder::<BoolType>::new();
+        assert!(decoder.set_data(Bytes::from(vec![0]), 1).is_err());
+    }
+
+    #[test]
+    fn test_rle_value_decode_missing_data() {
+        let mut decoder = RleValueDecoder::<BoolType>::new();
+        assert!(decoder.set_data(Bytes::from(vec![5, 0, 0, 0]), 1).is_err());
+    }
+
     #[test]
     fn test_rle_value_decode_bool_decode() {
         // Test multiple 'put' calls on the same encoder