feat: replace apache-avro with serde_avro_fast and parallelize manifest reads (#203)

Switch Avro deserialization from apache-avro (Value intermediate repr) to
serde_avro_fast (direct bytes→struct), eliminating redundant allocations
for ~10-20x deserialization speedup. Read manifest files concurrently
with buffer_unordered(64) instead of sequentially.

Author: JingsongLi

Cargo.toml

@@ -31,6 +31,5 @@ rust-version = "1.86.0"
 arrow-array = { version = "57.0", features = ["ffi"] }
 arrow-schema = "57.0"
 arrow-cast = "57.0"
-arrow-select = "57.0"
 parquet = "57.0"
 tokio = "1.39.2"

crates/paimon/Cargo.toml

@@ -51,13 +51,12 @@ snafu = "0.9.0"
 typed-builder = "^0.19"
 opendal = { version = "0.55", features = ["services-fs"] }
 pretty_assertions = "1"
-apache-avro = { version = "0.17", features = ["snappy", "zstandard"] }
+serde_avro_fast = { version = "2.0.2", features = ["snappy", "zstandard"] }
 indexmap = "2.5.0"
 roaring = "0.11"
 arrow-array = { workspace = true }
 arrow-cast = { workspace = true }
 arrow-schema = { workspace = true }
-arrow-select = { workspace = true }
 futures = "0.3"
 parquet = { workspace = true, features = ["async", "zstd"] }
 async-stream = "0.3.6"
@@ -76,5 +75,5 @@ urlencoding = "2.1"
 [dev-dependencies]
 axum = { version = "0.7", features = ["macros", "tokio", "http1", "http2"] }
 rand = "0.8.5"
-serde_avro_fast = { version = "2.0.2", features = ["snappy"] }
+
 tempfile = "3"

crates/paimon/src/error.rs

@@ -68,14 +68,6 @@ pub enum Error {
         display("Paimon hitting invalid config: {}", message)
     )]
     ConfigInvalid { message: String },
-    #[snafu(
-        visibility(pub(crate)),
-        display("Paimon hitting unexpected avro error {}: {:?}", message, source)
-    )]
-    DataUnexpected {
-        message: String,
-        source: Box<apache_avro::Error>,
-    },
     #[snafu(
         visibility(pub(crate)),
         display("Paimon hitting invalid file index format: {}", message)
@@ -127,15 +119,6 @@ impl From<opendal::Error> for Error {
     }
 }
 
-impl From<apache_avro::Error> for Error {
-    fn from(source: apache_avro::Error) -> Self {
-        Error::DataUnexpected {
-            message: "".to_string(),
-            source: Box::new(source),
-        }
-    }
-}
-
 impl From<parquet::errors::ParquetError> for Error {
     fn from(source: parquet::errors::ParquetError) -> Self {
         Error::ParquetDataUnexpected {

crates/paimon/src/spec/data_file.rs

@@ -412,6 +412,14 @@ pub struct DataFileMeta {
         skip_serializing_if = "Option::is_none"
     )]
     pub write_cols: Option<Vec<String>>,
+
+    /// External path for the data file (e.g. when data is stored outside the table directory).
+    #[serde(
+        rename = "_EXTERNAL_PATH",
+        default,
+        skip_serializing_if = "Option::is_none"
+    )]
+    pub external_path: Option<String>,
 }
 
 impl Display for DataFileMeta {

crates/paimon/src/spec/index_manifest.rs

@@ -18,12 +18,11 @@
 use crate::io::FileIO;
 use crate::spec::manifest_common::FileKind;
 use crate::spec::IndexFileMeta;
-use apache_avro::types::Value;
-use apache_avro::{from_value, Reader};
 use serde::{Deserialize, Serialize};
+use serde_avro_fast::object_container_file_encoding::Reader;
+use snafu::ResultExt;
 use std::fmt::{Display, Formatter};
 
-use crate::Error;
 use crate::Result;
 
 /// Manifest entry for index file.
@@ -76,12 +75,12 @@ impl IndexManifest {
 
     /// Read index manifest entries from Avro-encoded bytes.
     pub fn read_from_bytes(bytes: &[u8]) -> Result<Vec<IndexManifestEntry>> {
-        let reader = Reader::new(bytes).map_err(Error::from)?;
-        let records = reader
-            .collect::<std::result::Result<Vec<Value>, _>>()
-            .map_err(Error::from)?;
-        let values = Value::Array(records);
-        from_value::<Vec<IndexManifestEntry>>(&values).map_err(Error::from)
+        let mut reader = Reader::from_slice(bytes)
+            .whatever_context::<_, crate::Error>("read index manifest avro")?;
+        reader
+            .deserialize::<IndexManifestEntry>()
+            .collect::<std::result::Result<Vec<_>, _>>()
+            .whatever_context::<_, crate::Error>("deserialize index manifest entry")
     }
 }
 

crates/paimon/src/spec/manifest.rs

@@ -17,10 +17,9 @@
 
 use crate::io::FileIO;
 use crate::spec::manifest_entry::ManifestEntry;
-use apache_avro::types::Value;
-use apache_avro::{from_value, Reader};
+use serde_avro_fast::object_container_file_encoding::Reader;
+use snafu::ResultExt;
 
-use crate::Error;
 use crate::Result;
 
 /// Manifest file reader and writer.
@@ -60,12 +59,12 @@ impl Manifest {
     /// # Returns
     /// A vector of ManifestEntry records
     fn read_from_bytes(bytes: &[u8]) -> Result<Vec<ManifestEntry>> {
-        let reader = Reader::new(bytes).map_err(Error::from)?;
-        let records = reader
-            .collect::<std::result::Result<Vec<Value>, _>>()
-            .map_err(Error::from)?;
-        let values = Value::Array(records);
-        from_value::<Vec<ManifestEntry>>(&values).map_err(Error::from)
+        let mut reader =
+            Reader::from_slice(bytes).whatever_context::<_, crate::Error>("read manifest avro")?;
+        reader
+            .deserialize::<ManifestEntry>()
+            .collect::<std::result::Result<Vec<_>, _>>()
+            .whatever_context::<_, crate::Error>("deserialize manifest entry")
     }
 }
 

crates/paimon/src/spec/manifest_entry.rs

@@ -28,6 +28,9 @@ pub struct Identifier {
     pub bucket: i32,
     pub level: i32,
     pub file_name: String,
+    pub extra_files: Vec<String>,
+    pub embedded_index: Option<Vec<u8>>,
+    pub external_path: Option<String>,
 }
 
 /// Entry of a manifest file, representing an addition / deletion of a data file.
@@ -90,6 +93,9 @@ impl ManifestEntry {
             bucket: self.bucket,
             level: self.file.level,
             file_name: self.file.file_name.clone(),
+            extra_files: self.file.extra_files.clone(),
+            embedded_index: self.file.embedded_index.clone(),
+            external_path: self.file.external_path.clone(),
         }
     }
 

crates/paimon/src/spec/mod.rs

@@ -47,6 +47,7 @@ pub use manifest::Manifest;
 mod manifest_common;
 pub use manifest_common::FileKind;
 mod manifest_entry;
+pub use manifest_entry::Identifier;
 pub use manifest_entry::ManifestEntry;
 mod objects_file;
 pub use objects_file::from_avro_bytes;

crates/paimon/src/spec/objects_file.rs

@@ -15,19 +15,18 @@
 // specific language governing permissions and limitations
 // under the License.
 
-use crate::Error;
-use apache_avro::types::Value;
-use apache_avro::{from_value, Reader};
 use serde::de::DeserializeOwned;
+use serde_avro_fast::object_container_file_encoding::Reader;
+use snafu::ResultExt;
 
 #[allow(dead_code)]
 pub fn from_avro_bytes<T: DeserializeOwned>(bytes: &[u8]) -> crate::Result<Vec<T>> {
-    let reader = Reader::new(bytes).map_err(Error::from)?;
-    let records = reader
-        .collect::<Result<Vec<Value>, _>>()
-        .map_err(Error::from)?;
-    let values = Value::Array(records);
-    from_value::<Vec<T>>(&values).map_err(Error::from)
+    let mut reader = Reader::from_slice(bytes)
+        .whatever_context::<_, crate::Error>("read avro object container")?;
+    reader
+        .deserialize::<T>()
+        .collect::<std::result::Result<Vec<_>, _>>()
+        .whatever_context::<_, crate::Error>("deserialize avro records")
 }
 
 #[cfg(test)]
@@ -122,6 +121,7 @@ mod tests {
                         embedded_index: None,
                         first_row_id: None,
                         write_cols: None,
+                        external_path: None,
                     },
                     2
                 ),
@@ -158,6 +158,7 @@ mod tests {
                         embedded_index: None,
                         first_row_id: None,
                         write_cols: None,
+                        external_path: None,
                     },
                     2
                 ),

crates/paimon/src/table/bin_pack.rs

@@ -98,6 +98,7 @@ mod tests {
             embedded_index: None,
             first_row_id: None,
             write_cols: None,
+            external_path: None,
         }
     }