[Parquet] ALP encoder/decoder support

parquet/Cargo.toml

@@ -89,7 +89,8 @@ zstd = { version = "0.13", default-features = false }
 serde_json = { version = "1.0", features = ["std"], default-features = false }
 arrow = { workspace = true, features = ["ipc", "test_utils", "prettyprint", "json"] }
 arrow-cast = { workspace = true }
-tokio = { version = "1.0", default-features = false, features = ["macros", "rt-multi-thread", "io-util", "fs", "sync"] }
+arrow-csv = { workspace = true }
+tokio = { version = "1.0", default-features = false, features = ["macros", "rt-multi-thread", "io-util", "fs"] }
 rand = { version = "0.9", default-features = false, features = ["std", "std_rng", "thread_rng"] }
 object_store = { workspace = true, features = ["azure", "fs"] }
 sysinfo = { version = "0.39.6", default-features = false, features = ["system"] }

parquet/src/arrow/arrow_writer/mod.rs

@@ -3236,7 +3236,7 @@ mod tests {
                 Encoding::BYTE_STREAM_SPLIT,
             ],
             DataType::Float32 | DataType::Float64 => {
-                vec![Encoding::PLAIN, Encoding::BYTE_STREAM_SPLIT]
+                vec![Encoding::PLAIN, Encoding::BYTE_STREAM_SPLIT, Encoding::ALP]
             }
             _ => vec![Encoding::PLAIN],
         };

parquet/src/basic.rs

@@ -449,6 +449,10 @@ enum Encoding {
   /// afterwards. Note that the use of this encoding with FIXED_LEN_BYTE_ARRAY(N) data may
   /// perform poorly for large values of N.
   BYTE_STREAM_SPLIT = 9;
+  /// Adaptive Lossless floating-Point encoding (ALP).
+  ///
+  /// Currently specified for FLOAT and DOUBLE.
+  ALP = 10;
 }
 );
 
@@ -469,6 +473,7 @@ impl FromStr for Encoding {
             "DELTA_BYTE_ARRAY" | "delta_byte_array" => Ok(Encoding::DELTA_BYTE_ARRAY),
             "RLE_DICTIONARY" | "rle_dictionary" => Ok(Encoding::RLE_DICTIONARY),
             "BYTE_STREAM_SPLIT" | "byte_stream_split" => Ok(Encoding::BYTE_STREAM_SPLIT),
+            "ALP" | "alp" => Ok(Encoding::ALP),
             _ => Err(general_err!("unknown encoding: {}", s)),
         }
     }
@@ -608,6 +613,7 @@ fn i32_to_encoding(val: i32) -> Encoding {
         7 => Encoding::DELTA_BYTE_ARRAY,
         8 => Encoding::RLE_DICTIONARY,
         9 => Encoding::BYTE_STREAM_SPLIT,
+        10 => Encoding::ALP,
         _ => panic!("Impossible encoding {val}"),
     }
 }
@@ -1866,6 +1872,7 @@ mod tests {
         );
         assert_eq!(Encoding::DELTA_BYTE_ARRAY.to_string(), "DELTA_BYTE_ARRAY");
         assert_eq!(Encoding::RLE_DICTIONARY.to_string(), "RLE_DICTIONARY");
+        assert_eq!(Encoding::ALP.to_string(), "ALP");
     }
 
     #[test]
@@ -2174,6 +2181,8 @@ mod tests {
         assert_eq!(encoding, Encoding::RLE_DICTIONARY);
         encoding = "BYTE_STREAM_SPLIT".parse().unwrap();
         assert_eq!(encoding, Encoding::BYTE_STREAM_SPLIT);
+        encoding = "alp".parse().unwrap();
+        assert_eq!(encoding, Encoding::ALP);
 
         // test lowercase
         encoding = "byte_stream_split".parse().unwrap();
@@ -2309,6 +2318,7 @@ mod tests {
             Encoding::PLAIN_DICTIONARY,
             Encoding::RLE_DICTIONARY,
             Encoding::BYTE_STREAM_SPLIT,
+            Encoding::ALP,
         ];
         encodings_roundtrip(encodings.into());
     }

parquet/src/data_type.rs

@@ -702,6 +702,7 @@ pub(crate) mod private {
         + Send
         + HeapSize
         + crate::encodings::decoding::private::GetDecoder
+        + crate::encodings::encoding::private::GetEncoder
         + crate::file::statistics::private::MakeStatistics
     {
         const PHYSICAL_TYPE: Type;