Code complete

Author: liurenjie1024

crates/iceberg/src/error.rs

@@ -325,6 +325,8 @@ define_from_err!(
     "Failed to convert decimal literal to rust decimal"
 );
 
+define_from_err!(std::io::Error, ErrorKind::Unexpected, "IO Operation failed");
+
 /// Helper macro to check arguments.
 ///
 ///

crates/iceberg/src/spec/manifest.rs

@@ -24,15 +24,14 @@ use super::{
 };
 use super::{Literal, UNASSIGNED_SEQUENCE_NUMBER};
 use crate::error::Result;
-use crate::io::{InputFile, OutputFile};
+use crate::io::OutputFile;
 use crate::spec::PartitionField;
 use crate::{Error, ErrorKind};
 use apache_avro::{from_value, to_value, Reader as AvroReader, Writer as AvroWriter};
 use futures::AsyncWriteExt;
 use serde_json::to_vec;
 use std::cmp::min;
 use std::collections::HashMap;
-use std::io::Read;
 use std::str::FromStr;
 use std::sync::Arc;
 
@@ -274,14 +273,16 @@ impl ManifestWriter {
             self.update_field_summary(&entry);
 
             let value = match manifest.metadata.format_version {
-                FormatVersion::V1 => {
-                    to_value(_serde::ManifestEntryV1::try_from(entry, &partition_type)?)?
-                        .resolve(&avro_schema)?
-                }
-                FormatVersion::V2 => {
-                    to_value(_serde::ManifestEntryV2::try_from(entry, &partition_type)?)?
-                        .resolve(&avro_schema)?
-                }
+                FormatVersion::V1 => to_value(_serde::ManifestEntryV1::try_from(
+                    (*entry).clone(),
+                    &partition_type,
+                )?)?
+                .resolve(&avro_schema)?,
+                FormatVersion::V2 => to_value(_serde::ManifestEntryV2::try_from(
+                    (*entry).clone(),
+                    &partition_type,
+                )?)?
+                .resolve(&avro_schema)?,
             };
 
             avro_writer.append(value)?;
@@ -803,6 +804,7 @@ impl ManifestMetadata {
     }
 }
 
+/// Reference to [`ManifestEntry`].
 pub type ManifestEntryRef = Arc<ManifestEntry>;
 
 /// A manifest is an immutable Avro file that lists data files or delete
@@ -1455,7 +1457,7 @@ mod tests {
                 format_version: FormatVersion::V2,
             },
             entries: vec![
-                ManifestEntry {
+                Arc::new(ManifestEntry {
                     status: ManifestStatus::Added,
                     snapshot_id: None,
                     sequence_number: None,
@@ -1478,7 +1480,7 @@ mod tests {
                         equality_ids: Vec::new(),
                         sort_order_id: None,
                     }
-                }
+                })
             ]
         };
 
@@ -1575,7 +1577,7 @@ mod tests {
                 content: ManifestContentType::Data,
                 format_version: FormatVersion::V2,
             },
-            entries: vec![ManifestEntry {
+            entries: vec![Arc::new(ManifestEntry {
                 status: ManifestStatus::Added,
                 snapshot_id: None,
                 sequence_number: None,
@@ -1640,7 +1642,7 @@ mod tests {
                     equality_ids: vec![],
                     sort_order_id: None,
                 },
-            }],
+            })],
         };
 
         let writer = |output_file: OutputFile| ManifestWriter::new(output_file, 1, vec![]);
@@ -1684,7 +1686,7 @@ mod tests {
                 content: ManifestContentType::Data,
                 format_version: FormatVersion::V1,
             },
-            entries: vec![ManifestEntry {
+            entries: vec![Arc::new(ManifestEntry {
                 status: ManifestStatus::Added,
                 snapshot_id: Some(0),
                 sequence_number: None,
@@ -1707,7 +1709,7 @@ mod tests {
                     equality_ids: vec![],
                     sort_order_id: Some(0),
                 }
-            }],
+            })],
         };
 
         let writer =
@@ -1754,7 +1756,7 @@ mod tests {
                 format_version: FormatVersion::V1,
             },
             entries: vec![
-                ManifestEntry {
+                Arc::new(ManifestEntry {
                     status: ManifestStatus::Added,
                     snapshot_id: Some(0),
                     sequence_number: None,
@@ -1795,7 +1797,7 @@ mod tests {
                         equality_ids: vec![],
                         sort_order_id: Some(0),
                     },
-                }
+                })
             ]
         };
 

crates/iceberg/src/spec/manifest_list.rs

@@ -29,7 +29,7 @@ use self::{
     _serde::{ManifestListEntryV1, ManifestListEntryV2},
 };
 
-use super::{FormatVersion, Manifest, ManifestEntry, StructType};
+use super::{FormatVersion, Manifest, StructType};
 use crate::error::Result;
 
 /// Placeholder for sequence number. The field with this value must be replaced with the actual sequence number before it write.
@@ -60,16 +60,16 @@ impl ManifestList {
         bs: &[u8],
         version: FormatVersion,
         partition_type: &StructType,
-    ) -> Result<ManifestList, Error> {
+    ) -> Result<ManifestList> {
         match version {
             FormatVersion::V1 => {
                 let reader = Reader::with_schema(&MANIFEST_LIST_AVRO_SCHEMA_V1, bs)?;
-                let values = Value::Array(reader.collect::<Result<Vec<Value>, _>>()?);
+                let values = Value::Array(reader.collect::<std::result::Result<Vec<Value>, _>>()?);
                 from_value::<_serde::ManifestListV1>(&values)?.try_into(partition_type)
             }
             FormatVersion::V2 => {
                 let reader = Reader::with_schema(&MANIFEST_LIST_AVRO_SCHEMA_V2, bs)?;
-                let values = Value::Array(reader.collect::<Result<Vec<Value>, _>>()?);
+                let values = Value::Array(reader.collect::<std::result::Result<Vec<Value>, _>>()?);
                 from_value::<_serde::ManifestListV2>(&values)?.try_into(partition_type)
             }
         }
@@ -169,7 +169,7 @@ impl ManifestListWriter {
     pub fn add_manifest_entries(
         &mut self,
         manifest_entries: impl Iterator<Item = ManifestListEntry>,
-    ) -> Result<(), Error> {
+    ) -> Result<()> {
         match self.format_version {
             FormatVersion::V1 => {
                 for manifest_entry in manifest_entries {
@@ -212,7 +212,7 @@ impl ManifestListWriter {
     }
 
     /// Write the manifest list to the output file.
-    pub async fn close(self) -> Result<(), Error> {
+    pub async fn close(self) -> Result<()> {
         let data = self.avro_writer.into_inner()?;
         let mut writer = self.output_file.writer().await?;
         writer.write_all(&data).await.unwrap();
@@ -591,7 +591,7 @@ pub enum ManifestContentType {
 impl FromStr for ManifestContentType {
     type Err = Error;
 
-    fn from_str(s: &str) -> Result<Self, Error> {
+    fn from_str(s: &str) -> Result<Self> {
         match s {
             "data" => Ok(ManifestContentType::Data),
             "deletes" => Ok(ManifestContentType::Deletes),
@@ -615,7 +615,7 @@ impl ToString for ManifestContentType {
 impl TryFrom<i32> for ManifestContentType {
     type Error = Error;
 
-    fn try_from(value: i32) -> Result<Self, Self::Error> {
+    fn try_from(value: i32) -> std::result::Result<Self, Self::Error> {
         match value {
             0 => Ok(ManifestContentType::Data),
             1 => Ok(ManifestContentType::Deletes),
@@ -637,16 +637,17 @@ impl ManifestListEntry {
     pub async fn load_manifest(&self, file_io: &FileIO) -> Result<Manifest> {
         let mut avro = Vec::new();
         file_io
-            .new_input(&self.manifest_path)
+            .new_input(&self.manifest_path)?
             .reader()
+            .await?
             .read_to_end(&mut avro)
             .await?;
 
-        let (metadata, mut entries) = Manifest::try_from_avro_bytes(&avro).await?;
+        let (metadata, mut entries) = Manifest::try_from_avro_bytes(&avro)?;
 
         // Let entries inherit values from the manifest list entry.
         for entry in &mut entries {
-            entry.inherit_data(&self);
+            entry.inherit_data(self);
         }
 
         Ok(Manifest::new(metadata, entries))

crates/iceberg/src/spec/snapshot.rs

@@ -21,15 +21,14 @@
 use crate::error::Result;
 use chrono::{DateTime, TimeZone, Utc};
 use futures::AsyncReadExt;
-use itertools::Format;
 use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
 use std::sync::Arc;
 use typed_builder::TypedBuilder;
 
 use super::table_metadata::SnapshotLog;
 use crate::io::FileIO;
-use crate::spec::{FormatVersion, ManifestList, SchemaId, TableMetadata};
+use crate::spec::{ManifestList, SchemaId, TableMetadata};
 use crate::{Error, ErrorKind};
 use _serde::SnapshotV2;
 
@@ -144,15 +143,16 @@ impl Snapshot {
     ) -> Result<ManifestList> {
         match &self.manifest_list {
             ManifestListLocation::ManifestListFile(file) => {
-                let mut manifest_list_content= Vec::with_capacity(1024);
+                let mut manifest_list_content= Vec::new();
                 file_io
                     .new_input(file)?
+                    .reader().await?
                     .read_to_end(&mut manifest_list_content)
                     .await?;
 
                 Ok(ManifestList::parse_with_version(&manifest_list_content, table_metadata.format_version(), table_metadata.default_partition_spec().unwrap().partition_type(table_metadata.current_schema()).as_ref().unwrap())?)
             }
-            ManifestListLocation::ManifestFiles(files) => Err(Error::new(
+            ManifestListLocation::ManifestFiles(_) => Err(Error::new(
                 ErrorKind::FeatureUnsupported,
                 "Loading manifests from `manifests` is currently not supported, we only support loading from `manifest-list` file, see https://iceberg.apache.org/spec/#snapshots for more information.",
             )),
@@ -176,6 +176,7 @@ pub(super) mod _serde {
 
     use serde::{Deserialize, Serialize};
 
+    use crate::spec::SchemaId;
     use crate::{Error, ErrorKind};
 
     use super::{ManifestListLocation, Operation, Snapshot, Summary};
@@ -192,7 +193,7 @@ pub(super) mod _serde {
         pub manifest_list: String,
         pub summary: Summary,
         #[serde(skip_serializing_if = "Option::is_none")]
-        pub schema_id: Option<i64>,
+        pub schema_id: Option<SchemaId>,
     }
 
     #[derive(Debug, Serialize, Deserialize, PartialEq, Eq)]
@@ -210,7 +211,7 @@ pub(super) mod _serde {
         #[serde(skip_serializing_if = "Option::is_none")]
         pub summary: Option<Summary>,
         #[serde(skip_serializing_if = "Option::is_none")]
-        pub schema_id: Option<i64>,
+        pub schema_id: Option<SchemaId>,
     }
 
     impl From<SnapshotV2> for Snapshot {