Merge branch 'JanKaul:main' into main

Author: SergeiPatiakin

catalogs/iceberg-rest-catalog/src/apis/configuration.rs

@@ -45,6 +45,7 @@ pub struct ApiKey {
 pub struct AWSv4Key {
     pub access_key: String,
     pub secret_key: SecretString,
+    pub session_token: Option<SecretString>,
     pub region: String,
     pub service: String,
 }
@@ -62,15 +63,17 @@ impl AWSv4Key {
             .body(body)
             .unwrap();
         let signing_settings = SigningSettings::default();
-        let signing_params = SigningParams::builder()
+        let mut builder = SigningParams::builder()
             .access_key(self.access_key.as_str())
             .secret_key(self.secret_key.expose_secret().as_str())
             .region(self.region.as_str())
             .service_name(self.service.as_str())
             .time(SystemTime::now())
-            .settings(signing_settings)
-            .build()
-            .unwrap();
+            .settings(signing_settings);
+        if let Some(session_token) = &self.session_token {
+            builder.set_security_token(Some(session_token.expose_secret().as_str()));
+        }
+        let signing_params = builder.build().unwrap();
         let signable_request = SignableRequest::from(&request);
         let (mut signing_instructions, _signature) =
             sign(signable_request, &signing_params)?.into_parts();

catalogs/iceberg-rest-catalog/src/catalog.rs

@@ -93,7 +93,7 @@ impl Catalog for RestCatalog {
         catalog_api_api::drop_namespace(
             &self.configuration,
             self.name.as_deref(),
-            &namespace.url_encode(),
+            &namespace.to_string(),
         )
         .await
         .map_err(Into::<Error>::into)?;
@@ -107,7 +107,7 @@ impl Catalog for RestCatalog {
         let response = catalog_api_api::load_namespace_metadata(
             &self.configuration,
             self.name.as_deref(),
-            &namespace.url_encode(),
+            &namespace.to_string(),
         )
         .await
         .map_err(Into::<Error>::into)?;
@@ -123,7 +123,7 @@ impl Catalog for RestCatalog {
         catalog_api_api::update_properties(
             &self.configuration,
             self.name.as_deref(),
-            &namespace.url_encode(),
+            &namespace.to_string(),
             models::UpdateNamespacePropertiesRequest { updates, removals },
         )
         .await
@@ -135,7 +135,7 @@ impl Catalog for RestCatalog {
         match catalog_api_api::namespace_exists(
             &self.configuration,
             self.name.as_deref(),
-            &namespace.url_encode(),
+            &namespace.to_string(),
         )
         .await
         {

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

@@ -47,9 +47,17 @@ impl Operation {
                 schema,
                 branch,
             } => {
+                let schema_changed = metadata.current_schema(branch.as_deref())
+                    .map(|s| schema != *s.fields())
+                    .unwrap_or(true);
+
                 let version = metadata.current_version(branch.as_deref())?;
                 let version_id = metadata.versions.keys().max().unwrap_or(&0) + 1;
-                let schema_id = metadata.schemas.keys().max().unwrap_or(&0) + 1;
+                let schema_id = if schema_changed {
+                    metadata.schemas.keys().max().unwrap_or(&0) + 1
+                } else {
+                    *metadata.current_schema(branch.as_deref()).unwrap().schema_id()
+                };
                 let last_column_id = schema.iter().map(|x| x.id).max().unwrap_or(0);
 
                 let version = Version {
@@ -72,28 +80,35 @@ impl Operation {
 
                 let branch_name = branch.unwrap_or("main".to_string());
 
+                let mut view_updates: Vec<ViewUpdate<T>> = if schema_changed {
+                    vec![ViewUpdate::AddSchema {
+                        schema: Schema::from_struct_type(schema, schema_id, None),
+                        last_column_id: Some(last_column_id),
+                    }]
+                } else {
+                    vec![]
+                };
+
+                view_updates.append(&mut vec![
+                    ViewUpdate::AddViewVersion {
+                        view_version: version,
+                    },
+                    ViewUpdate::SetCurrentViewVersion {
+                        view_version_id: version_id,
+                    },
+                    ViewUpdate::SetProperties {
+                        updates: HashMap::from_iter(vec![(
+                            REF_PREFIX.to_string() + &branch_name,
+                            version_id.to_string(),
+                        )]),
+                    },
+                ]);
+
                 Ok((
                     Some(ViewRequirement::AssertViewUuid {
                         uuid: metadata.view_uuid,
                     }),
-                    vec![
-                        ViewUpdate::AddViewVersion {
-                            view_version: version,
-                        },
-                        ViewUpdate::SetCurrentViewVersion {
-                            view_version_id: version_id,
-                        },
-                        ViewUpdate::AddSchema {
-                            schema: Schema::from_struct_type(schema, schema_id, None),
-                            last_column_id: Some(last_column_id),
-                        },
-                        ViewUpdate::SetProperties {
-                            updates: HashMap::from_iter(vec![(
-                                REF_PREFIX.to_string() + &branch_name,
-                                version_id.to_string(),
-                            )]),
-                        },
-                    ],
+                    view_updates,
                 ))
             }
             Operation::UpdateProperties(entries) => Ok((