Antalya 26.6 - Fixed refreshing of glue metadata after ALTER calls

src/Databases/DataLake/GlueCatalog.cpp

@@ -51,6 +51,7 @@
 #include <Common/FailPoint.h>
 #include <Storages/ObjectStorage/DataLakes/DataLakeConfiguration.h>
 #include <Storages/ObjectStorage/DataLakes/DataLakeStorageSettings.h>
+#include <Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.h>
 #include <Storages/ObjectStorage/DataLakes/Iceberg/SchemaProcessor.h>
 #include <Storages/ObjectStorage/DataLakes/Iceberg/Utils.h>
 #include <Common/ProxyConfigurationResolverProvider.h>
@@ -70,6 +71,7 @@ namespace DB::FailPoints
 
 namespace DB::Setting
 {
+    extern const SettingsBool allow_experimental_geo_types_in_iceberg;
     extern const SettingsUInt64 s3_max_connections;
     extern const SettingsUInt64 s3_max_redirects;
     extern const SettingsUInt64 s3_retry_attempts;
@@ -109,6 +111,87 @@ namespace CurrentMetrics
     extern const Metric MarkCacheFiles;
 }
 
+namespace
+{
+
+/// Convert an Iceberg JSON type (string or object) to the Glue type string.
+String icebergTypeToGlueType(const Poco::Dynamic::Var & type_value)
+{
+    if (type_value.isString())
+    {
+        String type = type_value.toString();
+        if (type == "timestamptz")
+            return "timestamp";
+        if (type == "timestamp_ns" || type == "timestamptz_ns")
+            return "timestamp_nano";
+        return type;
+    }
+
+    auto obj = type_value.extract<Poco::JSON::Object::Ptr>();
+    String complex_type = obj->getValue<String>(DB::Iceberg::f_type);
+
+    if (complex_type == DB::Iceberg::f_list)
+    {
+        return "array<" + icebergTypeToGlueType(obj->get(DB::Iceberg::f_element)) + ">";
+    }
+    if (complex_type == DB::Iceberg::f_map)
+    {
+        return "map<" + icebergTypeToGlueType(obj->get(DB::Iceberg::f_key)) + ", "
+             + icebergTypeToGlueType(obj->get(DB::Iceberg::f_value)) + ">";
+    }
+    if (complex_type == DB::Iceberg::f_struct)
+    {
+        auto fields = obj->getArray(DB::Iceberg::f_fields);
+        String result = "struct<";
+        for (UInt32 i = 0; i < fields->size(); ++i)
+        {
+            if (i > 0)
+                result += ", ";
+            auto field = fields->getObject(i);
+            result += field->getValue<String>(DB::Iceberg::f_name) + ":" + icebergTypeToGlueType(field->get(DB::Iceberg::f_type));
+        }
+        result += ">";
+        return result;
+    }
+
+    throw DB::Exception(DB::ErrorCodes::BAD_ARGUMENTS, "Unknown Iceberg complex type: {}", complex_type);
+}
+
+/// Build Glue Column objects from an Iceberg schema JSON object
+/// (which has "type": "struct", "fields": [...]).
+std::vector<Aws::Glue::Model::Column> icebergSchemaToGlueColumns(const Poco::JSON::Object::Ptr & schema)
+{
+    std::vector<Aws::Glue::Model::Column> glue_columns;
+    auto fields = schema->getArray(DB::Iceberg::f_fields);
+
+    for (UInt32 i = 0; i < fields->size(); ++i)
+    {
+        auto field = fields->getObject(i);
+        Aws::Glue::Model::Column col;
+        col.SetName(field->getValue<String>(DB::Iceberg::f_name));
+        col.SetType(icebergTypeToGlueType(field->get(DB::Iceberg::f_type)));
+
+        Aws::Map<Aws::String, Aws::String> params;
+        params["iceberg.field.id"] = std::to_string(field->getValue<Int32>(DB::Iceberg::f_id));
+        params["iceberg.field.optional"] = field->getValue<bool>(DB::Iceberg::f_required) ? "false" : "true";
+        params["iceberg.field.current"] = "true";
+        col.SetParameters(params);
+
+        glue_columns.push_back(std::move(col));
+    }
+
+    return glue_columns;
+}
+
+/// Extract the current schema object from full Iceberg metadata JSON.
+Poco::JSON::Object::Ptr getCurrentSchemaFromMetadata(const Poco::JSON::Object::Ptr & metadata)
+{
+    auto [schema, _] = DB::Iceberg::parseTableSchemaV2Method(metadata);
+    return schema;
+}
+
+}
+
 namespace DataLake
 {
 
@@ -335,7 +418,7 @@ bool GlueCatalog::existsTable(const std::string & database_name, const std::stri
 bool GlueCatalog::tryGetTableMetadata(
     const std::string & database_name,
     const std::string & table_name,
-    DB::ContextPtr /* context_ */,
+    DB::ContextPtr context_,
     TableMetadata & result) const
 {
     if (!isNamespaceAllowed(database_name))
@@ -417,26 +500,60 @@ bool GlueCatalog::tryGetTableMetadata(
         {
             DB::NamesAndTypesList schema;
             auto columns = table_outcome.GetStorageDescriptor().GetColumns();
-            for (const auto & column : columns)
+            if (!columns.empty())
             {
-                const auto column_params = column.GetParameters();
-                bool can_be_nullable = column_params.contains("iceberg.field.optional") && column_params.at("iceberg.field.optional") == "true";
-
-                /// Skip field if it's not "current" (for example Renamed). No idea how someone can utilize "non current fields" but for some reason
-                /// they are returned by Glue API. So if you do "RENAME COLUMN a to new_a" glue will return two fields: a and new_a.
-                /// And a will be marked as "non current" field.
-                if (column_params.contains("iceberg.field.current") && column_params.at("iceberg.field.current") == "false")
-                    continue;
-
-                String column_type = column.GetType();
-                if (column_type == "timestamp" || column_type == "timestamp_nano")
+                for (const auto & column : columns)
                 {
-                    if (!result.requiresDataLakeSpecificProperties())
-                        setup_specific_properties();
-                    column_type = getActualTimestampType(column.GetName(), result, column_type);
+                    const auto column_params = column.GetParameters();
+                    bool can_be_nullable = column_params.contains("iceberg.field.optional") && column_params.at("iceberg.field.optional") == "true";
+
+                    /// Skip field if it's not "current" (for example Renamed). No idea how someone can utilize "non current fields" but for some reason
+                    /// they are returned by Glue API. So if you do "RENAME COLUMN a to new_a" glue will return two fields: a and new_a.
+                    /// And a will be marked as "non current" field.
+                    if (column_params.contains("iceberg.field.current") && column_params.at("iceberg.field.current") == "false")
+                        continue;
+
+                    String column_type = column.GetType();
+                    if (column_type == "timestamp" || column_type == "timestamp_nano")
+                    {
+                        if (!result.requiresDataLakeSpecificProperties())
+                            setup_specific_properties();
+                        column_type = getActualTimestampType(column.GetName(), result, column_type);
+                    }
+
+                    schema.push_back({column.GetName(), getType(column_type, can_be_nullable, getContext())});
+                }
+            }
+            else
+            {
+                /// StorageDescriptor has no columns (e.g. table was created via ClickHouse DDL
+                /// which only sets metadata_location). Fall back to parsing the Iceberg metadata
+                /// file directly, same approach as the REST catalog.
+                if (!result.requiresDataLakeSpecificProperties())
+                    setup_specific_properties();
+
+                auto table_specific_properties = result.getDataLakeSpecificProperties();
+                if (table_specific_properties.has_value() && !table_specific_properties->iceberg_metadata_file_location.empty())
+                {
+                    const String & metadata_uri = table_specific_properties->iceberg_metadata_file_location;
+                    if (!metadata_objects.get(metadata_uri))
+                    {
+                        auto [object_storage, bucket_name, metadata_path] = createObjectStorageForEarlyTableAccess(metadata_uri, result);
+                        auto compression_method = DB::Iceberg::getCompressionMethodFromMetadataFile(metadata_uri);
+                        auto metadata_object = DB::Iceberg::getMetadataJSONObject(
+                            metadata_path, object_storage, nullptr, getContext(), log, compression_method, std::nullopt);
+                        metadata_objects.set(metadata_uri, std::make_shared<Poco::JSON::Object::Ptr>(metadata_object));
+                    }
+
+                    auto metadata_object = *metadata_objects.get(metadata_uri);
+                    const bool allow_geo_parser
+                        = getContext()->getSettingsRef()[DB::Setting::allow_experimental_geo_types_in_iceberg].value;
+                    auto schema_processor = DB::Iceberg::IcebergSchemaProcessor(context_, allow_geo_parser);
+                    auto id = DB::IcebergMetadata::parseTableSchema(metadata_object, schema_processor, context_, log);
+                    auto parsed_schema = schema_processor.getClickhouseTableSchemaById(id);
+                    if (parsed_schema)
+                        schema = *parsed_schema;
                 }
-
-                schema.push_back({column.GetName(), getType(column_type, can_be_nullable, getContext())});
             }
             result.setSchema(schema);
         }
@@ -652,7 +769,7 @@ void GlueCatalog::createNamespaceIfNotExists(const String & namespace_name) cons
     glue_client->CreateDatabase(create_request);
 }
 
-void GlueCatalog::createTable(const String & namespace_name, const String & table_name, const String & new_metadata_path, Poco::JSON::Object::Ptr /*metadata_content*/) const
+void GlueCatalog::createTable(const String & namespace_name, const String & table_name, const String & new_metadata_path, Poco::JSON::Object::Ptr metadata_content) const
 {
     if (!isNamespaceAllowed(namespace_name))
         throw DB::Exception(DB::ErrorCodes::CATALOG_NAMESPACE_DISABLED,
@@ -675,6 +792,20 @@ void GlueCatalog::createTable(const String & namespace_name, const String & tabl
 
     sd.SetLocation(grandparent.c_str());
 
+    if (metadata_content)
+    {
+        try
+        {
+            auto schema = getCurrentSchemaFromMetadata(metadata_content);
+            sd.SetColumns(icebergSchemaToGlueColumns(schema));
+        }
+        catch (...)
+        {
+            LOG_WARNING(log, "Failed to extract schema from metadata for table {}.{}: {}",
+                namespace_name, table_name, DB::getCurrentExceptionMessage(false));
+        }
+    }
+
     table_input.SetStorageDescriptor(sd);
     table_input.SetTableType("ICEBERG");
 
@@ -698,7 +829,11 @@ void GlueCatalog::createTable(const String & namespace_name, const String & tabl
         throw DB::Exception(DB::ErrorCodes::DATALAKE_DATABASE_ERROR, "Can not create metadata in glue catalog: {}", response.GetError().GetMessage());
 }
 
-bool GlueCatalog::updateMetadata(const String & namespace_name, const String & table_name, const String & new_metadata_path, Poco::JSON::Object::Ptr /*new_snapshot*/) const
+bool GlueCatalog::updateTableInGlue(
+    const String & namespace_name,
+    const String & table_name,
+    const String & new_metadata_path,
+    const std::vector<Aws::Glue::Model::Column> & columns) const
 {
     Aws::Glue::Model::UpdateTableRequest request;
     request.SetDatabaseName(namespace_name);
@@ -716,6 +851,9 @@ bool GlueCatalog::updateMetadata(const String & namespace_name, const String & t
     /// We should drop `metadata/v<i>-metadata.json` suffix to get location.
     sd.SetLocation(grandparent.c_str());
 
+    if (!columns.empty())
+        sd.SetColumns(columns);
+
     table_input.SetStorageDescriptor(sd);
     table_input.SetTableType("ICEBERG");
 
@@ -741,16 +879,34 @@ bool GlueCatalog::updateMetadata(const String & namespace_name, const String & t
     return true;
 }
 
+bool GlueCatalog::updateMetadata(const String & namespace_name, const String & table_name, const String & new_metadata_path, Poco::JSON::Object::Ptr /*new_snapshot*/) const
+{
+    return updateTableInGlue(namespace_name, table_name, new_metadata_path);
+}
+
 bool GlueCatalog::updateSchema(
     const String & namespace_name,
     const String & table_name,
     const String & new_metadata_path,
-    Poco::JSON::Object::Ptr /*new_schema*/,
+    Poco::JSON::Object::Ptr new_schema,
     Int32 /*previous_schema_id*/,
     Int32 /*new_last_column_id*/,
     Poco::JSON::Object::Ptr /*metadata*/) const
 {
-    return updateMetadata(namespace_name, table_name, new_metadata_path, nullptr);
+    std::vector<Aws::Glue::Model::Column> columns;
+    if (new_schema)
+    {
+        try
+        {
+            columns = icebergSchemaToGlueColumns(new_schema);
+        }
+        catch (...)
+        {
+            LOG_WARNING(log, "Failed to convert schema to Glue columns for table {}.{}: {}",
+                namespace_name, table_name, DB::getCurrentExceptionMessage(false));
+        }
+    }
+    return updateTableInGlue(namespace_name, table_name, new_metadata_path, columns);
 }
 
 void GlueCatalog::dropTable(const String & namespace_name, const String & table_name) const

src/Databases/DataLake/GlueCatalog.h

@@ -4,6 +4,7 @@
 #if USE_AWS_S3 && USE_AVRO
 
 #include <aws/core/auth/AWSCredentials.h>
+#include <aws/glue/model/Column.h>
 #include <Databases/DataLake/ICatalog.h>
 #include <Interpreters/Context_fwd.h>
 #include <Poco/JSON/Object.h>
@@ -128,6 +129,14 @@ class GlueCatalog final : public ICatalog, private DB::WithContext
 
     ObjectStorageWithPath createObjectStorageForEarlyTableAccess(const String & s3_location, const TableMetadata & table_metadata) const;
 
+    /// Shared implementation for updateMetadata / updateSchema that optionally
+    /// sets StorageDescriptor columns in the Glue UpdateTable call.
+    bool updateTableInGlue(
+        const String & namespace_name,
+        const String & table_name,
+        const String & new_metadata_path,
+        const std::vector<Aws::Glue::Model::Column> & columns = {}) const;
+
     mutable DB::CacheBase<String, Poco::JSON::Object::Ptr> metadata_objects;
 };
 

tests/integration/test_database_glue/test.py

@@ -724,6 +724,55 @@ def test_create(started_cluster):
     assert node.query(f"SELECT * FROM {CATALOG_NAME}.`{root_namespace}.{table_name}`") == "AAPL\n"
 
 
+def test_schema_evolution_show_create_and_drop(started_cluster):
+    """SHOW CREATE TABLE must reflect columns added/dropped via ALTER.
+
+    Reproducer for the bug where GlueCatalog::updateSchema only updated
+    metadata_location but not StorageDescriptor.Columns, causing
+    SHOW CREATE TABLE to return a stale schema and DROP COLUMN to fail
+    with NOT_FOUND_COLUMN_IN_BLOCK.
+    """
+    node = started_cluster.instances["node1"]
+
+    test_ref = f"test_show_create_drop_{uuid.uuid4()}"
+    table_name = f"{test_ref}_table"
+    root_namespace = f"{test_ref}_namespace"
+    table_ref = f"{CATALOG_NAME}.`{root_namespace}.{table_name}`"
+    write_settings = {"allow_insert_into_iceberg": 1, "write_full_path_in_iceberg_metadata": 1}
+
+    create_clickhouse_glue_database(started_cluster, node, CATALOG_NAME)
+    create_clickhouse_glue_table(started_cluster, node, root_namespace, table_name, "(name Nullable(String))")
+
+    node.query(f"INSERT INTO {table_ref} VALUES ('Alice');", settings=write_settings)
+
+    node.query(f"ALTER TABLE {table_ref} ADD COLUMN column_a Nullable(String);", settings=write_settings)
+    node.query(f"ALTER TABLE {table_ref} ADD COLUMN column_b Nullable(Int64);", settings=write_settings)
+
+    assert node.query(f"SELECT * FROM {table_ref}") == "Alice\t\\N\t\\N\n"
+
+    show_create = node.query(f"SHOW CREATE TABLE {table_ref}")
+    assert "column_a" in show_create, f"column_a missing from SHOW CREATE:\n{show_create}"
+    assert "column_b" in show_create, f"column_b missing from SHOW CREATE:\n{show_create}"
+
+    node.query(f"ALTER TABLE {table_ref} DROP COLUMN column_a;", settings=write_settings)
+
+    show_create = node.query(f"SHOW CREATE TABLE {table_ref}")
+    assert "column_a" not in show_create, f"column_a still in SHOW CREATE after DROP:\n{show_create}"
+    assert "column_b" in show_create, f"column_b missing from SHOW CREATE after DROP:\n{show_create}"
+
+    assert node.query(f"SELECT name, column_b FROM {table_ref}") == "Alice\t\\N\n"
+
+    node.query(f"ALTER TABLE {table_ref} ADD COLUMN column_c Nullable(String);", settings=write_settings)
+    node.query(f"INSERT INTO {table_ref} (name, column_b, column_c) VALUES ('Bob', 42, 'hello');", settings=write_settings)
+
+    show_create = node.query(f"SHOW CREATE TABLE {table_ref}")
+    assert "column_c" in show_create, f"column_c missing from SHOW CREATE:\n{show_create}"
+    assert "column_a" not in show_create, f"column_a reappeared in SHOW CREATE:\n{show_create}"
+
+    result = node.query(f"SELECT name, column_b, column_c FROM {table_ref} ORDER BY name")
+    assert result == "Alice\t\\N\t\\N\nBob\t42\thello\n"
+
+
 def test_schema_evolution(started_cluster):
     node = started_cluster.instances["node1"]