Merge branch 'main' into fix-avro-schema

crates/iceberg/src/spec/manifest.rs

@@ -661,7 +661,7 @@ mod _const_schema {
                 ])),
             )),
         ];
-        let schema = Schema::builder().with_fields(fields).build().unwrap();
+        let schema = Schema::builder().with_fields(fields).build()?;
         schema_to_avro_schema("manifest_entry", &schema)
     }
 

crates/iceberg/src/spec/manifest_list.rs

@@ -215,8 +215,8 @@ impl ManifestListWriter {
     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();
-        writer.close().await.unwrap();
+        writer.write_all(&data).await?;
+        writer.close().await?;
         Ok(())
     }
 }

crates/iceberg/src/spec/table_metadata.rs

@@ -44,8 +44,8 @@ pub(crate) static INITIAL_SEQUENCE_NUMBER: i64 = 0;
 /// Reference to [`TableMetadata`].
 pub type TableMetadataRef = Arc<TableMetadata>;
 
-#[derive(Debug, PartialEq, Serialize, Deserialize, Eq, Clone)]
-#[serde(try_from = "TableMetadataEnum", into = "TableMetadataEnum")]
+#[derive(Debug, PartialEq, Deserialize, Eq, Clone)]
+#[serde(try_from = "TableMetadataEnum")]
 /// Fields for the version 2 of the table metadata.
 ///
 /// We assume that this data structure is always valid, so we will panic when invalid error happens.
@@ -379,6 +379,19 @@ pub(super) mod _serde {
     #[derive(Debug, PartialEq, Eq)]
     pub(super) struct VersionNumber<const V: u8>;
 
+    impl Serialize for TableMetadata {
+        fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+        where
+            S: serde::Serializer,
+        {
+            // we must do a clone here
+            let table_metadata_enum: TableMetadataEnum =
+                self.clone().try_into().map_err(serde::ser::Error::custom)?;
+
+            table_metadata_enum.serialize(serializer)
+        }
+    }
+
     impl<const V: u8> Serialize for VersionNumber<V> {
         fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
         where
@@ -412,12 +425,13 @@ pub(super) mod _serde {
         }
     }
 
-    impl From<TableMetadata> for TableMetadataEnum {
-        fn from(value: TableMetadata) -> Self {
-            match value.format_version {
+    impl TryFrom<TableMetadata> for TableMetadataEnum {
+        type Error = Error;
+        fn try_from(value: TableMetadata) -> Result<Self, Error> {
+            Ok(match value.format_version {
                 FormatVersion::V2 => TableMetadataEnum::V2(value.into()),
-                FormatVersion::V1 => TableMetadataEnum::V1(value.into()),
-            }
+                FormatVersion::V1 => TableMetadataEnum::V1(value.try_into()?),
+            })
         }
     }
 
@@ -676,9 +690,10 @@ pub(super) mod _serde {
         }
     }
 
-    impl From<TableMetadata> for TableMetadataV1 {
-        fn from(v: TableMetadata) -> Self {
-            TableMetadataV1 {
+    impl TryFrom<TableMetadata> for TableMetadataV1 {
+        type Error = Error;
+        fn try_from(v: TableMetadata) -> Result<Self, Error> {
+            Ok(TableMetadataV1 {
                 format_version: VersionNumber::<1>,
                 table_uuid: Some(v.table_uuid),
                 location: v.location,
@@ -687,7 +702,10 @@ pub(super) mod _serde {
                 schema: v
                     .schemas
                     .get(&v.current_schema_id)
-                    .unwrap()
+                    .ok_or(Error::new(
+                        ErrorKind::Unexpected,
+                        "current_schema_id not found in schemas",
+                    ))?
                     .as_ref()
                     .clone()
                     .into(),
@@ -748,7 +766,7 @@ pub(super) mod _serde {
                         .collect(),
                 ),
                 default_sort_order_id: Some(v.default_sort_order_id),
-            }
+            })
         }
     }
 }
@@ -867,12 +885,12 @@ mod tests {
                     {
                         "spec-id": 1,
                         "fields": [
-                            {  
-                                "source-id": 4,  
-                                "field-id": 1000,  
-                                "name": "ts_day",  
+                            {
+                                "source-id": 4,
+                                "field-id": 1000,
+                                "name": "ts_day",
                                 "transform": "day"
-                            } 
+                            }
                         ]
                     }
                 ],
@@ -882,8 +900,8 @@ mod tests {
                     "commit.retry.num-retries": "1"
                 },
                 "metadata-log": [
-                    {  
-                        "metadata-file": "s3://bucket/.../v1.json",  
+                    {
+                        "metadata-file": "s3://bucket/.../v1.json",
                         "timestamp-ms": 1515100
                     }
                 ],

crates/iceberg/src/transaction.rs

@@ -160,7 +160,7 @@ impl<'a> ReplaceSortOrderAction<'a> {
                     .table
                     .metadata()
                     .default_sort_order()
-                    .unwrap()
+                    .expect("default sort order impossible to be None")
                     .order_id,
             },
         ];