diff --git a/src/Databases/DataLake/GlueCatalog.cpp b/src/Databases/DataLake/GlueCatalog.cpp index a8dda240e7ce..2590db1898a6 100644 --- a/src/Databases/DataLake/GlueCatalog.cpp +++ b/src/Databases/DataLake/GlueCatalog.cpp @@ -51,6 +51,7 @@ #include #include #include +#include #include #include #include @@ -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(); + String complex_type = obj->getValue(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(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 icebergSchemaToGlueColumns(const Poco::JSON::Object::Ptr & schema) +{ + std::vector 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(DB::Iceberg::f_name)); + col.SetType(icebergTypeToGlueType(field->get(DB::Iceberg::f_type))); + + Aws::Map params; + params["iceberg.field.id"] = std::to_string(field->getValue(DB::Iceberg::f_id)); + params["iceberg.field.optional"] = field->getValue(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(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 & 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-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 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 diff --git a/src/Databases/DataLake/GlueCatalog.h b/src/Databases/DataLake/GlueCatalog.h index 8d10ba0c8667..f2f8a4215013 100644 --- a/src/Databases/DataLake/GlueCatalog.h +++ b/src/Databases/DataLake/GlueCatalog.h @@ -4,6 +4,7 @@ #if USE_AWS_S3 && USE_AVRO #include +#include #include #include #include @@ -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 & columns = {}) const; + mutable DB::CacheBase metadata_objects; }; 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"]