feat: Support all decompression types (#20)

* Support all decompression types

* Move default compression block size to const

Cargo.toml

@@ -25,9 +25,12 @@ flate2 = "1"
 futures = { version = "0.3", default-features = false, features = ["std"] }
 futures-util = "0.3"
 lazy_static = "1.4"
+lz4_flex = "0.11"
+lzokay-native = "0.1"
 paste = "1.0"
 prost = { version = "0.11" }
 snafu = "0.7"
+snap = "1.1"
 tokio = { version = "1.28", features = [
     "io-util",
     "sync",

src/arrow_reader.rs

@@ -26,6 +26,7 @@ use crate::arrow_reader::column::timestamp::new_timestamp_iter;
 use crate::arrow_reader::column::NullableIterator;
 use crate::error::{self, Result};
 use crate::proto::{StripeFooter, StripeInformation};
+use crate::reader::decompress::Compression;
 use crate::reader::schema::{create_field, TypeDescription};
 use crate::reader::Reader;
 
@@ -427,7 +428,10 @@ impl Stripe {
     ) -> Result<Self> {
         let footer = Arc::new(r.stripe_footer(stripe).clone());
 
-        let compression = r.metadata().postscript.compression();
+        let compression = Compression::from_proto(
+            r.metadata().postscript.compression(),
+            r.metadata().postscript.compression_block_size,
+        );
         //TODO(weny): add tz
         let columns = columns
             .iter()

src/arrow_reader/column.rs

@@ -7,8 +7,8 @@ use tokio::io::{AsyncRead, AsyncReadExt, AsyncSeek, AsyncSeekExt};
 
 use crate::error::{self, Result};
 use crate::proto::stream::Kind;
-use crate::proto::{ColumnEncoding, CompressionKind, StripeFooter, StripeInformation};
-use crate::reader::decompress::Decompressor;
+use crate::proto::{ColumnEncoding, StripeFooter, StripeInformation};
+use crate::reader::decompress::{Compression, Decompressor};
 use crate::reader::schema::TypeDescription;
 use crate::reader::Reader;
 
@@ -25,7 +25,7 @@ pub mod timestamp;
 pub struct Column {
     data: Bytes,
     number_of_rows: u64,
-    compression: CompressionKind,
+    compression: Option<Compression>,
     footer: Arc<StripeFooter>,
     name: String,
     column: Arc<TypeDescription>,
@@ -99,7 +99,7 @@ impl Column {
 
     pub fn new<R: Read + Seek>(
         reader: &mut Reader<R>,
-        compression: CompressionKind,
+        compression: Option<Compression>,
         name: &str,
         column: &Arc<TypeDescription>,
         footer: &Arc<StripeFooter>,
@@ -120,7 +120,7 @@ impl Column {
 
     pub async fn new_async<R: AsyncRead + AsyncSeek + Unpin + Send>(
         reader: &mut Reader<R>,
-        compression: CompressionKind,
+        compression: Option<Compression>,
         name: &str,
         column: &Arc<TypeDescription>,
         footer: &Arc<StripeFooter>,

src/arrow_reader/column/date.rs

@@ -14,12 +14,15 @@ pub struct DateIterator {
 }
 
 pub fn convert_date(data: i64) -> Result<NaiveDate> {
-    let date = NaiveDate::from_ymd_opt(1970, 1, 1)
-        .context(error::InvalidDateSnafu)?
-        .checked_add_days(Days::new(data as u64))
-        .context(error::AddDaysSnafu)?;
-
-    Ok(date)
+    let days = Days::new(data.unsigned_abs());
+    // safe unwrap as is valid date
+    let epoch = NaiveDate::from_ymd_opt(1970, 1, 1).unwrap();
+    let date = if data.is_negative() {
+        epoch.checked_sub_days(days)
+    } else {
+        epoch.checked_add_days(days)
+    };
+    date.context(error::AddDaysSnafu)
 }
 
 impl Iterator for DateIterator {

src/async_arrow_reader.rs

@@ -17,6 +17,7 @@ use crate::arrow_reader::{
 };
 use crate::error::Result;
 use crate::proto::StripeInformation;
+use crate::reader::decompress::Compression;
 use crate::reader::schema::TypeDescription;
 use crate::reader::Reader;
 
@@ -188,7 +189,10 @@ impl Stripe {
     ) -> Result<Self> {
         let footer = Arc::new(r.stripe_footer(stripe).clone());
 
-        let compression = r.metadata().postscript.compression();
+        let compression = Compression::from_proto(
+            r.metadata().postscript.compression(),
+            r.metadata().postscript.compression_block_size,
+        );
         //TODO(weny): add tz
         let mut columns = Vec::with_capacity(column_defs.len());
         for (name, typ) in column_defs.iter() {

src/error.rs

@@ -87,6 +87,24 @@ pub enum Error {
         location: Location,
         source: io::Error,
     },
+
+    #[snafu(display("Failed to build snappy decoder: {}", source))]
+    BuildSnappyDecoder {
+        location: Location,
+        source: snap::Error,
+    },
+
+    #[snafu(display("Failed to build lzo decoder: {}", source))]
+    BuildLzoDecoder {
+        location: Location,
+        source: lzokay_native::Error,
+    },
+
+    #[snafu(display("Failed to build lz4 decoder: {}", source))]
+    BuildLz4Decoder {
+        location: Location,
+        source: lz4_flex::block::DecompressError,
+    },
 }
 
 pub type Result<T> = std::result::Result<T, Error>;