Respect other dialects when updating representation

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

@@ -35,6 +35,35 @@ 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
+}
+
 impl Operation {
     /// Execute operation
     pub async fn execute<T: Materialization>(
@@ -47,7 +76,8 @@ impl Operation {
                 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 +86,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 +101,10 @@ impl Operation {
                         engine_name: None,
                         engine_version: None,
                     },
-                    representations: vec![representation],
+                    representations: upsert_representation(
+                        version.representations(),
+                        representation,
+                    ),
                     default_catalog: version.default_catalog.clone(),
                     default_namespace: version.default_namespace.clone(),
                     timestamp_ms: SystemTime::now()