diff --git a/src/Databases/DataLake/GlueCatalog.cpp b/src/Databases/DataLake/GlueCatalog.cpp index a8dda240e7ce..af8638ff5661 100644 --- a/src/Databases/DataLake/GlueCatalog.cpp +++ b/src/Databases/DataLake/GlueCatalog.cpp @@ -652,7 +652,99 @@ 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 +/// Converts an Iceberg type (primitive string or complex JSON object) to a +/// Glue-compatible type string such as "string", "array", "map", +/// or "struct". +static String icebergTypeToGlueType(const Poco::Dynamic::Var & type_var) +{ + if (type_var.isString()) + return type_var.extract(); + + auto type_obj = type_var.extract(); + String type_name = type_obj->getValue(DB::Iceberg::f_type); + + if (type_name == DB::Iceberg::f_list) + { + auto element_type = icebergTypeToGlueType(type_obj->get(DB::Iceberg::f_element)); + return "array<" + element_type + ">"; + } + + if (type_name == DB::Iceberg::f_map) + { + auto key_type = icebergTypeToGlueType(type_obj->get(DB::Iceberg::f_key)); + auto value_type = icebergTypeToGlueType(type_obj->get(DB::Iceberg::f_value)); + return "map<" + key_type + "," + value_type + ">"; + } + + if (type_name == DB::Iceberg::f_struct) + { + auto fields = type_obj->getArray(DB::Iceberg::f_fields); + String result = "struct<"; + for (size_t i = 0; i < fields->size(); ++i) + { + if (i > 0) + result += ","; + auto field = fields->getObject(static_cast(i)); + result += field->getValue(DB::Iceberg::f_name) + ":" + + icebergTypeToGlueType(field->get(DB::Iceberg::f_type)); + } + result += ">"; + return result; + } + + return type_name; +} + +/// Populates a Glue StorageDescriptor's column list from an Iceberg schema JSON +/// object (the one with "type": "struct", "fields": [...]). +static void setGlueColumnsFromIcebergSchema(Aws::Glue::Model::StorageDescriptor & sd, const Poco::JSON::Object::Ptr & iceberg_schema) +{ + if (!iceberg_schema || !iceberg_schema->has(DB::Iceberg::f_fields)) + return; + + auto fields = iceberg_schema->getArray(DB::Iceberg::f_fields); + Aws::Vector columns; + columns.reserve(fields->size()); + + for (size_t i = 0; i < fields->size(); ++i) + { + auto field = fields->getObject(static_cast(i)); + Aws::Glue::Model::Column col; + col.SetName(field->getValue(DB::Iceberg::f_name)); + col.SetType(icebergTypeToGlueType(field->get(DB::Iceberg::f_type))); + + Aws::Map col_params; + bool is_optional = !field->getValue(DB::Iceberg::f_required); + col_params["iceberg.field.optional"] = is_optional ? "true" : "false"; + col_params["iceberg.field.current"] = "true"; + col_params["iceberg.field.id"] = std::to_string(field->getValue(DB::Iceberg::f_id)); + col.SetParameters(col_params); + + columns.push_back(std::move(col)); + } + + sd.SetColumns(std::move(columns)); +} + +/// Extracts the current schema from an Iceberg metadata JSON (the top-level +/// object that contains "current-schema-id" and "schemas"). +static Poco::JSON::Object::Ptr getCurrentSchemaFromMetadata(const Poco::JSON::Object::Ptr & metadata) +{ + if (!metadata) + return nullptr; + + auto current_schema_id = metadata->getValue(DB::Iceberg::f_current_schema_id); + auto schemas = metadata->getArray(DB::Iceberg::f_schemas); + for (size_t i = schemas->size(); i > 0; --i) + { + auto candidate = schemas->getObject(static_cast(i - 1)); + if (candidate->getValue(DB::Iceberg::f_schema_id) == current_schema_id) + return candidate; + } + return nullptr; +} + +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 +767,9 @@ void GlueCatalog::createTable(const String & namespace_name, const String & tabl sd.SetLocation(grandparent.c_str()); + if (auto schema = getCurrentSchemaFromMetadata(metadata_content)) + setGlueColumnsFromIcebergSchema(sd, schema); + table_input.SetStorageDescriptor(sd); table_input.SetTableType("ICEBERG"); @@ -745,12 +840,50 @@ 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); + Aws::Glue::Model::UpdateTableRequest request; + request.SetDatabaseName(namespace_name); + + Aws::Glue::Model::TableInput table_input; + table_input.SetName(table_name); + + Aws::Glue::Model::StorageDescriptor sd; + fs::path original_path = new_metadata_path; + + fs::path parent = original_path.parent_path(); + fs::path grandparent = parent.parent_path(); + + sd.SetLocation(grandparent.c_str()); + + setGlueColumnsFromIcebergSchema(sd, new_schema); + + table_input.SetStorageDescriptor(sd); + table_input.SetTableType("ICEBERG"); + + Aws::Map parameters; + parameters["metadata_location"] = new_metadata_path; + parameters["table_type"] = "ICEBERG"; + + table_input.SetParameters(parameters); + + request.SetTableInput(table_input); + + Aws::Glue::Model::UpdateTableOutcome response; + + { + ProfileEvents::increment(ProfileEvents::DataLakeGlueCatalogUpdateTable); + auto timer = DB::CurrentThread::getProfileEvents().timer(ProfileEvents::DataLakeGlueCatalogUpdateTableMicroseconds); + response = glue_client->UpdateTable(request); + } + + if (!response.IsSuccess()) + throw DB::Exception(DB::ErrorCodes::DATALAKE_DATABASE_ERROR, "Can not update schema in glue catalog {}", response.GetError().GetMessage()); + + return true; } void GlueCatalog::dropTable(const String & namespace_name, const String & table_name) const diff --git a/tests/integration/test_database_glue/test.py b/tests/integration/test_database_glue/test.py index 665aee5cd205..d366f0f02570 100644 --- a/tests/integration/test_database_glue/test.py +++ b/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"]