diff --git a/src/Common/FailPoint.cpp b/src/Common/FailPoint.cpp
index 71b6b731d871..e885c3134462 100644
--- a/src/Common/FailPoint.cpp
+++ b/src/Common/FailPoint.cpp
@@ -162,6 +162,9 @@ static struct InitFiu
     ONCE(write_file_operation_fail_on_read) \
     REGULAR(slowdown_parallel_replicas_local_plan_read) \
     ONCE(iceberg_writes_cleanup) \
+    ONCE(iceberg_alter_catalog_update_metadata_fail) \
+    REGULAR(iceberg_alter_orphan_metadata_cleanup_fail) \
+    REGULAR(datalake_iceberg_metadata_create_fail) \
     REGULAR(storage_cluster_read_sleep) \
     ONCE(backup_add_empty_memory_table) \
     PAUSEABLE_ONCE(backup_pause_on_start) \
diff --git a/src/Databases/DataLake/GlueCatalog.cpp b/src/Databases/DataLake/GlueCatalog.cpp
index 53ac171c79ff..966fd334a521 100644
--- a/src/Databases/DataLake/GlueCatalog.cpp
+++ b/src/Databases/DataLake/GlueCatalog.cpp
@@ -679,7 +679,8 @@ bool GlueCatalog::updateSchema(
     const String & table_name,
     const String & new_metadata_path,
     Poco::JSON::Object::Ptr /*new_schema*/,
-    Int32 /*previous_schema_id*/) const
+    Int32 /*previous_schema_id*/,
+    Poco::JSON::Object::Ptr /*full_metadata*/) const
 {
     return updateMetadata(namespace_name, table_name, new_metadata_path, nullptr);
 }
diff --git a/src/Databases/DataLake/GlueCatalog.h b/src/Databases/DataLake/GlueCatalog.h
index 919b13a5669f..d5b566050469 100644
--- a/src/Databases/DataLake/GlueCatalog.h
+++ b/src/Databases/DataLake/GlueCatalog.h
@@ -73,7 +73,8 @@ class GlueCatalog final : public ICatalog, private DB::WithContext
         const String & table_name,
         const String & new_metadata_path,
         Poco::JSON::Object::Ptr new_schema,
-        Int32 previous_schema_id) const override;
+        Int32 previous_schema_id,
+        Poco::JSON::Object::Ptr full_metadata = nullptr) const override;
 
     void dropTable(const String & namespace_name, const String & table_name) const override;
 
diff --git a/src/Databases/DataLake/ICatalog.cpp b/src/Databases/DataLake/ICatalog.cpp
index 432b4d8b61c5..70eccb5fc113 100644
--- a/src/Databases/DataLake/ICatalog.cpp
+++ b/src/Databases/DataLake/ICatalog.cpp
@@ -325,7 +325,8 @@ bool ICatalog::updateSchema(
     const String & /*table_name*/,
     const String & /*new_metadata_path*/,
     Poco::JSON::Object::Ptr /*new_schema*/,
-    Int32 /*previous_schema_id*/) const
+    Int32 /*previous_schema_id*/,
+    Poco::JSON::Object::Ptr /*full_metadata*/) const
 {
     throw DB::Exception(DB::ErrorCodes::NOT_IMPLEMENTED, "updateSchema is not implemented");
 }
diff --git a/src/Databases/DataLake/ICatalog.h b/src/Databases/DataLake/ICatalog.h
index e14b00ac3732..8fd5ba85666c 100644
--- a/src/Databases/DataLake/ICatalog.h
+++ b/src/Databases/DataLake/ICatalog.h
@@ -196,7 +196,8 @@ class ICatalog
         const String & table_name,
         const String & new_metadata_path,
         Poco::JSON::Object::Ptr new_schema,
-        Int32 previous_schema_id) const;
+        Int32 previous_schema_id,
+        Poco::JSON::Object::Ptr full_metadata = nullptr) const;
 
     /// Drop table from catalog.
     virtual void dropTable(const String & namespace_name, const String & table_name) const;
diff --git a/src/Databases/DataLake/RestCatalog.cpp b/src/Databases/DataLake/RestCatalog.cpp
index 9f85e9d80d6d..a19148b7e124 100644
--- a/src/Databases/DataLake/RestCatalog.cpp
+++ b/src/Databases/DataLake/RestCatalog.cpp
@@ -8,6 +8,8 @@
 #include <Storages/ObjectStorage/DataLakes/Iceberg/IcebergWrites.h>
 #include <mutex>
 #include <chrono>
+#include <optional>
+#include <sstream>
 #include <unordered_set>
 #include <Core/SettingsEnums.h>
 #include "config.h"
@@ -34,6 +36,7 @@
 #include <Interpreters/Context.h>
 #include <filesystem>
 
+#include <Storages/ObjectStorage/DataLakes/Iceberg/Constant.h>
 #include <Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.h>
 #include <Server/HTTP/HTMLForm.h>
 #include <Formats/FormatFactory.h>
@@ -66,6 +69,7 @@ namespace DB::Setting
 namespace DB::FailPoints
 {
     extern const char check_database_datalake_negative[];
+    extern const char iceberg_alter_catalog_update_metadata_fail[];
 }
 
 namespace DataLake
@@ -149,6 +153,305 @@ std::unordered_set<std::string> getAllowedBigLakeMetadataServiceHosts(
 
 }
 
+namespace
+{
+
+Poco::JSON::Object::Ptr cloneJsonObject(const Poco::JSON::Object::Ptr & obj)
+{
+    std::ostringstream oss; // STYLE_CHECK_ALLOW_STD_STRING_STREAM
+    obj->stringify(oss);
+    Poco::JSON::Parser parser;
+    return parser.parse(oss.str()).extract<Poco::JSON::Object::Ptr>();
+}
+
+bool icebergJsonValueEquals(const Poco::Dynamic::Var & lhs, const Poco::Dynamic::Var & rhs);
+
+bool icebergJsonObjectEquals(const Poco::JSON::Object::Ptr & lhs, const Poco::JSON::Object::Ptr & rhs)
+{
+    if (lhs.isNull() || rhs.isNull())
+        return lhs.isNull() && rhs.isNull();
+    if (lhs->size() != rhs->size())
+        return false;
+    for (auto it = lhs->begin(); it != lhs->end(); ++it)
+    {
+        if (!rhs->has(it->first))
+            return false;
+        if (!icebergJsonValueEquals(it->second, rhs->get(it->first)))
+            return false;
+    }
+    return true;
+}
+
+bool icebergJsonArrayEquals(const Poco::JSON::Array::Ptr & lhs, const Poco::JSON::Array::Ptr & rhs)
+{
+    if (lhs.isNull() || rhs.isNull())
+        return lhs.isNull() && rhs.isNull();
+    if (lhs->size() != rhs->size())
+        return false;
+    for (UInt32 i = 0; i < lhs->size(); ++i)
+        if (!icebergJsonValueEquals(lhs->get(i), rhs->get(i)))
+            return false;
+    return true;
+}
+
+/// Structural, key-order-independent comparison of two parsed JSON values.
+bool icebergJsonValueEquals(const Poco::Dynamic::Var & lhs, const Poco::Dynamic::Var & rhs)
+{
+    const bool lhs_is_object = lhs.type() == typeid(Poco::JSON::Object::Ptr);
+    const bool rhs_is_object = rhs.type() == typeid(Poco::JSON::Object::Ptr);
+    if (lhs_is_object || rhs_is_object)
+    {
+        if (!(lhs_is_object && rhs_is_object))
+            return false;
+        return icebergJsonObjectEquals(lhs.extract<Poco::JSON::Object::Ptr>(), rhs.extract<Poco::JSON::Object::Ptr>());
+    }
+    const bool lhs_is_array = lhs.type() == typeid(Poco::JSON::Array::Ptr);
+    const bool rhs_is_array = rhs.type() == typeid(Poco::JSON::Array::Ptr);
+    if (lhs_is_array || rhs_is_array)
+    {
+        if (!(lhs_is_array && rhs_is_array))
+            return false;
+        return icebergJsonArrayEquals(lhs.extract<Poco::JSON::Array::Ptr>(), rhs.extract<Poco::JSON::Array::Ptr>());
+    }
+    return lhs.toString() == rhs.toString();
+}
+
+/// Two Iceberg schemas are equivalent when they differ only by their `schema-id`.
+bool schemasEquivalentIgnoringId(const Poco::JSON::Object::Ptr & lhs, const Poco::JSON::Object::Ptr & rhs)
+{
+    Poco::JSON::Object::Ptr lhs_copy = cloneJsonObject(lhs);
+    Poco::JSON::Object::Ptr rhs_copy = cloneJsonObject(rhs);
+    lhs_copy->remove(DB::Iceberg::f_schema_id);
+    rhs_copy->remove(DB::Iceberg::f_schema_id);
+    return icebergJsonObjectEquals(lhs_copy, rhs_copy);
+}
+
+void collectSchemaFieldIdsFromFields(const Poco::JSON::Array::Ptr & fields, std::unordered_set<Int32> & ids)
+{
+    for (UInt32 i = 0; i < fields->size(); ++i)
+    {
+        auto field = fields->getObject(i);
+        if (field->has(DB::Iceberg::f_id))
+            ids.insert(field->getValue<Int32>(DB::Iceberg::f_id));
+    }
+}
+
+/// Returns true when the default sort order references field ids that are absent
+/// from the new schema (i.e. the sort order became incompatible after a column drop).
+bool sortOrderIncompatibleWithSchema(
+    const Poco::JSON::Object::Ptr & metadata_obj,
+    const Poco::JSON::Object::Ptr & new_schema_obj)
+{
+    if (!metadata_obj->has(DB::Iceberg::f_sort_orders) || !metadata_obj->has(DB::Iceberg::f_default_sort_order_id))
+        return false;
+
+    const Int64 default_sort_order_id = metadata_obj->getValue<Int64>(DB::Iceberg::f_default_sort_order_id);
+    if (default_sort_order_id == 0)
+        return false;
+
+    auto sort_orders = metadata_obj->getArray(DB::Iceberg::f_sort_orders);
+    Poco::JSON::Object::Ptr default_sort_order;
+    for (UInt32 i = 0; i < sort_orders->size(); ++i)
+    {
+        auto sort_order = sort_orders->getObject(i);
+        if (sort_order->getValue<Int64>(DB::Iceberg::f_order_id) == default_sort_order_id)
+        {
+            default_sort_order = sort_order;
+            break;
+        }
+    }
+
+    if (!default_sort_order || !default_sort_order->has(DB::Iceberg::f_fields))
+        return false;
+
+    auto sort_fields = default_sort_order->getArray(DB::Iceberg::f_fields);
+    if (sort_fields->size() == 0)
+        return false;
+
+    std::unordered_set<Int32> new_schema_field_ids;
+    if (new_schema_obj->has(DB::Iceberg::f_fields))
+        collectSchemaFieldIdsFromFields(new_schema_obj->getArray(DB::Iceberg::f_fields), new_schema_field_ids);
+
+    for (UInt32 i = 0; i < sort_fields->size(); ++i)
+    {
+        auto field = sort_fields->getObject(i);
+        if (!field->has(DB::Iceberg::f_source_id))
+            continue;
+
+        const Int32 source_id = field->getValue<Int32>(DB::Iceberg::f_source_id);
+        if (!new_schema_field_ids.contains(source_id))
+            return true;
+    }
+
+    return false;
+}
+
+}
+
+Poco::JSON::Object::Ptr buildUpdateMetadataRequestBody(
+    const String & namespace_name, const String & table_name, Poco::JSON::Object::Ptr new_snapshot)
+{
+    if (!new_snapshot)
+        return nullptr;
+
+    Poco::JSON::Object::Ptr request_body = new Poco::JSON::Object;
+    {
+        Poco::JSON::Object::Ptr identifier = new Poco::JSON::Object;
+        identifier->set("name", table_name);
+        Poco::JSON::Array::Ptr namespaces = new Poco::JSON::Array;
+        namespaces->add(namespace_name);
+        identifier->set("namespace", namespaces);
+
+        request_body->set("identifier", identifier);
+    }
+
+    if (new_snapshot->has(DB::Iceberg::f_schemas))
+    {
+        if (!new_snapshot->has(DB::Iceberg::f_current_schema_id))
+            throw DB::Exception(
+                DB::ErrorCodes::DATALAKE_DATABASE_ERROR,
+                "Iceberg update-metadata for {}.{} is missing '{}' field",
+                namespace_name, table_name, DB::Iceberg::f_current_schema_id);
+
+        const Int32 new_schema_id = new_snapshot->getValue<Int32>(DB::Iceberg::f_current_schema_id);
+        const Int32 old_schema_id = new_schema_id - 1;
+
+        Poco::JSON::Object::Ptr new_schema_obj;
+        auto schemas = new_snapshot->getArray(DB::Iceberg::f_schemas);
+        for (UInt32 i = 0; i < schemas->size(); ++i)
+        {
+            auto s = schemas->getObject(i);
+            if (s->getValue<Int32>(DB::Iceberg::f_schema_id) == new_schema_id)
+            {
+                new_schema_obj = s;
+                break;
+            }
+        }
+        if (!new_schema_obj)
+            throw DB::Exception(
+                DB::ErrorCodes::DATALAKE_DATABASE_ERROR,
+                "Iceberg update-metadata for {}.{}: no schema object matching current-schema-id={}",
+                namespace_name, table_name, new_schema_id);
+
+        Poco::JSON::Object::Ptr schema_for_rest = cloneJsonObject(new_schema_obj);
+        if (!schema_for_rest->has("identifier-field-ids"))
+        {
+            Poco::JSON::Array::Ptr empty_identifier_field_ids = new Poco::JSON::Array;
+            schema_for_rest->set("identifier-field-ids", empty_identifier_field_ids);
+        }
+
+        if (old_schema_id >= 0)
+        {
+            Poco::JSON::Object::Ptr requirement = new Poco::JSON::Object;
+            requirement->set("type", "assert-current-schema-id");
+            requirement->set("current-schema-id", old_schema_id);
+
+            Poco::JSON::Array::Ptr requirements = new Poco::JSON::Array;
+            requirements->add(requirement);
+            request_body->set("requirements", requirements);
+        }
+
+        /// The target schema may be identical to a schema already present in the table's
+        /// schema history. The Iceberg catalog deduplicates identical schemas, so an
+        /// `add-schema` update becomes a no-op and a subsequent `set-current-schema: -1`
+        /// is rejected. In that case we point `set-current-schema` at the existing id.
+        std::optional<Int32> existing_equivalent_schema_id;
+        for (UInt32 i = 0; i < schemas->size(); ++i)
+        {
+            auto existing_schema = schemas->getObject(i);
+            if (existing_schema->getValue<Int32>(DB::Iceberg::f_schema_id) == new_schema_id)
+                continue;
+            if (schemasEquivalentIgnoringId(existing_schema, new_schema_obj))
+            {
+                existing_equivalent_schema_id = existing_schema->getValue<Int32>(DB::Iceberg::f_schema_id);
+                break;
+            }
+        }
+
+        Poco::JSON::Array::Ptr updates = new Poco::JSON::Array;
+        if (existing_equivalent_schema_id.has_value())
+        {
+            Poco::JSON::Object::Ptr set_current_schema = new Poco::JSON::Object;
+            set_current_schema->set("action", "set-current-schema");
+            set_current_schema->set("schema-id", *existing_equivalent_schema_id);
+            updates->add(set_current_schema);
+        }
+        else
+        {
+            {
+                Poco::JSON::Object::Ptr add_schema = new Poco::JSON::Object;
+                add_schema->set("action", "add-schema");
+                add_schema->set("schema", schema_for_rest);
+                if (new_snapshot->has(DB::Iceberg::f_last_column_id))
+                    add_schema->set("last-column-id", new_snapshot->getValue<Int32>(DB::Iceberg::f_last_column_id));
+                updates->add(add_schema);
+            }
+            {
+                Poco::JSON::Object::Ptr set_current_schema = new Poco::JSON::Object;
+                set_current_schema->set("action", "set-current-schema");
+                set_current_schema->set("schema-id", -1);
+                updates->add(set_current_schema);
+            }
+        }
+
+        if (sortOrderIncompatibleWithSchema(new_snapshot, new_schema_obj))
+        {
+            Poco::JSON::Object::Ptr unsorted_sort_order = new Poco::JSON::Object;
+            unsorted_sort_order->set(DB::Iceberg::f_order_id, 0);
+            unsorted_sort_order->set(DB::Iceberg::f_fields, Poco::JSON::Array::Ptr(new Poco::JSON::Array));
+
+            Poco::JSON::Object::Ptr add_sort_order = new Poco::JSON::Object;
+            add_sort_order->set("action", "add-sort-order");
+            add_sort_order->set("sort-order", unsorted_sort_order);
+            updates->add(add_sort_order);
+
+            Poco::JSON::Object::Ptr set_default_sort_order = new Poco::JSON::Object;
+            set_default_sort_order->set("action", "set-default-sort-order");
+            set_default_sort_order->set("sort-order-id", -1);
+            updates->add(set_default_sort_order);
+        }
+
+        request_body->set("updates", updates);
+    }
+    else
+    {
+        if (new_snapshot->has("parent-snapshot-id"))
+        {
+            auto parent_snapshot_id = new_snapshot->getValue<Int64>("parent-snapshot-id");
+            if (parent_snapshot_id != -1)
+            {
+                Poco::JSON::Object::Ptr requirement = new Poco::JSON::Object;
+                requirement->set("type", "assert-ref-snapshot-id");
+                requirement->set("ref", "main");
+                requirement->set("snapshot-id", parent_snapshot_id);
+
+                Poco::JSON::Array::Ptr requirements = new Poco::JSON::Array;
+                requirements->add(requirement);
+                request_body->set("requirements", requirements);
+            }
+        }
+
+        Poco::JSON::Array::Ptr updates = new Poco::JSON::Array;
+        {
+            Poco::JSON::Object::Ptr add_snapshot = new Poco::JSON::Object;
+            add_snapshot->set("action", "add-snapshot");
+            add_snapshot->set("snapshot", new_snapshot);
+            updates->add(add_snapshot);
+        }
+        {
+            Poco::JSON::Object::Ptr set_snapshot = new Poco::JSON::Object;
+            set_snapshot->set("action", "set-snapshot-ref");
+            set_snapshot->set("ref-name", "main");
+            set_snapshot->set("type", "branch");
+            set_snapshot->set("snapshot-id", new_snapshot->getValue<Int64>("snapshot-id"));
+            updates->add(set_snapshot);
+        }
+        request_body->set("updates", updates);
+    }
+
+    return request_body;
+}
+
 std::string RestCatalog::Config::toString() const
 {
     DB::WriteBufferFromOwnString wb;
@@ -1240,57 +1543,13 @@ void RestCatalog::createTable(const String & namespace_name, const String & tabl
 
 bool RestCatalog::updateMetadata(const String & namespace_name, const String & table_name, const String & /*new_metadata_path*/, Poco::JSON::Object::Ptr new_snapshot) const
 {
-    const std::string endpoint = (base_url / config.prefix / NAMESPACES_ENDPOINT / encodeNamespaceForURI(namespace_name) / "tables" / table_name).generic_string();
-
-    Poco::JSON::Object::Ptr request_body = new Poco::JSON::Object;
-    {
-        Poco::JSON::Object::Ptr identifier = new Poco::JSON::Object;
-        identifier->set("name", table_name);
-        Poco::JSON::Array::Ptr namespaces = new Poco::JSON::Array;
-        namespaces->add(namespace_name);
-        identifier->set("namespace", namespaces);
-
-        request_body->set("identifier", identifier);
-    }
-
-    if (new_snapshot->has("parent-snapshot-id"))
-    {
-        auto parent_snapshot_id = new_snapshot->getValue<Int64>("parent-snapshot-id");
-        if (parent_snapshot_id != -1)
-        {
-            Poco::JSON::Object::Ptr requirement = new Poco::JSON::Object;
-            requirement->set("type", "assert-ref-snapshot-id");
-            requirement->set("ref", "main");
-            requirement->set("snapshot-id", parent_snapshot_id);
-
-            Poco::JSON::Array::Ptr requirements = new Poco::JSON::Array;
-            requirements->add(requirement);
+    fiu_do_on(DB::FailPoints::iceberg_alter_catalog_update_metadata_fail, { return false; });
 
-            request_body->set("requirements", requirements);
-        }
-    }
-
-    {
-        Poco::JSON::Array::Ptr updates = new Poco::JSON::Array;
-
-        {
-            Poco::JSON::Object::Ptr add_snapshot = new Poco::JSON::Object;
-            add_snapshot->set("action", "add-snapshot");
-            add_snapshot->set("snapshot", new_snapshot);
-            updates->add(add_snapshot);
-        }
-
-        {
-            Poco::JSON::Object::Ptr set_snapshot = new Poco::JSON::Object;
-            set_snapshot->set("action", "set-snapshot-ref");
-            set_snapshot->set("ref-name", "main");
-            set_snapshot->set("type", "branch");
-            set_snapshot->set("snapshot-id", new_snapshot->getValue<Int64>("snapshot-id"));
+    const std::string endpoint = (base_url / config.prefix / NAMESPACES_ENDPOINT / encodeNamespaceForURI(namespace_name) / "tables" / table_name).generic_string();
 
-            updates->add(set_snapshot);
-        }
-        request_body->set("updates", updates);
-    }
+    auto request_body = buildUpdateMetadataRequestBody(namespace_name, table_name, new_snapshot);
+    if (!request_body)
+        return true;
 
     try
     {
@@ -1298,7 +1557,8 @@ bool RestCatalog::updateMetadata(const String & namespace_name, const String & t
     }
     catch (const DB::HTTPException & ex)
     {
-        LOG_TRACE(log, "Unsucceeded request {}", ex.what());
+        LOG_WARNING(log, "Iceberg REST updateMetadata for {}.{} failed: {}",
+            namespace_name, table_name, ex.displayText());
         return false;
     }
     return true;
@@ -1309,49 +1569,59 @@ bool RestCatalog::updateSchema(
     const String & table_name,
     const String & /*new_metadata_path*/,
     Poco::JSON::Object::Ptr new_schema,
-    Int32 previous_schema_id) const
+    Int32 previous_schema_id,
+    Poco::JSON::Object::Ptr full_metadata) const
 {
     const std::string endpoint = (base_url / config.prefix / NAMESPACES_ENDPOINT / encodeNamespaceForURI(namespace_name) / "tables" / table_name).generic_string();
 
-    Poco::JSON::Object::Ptr request_body = new Poco::JSON::Object;
-    {
-        Poco::JSON::Object::Ptr identifier = new Poco::JSON::Object;
-        identifier->set("name", table_name);
-        Poco::JSON::Array::Ptr namespaces = new Poco::JSON::Array;
-        namespaces->add(namespace_name);
-        identifier->set("namespace", namespaces);
-
-        request_body->set("identifier", identifier);
-    }
+    Poco::JSON::Object::Ptr request_body;
 
+    /// When full metadata is available, use the richer builder which handles
+    /// equivalent-schema dedup and sort-order reset.
+    if (full_metadata)
     {
-        Poco::JSON::Object::Ptr requirement = new Poco::JSON::Object;
-        requirement->set("type", "assert-current-schema-id");
-        requirement->set("current-schema-id", previous_schema_id);
-
-        Poco::JSON::Array::Ptr requirements = new Poco::JSON::Array;
-        requirements->add(requirement);
-        request_body->set("requirements", requirements);
+        request_body = buildUpdateMetadataRequestBody(namespace_name, table_name, full_metadata);
+        if (!request_body)
+            return true;
     }
-
+    else
     {
-        Poco::JSON::Array::Ptr updates = new Poco::JSON::Array;
-
+        request_body = new Poco::JSON::Object;
         {
-            Poco::JSON::Object::Ptr add_schema = new Poco::JSON::Object;
-            add_schema->set("action", "add-schema");
-            add_schema->set("schema", new_schema);
-            updates->add(add_schema);
+            Poco::JSON::Object::Ptr identifier = new Poco::JSON::Object;
+            identifier->set("name", table_name);
+            Poco::JSON::Array::Ptr namespaces = new Poco::JSON::Array;
+            namespaces->add(namespace_name);
+            identifier->set("namespace", namespaces);
+            request_body->set("identifier", identifier);
         }
 
         {
-            Poco::JSON::Object::Ptr set_current_schema = new Poco::JSON::Object;
-            set_current_schema->set("action", "set-current-schema");
-            set_current_schema->set("schema-id", -1);
-            updates->add(set_current_schema);
+            Poco::JSON::Object::Ptr requirement = new Poco::JSON::Object;
+            requirement->set("type", "assert-current-schema-id");
+            requirement->set("current-schema-id", previous_schema_id);
+
+            Poco::JSON::Array::Ptr requirements = new Poco::JSON::Array;
+            requirements->add(requirement);
+            request_body->set("requirements", requirements);
         }
 
-        request_body->set("updates", updates);
+        {
+            Poco::JSON::Array::Ptr updates = new Poco::JSON::Array;
+            {
+                Poco::JSON::Object::Ptr add_schema = new Poco::JSON::Object;
+                add_schema->set("action", "add-schema");
+                add_schema->set("schema", new_schema);
+                updates->add(add_schema);
+            }
+            {
+                Poco::JSON::Object::Ptr set_current_schema = new Poco::JSON::Object;
+                set_current_schema->set("action", "set-current-schema");
+                set_current_schema->set("schema-id", -1);
+                updates->add(set_current_schema);
+            }
+            request_body->set("updates", updates);
+        }
     }
 
     try
@@ -1360,7 +1630,8 @@ bool RestCatalog::updateSchema(
     }
     catch (const DB::HTTPException & ex)
     {
-        LOG_TRACE(log, "Unsucceeded request {}", ex.what());
+        LOG_WARNING(log, "Iceberg REST updateSchema for {}.{} failed: {}",
+            namespace_name, table_name, ex.displayText());
         return false;
     }
     return true;
diff --git a/src/Databases/DataLake/RestCatalog.h b/src/Databases/DataLake/RestCatalog.h
index 982475ee2c96..00799d781142 100644
--- a/src/Databases/DataLake/RestCatalog.h
+++ b/src/Databases/DataLake/RestCatalog.h
@@ -79,7 +79,8 @@ class RestCatalog : public ICatalog, public DB::WithContext
         const String & table_name,
         const String & new_metadata_path,
         Poco::JSON::Object::Ptr new_schema,
-        Int32 previous_schema_id) const override;
+        Int32 previous_schema_id,
+        Poco::JSON::Object::Ptr full_metadata = nullptr) const override;
 
     bool isTransactional() const override { return true; }
 
@@ -243,6 +244,16 @@ class BigLakeCatalog : public RestCatalog
     AccessToken retrieveGoogleCloudAccessTokenFromRefreshToken() const;
 };
 
+/// Builds the JSON body for `POST .../namespaces/{ns}/tables/{table}` (Iceberg REST update).
+///
+/// Returns `nullptr` when `new_snapshot` is null (nothing to commit). Throws
+/// `DB::Exception(DATALAKE_DATABASE_ERROR)` with a specific message when the metadata
+/// blob is malformed (e.g. missing `current-schema-id`, no schema object matching it).
+Poco::JSON::Object::Ptr buildUpdateMetadataRequestBody(
+    const String & namespace_name,
+    const String & table_name,
+    Poco::JSON::Object::Ptr new_snapshot);
+
 }
 
 #endif
diff --git a/src/Databases/DataLake/tests/gtest_rest_catalog_update_metadata.cpp b/src/Databases/DataLake/tests/gtest_rest_catalog_update_metadata.cpp
new file mode 100644
index 000000000000..11fcc085990c
--- /dev/null
+++ b/src/Databases/DataLake/tests/gtest_rest_catalog_update_metadata.cpp
@@ -0,0 +1,186 @@
+#include "config.h"
+
+#if USE_AVRO
+
+#include <gtest/gtest.h>
+#include <Databases/DataLake/RestCatalog.h>
+#include <Storages/ObjectStorage/DataLakes/Iceberg/Constant.h>
+#include <Common/Exception.h>
+#include <Poco/JSON/Array.h>
+#include <Poco/JSON/Object.h>
+#include <base/types.h>
+#include <sstream>
+
+using namespace DB;
+
+namespace
+{
+Poco::JSON::Object::Ptr findUpdateByAction(const Poco::JSON::Array::Ptr & updates, const std::string & action)
+{
+    for (unsigned int i = 0; i < updates->size(); ++i)
+    {
+        auto o = updates->getObject(i);
+        if (o->getValue<std::string>("action") == action)
+            return o;
+    }
+    return nullptr;
+}
+}
+
+TEST(RestCatalogUpdateMetadataBody, NullSnapshotReturnsNull)
+{
+    auto body = DataLake::buildUpdateMetadataRequestBody("ns", "t", nullptr);
+    EXPECT_FALSE(body);
+}
+
+TEST(RestCatalogUpdateMetadataBody, SchemaUpdateValid)
+{
+    Poco::JSON::Object::Ptr snapshot = new Poco::JSON::Object;
+    Poco::JSON::Array::Ptr schemas = new Poco::JSON::Array;
+    Poco::JSON::Object::Ptr schema = new Poco::JSON::Object;
+    schema->set(Iceberg::f_schema_id, 1);
+    schema->set(Iceberg::f_type, "struct");
+    schema->set(Iceberg::f_fields, Poco::JSON::Array::Ptr(new Poco::JSON::Array));
+    schemas->add(schema);
+    snapshot->set(Iceberg::f_schemas, schemas);
+    snapshot->set(Iceberg::f_current_schema_id, 1);
+    snapshot->set(Iceberg::f_last_column_id, 3);
+
+    auto body = DataLake::buildUpdateMetadataRequestBody("my.ns", "tbl", snapshot);
+    ASSERT_TRUE(body);
+
+    auto id = body->getObject("identifier");
+    EXPECT_EQ(id->getValue<std::string>("name"), "tbl");
+    auto ns = id->getArray("namespace");
+    ASSERT_EQ(ns->size(), 1u);
+    EXPECT_EQ(ns->getElement<std::string>(0), "my.ns");
+
+    ASSERT_TRUE(body->has("requirements"));
+    auto req = body->getArray("requirements")->getObject(0);
+    EXPECT_EQ(req->getValue<std::string>("type"), "assert-current-schema-id");
+    EXPECT_EQ(req->getValue<Int32>("current-schema-id"), 0);
+
+    auto updates = body->getArray("updates");
+    auto add_schema = findUpdateByAction(updates, "add-schema");
+    ASSERT_TRUE(add_schema);
+    EXPECT_TRUE(add_schema->has("schema"));
+    EXPECT_EQ(add_schema->getValue<Int32>("last-column-id"), 3);
+
+    auto set_schema = findUpdateByAction(updates, "set-current-schema");
+    ASSERT_TRUE(set_schema);
+    EXPECT_EQ(set_schema->getValue<Int32>("schema-id"), -1);
+}
+
+TEST(RestCatalogUpdateMetadataBody, SchemaUpdateCurrentIdZeroNoRequirement)
+{
+    Poco::JSON::Object::Ptr snapshot = new Poco::JSON::Object;
+    Poco::JSON::Array::Ptr schemas = new Poco::JSON::Array;
+    Poco::JSON::Object::Ptr schema = new Poco::JSON::Object;
+    schema->set(Iceberg::f_schema_id, 0);
+    schema->set(Iceberg::f_type, "struct");
+    schema->set(Iceberg::f_fields, Poco::JSON::Array::Ptr(new Poco::JSON::Array));
+    schemas->add(schema);
+    snapshot->set(Iceberg::f_schemas, schemas);
+    snapshot->set(Iceberg::f_current_schema_id, 0);
+
+    auto body = DataLake::buildUpdateMetadataRequestBody("ns", "t", snapshot);
+    ASSERT_TRUE(body);
+    EXPECT_FALSE(body->has("requirements"));
+}
+
+TEST(RestCatalogUpdateMetadataBody, SchemaUpdateBodyIsStringifiable)
+{
+    Poco::JSON::Object::Ptr snapshot = new Poco::JSON::Object;
+    Poco::JSON::Array::Ptr schemas = new Poco::JSON::Array;
+    Poco::JSON::Object::Ptr schema = new Poco::JSON::Object;
+    schema->set(Iceberg::f_schema_id, 1);
+    schema->set(Iceberg::f_type, "struct");
+    schema->set(Iceberg::f_fields, Poco::JSON::Array::Ptr(new Poco::JSON::Array));
+    schemas->add(schema);
+    snapshot->set(Iceberg::f_schemas, schemas);
+    snapshot->set(Iceberg::f_current_schema_id, 1);
+
+    auto body = DataLake::buildUpdateMetadataRequestBody("ns", "t", snapshot);
+    ASSERT_TRUE(body);
+
+    std::ostringstream oss; // STYLE_CHECK_ALLOW_STD_STRING_STREAM
+    ASSERT_NO_THROW(body->stringify(oss));
+    EXPECT_NE(oss.str().find("\"identifier-field-ids\""), std::string::npos);
+    EXPECT_NE(oss.str().find("\"add-schema\""), std::string::npos);
+}
+
+TEST(RestCatalogUpdateMetadataBody, SchemaUpdateMissingCurrentSchemaIdThrows)
+{
+    Poco::JSON::Object::Ptr snapshot = new Poco::JSON::Object;
+    snapshot->set(Iceberg::f_schemas, Poco::JSON::Array::Ptr(new Poco::JSON::Array));
+
+    EXPECT_THROW(DataLake::buildUpdateMetadataRequestBody("ns", "t", snapshot), DB::Exception);
+}
+
+TEST(RestCatalogUpdateMetadataBody, SchemaUpdateNoMatchingSchemaIdThrows)
+{
+    Poco::JSON::Object::Ptr snapshot = new Poco::JSON::Object;
+    Poco::JSON::Array::Ptr schemas = new Poco::JSON::Array;
+    Poco::JSON::Object::Ptr schema = new Poco::JSON::Object;
+    schema->set(Iceberg::f_schema_id, 1);
+    schema->set(Iceberg::f_type, "struct");
+    schema->set(Iceberg::f_fields, Poco::JSON::Array::Ptr(new Poco::JSON::Array));
+    schemas->add(schema);
+    snapshot->set(Iceberg::f_schemas, schemas);
+    snapshot->set(Iceberg::f_current_schema_id, 99);
+
+    EXPECT_THROW(DataLake::buildUpdateMetadataRequestBody("ns", "t", snapshot), DB::Exception);
+}
+
+TEST(RestCatalogUpdateMetadataBody, SnapshotUpdateWithParent)
+{
+    Poco::JSON::Object::Ptr snapshot = new Poco::JSON::Object;
+    snapshot->set("snapshot-id", static_cast<Int64>(12345));
+    snapshot->set("parent-snapshot-id", static_cast<Int64>(12344));
+    snapshot->set(Iceberg::f_timestamp_ms, static_cast<Int64>(1700000000000LL));
+
+    auto body = DataLake::buildUpdateMetadataRequestBody("ns", "t", snapshot);
+    ASSERT_TRUE(body);
+
+    ASSERT_TRUE(body->has("requirements"));
+    auto req = body->getArray("requirements")->getObject(0);
+    EXPECT_EQ(req->getValue<std::string>("type"), "assert-ref-snapshot-id");
+    EXPECT_EQ(req->getValue<std::string>("ref"), "main");
+    EXPECT_EQ(req->getValue<Int64>("snapshot-id"), 12344);
+
+    auto updates = body->getArray("updates");
+    auto add_snap = findUpdateByAction(updates, "add-snapshot");
+    ASSERT_TRUE(add_snap);
+    EXPECT_EQ(add_snap->getObject("snapshot")->getValue<Int64>("snapshot-id"), 12345);
+
+    auto set_ref = findUpdateByAction(updates, "set-snapshot-ref");
+    ASSERT_TRUE(set_ref);
+    EXPECT_EQ(set_ref->getValue<Int64>("snapshot-id"), 12345);
+}
+
+TEST(RestCatalogUpdateMetadataBody, SnapshotUpdateWithoutParent)
+{
+    Poco::JSON::Object::Ptr snapshot = new Poco::JSON::Object;
+    snapshot->set("snapshot-id", static_cast<Int64>(999));
+
+    auto body = DataLake::buildUpdateMetadataRequestBody("ns", "t", snapshot);
+    ASSERT_TRUE(body);
+    EXPECT_FALSE(body->has("requirements"));
+
+    auto updates = body->getArray("updates");
+    ASSERT_TRUE(findUpdateByAction(updates, "add-snapshot"));
+    ASSERT_TRUE(findUpdateByAction(updates, "set-snapshot-ref"));
+}
+
+TEST(RestCatalogUpdateMetadataBody, SnapshotUpdateParentMinusOneNoRequirement)
+{
+    Poco::JSON::Object::Ptr snapshot = new Poco::JSON::Object;
+    snapshot->set("snapshot-id", static_cast<Int64>(1));
+    snapshot->set("parent-snapshot-id", static_cast<Int64>(-1));
+
+    auto body = DataLake::buildUpdateMetadataRequestBody("ns", "t", snapshot);
+    ASSERT_TRUE(body);
+    EXPECT_FALSE(body->has("requirements"));
+}
+
+#endif
diff --git a/src/Storages/ObjectStorage/DataLakes/DataLakeConfiguration.h b/src/Storages/ObjectStorage/DataLakes/DataLakeConfiguration.h
index f22078d2799c..bdc5e173addc 100644
--- a/src/Storages/ObjectStorage/DataLakes/DataLakeConfiguration.h
+++ b/src/Storages/ObjectStorage/DataLakes/DataLakeConfiguration.h
@@ -27,6 +27,7 @@
 #include <type_traits>
 
 #include <Common/ErrorCodes.h>
+#include <Common/FailPoint.h>
 #include <Common/filesystemHelpers.h>
 #include <Disks/DiskType.h>
 #include <Disks/DiskObjectStorage/ObjectStorages/IObjectStorage.h>
@@ -48,9 +49,15 @@ namespace ErrorCodes
 {
     extern const int BAD_ARGUMENTS;
     extern const int LOGICAL_ERROR;
+    extern const int NOT_INITIALIZED;
     extern const int PATH_ACCESS_DENIED;
 }
 
+namespace FailPoints
+{
+    extern const char datalake_iceberg_metadata_create_fail[];
+}
+
 namespace DataLakeStorageSetting
 {
     extern DataLakeStorageSettingsDatabaseDataLakeCatalogType storage_catalog_type;
@@ -122,6 +129,7 @@ class DataLakeConfiguration : public BaseStorageConfiguration, public std::enabl
     {
         if (current_metadata != nullptr)
             return;
+        fiu_do_on(FailPoints::datalake_iceberg_metadata_create_fail, { return; });
         BaseStorageConfiguration::update(object_storage, local_context);
         assertLocalPathCorrect(object_storage, local_context);
         current_metadata = DataLakeMetadata::create(object_storage, weak_from_this(), local_context);
@@ -424,7 +432,7 @@ class DataLakeConfiguration : public BaseStorageConfiguration, public std::enabl
     void assertInitialized() const
     {
         if (!current_metadata)
-            throw Exception(ErrorCodes::LOGICAL_ERROR, "Metadata is not initialized");
+            throw Exception(ErrorCodes::NOT_INITIALIZED, "Metadata is not initialized");
     }
 
     ReadFromFormatInfo prepareReadingFromFormat(
diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/Compaction.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/Compaction.cpp
index 66f07c521b27..fcc188dddcb6 100644
--- a/src/Storages/ObjectStorage/DataLakes/Iceberg/Compaction.cpp
+++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/Compaction.cpp
@@ -138,7 +138,10 @@ static Plan getPlan(
         context,
         log.get(),
         persistent_table_components.table_uuid,
-        persistent_table_components.metadata_compression_method);
+        persistent_table_components.metadata_compression_method,
+        /* force_fetch_latest_metadata */ true,
+        /* ignore_explicit_metadata_file_path */ false,
+        /* select_by_table_uuid */ true);
 
     Poco::JSON::Object::Ptr initial_metadata_object
         = getMetadataJSONObject(metadata_file_path, object_storage, persistent_table_components.metadata_cache, context, log, compression_method, persistent_table_components.table_uuid);
diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/MetadataGenerator.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/MetadataGenerator.cpp
index 85f5127c21c4..9a14cec78e9a 100644
--- a/src/Storages/ObjectStorage/DataLakes/Iceberg/MetadataGenerator.cpp
+++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/MetadataGenerator.cpp
@@ -1,3 +1,4 @@
+#include <IO/ReadBufferFromString.h>
 #include <IO/ReadHelpers.h>
 #include <Storages/ObjectStorage/DataLakes/Iceberg/MetadataGenerator.h>
 
@@ -54,6 +55,32 @@ bool checkValidSchemaEvolution(Poco::Dynamic::Var old_type, Poco::Dynamic::Var n
         return true;
     }
 
+    if (old_type.isString() && new_type.isString())
+    {
+        auto old_str = old_type.extract<String>();
+        auto new_str = new_type.extract<String>();
+        if (old_str.starts_with("decimal(") && old_str.ends_with(')')
+            && new_str.starts_with("decimal(") && new_str.ends_with(')'))
+        {
+            auto parse = [](const String & s) -> std::pair<size_t, size_t>
+            {
+                DB::ReadBufferFromString buf(std::string_view(s.begin() + 8, s.end() - 1));
+                size_t p = 0, sc = 0;
+                readIntText(p, buf);
+                skipWhitespaceIfAny(buf);
+                assertChar(',', buf);
+                skipWhitespaceIfAny(buf);
+                tryReadIntText(sc, buf);
+                return {p, sc};
+            };
+            auto [old_precision, old_scale] = parse(old_str);
+            auto [new_precision, new_scale] = parse(new_str);
+            if (old_precision <= new_precision && old_scale <= new_scale)
+                return true;
+        }
+    }
+
+    if (!old_type.isString() && !new_type.isString())
     {
         auto old_complex_type = old_type.extract<Poco::JSON::Object::Ptr>();
         auto new_complex_type = new_type.extract<Poco::JSON::Object::Ptr>();
@@ -69,6 +96,23 @@ bool checkValidSchemaEvolution(Poco::Dynamic::Var old_type, Poco::Dynamic::Var n
     return false;
 }
 
+bool icebergTypesEqual(Poco::Dynamic::Var old_type, Poco::Dynamic::Var new_type)
+{
+    if (old_type.isString() && new_type.isString())
+        return old_type.extract<String>() == new_type.extract<String>();
+
+    if (!old_type.isString() && !new_type.isString())
+    {
+        std::ostringstream oss_old; // STYLE_CHECK_ALLOW_STD_STRING_STREAM
+        std::ostringstream oss_new; // STYLE_CHECK_ALLOW_STD_STRING_STREAM
+        old_type.extract<Poco::JSON::Object::Ptr>()->stringify(oss_old);
+        new_type.extract<Poco::JSON::Object::Ptr>()->stringify(oss_new);
+        return oss_old.str() == oss_new.str();
+    }
+
+    return false;
+}
+
 }
 
 MetadataGenerator::MetadataGenerator(Poco::JSON::Object::Ptr metadata_object_)
@@ -317,10 +361,9 @@ void MetadataGenerator::generateAddColumnMetadata(const String & column_name, Da
     metadata_object->getArray(Iceberg::f_schemas)->add(current_schema);
 }
 
-void MetadataGenerator::generateModifyColumnMetadata(const String & column_name, DataTypePtr type)
+bool MetadataGenerator::generateModifyColumnMetadata(const String & column_name, DataTypePtr type)
 {
     auto current_schema_id = metadata_object->getValue<Int32>(Iceberg::f_current_schema_id);
-    metadata_object->set(Iceberg::f_current_schema_id, current_schema_id + 1);
 
     Poco::JSON::Object::Ptr current_schema;
     auto schemas = metadata_object->getArray(Iceberg::f_schemas);
@@ -335,37 +378,41 @@ void MetadataGenerator::generateModifyColumnMetadata(const String & column_name,
 
     if (!current_schema)
         throw Exception(ErrorCodes::BAD_ARGUMENTS, "Not found schema with id {}", current_schema_id);
-    current_schema = deepCopy(current_schema);
-    auto last_column_id = metadata_object->getValue<Int32>(Iceberg::f_last_column_id);
 
+    auto last_column_id = metadata_object->getValue<Int32>(Iceberg::f_last_column_id);
     auto new_type = Iceberg::getIcebergType(type, last_column_id);
     auto schema_fields = current_schema->getArray(Iceberg::f_fields);
 
-    bool found = false;
     for (UInt32 i = 0; i < schema_fields->size(); ++i)
     {
         auto current_field = schema_fields->getObject(i);
         if (current_field->getValue<String>(Iceberg::f_name) == column_name)
         {
+            if (current_field->getValue<bool>(Iceberg::f_required) == new_type.second
+                && icebergTypesEqual(current_field->get(Iceberg::f_type), new_type.first))
+                return false;
+
             if (!checkValidSchemaEvolution(current_field->get(Iceberg::f_type), new_type.first))
                 throw Exception(ErrorCodes::BAD_ARGUMENTS, "Iceberg spec doesn't allow schema evolution to type {}", type->getPrettyName());
 
-            auto old_type = deepCopy(current_field);
-            current_field->set(Iceberg::f_type, new_type.first);
             if (!current_field->getValue<bool>(Iceberg::f_required) && !type->isNullable())
                 throw Exception(ErrorCodes::BAD_ARGUMENTS, "Iceberg spec doesn't allow change type from nullable to non-nullable {}", type->getPrettyName());
 
+            current_schema = deepCopy(current_schema);
+            schema_fields = current_schema->getArray(Iceberg::f_fields);
+            current_field = schema_fields->getObject(i);
+
+            current_field->set(Iceberg::f_type, new_type.first);
             current_field->set(Iceberg::f_required, new_type.second);
-            found = true;
-            break;
+
+            metadata_object->set(Iceberg::f_current_schema_id, current_schema_id + 1);
+            current_schema->set(Iceberg::f_schema_id, current_schema_id + 1);
+            metadata_object->getArray(Iceberg::f_schemas)->add(current_schema);
+            return true;
         }
     }
 
-    if (!found)
-        throw Exception(ErrorCodes::BAD_ARGUMENTS, "Not found column {}", column_name);
-
-    current_schema->set(Iceberg::f_schema_id, current_schema_id + 1);
-    metadata_object->getArray(Iceberg::f_schemas)->add(current_schema);
+    throw Exception(ErrorCodes::BAD_ARGUMENTS, "Column {} not found in schema", column_name);
 }
 
 void MetadataGenerator::generateRenameColumnMetadata(const String & column_name, const String & new_column_name)
diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/MetadataGenerator.h b/src/Storages/ObjectStorage/DataLakes/Iceberg/MetadataGenerator.h
index de7cbc86d99f..676185c4ae63 100644
--- a/src/Storages/ObjectStorage/DataLakes/Iceberg/MetadataGenerator.h
+++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/MetadataGenerator.h
@@ -43,7 +43,8 @@ class MetadataGenerator
 
     void generateAddColumnMetadata(const String & column_name, DataTypePtr type);
     void generateDropColumnMetadata(const String & column_name);
-    void generateModifyColumnMetadata(const String & column_name, DataTypePtr type);
+    /// Returns false when the column already has the requested type (no metadata change).
+    bool generateModifyColumnMetadata(const String & column_name, DataTypePtr type);
     void generateRenameColumnMetadata(const String & column_name, const String & new_column_name);
 
 private:
diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.cpp
index 3ac3042dc50c..fcd7a1c59680 100644
--- a/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.cpp
+++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.cpp
@@ -510,6 +510,15 @@ std::pair<Poco::Dynamic::Var, bool> getIcebergType(DataTypePtr type, Int32 & ite
 {
     switch (type->getTypeId())
     {
+        case TypeIndex::UInt8:
+        {
+            if (isBool(type))
+                return {"boolean", true};
+            return {"int", true};
+        }
+        case TypeIndex::Int8:
+        case TypeIndex::UInt16:
+        case TypeIndex::Int16:
         case TypeIndex::UInt32:
         case TypeIndex::Int32:
             return {"int", true};
@@ -536,6 +545,11 @@ std::pair<Poco::Dynamic::Var, bool> getIcebergType(DataTypePtr type, Int32 & ite
             return {"string", true};
         case TypeIndex::UUID:
             return {"uuid", true};
+        case TypeIndex::Decimal32:
+        case TypeIndex::Decimal64:
+        case TypeIndex::Decimal128:
+        case TypeIndex::Decimal256:
+            return {"decimal(" + std::to_string(getDecimalPrecision(*type)) + ", " + std::to_string(getDecimalScale(*type)) + ")", true};
         case TypeIndex::Tuple:
         {
             auto type_tuple = std::static_pointer_cast<const DataTypeTuple>(type);
@@ -1244,7 +1258,8 @@ MetadataFileWithInfo getLatestOrExplicitMetadataFileAndVersion(
     const std::optional<String> & table_uuid,
     CompressionMethod known_compression_method,
     bool force_fetch_latest_metadata,
-    bool ignore_explicit_metadata_file_path)
+    bool ignore_explicit_metadata_file_path,
+    bool select_by_table_uuid)
 {
     if (data_lake_settings[DataLakeStorageSetting::iceberg_metadata_file_path].changed && !ignore_explicit_metadata_file_path)
     {
@@ -1304,7 +1319,14 @@ MetadataFileWithInfo getLatestOrExplicitMetadataFileAndVersion(
 
     {
         return getLatestMetadataFileAndVersion(
-            object_storage, table_path, data_lake_settings, metadata_cache, local_context, table_uuid, false, force_fetch_latest_metadata);
+            object_storage,
+            table_path,
+            data_lake_settings,
+            metadata_cache,
+            local_context,
+            table_uuid,
+            select_by_table_uuid && table_uuid.has_value(),
+            force_fetch_latest_metadata);
     }
 }
 
diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.h b/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.h
index 5095974c7e26..42aeda16367b 100644
--- a/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.h
+++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.h
@@ -96,7 +96,8 @@ MetadataFileWithInfo getLatestOrExplicitMetadataFileAndVersion(
     const std::optional<String> & table_uuid,
     CompressionMethod known_compression_method,
     bool force_fetch_latest_metadata = true,
-    bool ignore_explicit_metadata_file_path = false);
+    bool ignore_explicit_metadata_file_path = false,
+    bool select_by_table_uuid = false);
 
 std::pair<Poco::JSON::Object::Ptr, Int32> parseTableSchemaV1Method(const Poco::JSON::Object::Ptr & metadata_object);
 std::pair<Poco::JSON::Object::Ptr, Int32> parseTableSchemaV2Method(const Poco::JSON::Object::Ptr & metadata_object);
diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_type_mapping.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_type_mapping.cpp
new file mode 100644
index 000000000000..0379efd819fd
--- /dev/null
+++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_type_mapping.cpp
@@ -0,0 +1,112 @@
+#include "config.h"
+
+#if USE_AVRO
+
+#include <gtest/gtest.h>
+
+#include <DataTypes/DataTypeFactory.h>
+#include <DataTypes/DataTypeNullable.h>
+#include <DataTypes/DataTypesDecimal.h>
+#include <DataTypes/DataTypesNumber.h>
+#include <DataTypes/IDataType.h>
+#include <Storages/ObjectStorage/DataLakes/Iceberg/Utils.h>
+
+using namespace DB;
+using namespace DB::Iceberg;
+
+TEST(IcebergTypeMapping, BoolMapsToBoolean)
+{
+    auto bool_type = DataTypeFactory::instance().get("Bool");
+    Int32 iter = 0;
+    auto [iceberg_type, required] = getIcebergType(bool_type, iter);
+    ASSERT_TRUE(iceberg_type.isString());
+    EXPECT_EQ(iceberg_type.extract<String>(), "boolean");
+    EXPECT_TRUE(required);
+}
+
+TEST(IcebergTypeMapping, NullableBoolMapsToBoolean)
+{
+    auto bool_type = makeNullable(DataTypeFactory::instance().get("Bool"));
+    Int32 iter = 0;
+    auto [iceberg_type, required] = getIcebergType(bool_type, iter);
+    ASSERT_TRUE(iceberg_type.isString());
+    EXPECT_EQ(iceberg_type.extract<String>(), "boolean");
+    EXPECT_FALSE(required);
+}
+
+TEST(IcebergTypeMapping, UInt8MapsToInt)
+{
+    auto type = std::make_shared<DataTypeUInt8>();
+    Int32 iter = 0;
+    auto [iceberg_type, required] = getIcebergType(type, iter);
+    ASSERT_TRUE(iceberg_type.isString());
+    EXPECT_EQ(iceberg_type.extract<String>(), "int");
+    EXPECT_TRUE(required);
+}
+
+TEST(IcebergTypeMapping, Int8MapsToInt)
+{
+    auto type = std::make_shared<DataTypeInt8>();
+    Int32 iter = 0;
+    auto [iceberg_type, required] = getIcebergType(type, iter);
+    ASSERT_TRUE(iceberg_type.isString());
+    EXPECT_EQ(iceberg_type.extract<String>(), "int");
+}
+
+TEST(IcebergTypeMapping, UInt16MapsToInt)
+{
+    auto type = std::make_shared<DataTypeUInt16>();
+    Int32 iter = 0;
+    auto [iceberg_type, required] = getIcebergType(type, iter);
+    ASSERT_TRUE(iceberg_type.isString());
+    EXPECT_EQ(iceberg_type.extract<String>(), "int");
+}
+
+TEST(IcebergTypeMapping, Int16MapsToInt)
+{
+    auto type = std::make_shared<DataTypeInt16>();
+    Int32 iter = 0;
+    auto [iceberg_type, required] = getIcebergType(type, iter);
+    ASSERT_TRUE(iceberg_type.isString());
+    EXPECT_EQ(iceberg_type.extract<String>(), "int");
+}
+
+TEST(IcebergTypeMapping, Decimal32MapsToDecimal)
+{
+    auto type = std::make_shared<DataTypeDecimal<Decimal32>>(9, 2);
+    Int32 iter = 0;
+    auto [iceberg_type, required] = getIcebergType(type, iter);
+    ASSERT_TRUE(iceberg_type.isString());
+    EXPECT_EQ(iceberg_type.extract<String>(), "decimal(9, 2)");
+    EXPECT_TRUE(required);
+}
+
+TEST(IcebergTypeMapping, Decimal64MapsToDecimal)
+{
+    auto type = std::make_shared<DataTypeDecimal<Decimal64>>(18, 5);
+    Int32 iter = 0;
+    auto [iceberg_type, required] = getIcebergType(type, iter);
+    ASSERT_TRUE(iceberg_type.isString());
+    EXPECT_EQ(iceberg_type.extract<String>(), "decimal(18, 5)");
+}
+
+TEST(IcebergTypeMapping, Decimal128MapsToDecimal)
+{
+    auto type = std::make_shared<DataTypeDecimal<Decimal128>>(38, 10);
+    Int32 iter = 0;
+    auto [iceberg_type, required] = getIcebergType(type, iter);
+    ASSERT_TRUE(iceberg_type.isString());
+    EXPECT_EQ(iceberg_type.extract<String>(), "decimal(38, 10)");
+}
+
+TEST(IcebergTypeMapping, NullableDecimalMapsToDecimalNotRequired)
+{
+    auto type = makeNullable(std::make_shared<DataTypeDecimal<Decimal32>>(7, 3));
+    Int32 iter = 0;
+    auto [iceberg_type, required] = getIcebergType(type, iter);
+    ASSERT_TRUE(iceberg_type.isString());
+    EXPECT_EQ(iceberg_type.extract<String>(), "decimal(7, 3)");
+    EXPECT_FALSE(required);
+}
+
+#endif
diff --git a/tests/integration/test_database_iceberg/test.py b/tests/integration/test_database_iceberg/test.py
index 5d233f89d7ca..c451a42d8d55 100644
--- a/tests/integration/test_database_iceberg/test.py
+++ b/tests/integration/test_database_iceberg/test.py
@@ -1335,3 +1335,4 @@ def test_partitioning_by_string(started_cluster):
     create_clickhouse_iceberg_database(started_cluster, node, CATALOG_NAME)
 
     assert node.query(f"SELECT * FROM {CATALOG_NAME}.`{namespace}.{table_name}`") == "a:b,c[d=e/f%g?h\ttest\t12:00:00.000000\n"
+
diff --git a/tests/integration/test_storage_iceberg_no_spark/test_writes_add_column.py b/tests/integration/test_storage_iceberg_no_spark/test_writes_add_column.py
new file mode 100644
index 000000000000..b1c8a08407cc
--- /dev/null
+++ b/tests/integration/test_storage_iceberg_no_spark/test_writes_add_column.py
@@ -0,0 +1,106 @@
+import pytest
+
+from helpers.iceberg_utils import (
+    create_iceberg_table,
+    get_uuid_str,
+)
+
+INSERT_SETTINGS = {"allow_insert_into_iceberg": 1}
+
+
+@pytest.mark.parametrize("format_version", [1, 2])
+@pytest.mark.parametrize("storage_type", ["local", "s3"])
+def test_add_column_basic(started_cluster_iceberg_no_spark, format_version, storage_type):
+    """ADD COLUMN (nullable): existing rows read with NULL in the new column; new inserts can set it."""
+    instance = started_cluster_iceberg_no_spark.instances["node1"]
+    TABLE_NAME = "test_add_column_basic_" + storage_type + "_" + get_uuid_str()
+
+    create_iceberg_table(
+        storage_type,
+        instance,
+        TABLE_NAME,
+        started_cluster_iceberg_no_spark,
+        "(id Int32, value Nullable(String))",
+        format_version,
+    )
+
+    instance.query(f"INSERT INTO {TABLE_NAME} VALUES (1, 'hello'), (2, 'world');", settings=INSERT_SETTINGS)
+    assert instance.query(f"SELECT id, value FROM {TABLE_NAME} ORDER BY id") == "1\thello\n2\tworld\n"
+
+    instance.query(f"ALTER TABLE {TABLE_NAME} ADD COLUMN extra Nullable(Int32);", settings=INSERT_SETTINGS)
+
+    assert instance.query(f"SELECT id, value, extra FROM {TABLE_NAME} ORDER BY id") == (
+        "1\thello\t\\N\n2\tworld\t\\N\n"
+    )
+
+    instance.query(f"INSERT INTO {TABLE_NAME} VALUES (3, 'foo', 7);", settings=INSERT_SETTINGS)
+    assert instance.query(f"SELECT id, value, extra FROM {TABLE_NAME} ORDER BY id") == (
+        "1\thello\t\\N\n2\tworld\t\\N\n3\tfoo\t7\n"
+    )
+
+
+@pytest.mark.parametrize("format_version", [1, 2])
+@pytest.mark.parametrize("storage_type", ["local", "s3"])
+def test_add_column_errors(started_cluster_iceberg_no_spark, format_version, storage_type):
+    """Non-nullable ADD COLUMN and duplicate name must fail; schema unchanged."""
+    instance = started_cluster_iceberg_no_spark.instances["node1"]
+    TABLE_NAME = "test_add_column_errors_" + storage_type + "_" + get_uuid_str()
+
+    create_iceberg_table(
+        storage_type,
+        instance,
+        TABLE_NAME,
+        started_cluster_iceberg_no_spark,
+        "(id Int32, value Nullable(String))",
+        format_version,
+    )
+
+    error = instance.query_and_get_error(
+        f"ALTER TABLE {TABLE_NAME} ADD COLUMN bad Int32;",
+        settings=INSERT_SETTINGS,
+    )
+    assert "non-nullable" in error.lower() or "doesn't allow" in error.lower()
+
+    error = instance.query_and_get_error(
+        f"ALTER TABLE {TABLE_NAME} ADD COLUMN value Nullable(Int32);",
+        settings=INSERT_SETTINGS,
+    )
+    assert "DUPLICATE_COLUMN" in error or "already exists" in error
+
+    assert instance.query(
+        f"SELECT name FROM system.columns WHERE database = currentDatabase() AND table = '{TABLE_NAME}' ORDER BY name"
+    ) == "id\nvalue\n"
+
+
+@pytest.mark.parametrize("format_version", [1, 2])
+@pytest.mark.parametrize("storage_type", ["local", "s3"])
+def test_add_column_bool_and_decimal(started_cluster_iceberg_no_spark, format_version, storage_type):
+    """ADD COLUMN with Bool (Iceberg boolean) and Decimal (Iceberg decimal) types."""
+    instance = started_cluster_iceberg_no_spark.instances["node1"]
+    TABLE_NAME = "test_add_column_bool_dec_" + storage_type + "_" + get_uuid_str()
+
+    create_iceberg_table(
+        storage_type,
+        instance,
+        TABLE_NAME,
+        started_cluster_iceberg_no_spark,
+        "(id Int32, value Nullable(String))",
+        format_version,
+    )
+
+    instance.query(f"INSERT INTO {TABLE_NAME} VALUES (1, 'a'), (2, 'b');", settings=INSERT_SETTINGS)
+
+    instance.query(f"ALTER TABLE {TABLE_NAME} ADD COLUMN flag Nullable(Bool);", settings=INSERT_SETTINGS)
+    instance.query(f"ALTER TABLE {TABLE_NAME} ADD COLUMN price Nullable(Decimal(10, 2));", settings=INSERT_SETTINGS)
+
+    assert instance.query(f"SELECT id, value, flag, price FROM {TABLE_NAME} ORDER BY id") == (
+        "1\ta\t\\N\t\\N\n2\tb\t\\N\t\\N\n"
+    )
+
+    instance.query(
+        f"INSERT INTO {TABLE_NAME} VALUES (3, 'c', true, 99.95), (4, 'd', false, 123.40);",
+        settings=INSERT_SETTINGS,
+    )
+    assert instance.query(f"SELECT id, flag, price FROM {TABLE_NAME} ORDER BY id") == (
+        "1\t\\N\t\\N\n2\t\\N\t\\N\n3\ttrue\t99.95\n4\tfalse\t123.40\n"
+    )
diff --git a/tests/integration/test_storage_iceberg_no_spark/test_writes_drop_column.py b/tests/integration/test_storage_iceberg_no_spark/test_writes_drop_column.py
new file mode 100644
index 000000000000..f29f8903e2c6
--- /dev/null
+++ b/tests/integration/test_storage_iceberg_no_spark/test_writes_drop_column.py
@@ -0,0 +1,62 @@
+import pytest
+
+from helpers.iceberg_utils import (
+    create_iceberg_table,
+    get_uuid_str,
+)
+
+INSERT_SETTINGS = {"allow_insert_into_iceberg": 1}
+
+
+@pytest.mark.parametrize("format_version", [1, 2])
+@pytest.mark.parametrize("storage_type", ["local", "s3"])
+def test_drop_column_basic(started_cluster_iceberg_no_spark, format_version, storage_type):
+    """DROP COLUMN removes the column from reads and inserts; remaining columns unchanged."""
+    instance = started_cluster_iceberg_no_spark.instances["node1"]
+    TABLE_NAME = "test_drop_column_basic_" + storage_type + "_" + get_uuid_str()
+
+    create_iceberg_table(
+        storage_type,
+        instance,
+        TABLE_NAME,
+        started_cluster_iceberg_no_spark,
+        "(id Int32, value Nullable(String))",
+        format_version,
+    )
+
+    instance.query(f"INSERT INTO {TABLE_NAME} VALUES (1, 'hello'), (2, 'world');", settings=INSERT_SETTINGS)
+    assert instance.query(f"SELECT id, value FROM {TABLE_NAME} ORDER BY id") == "1\thello\n2\tworld\n"
+
+    instance.query(f"ALTER TABLE {TABLE_NAME} DROP COLUMN value;", settings=INSERT_SETTINGS)
+
+    assert instance.query(f"SELECT id FROM {TABLE_NAME} ORDER BY id") == "1\n2\n"
+
+    instance.query(f"INSERT INTO {TABLE_NAME} VALUES (3);", settings=INSERT_SETTINGS)
+    assert instance.query(f"SELECT id FROM {TABLE_NAME} ORDER BY id") == "1\n2\n3\n"
+
+
+@pytest.mark.parametrize("format_version", [1, 2])
+@pytest.mark.parametrize("storage_type", ["local", "s3"])
+def test_drop_column_errors(started_cluster_iceberg_no_spark, format_version, storage_type):
+    """Dropping a non-existent column must fail; table structure unchanged."""
+    instance = started_cluster_iceberg_no_spark.instances["node1"]
+    TABLE_NAME = "test_drop_column_errors_" + storage_type + "_" + get_uuid_str()
+
+    create_iceberg_table(
+        storage_type,
+        instance,
+        TABLE_NAME,
+        started_cluster_iceberg_no_spark,
+        "(id Int32, value Nullable(String))",
+        format_version,
+    )
+
+    error = instance.query_and_get_error(
+        f"ALTER TABLE {TABLE_NAME} DROP COLUMN nonexistent;",
+        settings=INSERT_SETTINGS,
+    )
+    assert "nonexistent" in error
+
+    assert instance.query(
+        f"SELECT name FROM system.columns WHERE database = currentDatabase() AND table = '{TABLE_NAME}' ORDER BY name"
+    ) == "id\nvalue\n"
diff --git a/tests/integration/test_storage_iceberg_no_spark/test_writes_modify_column.py b/tests/integration/test_storage_iceberg_no_spark/test_writes_modify_column.py
new file mode 100644
index 000000000000..314927cd586b
--- /dev/null
+++ b/tests/integration/test_storage_iceberg_no_spark/test_writes_modify_column.py
@@ -0,0 +1,70 @@
+import pytest
+
+from helpers.iceberg_utils import (
+    create_iceberg_table,
+    get_uuid_str,
+)
+
+INSERT_SETTINGS = {"allow_insert_into_iceberg": 1}
+
+
+@pytest.mark.parametrize("format_version", [1, 2])
+@pytest.mark.parametrize("storage_type", ["local", "s3"])
+def test_modify_column_basic(started_cluster_iceberg_no_spark, format_version, storage_type):
+    """Widen Int32 to Int64 (Iceberg int→long); existing and new rows read correctly."""
+    instance = started_cluster_iceberg_no_spark.instances["node1"]
+    TABLE_NAME = "test_modify_column_basic_" + storage_type + "_" + get_uuid_str()
+
+    create_iceberg_table(
+        storage_type,
+        instance,
+        TABLE_NAME,
+        started_cluster_iceberg_no_spark,
+        "(id Int32, value Nullable(String))",
+        format_version,
+    )
+
+    instance.query(f"INSERT INTO {TABLE_NAME} VALUES (1, 'hello'), (2, 'world');", settings=INSERT_SETTINGS)
+    assert instance.query(f"SELECT id, value FROM {TABLE_NAME} ORDER BY id") == "1\thello\n2\tworld\n"
+
+    instance.query(f"ALTER TABLE {TABLE_NAME} MODIFY COLUMN id Int64;", settings=INSERT_SETTINGS)
+
+    assert instance.query(f"SELECT id, value FROM {TABLE_NAME} ORDER BY id") == "1\thello\n2\tworld\n"
+
+    instance.query(f"INSERT INTO {TABLE_NAME} VALUES (3000000000, 'foo');", settings=INSERT_SETTINGS)
+    assert instance.query(f"SELECT id, value FROM {TABLE_NAME} ORDER BY id") == "1\thello\n2\tworld\n3000000000\tfoo\n"
+
+
+@pytest.mark.parametrize("format_version", [1, 2])
+@pytest.mark.parametrize("storage_type", ["local", "s3"])
+def test_modify_column_errors(started_cluster_iceberg_no_spark, format_version, storage_type):
+    """Invalid schema evolution (e.g. String→Int64) must fail; columns unchanged."""
+    instance = started_cluster_iceberg_no_spark.instances["node1"]
+    TABLE_NAME = "test_modify_column_errors_" + storage_type + "_" + get_uuid_str()
+
+    create_iceberg_table(
+        storage_type,
+        instance,
+        TABLE_NAME,
+        started_cluster_iceberg_no_spark,
+        "(id Int32, value Nullable(String))",
+        format_version,
+    )
+
+    error = instance.query_and_get_error(
+        f"ALTER TABLE {TABLE_NAME} MODIFY COLUMN value Int64;",
+        settings=INSERT_SETTINGS,
+    )
+    el = error.lower()
+    # String→integer: mismatched Poco::Var kinds in checkValidSchemaEvolution → BadCastException
+    assert (
+        "bad cast" in el
+        or "can not convert" in el
+        or "cannot convert" in el
+        or "schema evolution" in el
+        or "doesn't allow" in el
+    )
+
+    assert instance.query(
+        f"SELECT name FROM system.columns WHERE database = currentDatabase() AND table = '{TABLE_NAME}' ORDER BY name"
+    ) == "id\nvalue\n"