Merge remote-tracking branch 'upstream/main'

Author: gruuya

iceberg-rust/src/materialized_view/transaction/mod.rs

@@ -41,14 +41,14 @@ impl<'view> Transaction<'view> {
     }
 
     /// Update the schmema of the view
-    pub fn update_representation(
+    pub fn update_representations(
         mut self,
-        representation: ViewRepresentation,
+        representations: Vec<ViewRepresentation>,
         schema: StructType,
     ) -> Self {
         self.view_operations
-            .push(ViewOperation::UpdateRepresentation {
-                representation,
+            .push(ViewOperation::UpdateRepresentations {
+                representations,
                 schema,
                 branch: self.branch.clone(),
             });

iceberg-rust/src/object_store/store.rs

@@ -272,7 +272,6 @@ mod tests {
         assert!(result.is_ok());
         let metadata = result.unwrap();
         if let TabularMetadata::Table(table_metadata) = metadata {
-            // Add specific checks for `table_metadata` fields if needed
             assert_eq!(table_metadata.table_uuid.to_string(), "fb072c92-a02b-11e9-ae9c-1bb7bc9eca94");
         } else {
             panic!("Expected TabularMetadata::Table variant");
@@ -336,7 +335,6 @@ mod tests {
         assert!(result.is_ok());
         let metadata = result.unwrap();
         if let TabularMetadata::View(view_metadata) = metadata {
-            // Add specific checks for `view_metadata` fields if needed
             assert_eq!(view_metadata.view_uuid.to_string(), "fa6506c3-7681-40c8-86dc-e36561f83385");
         } else {
             panic!("Expected TabularMetadata::View variant");
@@ -403,7 +401,6 @@ mod tests {
         assert!(result.is_ok());
         let metadata = result.unwrap();
         if let TabularMetadata::View(view_metadata) = metadata {
-            // Add specific checks for `view_metadata` fields if needed
             assert_eq!(view_metadata.view_uuid.to_string(), "fa6506c3-7681-40c8-86dc-e36561f83385");
         } else {
             panic!("Expected TabularMetadata::View variant");
@@ -413,7 +410,7 @@ mod tests {
     #[test]
     fn test_parse_metadata_invalid_json() {
         let location = "/path/to/metadata/v1.metadata.json";
-        let invalid_json_data = r#"{"key": "value""#; // Missing closing brace
+        let invalid_json_data = r#"{"key": "value""#;
         let bytes = invalid_json_data.as_bytes();
 
         let result = parse_metadata(location, bytes);

iceberg-rust/src/view/transaction/mod.rs

@@ -28,13 +28,13 @@ impl<'view> Transaction<'view> {
         }
     }
     /// Update the schmema of the view
-    pub fn update_representation(
+    pub fn update_representations(
         mut self,
-        representation: ViewRepresentation,
+        representations: Vec<ViewRepresentation>,
         schema: StructType,
     ) -> Self {
-        self.operations.push(ViewOperation::UpdateRepresentation {
-            representation,
+        self.operations.push(ViewOperation::UpdateRepresentations {
+            representations,
             schema,
             branch: self.branch.clone(),
         });

iceberg-rust/src/view/transaction/operation.rs

@@ -23,9 +23,9 @@ use crate::{
 /// View operation
 pub enum Operation {
     /// Update vresion
-    UpdateRepresentation {
+    UpdateRepresentations {
         /// Representation to add
-        representation: ViewRepresentation,
+        representations: Vec<ViewRepresentation>,
         /// Schema of the representation
         schema: StructType,
         /// Branch where to add the representation
@@ -35,19 +35,60 @@ pub enum Operation {
     UpdateProperties(Vec<(String, String)>),
 }
 
+// Tries to preserve dialect order
+fn upsert_representation(
+    current_representations: &[ViewRepresentation],
+    new_representation: ViewRepresentation,
+) -> Vec<ViewRepresentation> {
+    let ViewRepresentation::Sql {
+        dialect: new_dialect,
+        ..
+    } = &new_representation;
+    let mut updated = false;
+    let mut representations: Vec<ViewRepresentation> = current_representations
+        .iter()
+        .map(
+            |current_representation @ ViewRepresentation::Sql { dialect, .. }| {
+                if dialect == new_dialect {
+                    updated = true;
+                    new_representation.clone()
+                } else {
+                    current_representation.clone()
+                }
+            },
+        )
+        .collect();
+    if !updated {
+        representations.push(new_representation);
+    }
+    representations
+}
+
+fn upsert_representations(
+    current_representations: &[ViewRepresentation],
+    new_representations: &[ViewRepresentation],
+) -> Vec<ViewRepresentation> {
+    let mut representations: Vec<ViewRepresentation> = current_representations.into();
+    for r in new_representations {
+        representations = upsert_representation(&representations, r.clone());
+    }
+    representations
+}
+
 impl Operation {
     /// Execute operation
     pub async fn execute<T: Materialization>(
         self,
         metadata: &GeneralViewMetadata<T>,
     ) -> Result<(Option<ViewRequirement>, Vec<ViewUpdate<T>>), Error> {
         match self {
-            Operation::UpdateRepresentation {
-                representation,
+            Operation::UpdateRepresentations {
+                representations,
                 schema,
                 branch,
             } => {
-                let schema_changed = metadata.current_schema(branch.as_deref())
+                let schema_changed = metadata
+                    .current_schema(branch.as_deref())
                     .map(|s| schema != *s.fields())
                     .unwrap_or(true);
 
@@ -56,7 +97,10 @@ impl Operation {
                 let schema_id = if schema_changed {
                     metadata.schemas.keys().max().unwrap_or(&0) + 1
                 } else {
-                    *metadata.current_schema(branch.as_deref()).unwrap().schema_id()
+                    *metadata
+                        .current_schema(branch.as_deref())
+                        .unwrap()
+                        .schema_id()
                 };
                 let last_column_id = schema.iter().map(|x| x.id).max().unwrap_or(0);
 
@@ -68,7 +112,10 @@ impl Operation {
                         engine_name: None,
                         engine_version: None,
                     },
-                    representations: vec![representation],
+                    representations: upsert_representations(
+                        version.representations(),
+                        &representations,
+                    ),
                     default_catalog: version.default_catalog.clone(),
                     default_namespace: version.default_namespace.clone(),
                     timestamp_ms: SystemTime::now()
@@ -120,3 +167,48 @@ impl Operation {
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use iceberg_rust_spec::view_metadata::ViewRepresentation;
+
+    use crate::view::transaction::operation::upsert_representations;
+
+    #[test]
+    fn test_upsert_representations() {
+        assert_eq!(
+            upsert_representations(
+                &[
+                    ViewRepresentation::sql("a1", Some("a")),
+                    ViewRepresentation::sql("b1", Some("b"))
+                ],
+                &[
+                    ViewRepresentation::sql("b2", Some("b")),
+                    ViewRepresentation::sql("c2", Some("c"))
+                ]
+            ),
+            vec![
+                ViewRepresentation::sql("a1", Some("a")),
+                ViewRepresentation::sql("b2", Some("b")),
+                ViewRepresentation::sql("c2", Some("c")),
+            ]
+        );
+        assert_eq!(
+            upsert_representations(
+                &[
+                    ViewRepresentation::sql("a1", Some("a")),
+                    ViewRepresentation::sql("b1", Some("b"))
+                ],
+                &[
+                    ViewRepresentation::sql("c2", Some("c")),
+                    ViewRepresentation::sql("a2", Some("a"))
+                ]
+            ),
+            vec![
+                ViewRepresentation::sql("a2", Some("a")),
+                ViewRepresentation::sql("b1", Some("b")),
+                ViewRepresentation::sql("c2", Some("c")),
+            ]
+        );
+    }
+}