From e93ce1c26f38826880f5f2cd3d1dfedd5f51e174 Mon Sep 17 00:00:00 2001 From: Kanthi Subramanian Date: Tue, 4 Aug 2026 00:51:39 +0200 Subject: [PATCH 1/6] Fix alter operations for iceberg --- src/Common/FailPoint.cpp | 3 + src/Databases/DataLake/GlueCatalog.cpp | 3 +- src/Databases/DataLake/GlueCatalog.h | 3 +- src/Databases/DataLake/ICatalog.cpp | 3 +- src/Databases/DataLake/ICatalog.h | 3 +- src/Databases/DataLake/RestCatalog.cpp | 433 ++++++++++++++---- src/Databases/DataLake/RestCatalog.h | 13 +- .../gtest_rest_catalog_update_metadata.cpp | 186 ++++++++ .../DataLakes/DataLakeConfiguration.h | 10 +- .../DataLakes/Iceberg/Compaction.cpp | 5 +- .../DataLakes/Iceberg/MetadataGenerator.cpp | 49 +- .../DataLakes/Iceberg/MetadataGenerator.h | 3 +- .../DataLakes/Iceberg/Mutations.cpp | 78 +++- .../ObjectStorage/DataLakes/Iceberg/Utils.cpp | 12 +- .../ObjectStorage/DataLakes/Iceberg/Utils.h | 3 +- .../integration/test_database_iceberg/test.py | 391 +++++++++++++++- .../test_writes_add_column.py | 72 +++ .../test_writes_drop_column.py | 62 +++ .../test_writes_modify_column.py | 70 +++ 19 files changed, 1272 insertions(+), 130 deletions(-) create mode 100644 src/Databases/DataLake/tests/gtest_rest_catalog_update_metadata.cpp create mode 100644 tests/integration/test_storage_iceberg_no_spark/test_writes_add_column.py create mode 100644 tests/integration/test_storage_iceberg_no_spark/test_writes_drop_column.py create mode 100644 tests/integration/test_storage_iceberg_no_spark/test_writes_modify_column.py 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 28c1195082e4..3d0a34885a99 100644 --- a/src/Databases/DataLake/RestCatalog.cpp +++ b/src/Databases/DataLake/RestCatalog.cpp @@ -8,6 +8,8 @@ #include #include #include +#include +#include #include #include #include "config.h" @@ -34,6 +36,7 @@ #include #include +#include #include #include #include @@ -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 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(); +} + +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(), rhs.extract()); + } + 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(), rhs.extract()); + } + 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 & 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(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(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(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 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(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(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(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 existing_equivalent_schema_id; + for (UInt32 i = 0; i < schemas->size(); ++i) + { + auto existing_schema = schemas->getObject(i); + if (existing_schema->getValue(DB::Iceberg::f_schema_id) == new_schema_id) + continue; + if (schemasEquivalentIgnoringId(existing_schema, new_schema_obj)) + { + existing_equivalent_schema_id = existing_schema->getValue(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(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("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("snapshot-id")); + updates->add(set_snapshot); + } + request_body->set("updates", updates); + } + + return request_body; +} + std::string RestCatalog::Config::toString() const { DB::WriteBufferFromOwnString wb; @@ -1233,57 +1536,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("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("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 { @@ -1291,7 +1550,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; @@ -1302,49 +1562,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 @@ -1353,7 +1623,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 +#include +#include +#include +#include +#include +#include +#include + +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("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("name"), "tbl"); + auto ns = id->getArray("namespace"); + ASSERT_EQ(ns->size(), 1u); + EXPECT_EQ(ns->getElement(0), "my.ns"); + + ASSERT_TRUE(body->has("requirements")); + auto req = body->getArray("requirements")->getObject(0); + EXPECT_EQ(req->getValue("type"), "assert-current-schema-id"); + EXPECT_EQ(req->getValue("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("last-column-id"), 3); + + auto set_schema = findUpdateByAction(updates, "set-current-schema"); + ASSERT_TRUE(set_schema); + EXPECT_EQ(set_schema->getValue("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(12345)); + snapshot->set("parent-snapshot-id", static_cast(12344)); + snapshot->set(Iceberg::f_timestamp_ms, static_cast(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("type"), "assert-ref-snapshot-id"); + EXPECT_EQ(req->getValue("ref"), "main"); + EXPECT_EQ(req->getValue("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("snapshot-id"), 12345); + + auto set_ref = findUpdateByAction(updates, "set-snapshot-ref"); + ASSERT_TRUE(set_ref); + EXPECT_EQ(set_ref->getValue("snapshot-id"), 12345); +} + +TEST(RestCatalogUpdateMetadataBody, SnapshotUpdateWithoutParent) +{ + Poco::JSON::Object::Ptr snapshot = new Poco::JSON::Object; + snapshot->set("snapshot-id", static_cast(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(1)); + snapshot->set("parent-snapshot-id", static_cast(-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 #include +#include #include #include #include @@ -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..f85b14764067 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/MetadataGenerator.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/MetadataGenerator.cpp @@ -54,6 +54,7 @@ bool checkValidSchemaEvolution(Poco::Dynamic::Var old_type, Poco::Dynamic::Var n return true; } + if (!old_type.isString() && !new_type.isString()) { auto old_complex_type = old_type.extract(); auto new_complex_type = new_type.extract(); @@ -69,6 +70,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() == new_type.extract(); + + 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()->stringify(oss_old); + new_type.extract()->stringify(oss_new); + return oss_old.str() == oss_new.str(); + } + + return false; +} + } MetadataGenerator::MetadataGenerator(Poco::JSON::Object::Ptr metadata_object_) @@ -317,10 +335,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(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 +352,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(Iceberg::f_last_column_id); + auto last_column_id = metadata_object->getValue(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(Iceberg::f_name) == column_name) { + if (current_field->getValue(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(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/Mutations.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/Mutations.cpp index 8a04b5c83b3e..02e698c82170 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/Mutations.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/Mutations.cpp @@ -39,6 +39,7 @@ namespace DB::ErrorCodes { extern const int BAD_ARGUMENTS; +extern const int DATALAKE_DATABASE_ERROR; extern const int LOGICAL_ERROR; extern const int LIMIT_EXCEEDED; } @@ -52,6 +53,7 @@ extern const DataLakeStorageSettingsString iceberg_metadata_file_path; namespace DB::FailPoints { extern const char iceberg_writes_cleanup[]; +extern const char iceberg_alter_orphan_metadata_cleanup_fail[]; } namespace DB::Iceberg @@ -590,7 +592,8 @@ void mutate( persistent_table_components.table_uuid, persistent_table_components.metadata_compression_method, /* force_fetch_latest_metadata */ true, - /* ignore_explicit_metadata_file_path */ true); + /* ignore_explicit_metadata_file_path */ true, + /* select_by_table_uuid */ true); FileNamesGenerator filename_generator(persistent_table_components.path_resolver.getTableLocation(), false, CompressionMethod::None, write_format); filename_generator.setVersion(last_version + 1); @@ -720,11 +723,14 @@ void alter( std::shared_ptr catalog) { if (params.size() != 1) - throw Exception(ErrorCodes::BAD_ARGUMENTS, "Params with size 1 is not supported"); + throw Exception(ErrorCodes::BAD_ARGUMENTS, "Iceberg alter supports exactly one command at a time, got {}", params.size()); - size_t i = 0; - bool succeeded = false; - while (i < MAX_TRANSACTION_RETRIES) + /// The command was marked as a no-op by AlterCommands::prepare (e.g. RENAME/DROP COLUMN IF EXISTS + /// for a missing column, or ADD COLUMN IF NOT EXISTS for an existing one). + if (params[0].ignore) + return; + + for (size_t i = 0; i < MAX_TRANSACTION_RETRIES; ++i) { auto log = getLogger("IcebergMutations"); @@ -743,7 +749,8 @@ void alter( persistent_table_components.table_uuid, persistent_table_components.metadata_compression_method, /* force_fetch_latest_metadata */ true, - /* ignore_explicit_metadata_file_path */ true); + /* ignore_explicit_metadata_file_path */ true, + /* select_by_table_uuid */ true); last_version = last_version_info.version; metadata_path = last_version_info.path; compression_method = last_version_info.compression_method; @@ -776,13 +783,18 @@ void alter( persistent_table_components.table_uuid, persistent_table_components.metadata_compression_method, /* force_fetch_latest_metadata */ true, - /* ignore_explicit_metadata_file_path */ false); + /* ignore_explicit_metadata_file_path */ false, + /* select_by_table_uuid */ true); last_version = last_version_info.version; metadata_path = last_version_info.path; compression_method = last_version_info.compression_method; } - FileNamesGenerator filename_generator(persistent_table_components.path_resolver.getTableLocation(), false, CompressionMethod::None, write_format); + FileNamesGenerator filename_generator( + persistent_table_components.path_resolver.getTableLocation(), + catalog && catalog->isTransactional(), + CompressionMethod::None, + write_format); filename_generator.setVersion(last_version + 1); filename_generator.setCompressionMethod(compression_method); @@ -803,7 +815,8 @@ void alter( metadata_json_generator.generateDropColumnMetadata(params[0].column_name); break; case AlterCommand::Type::MODIFY_COLUMN: - metadata_json_generator.generateModifyColumnMetadata(params[0].column_name, params[0].data_type); + if (!metadata_json_generator.generateModifyColumnMetadata(params[0].column_name, params[0].data_type)) + return; break; case AlterCommand::Type::RENAME_COLUMN: metadata_json_generator.generateRenameColumnMetadata(params[0].column_name, params[0].rename_to); @@ -843,7 +856,7 @@ void alter( context, data_lake_settings[DataLakeStorageSetting::iceberg_use_version_hint])) { - ++i; + LOG_WARNING(log, "Iceberg alter: failed to write metadata (attempt {}), retrying", i + 1); continue; } @@ -851,24 +864,45 @@ void alter( { auto catalog_filename = persistent_table_components.path_resolver.resolveForCatalog(metadata_info.path); const auto & [namespace_name, table_name] = DataLake::parseTableName(storage_id.getTableName()); - if (!catalog->updateSchema(namespace_name, table_name, catalog_filename, new_schema, previous_schema_id)) + if (!catalog->updateSchema(namespace_name, table_name, catalog_filename, new_schema, previous_schema_id, metadata)) { - ++i; - continue; + auto storage_metadata_name = persistent_table_components.path_resolver.resolve(metadata_info.path); + String orphan_cleanup_error; + try + { + fiu_do_on(FailPoints::iceberg_alter_orphan_metadata_cleanup_fail, + { + throw Exception(ErrorCodes::DATALAKE_DATABASE_ERROR, "Failpoint: orphan metadata cleanup failed"); + }); + object_storage->removeObjectIfExists(StoredObject(storage_metadata_name)); + } + catch (...) + { + orphan_cleanup_error = getCurrentExceptionMessage(false); + tryLogCurrentException(log, "Iceberg alter: failed to remove orphan metadata file after catalog commit failure"); + } + if (orphan_cleanup_error.empty()) + { + throw Exception( + ErrorCodes::DATALAKE_DATABASE_ERROR, + "Iceberg alter: catalog commit failed for '{}' after metadata file was written successfully", + catalog_filename); + } + throw Exception( + ErrorCodes::DATALAKE_DATABASE_ERROR, + "Iceberg alter: catalog commit failed for '{}' after metadata file was written successfully. " + "Failed to remove orphan metadata file '{}': {}", + catalog_filename, + storage_metadata_name, + orphan_cleanup_error); } } - succeeded = true; - break; + persistent_table_components.invalidateMetadataCache(); + return; } - if (!succeeded) - throw Exception(ErrorCodes::LIMIT_EXCEEDED, "Too many unsuccessed retries to alter iceberg table"); - - /// Invalidate the metadata files cache so that subsequent operations on this table see the - /// schema we just wrote. See `PersistentTableComponents::invalidateMetadataCache` for the - /// rationale. - persistent_table_components.invalidateMetadataCache(); + throw Exception(ErrorCodes::LIMIT_EXCEEDED, "Too many unsuccessful retries to alter iceberg table"); } #endif diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.cpp index f7a3f164ac1d..4c39350ebb19 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.cpp @@ -1210,7 +1210,8 @@ MetadataFileWithInfo getLatestOrExplicitMetadataFileAndVersion( const std::optional & 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) { @@ -1270,7 +1271,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 43d2c040ad59..8fb909f328a9 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.h +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.h @@ -94,7 +94,8 @@ MetadataFileWithInfo getLatestOrExplicitMetadataFileAndVersion( const std::optional & 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 parseTableSchemaV1Method(const Poco::JSON::Object::Ptr & metadata_object); std::pair parseTableSchemaV2Method(const Poco::JSON::Object::Ptr & metadata_object); diff --git a/tests/integration/test_database_iceberg/test.py b/tests/integration/test_database_iceberg/test.py index 31ec8882a357..3c167b73816f 100644 --- a/tests/integration/test_database_iceberg/test.py +++ b/tests/integration/test_database_iceberg/test.py @@ -13,10 +13,13 @@ from pyiceberg.catalog import load_catalog from pyiceberg.partitioning import PartitionField, PartitionSpec from pyiceberg.schema import Schema -from pyiceberg.table.sorting import SortField, SortOrder +from pyiceberg.table.sorting import SortField, SortOrder, UNSORTED_SORT_ORDER from pyiceberg.transforms import DayTransform, IdentityTransform from pyiceberg.types import ( DoubleType, + IntegerType, + LongType, + FloatType, NestedField, StringType, StructType, @@ -24,9 +27,11 @@ TimestamptzType ) +from minio import Minio from helpers.cluster import ClickHouseCluster from helpers.config_cluster import minio_secret_key, minio_access_key from helpers.client import QueryRuntimeException +from helpers.s3_tools import list_s3_objects BASE_URL = "http://rest:8181/v1" @@ -95,11 +100,12 @@ def create_table( schema=DEFAULT_SCHEMA, partition_spec=DEFAULT_PARTITION_SPEC, sort_order=DEFAULT_SORT_ORDER, + location="s3://warehouse-rest/data", ): return catalog.create_table( identifier=f"{namespace}.{table}", schema=schema, - location="s3://warehouse-rest/data", + location=location, partition_spec=partition_spec, sort_order=sort_order, ) @@ -1238,3 +1244,384 @@ def test_iceberg_file_progress_callback(started_cluster): f"`IcebergIterator::next` did not invoke the file-progress callback " f"(regression of PR #105413 wiring)." ) + + +def test_alter_drop_column_without_reload(started_cluster): + node = started_cluster.instances["node1"] + + test_ref = f"test_alter_drop_column_{uuid.uuid4()}" + table_name = f"{test_ref}_table" + root_namespace = f"{test_ref}_namespace" + + schema = Schema( + NestedField(field_id=1, name="x", field_type=StringType(), required=False), + NestedField(field_id=2, name="y", field_type=StringType(), required=False), + ) + + catalog = load_catalog_impl(started_cluster) + catalog.create_namespace(root_namespace) + create_table( + catalog, + root_namespace, + table_name, + schema, + PartitionSpec(), + DEFAULT_SORT_ORDER, + location=f"s3://warehouse-rest/data/{root_namespace}/{table_name}", + ) + + create_clickhouse_iceberg_database(started_cluster, node, CATALOG_NAME) + node.query( + f"INSERT INTO {CATALOG_NAME}.`{root_namespace}.{table_name}` VALUES ('a', 'b');", + settings={"allow_insert_into_iceberg": 1, "write_full_path_in_iceberg_metadata": 1}, + ) + assert ( + node.query(f"SELECT * FROM {CATALOG_NAME}.`{root_namespace}.{table_name}`") + == "a\tb\n" + ) + + node.query( + f"ALTER TABLE {CATALOG_NAME}.`{root_namespace}.{table_name}` DROP COLUMN y;", + settings={"allow_insert_into_iceberg": 1}, + ) + assert ( + node.query(f"SELECT * FROM {CATALOG_NAME}.`{root_namespace}.{table_name}`") + == "a\n" + ) + assert "`y`" not in node.query( + f"SHOW CREATE TABLE {CATALOG_NAME}.`{root_namespace}.{table_name}`" + ) + + +def test_alter_modify_column_rest_catalog(started_cluster): + node = started_cluster.instances["node1"] + + test_ref = f"test_alter_modify_column_{uuid.uuid4()}" + table_name = f"{test_ref}_table" + root_namespace = f"{test_ref}_namespace" + + schema = Schema( + NestedField(field_id=1, name="id", field_type=IntegerType(), required=False), + NestedField(field_id=2, name="value", field_type=StringType(), required=False), + ) + + catalog = load_catalog_impl(started_cluster) + catalog.create_namespace(root_namespace) + create_table( + catalog, + root_namespace, + table_name, + schema, + PartitionSpec(), + DEFAULT_SORT_ORDER, + location=f"s3://warehouse-rest/data/{root_namespace}/{table_name}", + ) + + create_clickhouse_iceberg_database(started_cluster, node, CATALOG_NAME) + node.query( + f"INSERT INTO {CATALOG_NAME}.`{root_namespace}.{table_name}` VALUES (1, 'hello'), (2, 'world');", + settings={"allow_insert_into_iceberg": 1, "write_full_path_in_iceberg_metadata": 1}, + ) + assert ( + node.query(f"SELECT id, value FROM {CATALOG_NAME}.`{root_namespace}.{table_name}` ORDER BY id") + == "1\thello\n2\tworld\n" + ) + + node.query( + f"ALTER TABLE {CATALOG_NAME}.`{root_namespace}.{table_name}` ADD COLUMN newCol Nullable(Int64);", + settings={"allow_insert_into_iceberg": 1}, + ) + + node.query( + f"ALTER TABLE {CATALOG_NAME}.`{root_namespace}.{table_name}` MODIFY COLUMN newCol Nullable(Int64);", + settings={"allow_insert_into_iceberg": 1}, + ) + + node.query( + f"ALTER TABLE {CATALOG_NAME}.`{root_namespace}.{table_name}` MODIFY COLUMN id Nullable(Int64);", + settings={"allow_insert_into_iceberg": 1}, + ) + + assert ( + node.query(f"SELECT id, value FROM {CATALOG_NAME}.`{root_namespace}.{table_name}` ORDER BY id") + == "1\thello\n2\tworld\n" + ) + + node.query( + f"INSERT INTO {CATALOG_NAME}.`{root_namespace}.{table_name}` VALUES (3000000000, 'foo', NULL);", + settings={"allow_insert_into_iceberg": 1, "write_full_path_in_iceberg_metadata": 1}, + ) + assert ( + node.query( + f"SELECT id, value, newCol FROM {CATALOG_NAME}.`{root_namespace}.{table_name}` ORDER BY id" + ) + == "1\thello\t\\N\n2\tworld\t\\N\n3000000000\tfoo\t\\N\n" + ) + + iceberg_table = catalog.load_table(f"{root_namespace}.{table_name}") + current_schema = iceberg_table.schema() + assert isinstance(current_schema.find_field("id").field_type, LongType) + assert isinstance(current_schema.find_field("newCol").field_type, LongType) + + +def test_alter_orphan_metadata_cleanup_on_catalog_failure(started_cluster): + node = started_cluster.instances["node1"] + + test_ref = f"test_alter_orphan_cleanup_{uuid.uuid4()}" + table_name = f"{test_ref}_table" + root_namespace = f"{test_ref}_namespace" + + schema = Schema( + NestedField(field_id=1, name="x", field_type=StringType(), required=False), + NestedField(field_id=2, name="y", field_type=StringType(), required=False), + ) + + catalog = load_catalog_impl(started_cluster) + catalog.create_namespace(root_namespace) + create_table( + catalog, + root_namespace, + table_name, + schema, + PartitionSpec(), + DEFAULT_SORT_ORDER, + location=f"s3://warehouse-rest/data/{root_namespace}/{table_name}", + ) + + create_clickhouse_iceberg_database(started_cluster, node, CATALOG_NAME) + node.query( + f"INSERT INTO {CATALOG_NAME}.`{root_namespace}.{table_name}` VALUES ('a', 'b');", + settings={"allow_insert_into_iceberg": 1, "write_full_path_in_iceberg_metadata": 1}, + ) + + iceberg_table = catalog.load_table(f"{root_namespace}.{table_name}") + metadata_location_before = iceberg_table.metadata_location + metadata_prefix = metadata_location_before.replace("s3://warehouse-rest/", "").rsplit("/", 1)[0] + "/" + + minio_client = Minio( + f"{started_cluster.get_instance_ip('minio')}:9000", + access_key=minio_access_key, + secret_key=minio_secret_key, + secure=False, + ) + + def count_metadata_files(): + return len( + [ + f + for f in list_s3_objects(minio_client, "warehouse-rest", prefix=metadata_prefix) + if f.endswith(".metadata.json") + ] + ) + + metadata_files_before = count_metadata_files() + + node.query("SYSTEM ENABLE FAILPOINT iceberg_alter_catalog_update_metadata_fail") + try: + with pytest.raises(QueryRuntimeException, match="catalog commit failed"): + node.query( + f"ALTER TABLE {CATALOG_NAME}.`{root_namespace}.{table_name}` DROP COLUMN y;", + settings={"allow_insert_into_iceberg": 1}, + ) + finally: + node.query("SYSTEM DISABLE FAILPOINT iceberg_alter_catalog_update_metadata_fail") + + assert count_metadata_files() == metadata_files_before + catalog.load_table(f"{root_namespace}.{table_name}") + assert catalog.load_table(f"{root_namespace}.{table_name}").metadata_location == metadata_location_before + assert ( + node.query(f"SELECT * FROM {CATALOG_NAME}.`{root_namespace}.{table_name}`") + == "a\tb\n" + ) + + +def test_alter_fails_when_metadata_not_initialized(started_cluster): + node = started_cluster.instances["node1"] + + test_ref = f"test_alter_uninit_metadata_{uuid.uuid4()}" + table_name = f"{test_ref}_table" + root_namespace = f"{test_ref}_namespace" + + schema = Schema( + NestedField(field_id=1, name="x", field_type=StringType(), required=False), + NestedField(field_id=2, name="y", field_type=StringType(), required=False), + ) + + catalog = load_catalog_impl(started_cluster) + catalog.create_namespace(root_namespace) + create_table( + catalog, + root_namespace, + table_name, + schema, + PartitionSpec(), + DEFAULT_SORT_ORDER, + location=f"s3://warehouse-rest/data/{root_namespace}/{table_name}", + ) + + create_clickhouse_iceberg_database(started_cluster, node, CATALOG_NAME) + + node.query("SYSTEM ENABLE FAILPOINT datalake_iceberg_metadata_create_fail") + try: + node.query(f"DROP DATABASE IF EXISTS {CATALOG_NAME}") + create_clickhouse_iceberg_database(started_cluster, node, CATALOG_NAME) + + with pytest.raises(QueryRuntimeException, match="Metadata is not initialized"): + node.query( + f"ALTER TABLE {CATALOG_NAME}.`{root_namespace}.{table_name}` DROP COLUMN y;", + settings={"allow_insert_into_iceberg": 1}, + ) + finally: + node.query("SYSTEM DISABLE FAILPOINT datalake_iceberg_metadata_create_fail") + node.query(f"DROP DATABASE IF EXISTS {CATALOG_NAME}") + create_clickhouse_iceberg_database(started_cluster, node, CATALOG_NAME) + + +def test_alter_orphan_cleanup_failure_reported(started_cluster): + node = started_cluster.instances["node1"] + + test_ref = f"test_alter_orphan_cleanup_fail_{uuid.uuid4()}" + table_name = f"{test_ref}_table" + root_namespace = f"{test_ref}_namespace" + + schema = Schema( + NestedField(field_id=1, name="x", field_type=StringType(), required=False), + NestedField(field_id=2, name="y", field_type=StringType(), required=False), + ) + + catalog = load_catalog_impl(started_cluster) + catalog.create_namespace(root_namespace) + create_table( + catalog, + root_namespace, + table_name, + schema, + PartitionSpec(), + DEFAULT_SORT_ORDER, + location=f"s3://warehouse-rest/data/{root_namespace}/{table_name}", + ) + + create_clickhouse_iceberg_database(started_cluster, node, CATALOG_NAME) + node.query( + f"INSERT INTO {CATALOG_NAME}.`{root_namespace}.{table_name}` VALUES ('a', 'b');", + settings={"allow_insert_into_iceberg": 1, "write_full_path_in_iceberg_metadata": 1}, + ) + + node.query("SYSTEM ENABLE FAILPOINT iceberg_alter_catalog_update_metadata_fail") + node.query("SYSTEM ENABLE FAILPOINT iceberg_alter_orphan_metadata_cleanup_fail") + try: + with pytest.raises(QueryRuntimeException) as exc_info: + node.query( + f"ALTER TABLE {CATALOG_NAME}.`{root_namespace}.{table_name}` DROP COLUMN y;", + settings={"allow_insert_into_iceberg": 1}, + ) + error = exc_info.value.args[0].lower() + assert "catalog commit failed" in error + assert "failed to remove orphan metadata file" in error + finally: + node.query("SYSTEM DISABLE FAILPOINT iceberg_alter_orphan_metadata_cleanup_fail") + node.query("SYSTEM DISABLE FAILPOINT iceberg_alter_catalog_update_metadata_fail") + + +def test_alter_sequential_add_drop_shared_location(started_cluster): + """ + Two Iceberg tables share the same storage location, so their metadata files + land in the same folder. Running a sequence of ALTER ADD/DROP COLUMN + statements on one table must keep selecting that table's own metadata + (by table-uuid) instead of the globally highest-version metadata file. + """ + node = started_cluster.instances["node1"] + + test_ref = f"test_alter_sequential_{uuid.uuid4()}" + table_a = f"{test_ref}_table_a" + table_b = f"{test_ref}_table_b" + root_namespace = f"{test_ref}_namespace" + + schema = Schema( + NestedField(field_id=1, name="x", field_type=StringType(), required=False), + ) + + shared_location = f"s3://warehouse-rest/data/{root_namespace}/shared" + + catalog = load_catalog_impl(started_cluster) + catalog.create_namespace(root_namespace) + create_table( + catalog, + root_namespace, + table_a, + schema, + PartitionSpec(), + UNSORTED_SORT_ORDER, + location=shared_location, + ) + create_table( + catalog, + root_namespace, + table_b, + schema, + PartitionSpec(), + UNSORTED_SORT_ORDER, + location=shared_location, + ) + + create_clickhouse_iceberg_database(started_cluster, node, CATALOG_NAME) + + node.query( + f"INSERT INTO {CATALOG_NAME}.`{root_namespace}.{table_a}` VALUES ('a');", + settings={"allow_insert_into_iceberg": 1, "write_full_path_in_iceberg_metadata": 1}, + ) + node.query( + f"INSERT INTO {CATALOG_NAME}.`{root_namespace}.{table_b}` VALUES ('b');", + settings={"allow_insert_into_iceberg": 1, "write_full_path_in_iceberg_metadata": 1}, + ) + + table_b_columns = ["b_col1", "b_col2", "b_col3", "b_col4", "b_col5", "b_col6"] + for column in table_b_columns: + node.query( + f"ALTER TABLE {CATALOG_NAME}.`{root_namespace}.{table_b}` ADD COLUMN IF NOT EXISTS {column} Nullable(String);", + settings={"allow_insert_into_iceberg": 1}, + ) + + alter_statements = [ + f"ALTER TABLE {CATALOG_NAME}.`{root_namespace}.{table_a}` ADD COLUMN IF NOT EXISTS name Nullable(String);", + f"ALTER TABLE {CATALOG_NAME}.`{root_namespace}.{table_a}` ADD COLUMN IF NOT EXISTS age Nullable(UInt64);", + f"ALTER TABLE {CATALOG_NAME}.`{root_namespace}.{table_a}` ADD COLUMN IF NOT EXISTS email Nullable(String);", + f"ALTER TABLE {CATALOG_NAME}.`{root_namespace}.{table_a}` ADD COLUMN IF NOT EXISTS `double` Nullable(Float64);", + f"ALTER TABLE {CATALOG_NAME}.`{root_namespace}.{table_a}` ADD COLUMN IF NOT EXISTS `integer` Nullable(UInt64);", + f"ALTER TABLE {CATALOG_NAME}.`{root_namespace}.{table_a}` DROP COLUMN IF EXISTS name;", + f"ALTER TABLE {CATALOG_NAME}.`{root_namespace}.{table_a}` DROP COLUMN IF EXISTS age;", + f"ALTER TABLE {CATALOG_NAME}.`{root_namespace}.{table_a}` DROP COLUMN IF EXISTS email;", + f"ALTER TABLE {CATALOG_NAME}.`{root_namespace}.{table_a}` DROP COLUMN IF EXISTS `double`;", + f"ALTER TABLE {CATALOG_NAME}.`{root_namespace}.{table_a}` DROP COLUMN IF EXISTS `integer`;", + ] + for i, statement in enumerate(alter_statements): + if i > 0: + time.sleep(2) + node.query(statement, settings={"allow_insert_into_iceberg": 1}) + + show_create_a = node.query( + f"SHOW CREATE TABLE {CATALOG_NAME}.`{root_namespace}.{table_a}`" + ) + assert "`x`" in show_create_a + for column in ["name", "age", "email", "double", "integer"]: + assert f"`{column}`" not in show_create_a + + assert ( + node.query(f"SELECT * FROM {CATALOG_NAME}.`{root_namespace}.{table_a}`") == "a\n" + ) + + schema_a = catalog.load_table(f"{root_namespace}.{table_a}").schema() + assert [field.name for field in schema_a.fields] == ["x"] + + show_create_b = node.query( + f"SHOW CREATE TABLE {CATALOG_NAME}.`{root_namespace}.{table_b}`" + ) + for column in table_b_columns: + assert f"`{column}`" in show_create_b + + assert ( + node.query(f"SELECT x FROM {CATALOG_NAME}.`{root_namespace}.{table_b}`") == "b\n" + ) + + schema_b = catalog.load_table(f"{root_namespace}.{table_b}").schema() + assert [field.name for field in schema_b.fields] == ["x"] + table_b_columns 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..630cafdf3a06 --- /dev/null +++ b/tests/integration/test_storage_iceberg_no_spark/test_writes_add_column.py @@ -0,0 +1,72 @@ +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" 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" From 0eb31fc3534628cf1eb82186269cfe8e19179f67 Mon Sep 17 00:00:00 2001 From: Kanthi Subramanian Date: Thu, 6 Aug 2026 05:37:42 +0200 Subject: [PATCH 2/6] fix medium defects with retries --- .../DataLakes/Iceberg/Mutations.cpp | 41 +++++++------------ .../integration/test_database_iceberg/test.py | 7 +--- 2 files changed, 17 insertions(+), 31 deletions(-) diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/Mutations.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/Mutations.cpp index 02e698c82170..274757c2adc9 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/Mutations.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/Mutations.cpp @@ -866,35 +866,24 @@ void alter( const auto & [namespace_name, table_name] = DataLake::parseTableName(storage_id.getTableName()); if (!catalog->updateSchema(namespace_name, table_name, catalog_filename, new_schema, previous_schema_id, metadata)) { - auto storage_metadata_name = persistent_table_components.path_resolver.resolve(metadata_info.path); - String orphan_cleanup_error; - try + if (!catalog_writes_metadata_file) { - fiu_do_on(FailPoints::iceberg_alter_orphan_metadata_cleanup_fail, + try { - throw Exception(ErrorCodes::DATALAKE_DATABASE_ERROR, "Failpoint: orphan metadata cleanup failed"); - }); - object_storage->removeObjectIfExists(StoredObject(storage_metadata_name)); - } - catch (...) - { - orphan_cleanup_error = getCurrentExceptionMessage(false); - tryLogCurrentException(log, "Iceberg alter: failed to remove orphan metadata file after catalog commit failure"); - } - if (orphan_cleanup_error.empty()) - { - throw Exception( - ErrorCodes::DATALAKE_DATABASE_ERROR, - "Iceberg alter: catalog commit failed for '{}' after metadata file was written successfully", - catalog_filename); + fiu_do_on(FailPoints::iceberg_alter_orphan_metadata_cleanup_fail, + { + throw Exception(ErrorCodes::DATALAKE_DATABASE_ERROR, "Failpoint: orphan metadata cleanup failed"); + }); + auto storage_metadata_name = persistent_table_components.path_resolver.resolve(metadata_info.path); + object_storage->removeObjectIfExists(StoredObject(storage_metadata_name)); + } + catch (...) + { + tryLogCurrentException(log, "Iceberg alter: failed to remove orphan metadata file after catalog commit failure"); + } } - throw Exception( - ErrorCodes::DATALAKE_DATABASE_ERROR, - "Iceberg alter: catalog commit failed for '{}' after metadata file was written successfully. " - "Failed to remove orphan metadata file '{}': {}", - catalog_filename, - storage_metadata_name, - orphan_cleanup_error); + LOG_WARNING(log, "Iceberg alter: catalog commit failed (attempt {}), retrying", i + 1); + continue; } } diff --git a/tests/integration/test_database_iceberg/test.py b/tests/integration/test_database_iceberg/test.py index 0e501fe8deb9..ebb43069f96c 100644 --- a/tests/integration/test_database_iceberg/test.py +++ b/tests/integration/test_database_iceberg/test.py @@ -1419,7 +1419,7 @@ def count_metadata_files(): node.query("SYSTEM ENABLE FAILPOINT iceberg_alter_catalog_update_metadata_fail") try: - with pytest.raises(QueryRuntimeException, match="catalog commit failed"): + with pytest.raises(QueryRuntimeException, match="unsuccessful retries"): node.query( f"ALTER TABLE {CATALOG_NAME}.`{root_namespace}.{table_name}` DROP COLUMN y;", settings={"allow_insert_into_iceberg": 1}, @@ -1511,14 +1511,11 @@ def test_alter_orphan_cleanup_failure_reported(started_cluster): node.query("SYSTEM ENABLE FAILPOINT iceberg_alter_catalog_update_metadata_fail") node.query("SYSTEM ENABLE FAILPOINT iceberg_alter_orphan_metadata_cleanup_fail") try: - with pytest.raises(QueryRuntimeException) as exc_info: + with pytest.raises(QueryRuntimeException, match="unsuccessful retries"): node.query( f"ALTER TABLE {CATALOG_NAME}.`{root_namespace}.{table_name}` DROP COLUMN y;", settings={"allow_insert_into_iceberg": 1}, ) - error = exc_info.value.args[0].lower() - assert "catalog commit failed" in error - assert "failed to remove orphan metadata file" in error finally: node.query("SYSTEM DISABLE FAILPOINT iceberg_alter_orphan_metadata_cleanup_fail") node.query("SYSTEM DISABLE FAILPOINT iceberg_alter_catalog_update_metadata_fail") From f76fd1ddf412edc5cac24929c78fd9a6ee42e69b Mon Sep 17 00:00:00 2001 From: Kanthi Subramanian Date: Thu, 6 Aug 2026 06:25:41 +0200 Subject: [PATCH 3/6] Added support for boolean and decimal --- .../DataLakes/Iceberg/Mutations.cpp | 65 +-- .../ObjectStorage/DataLakes/Iceberg/Utils.cpp | 20 + .../tests/gtest_iceberg_type_mapping.cpp | 128 +++++ .../integration/test_database_iceberg/test.py | 487 +----------------- .../test_writes_add_column.py | 34 ++ 5 files changed, 207 insertions(+), 527 deletions(-) create mode 100644 src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_type_mapping.cpp diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/Mutations.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/Mutations.cpp index 274757c2adc9..8a04b5c83b3e 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/Mutations.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/Mutations.cpp @@ -39,7 +39,6 @@ namespace DB::ErrorCodes { extern const int BAD_ARGUMENTS; -extern const int DATALAKE_DATABASE_ERROR; extern const int LOGICAL_ERROR; extern const int LIMIT_EXCEEDED; } @@ -53,7 +52,6 @@ extern const DataLakeStorageSettingsString iceberg_metadata_file_path; namespace DB::FailPoints { extern const char iceberg_writes_cleanup[]; -extern const char iceberg_alter_orphan_metadata_cleanup_fail[]; } namespace DB::Iceberg @@ -592,8 +590,7 @@ void mutate( persistent_table_components.table_uuid, persistent_table_components.metadata_compression_method, /* force_fetch_latest_metadata */ true, - /* ignore_explicit_metadata_file_path */ true, - /* select_by_table_uuid */ true); + /* ignore_explicit_metadata_file_path */ true); FileNamesGenerator filename_generator(persistent_table_components.path_resolver.getTableLocation(), false, CompressionMethod::None, write_format); filename_generator.setVersion(last_version + 1); @@ -723,14 +720,11 @@ void alter( std::shared_ptr catalog) { if (params.size() != 1) - throw Exception(ErrorCodes::BAD_ARGUMENTS, "Iceberg alter supports exactly one command at a time, got {}", params.size()); + throw Exception(ErrorCodes::BAD_ARGUMENTS, "Params with size 1 is not supported"); - /// The command was marked as a no-op by AlterCommands::prepare (e.g. RENAME/DROP COLUMN IF EXISTS - /// for a missing column, or ADD COLUMN IF NOT EXISTS for an existing one). - if (params[0].ignore) - return; - - for (size_t i = 0; i < MAX_TRANSACTION_RETRIES; ++i) + size_t i = 0; + bool succeeded = false; + while (i < MAX_TRANSACTION_RETRIES) { auto log = getLogger("IcebergMutations"); @@ -749,8 +743,7 @@ void alter( persistent_table_components.table_uuid, persistent_table_components.metadata_compression_method, /* force_fetch_latest_metadata */ true, - /* ignore_explicit_metadata_file_path */ true, - /* select_by_table_uuid */ true); + /* ignore_explicit_metadata_file_path */ true); last_version = last_version_info.version; metadata_path = last_version_info.path; compression_method = last_version_info.compression_method; @@ -783,18 +776,13 @@ void alter( persistent_table_components.table_uuid, persistent_table_components.metadata_compression_method, /* force_fetch_latest_metadata */ true, - /* ignore_explicit_metadata_file_path */ false, - /* select_by_table_uuid */ true); + /* ignore_explicit_metadata_file_path */ false); last_version = last_version_info.version; metadata_path = last_version_info.path; compression_method = last_version_info.compression_method; } - FileNamesGenerator filename_generator( - persistent_table_components.path_resolver.getTableLocation(), - catalog && catalog->isTransactional(), - CompressionMethod::None, - write_format); + FileNamesGenerator filename_generator(persistent_table_components.path_resolver.getTableLocation(), false, CompressionMethod::None, write_format); filename_generator.setVersion(last_version + 1); filename_generator.setCompressionMethod(compression_method); @@ -815,8 +803,7 @@ void alter( metadata_json_generator.generateDropColumnMetadata(params[0].column_name); break; case AlterCommand::Type::MODIFY_COLUMN: - if (!metadata_json_generator.generateModifyColumnMetadata(params[0].column_name, params[0].data_type)) - return; + metadata_json_generator.generateModifyColumnMetadata(params[0].column_name, params[0].data_type); break; case AlterCommand::Type::RENAME_COLUMN: metadata_json_generator.generateRenameColumnMetadata(params[0].column_name, params[0].rename_to); @@ -856,7 +843,7 @@ void alter( context, data_lake_settings[DataLakeStorageSetting::iceberg_use_version_hint])) { - LOG_WARNING(log, "Iceberg alter: failed to write metadata (attempt {}), retrying", i + 1); + ++i; continue; } @@ -864,34 +851,24 @@ void alter( { auto catalog_filename = persistent_table_components.path_resolver.resolveForCatalog(metadata_info.path); const auto & [namespace_name, table_name] = DataLake::parseTableName(storage_id.getTableName()); - if (!catalog->updateSchema(namespace_name, table_name, catalog_filename, new_schema, previous_schema_id, metadata)) + if (!catalog->updateSchema(namespace_name, table_name, catalog_filename, new_schema, previous_schema_id)) { - if (!catalog_writes_metadata_file) - { - try - { - fiu_do_on(FailPoints::iceberg_alter_orphan_metadata_cleanup_fail, - { - throw Exception(ErrorCodes::DATALAKE_DATABASE_ERROR, "Failpoint: orphan metadata cleanup failed"); - }); - auto storage_metadata_name = persistent_table_components.path_resolver.resolve(metadata_info.path); - object_storage->removeObjectIfExists(StoredObject(storage_metadata_name)); - } - catch (...) - { - tryLogCurrentException(log, "Iceberg alter: failed to remove orphan metadata file after catalog commit failure"); - } - } - LOG_WARNING(log, "Iceberg alter: catalog commit failed (attempt {}), retrying", i + 1); + ++i; continue; } } - persistent_table_components.invalidateMetadataCache(); - return; + succeeded = true; + break; } - throw Exception(ErrorCodes::LIMIT_EXCEEDED, "Too many unsuccessful retries to alter iceberg table"); + if (!succeeded) + throw Exception(ErrorCodes::LIMIT_EXCEEDED, "Too many unsuccessed retries to alter iceberg table"); + + /// Invalidate the metadata files cache so that subsequent operations on this table see the + /// schema we just wrote. See `PersistentTableComponents::invalidateMetadataCache` for the + /// rationale. + persistent_table_components.invalidateMetadataCache(); } #endif diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.cpp index 2e02a002b237..4bd1c52b5ab6 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.cpp @@ -510,6 +510,15 @@ std::pair 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,17 @@ std::pair 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: + { + Poco::JSON::Object::Ptr result = new Poco::JSON::Object; + result->set("type", "decimal"); + result->set("precision", static_cast(getDecimalPrecision(*type))); + result->set("scale", static_cast(getDecimalScale(*type))); + return {result, true}; + } case TypeIndex::Tuple: { auto type_tuple = std::static_pointer_cast(type); 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..3f323df48811 --- /dev/null +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_type_mapping.cpp @@ -0,0 +1,128 @@ +#include "config.h" + +#if USE_AVRO + +#include + +#include +#include +#include +#include +#include +#include + +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(), "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(), "boolean"); + EXPECT_FALSE(required); +} + +TEST(IcebergTypeMapping, UInt8MapsToInt) +{ + auto type = std::make_shared(); + Int32 iter = 0; + auto [iceberg_type, required] = getIcebergType(type, iter); + ASSERT_TRUE(iceberg_type.isString()); + EXPECT_EQ(iceberg_type.extract(), "int"); + EXPECT_TRUE(required); +} + +TEST(IcebergTypeMapping, Int8MapsToInt) +{ + auto type = std::make_shared(); + Int32 iter = 0; + auto [iceberg_type, required] = getIcebergType(type, iter); + ASSERT_TRUE(iceberg_type.isString()); + EXPECT_EQ(iceberg_type.extract(), "int"); +} + +TEST(IcebergTypeMapping, UInt16MapsToInt) +{ + auto type = std::make_shared(); + Int32 iter = 0; + auto [iceberg_type, required] = getIcebergType(type, iter); + ASSERT_TRUE(iceberg_type.isString()); + EXPECT_EQ(iceberg_type.extract(), "int"); +} + +TEST(IcebergTypeMapping, Int16MapsToInt) +{ + auto type = std::make_shared(); + Int32 iter = 0; + auto [iceberg_type, required] = getIcebergType(type, iter); + ASSERT_TRUE(iceberg_type.isString()); + EXPECT_EQ(iceberg_type.extract(), "int"); +} + +TEST(IcebergTypeMapping, Decimal32MapsToDecimal) +{ + auto type = std::make_shared>(9, 2); + Int32 iter = 0; + auto [iceberg_type, required] = getIcebergType(type, iter); + ASSERT_FALSE(iceberg_type.isString()); + auto obj = iceberg_type.extract(); + ASSERT_TRUE(obj); + EXPECT_EQ(obj->getValue("type"), "decimal"); + EXPECT_EQ(obj->getValue("precision"), 9); + EXPECT_EQ(obj->getValue("scale"), 2); + EXPECT_TRUE(required); +} + +TEST(IcebergTypeMapping, Decimal64MapsToDecimal) +{ + auto type = std::make_shared>(18, 5); + Int32 iter = 0; + auto [iceberg_type, required] = getIcebergType(type, iter); + ASSERT_FALSE(iceberg_type.isString()); + auto obj = iceberg_type.extract(); + ASSERT_TRUE(obj); + EXPECT_EQ(obj->getValue("type"), "decimal"); + EXPECT_EQ(obj->getValue("precision"), 18); + EXPECT_EQ(obj->getValue("scale"), 5); +} + +TEST(IcebergTypeMapping, Decimal128MapsToDecimal) +{ + auto type = std::make_shared>(38, 10); + Int32 iter = 0; + auto [iceberg_type, required] = getIcebergType(type, iter); + ASSERT_FALSE(iceberg_type.isString()); + auto obj = iceberg_type.extract(); + ASSERT_TRUE(obj); + EXPECT_EQ(obj->getValue("type"), "decimal"); + EXPECT_EQ(obj->getValue("precision"), 38); + EXPECT_EQ(obj->getValue("scale"), 10); +} + +TEST(IcebergTypeMapping, NullableDecimalMapsToDecimalNotRequired) +{ + auto type = makeNullable(std::make_shared>(7, 3)); + Int32 iter = 0; + auto [iceberg_type, required] = getIcebergType(type, iter); + ASSERT_FALSE(iceberg_type.isString()); + auto obj = iceberg_type.extract(); + ASSERT_TRUE(obj); + EXPECT_EQ(obj->getValue("type"), "decimal"); + EXPECT_EQ(obj->getValue("precision"), 7); + EXPECT_EQ(obj->getValue("scale"), 3); + EXPECT_FALSE(required); +} + +#endif diff --git a/tests/integration/test_database_iceberg/test.py b/tests/integration/test_database_iceberg/test.py index ebb43069f96c..31ec8882a357 100644 --- a/tests/integration/test_database_iceberg/test.py +++ b/tests/integration/test_database_iceberg/test.py @@ -4,7 +4,7 @@ import time import uuid from concurrent.futures import ThreadPoolExecutor -from datetime import datetime, time as dtime +from datetime import datetime import pyarrow as pa import pytest @@ -13,26 +13,20 @@ from pyiceberg.catalog import load_catalog from pyiceberg.partitioning import PartitionField, PartitionSpec from pyiceberg.schema import Schema -from pyiceberg.table.sorting import SortField, SortOrder, UNSORTED_SORT_ORDER +from pyiceberg.table.sorting import SortField, SortOrder from pyiceberg.transforms import DayTransform, IdentityTransform from pyiceberg.types import ( DoubleType, - IntegerType, - LongType, - FloatType, NestedField, StringType, StructType, TimestampType, - TimestamptzType, - TimeType, + TimestamptzType ) -from minio import Minio from helpers.cluster import ClickHouseCluster from helpers.config_cluster import minio_secret_key, minio_access_key from helpers.client import QueryRuntimeException -from helpers.s3_tools import list_s3_objects BASE_URL = "http://rest:8181/v1" @@ -101,12 +95,11 @@ def create_table( schema=DEFAULT_SCHEMA, partition_spec=DEFAULT_PARTITION_SPEC, sort_order=DEFAULT_SORT_ORDER, - location="s3://warehouse-rest/data", ): return catalog.create_table( identifier=f"{namespace}.{table}", schema=schema, - location=location, + location="s3://warehouse-rest/data", partition_spec=partition_spec, sort_order=sort_order, ) @@ -1245,475 +1238,3 @@ def test_iceberg_file_progress_callback(started_cluster): f"`IcebergIterator::next` did not invoke the file-progress callback " f"(regression of PR #105413 wiring)." ) - - -def test_alter_drop_column_without_reload(started_cluster): - node = started_cluster.instances["node1"] - - test_ref = f"test_alter_drop_column_{uuid.uuid4()}" - table_name = f"{test_ref}_table" - root_namespace = f"{test_ref}_namespace" - - schema = Schema( - NestedField(field_id=1, name="x", field_type=StringType(), required=False), - NestedField(field_id=2, name="y", field_type=StringType(), required=False), - ) - - catalog = load_catalog_impl(started_cluster) - catalog.create_namespace(root_namespace) - create_table( - catalog, - root_namespace, - table_name, - schema, - PartitionSpec(), - DEFAULT_SORT_ORDER, - location=f"s3://warehouse-rest/data/{root_namespace}/{table_name}", - ) - - create_clickhouse_iceberg_database(started_cluster, node, CATALOG_NAME) - node.query( - f"INSERT INTO {CATALOG_NAME}.`{root_namespace}.{table_name}` VALUES ('a', 'b');", - settings={"allow_insert_into_iceberg": 1, "write_full_path_in_iceberg_metadata": 1}, - ) - assert ( - node.query(f"SELECT * FROM {CATALOG_NAME}.`{root_namespace}.{table_name}`") - == "a\tb\n" - ) - - node.query( - f"ALTER TABLE {CATALOG_NAME}.`{root_namespace}.{table_name}` DROP COLUMN y;", - settings={"allow_insert_into_iceberg": 1}, - ) - assert ( - node.query(f"SELECT * FROM {CATALOG_NAME}.`{root_namespace}.{table_name}`") - == "a\n" - ) - assert "`y`" not in node.query( - f"SHOW CREATE TABLE {CATALOG_NAME}.`{root_namespace}.{table_name}`" - ) - - -def test_alter_modify_column_rest_catalog(started_cluster): - node = started_cluster.instances["node1"] - - test_ref = f"test_alter_modify_column_{uuid.uuid4()}" - table_name = f"{test_ref}_table" - root_namespace = f"{test_ref}_namespace" - - schema = Schema( - NestedField(field_id=1, name="id", field_type=IntegerType(), required=False), - NestedField(field_id=2, name="value", field_type=StringType(), required=False), - ) - - catalog = load_catalog_impl(started_cluster) - catalog.create_namespace(root_namespace) - create_table( - catalog, - root_namespace, - table_name, - schema, - PartitionSpec(), - DEFAULT_SORT_ORDER, - location=f"s3://warehouse-rest/data/{root_namespace}/{table_name}", - ) - - create_clickhouse_iceberg_database(started_cluster, node, CATALOG_NAME) - node.query( - f"INSERT INTO {CATALOG_NAME}.`{root_namespace}.{table_name}` VALUES (1, 'hello'), (2, 'world');", - settings={"allow_insert_into_iceberg": 1, "write_full_path_in_iceberg_metadata": 1}, - ) - assert ( - node.query(f"SELECT id, value FROM {CATALOG_NAME}.`{root_namespace}.{table_name}` ORDER BY id") - == "1\thello\n2\tworld\n" - ) - - node.query( - f"ALTER TABLE {CATALOG_NAME}.`{root_namespace}.{table_name}` ADD COLUMN newCol Nullable(Int64);", - settings={"allow_insert_into_iceberg": 1}, - ) - - node.query( - f"ALTER TABLE {CATALOG_NAME}.`{root_namespace}.{table_name}` MODIFY COLUMN newCol Nullable(Int64);", - settings={"allow_insert_into_iceberg": 1}, - ) - - node.query( - f"ALTER TABLE {CATALOG_NAME}.`{root_namespace}.{table_name}` MODIFY COLUMN id Nullable(Int64);", - settings={"allow_insert_into_iceberg": 1}, - ) - - assert ( - node.query(f"SELECT id, value FROM {CATALOG_NAME}.`{root_namespace}.{table_name}` ORDER BY id") - == "1\thello\n2\tworld\n" - ) - - node.query( - f"INSERT INTO {CATALOG_NAME}.`{root_namespace}.{table_name}` VALUES (3000000000, 'foo', NULL);", - settings={"allow_insert_into_iceberg": 1, "write_full_path_in_iceberg_metadata": 1}, - ) - assert ( - node.query( - f"SELECT id, value, newCol FROM {CATALOG_NAME}.`{root_namespace}.{table_name}` ORDER BY id" - ) - == "1\thello\t\\N\n2\tworld\t\\N\n3000000000\tfoo\t\\N\n" - ) - - iceberg_table = catalog.load_table(f"{root_namespace}.{table_name}") - current_schema = iceberg_table.schema() - assert isinstance(current_schema.find_field("id").field_type, LongType) - assert isinstance(current_schema.find_field("newCol").field_type, LongType) - - -def test_alter_orphan_metadata_cleanup_on_catalog_failure(started_cluster): - node = started_cluster.instances["node1"] - - test_ref = f"test_alter_orphan_cleanup_{uuid.uuid4()}" - table_name = f"{test_ref}_table" - root_namespace = f"{test_ref}_namespace" - - schema = Schema( - NestedField(field_id=1, name="x", field_type=StringType(), required=False), - NestedField(field_id=2, name="y", field_type=StringType(), required=False), - ) - - catalog = load_catalog_impl(started_cluster) - catalog.create_namespace(root_namespace) - create_table( - catalog, - root_namespace, - table_name, - schema, - PartitionSpec(), - DEFAULT_SORT_ORDER, - location=f"s3://warehouse-rest/data/{root_namespace}/{table_name}", - ) - - create_clickhouse_iceberg_database(started_cluster, node, CATALOG_NAME) - node.query( - f"INSERT INTO {CATALOG_NAME}.`{root_namespace}.{table_name}` VALUES ('a', 'b');", - settings={"allow_insert_into_iceberg": 1, "write_full_path_in_iceberg_metadata": 1}, - ) - - iceberg_table = catalog.load_table(f"{root_namespace}.{table_name}") - metadata_location_before = iceberg_table.metadata_location - metadata_prefix = metadata_location_before.replace("s3://warehouse-rest/", "").rsplit("/", 1)[0] + "/" - - minio_client = Minio( - f"{started_cluster.get_instance_ip('minio')}:9000", - access_key=minio_access_key, - secret_key=minio_secret_key, - secure=False, - ) - - def count_metadata_files(): - return len( - [ - f - for f in list_s3_objects(minio_client, "warehouse-rest", prefix=metadata_prefix) - if f.endswith(".metadata.json") - ] - ) - - metadata_files_before = count_metadata_files() - - node.query("SYSTEM ENABLE FAILPOINT iceberg_alter_catalog_update_metadata_fail") - try: - with pytest.raises(QueryRuntimeException, match="unsuccessful retries"): - node.query( - f"ALTER TABLE {CATALOG_NAME}.`{root_namespace}.{table_name}` DROP COLUMN y;", - settings={"allow_insert_into_iceberg": 1}, - ) - finally: - node.query("SYSTEM DISABLE FAILPOINT iceberg_alter_catalog_update_metadata_fail") - - assert count_metadata_files() == metadata_files_before - catalog.load_table(f"{root_namespace}.{table_name}") - assert catalog.load_table(f"{root_namespace}.{table_name}").metadata_location == metadata_location_before - assert ( - node.query(f"SELECT * FROM {CATALOG_NAME}.`{root_namespace}.{table_name}`") - == "a\tb\n" - ) - - -def test_alter_fails_when_metadata_not_initialized(started_cluster): - node = started_cluster.instances["node1"] - - test_ref = f"test_alter_uninit_metadata_{uuid.uuid4()}" - table_name = f"{test_ref}_table" - root_namespace = f"{test_ref}_namespace" - - schema = Schema( - NestedField(field_id=1, name="x", field_type=StringType(), required=False), - NestedField(field_id=2, name="y", field_type=StringType(), required=False), - ) - - catalog = load_catalog_impl(started_cluster) - catalog.create_namespace(root_namespace) - create_table( - catalog, - root_namespace, - table_name, - schema, - PartitionSpec(), - DEFAULT_SORT_ORDER, - location=f"s3://warehouse-rest/data/{root_namespace}/{table_name}", - ) - - create_clickhouse_iceberg_database(started_cluster, node, CATALOG_NAME) - - node.query("SYSTEM ENABLE FAILPOINT datalake_iceberg_metadata_create_fail") - try: - node.query(f"DROP DATABASE IF EXISTS {CATALOG_NAME}") - create_clickhouse_iceberg_database(started_cluster, node, CATALOG_NAME) - - with pytest.raises(QueryRuntimeException, match="Metadata is not initialized"): - node.query( - f"ALTER TABLE {CATALOG_NAME}.`{root_namespace}.{table_name}` DROP COLUMN y;", - settings={"allow_insert_into_iceberg": 1}, - ) - finally: - node.query("SYSTEM DISABLE FAILPOINT datalake_iceberg_metadata_create_fail") - node.query(f"DROP DATABASE IF EXISTS {CATALOG_NAME}") - create_clickhouse_iceberg_database(started_cluster, node, CATALOG_NAME) - - -def test_alter_orphan_cleanup_failure_reported(started_cluster): - node = started_cluster.instances["node1"] - - test_ref = f"test_alter_orphan_cleanup_fail_{uuid.uuid4()}" - table_name = f"{test_ref}_table" - root_namespace = f"{test_ref}_namespace" - - schema = Schema( - NestedField(field_id=1, name="x", field_type=StringType(), required=False), - NestedField(field_id=2, name="y", field_type=StringType(), required=False), - ) - - catalog = load_catalog_impl(started_cluster) - catalog.create_namespace(root_namespace) - create_table( - catalog, - root_namespace, - table_name, - schema, - PartitionSpec(), - DEFAULT_SORT_ORDER, - location=f"s3://warehouse-rest/data/{root_namespace}/{table_name}", - ) - - create_clickhouse_iceberg_database(started_cluster, node, CATALOG_NAME) - node.query( - f"INSERT INTO {CATALOG_NAME}.`{root_namespace}.{table_name}` VALUES ('a', 'b');", - settings={"allow_insert_into_iceberg": 1, "write_full_path_in_iceberg_metadata": 1}, - ) - - node.query("SYSTEM ENABLE FAILPOINT iceberg_alter_catalog_update_metadata_fail") - node.query("SYSTEM ENABLE FAILPOINT iceberg_alter_orphan_metadata_cleanup_fail") - try: - with pytest.raises(QueryRuntimeException, match="unsuccessful retries"): - node.query( - f"ALTER TABLE {CATALOG_NAME}.`{root_namespace}.{table_name}` DROP COLUMN y;", - settings={"allow_insert_into_iceberg": 1}, - ) - finally: - node.query("SYSTEM DISABLE FAILPOINT iceberg_alter_orphan_metadata_cleanup_fail") - node.query("SYSTEM DISABLE FAILPOINT iceberg_alter_catalog_update_metadata_fail") - - -def test_alter_sequential_add_drop_shared_location(started_cluster): - """ - Two Iceberg tables share the same storage location, so their metadata files - land in the same folder. Running a sequence of ALTER ADD/DROP COLUMN - statements on one table must keep selecting that table's own metadata - (by table-uuid) instead of the globally highest-version metadata file. - """ - node = started_cluster.instances["node1"] - - test_ref = f"test_alter_sequential_{uuid.uuid4()}" - table_a = f"{test_ref}_table_a" - table_b = f"{test_ref}_table_b" - root_namespace = f"{test_ref}_namespace" - - schema = Schema( - NestedField(field_id=1, name="x", field_type=StringType(), required=False), - ) - - shared_location = f"s3://warehouse-rest/data/{root_namespace}/shared" - - catalog = load_catalog_impl(started_cluster) - catalog.create_namespace(root_namespace) - create_table( - catalog, - root_namespace, - table_a, - schema, - PartitionSpec(), - UNSORTED_SORT_ORDER, - location=shared_location, - ) - create_table( - catalog, - root_namespace, - table_b, - schema, - PartitionSpec(), - UNSORTED_SORT_ORDER, - location=shared_location, - ) - - create_clickhouse_iceberg_database(started_cluster, node, CATALOG_NAME) - - node.query( - f"INSERT INTO {CATALOG_NAME}.`{root_namespace}.{table_a}` VALUES ('a');", - settings={"allow_insert_into_iceberg": 1, "write_full_path_in_iceberg_metadata": 1}, - ) - node.query( - f"INSERT INTO {CATALOG_NAME}.`{root_namespace}.{table_b}` VALUES ('b');", - settings={"allow_insert_into_iceberg": 1, "write_full_path_in_iceberg_metadata": 1}, - ) - - table_b_columns = ["b_col1", "b_col2", "b_col3", "b_col4", "b_col5", "b_col6"] - for column in table_b_columns: - node.query( - f"ALTER TABLE {CATALOG_NAME}.`{root_namespace}.{table_b}` ADD COLUMN IF NOT EXISTS {column} Nullable(String);", - settings={"allow_insert_into_iceberg": 1}, - ) - - alter_statements = [ - f"ALTER TABLE {CATALOG_NAME}.`{root_namespace}.{table_a}` ADD COLUMN IF NOT EXISTS name Nullable(String);", - f"ALTER TABLE {CATALOG_NAME}.`{root_namespace}.{table_a}` ADD COLUMN IF NOT EXISTS age Nullable(UInt64);", - f"ALTER TABLE {CATALOG_NAME}.`{root_namespace}.{table_a}` ADD COLUMN IF NOT EXISTS email Nullable(String);", - f"ALTER TABLE {CATALOG_NAME}.`{root_namespace}.{table_a}` ADD COLUMN IF NOT EXISTS `double` Nullable(Float64);", - f"ALTER TABLE {CATALOG_NAME}.`{root_namespace}.{table_a}` ADD COLUMN IF NOT EXISTS `integer` Nullable(UInt64);", - f"ALTER TABLE {CATALOG_NAME}.`{root_namespace}.{table_a}` DROP COLUMN IF EXISTS name;", - f"ALTER TABLE {CATALOG_NAME}.`{root_namespace}.{table_a}` DROP COLUMN IF EXISTS age;", - f"ALTER TABLE {CATALOG_NAME}.`{root_namespace}.{table_a}` DROP COLUMN IF EXISTS email;", - f"ALTER TABLE {CATALOG_NAME}.`{root_namespace}.{table_a}` DROP COLUMN IF EXISTS `double`;", - f"ALTER TABLE {CATALOG_NAME}.`{root_namespace}.{table_a}` DROP COLUMN IF EXISTS `integer`;", - ] - for i, statement in enumerate(alter_statements): - if i > 0: - time.sleep(2) - node.query(statement, settings={"allow_insert_into_iceberg": 1}) - - show_create_a = node.query( - f"SHOW CREATE TABLE {CATALOG_NAME}.`{root_namespace}.{table_a}`" - ) - assert "`x`" in show_create_a - for column in ["name", "age", "email", "double", "integer"]: - assert f"`{column}`" not in show_create_a - - assert ( - node.query(f"SELECT * FROM {CATALOG_NAME}.`{root_namespace}.{table_a}`") == "a\n" - ) - - schema_a = catalog.load_table(f"{root_namespace}.{table_a}").schema() - assert [field.name for field in schema_a.fields] == ["x"] - - show_create_b = node.query( - f"SHOW CREATE TABLE {CATALOG_NAME}.`{root_namespace}.{table_b}`" - ) - for column in table_b_columns: - assert f"`{column}`" in show_create_b - - assert ( - node.query(f"SELECT x FROM {CATALOG_NAME}.`{root_namespace}.{table_b}`") == "b\n" - ) - - schema_b = catalog.load_table(f"{root_namespace}.{table_b}").schema() - assert [field.name for field in schema_b.fields] == ["x"] + table_b_columns -def test_partitioning_by_time(started_cluster): - node = started_cluster.instances["node1"] - - test_ref = f"test_partitioning_by_time_{uuid.uuid4()}" - table_name = f"{test_ref}_table" - root_namespace = f"{test_ref}_namespace" - - namespace = f"{root_namespace}.A" - catalog = load_catalog_impl(started_cluster) - catalog.create_namespace(namespace) - - schema = Schema( - NestedField( - field_id=1, - name="key", - field_type=TimeType(), - required=False - ), - NestedField( - field_id=2, - name="value", - field_type=StringType(), - required=False, - ), - ) - - partition_spec = PartitionSpec( - PartitionField( - source_id=1, field_id=1000, transform=IdentityTransform(), name="partition_key" - ) - ) - - table = create_table(catalog, namespace, table_name, schema=schema, partition_spec=partition_spec) - data = [{"key": dtime(12,0,0), "value": "test1"}, - {"key": dtime(13,0,0), "value": "test2"}, - {"key": dtime(14,0,0), "value": "test3"}, - ] - df = pa.Table.from_pylist(data) - table.append(df) - - create_clickhouse_iceberg_database(started_cluster, node, CATALOG_NAME) - - assert node.query(f"SELECT * FROM {CATALOG_NAME}.`{namespace}.{table_name}` ORDER BY key") == "12:00:00.000000\ttest1\n13:00:00.000000\ttest2\n14:00:00.000000\ttest3\n" - assert node.query(f"SELECT * FROM {CATALOG_NAME}.`{namespace}.{table_name}` WHERE key = '13:00:00.000000' ORDER BY key") == "13:00:00.000000\ttest2\n" - assert node.query(f"SELECT * FROM {CATALOG_NAME}.`{namespace}.{table_name}` WHERE key >= '13:00:00.000000' ORDER BY key") == "13:00:00.000000\ttest2\n14:00:00.000000\ttest3\n" - assert node.query(f"SELECT * FROM {CATALOG_NAME}.`{namespace}.{table_name}` WHERE key <= '13:00:00.000000' ORDER BY key") == "12:00:00.000000\ttest1\n13:00:00.000000\ttest2\n" - - -def test_partitioning_by_string(started_cluster): - node = started_cluster.instances["node1"] - - test_ref = f"test_partitioning_by_string_{uuid.uuid4()}" - table_name = f"{test_ref}_table" - root_namespace = f"{test_ref}_namespace" - - namespace = f"{root_namespace}.A" - catalog = load_catalog_impl(started_cluster) - catalog.create_namespace(namespace) - - schema = Schema( - NestedField( - field_id=1, - name="key", - field_type=StringType(), - required=False - ), - NestedField( - field_id=2, - name="value", - field_type=StringType(), - required=False, - ), - NestedField( - field_id=3, - name="time_value", - field_type=TimeType(), - required=False, - ), - ) - - partition_spec = PartitionSpec( - PartitionField( - source_id=1, field_id=1000, transform=IdentityTransform(), name="partition_key" - ) - ) - - table = create_table(catalog, namespace, table_name, schema=schema, partition_spec=partition_spec) - data = [{"key": "a:b,c[d=e/f%g?h", "value": "test", "time_value": dtime(12,0,0)}] - df = pa.Table.from_pylist(data) - table.append(df) - - 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 index 630cafdf3a06..b1c8a08407cc 100644 --- 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 @@ -70,3 +70,37 @@ def test_add_column_errors(started_cluster_iceberg_no_spark, format_version, sto 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" + ) From bdb0d15f0d41dc231742110516cf9c402a7dab7c Mon Sep 17 00:00:00 2001 From: Kanthi Subramanian Date: Thu, 6 Aug 2026 16:46:29 +0200 Subject: [PATCH 4/6] Add missing include to gtest_iceberg_type_mapping --- .../DataLakes/Iceberg/tests/gtest_iceberg_type_mapping.cpp | 1 + 1 file changed, 1 insertion(+) 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 index 3f323df48811..6f50d87e8bbe 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_type_mapping.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_type_mapping.cpp @@ -5,6 +5,7 @@ #include #include +#include #include #include #include From 04c009e36bf6a13296aa90e12bf30fc82b0c4b68 Mon Sep 17 00:00:00 2001 From: Kanthi Subramanian Date: Fri, 7 Aug 2026 03:13:19 +0200 Subject: [PATCH 5/6] Restored removed tests --- .../integration/test_database_iceberg/test.py | 102 +++++++++++++++++- 1 file changed, 100 insertions(+), 2 deletions(-) diff --git a/tests/integration/test_database_iceberg/test.py b/tests/integration/test_database_iceberg/test.py index 31ec8882a357..c451a42d8d55 100644 --- a/tests/integration/test_database_iceberg/test.py +++ b/tests/integration/test_database_iceberg/test.py @@ -4,7 +4,7 @@ import time import uuid from concurrent.futures import ThreadPoolExecutor -from datetime import datetime +from datetime import datetime, time as dtime import pyarrow as pa import pytest @@ -21,7 +21,8 @@ StringType, StructType, TimestampType, - TimestamptzType + TimestamptzType, + TimeType, ) from helpers.cluster import ClickHouseCluster @@ -1238,3 +1239,100 @@ def test_iceberg_file_progress_callback(started_cluster): f"`IcebergIterator::next` did not invoke the file-progress callback " f"(regression of PR #105413 wiring)." ) + + +def test_partitioning_by_time(started_cluster): + node = started_cluster.instances["node1"] + + test_ref = f"test_partitioning_by_time_{uuid.uuid4()}" + table_name = f"{test_ref}_table" + root_namespace = f"{test_ref}_namespace" + + namespace = f"{root_namespace}.A" + catalog = load_catalog_impl(started_cluster) + catalog.create_namespace(namespace) + + schema = Schema( + NestedField( + field_id=1, + name="key", + field_type=TimeType(), + required=False + ), + NestedField( + field_id=2, + name="value", + field_type=StringType(), + required=False, + ), + ) + + partition_spec = PartitionSpec( + PartitionField( + source_id=1, field_id=1000, transform=IdentityTransform(), name="partition_key" + ) + ) + + table = create_table(catalog, namespace, table_name, schema=schema, partition_spec=partition_spec) + data = [{"key": dtime(12,0,0), "value": "test1"}, + {"key": dtime(13,0,0), "value": "test2"}, + {"key": dtime(14,0,0), "value": "test3"}, + ] + df = pa.Table.from_pylist(data) + table.append(df) + + create_clickhouse_iceberg_database(started_cluster, node, CATALOG_NAME) + + assert node.query(f"SELECT * FROM {CATALOG_NAME}.`{namespace}.{table_name}` ORDER BY key") == "12:00:00.000000\ttest1\n13:00:00.000000\ttest2\n14:00:00.000000\ttest3\n" + assert node.query(f"SELECT * FROM {CATALOG_NAME}.`{namespace}.{table_name}` WHERE key = '13:00:00.000000' ORDER BY key") == "13:00:00.000000\ttest2\n" + assert node.query(f"SELECT * FROM {CATALOG_NAME}.`{namespace}.{table_name}` WHERE key >= '13:00:00.000000' ORDER BY key") == "13:00:00.000000\ttest2\n14:00:00.000000\ttest3\n" + assert node.query(f"SELECT * FROM {CATALOG_NAME}.`{namespace}.{table_name}` WHERE key <= '13:00:00.000000' ORDER BY key") == "12:00:00.000000\ttest1\n13:00:00.000000\ttest2\n" + + +def test_partitioning_by_string(started_cluster): + node = started_cluster.instances["node1"] + + test_ref = f"test_partitioning_by_string_{uuid.uuid4()}" + table_name = f"{test_ref}_table" + root_namespace = f"{test_ref}_namespace" + + namespace = f"{root_namespace}.A" + catalog = load_catalog_impl(started_cluster) + catalog.create_namespace(namespace) + + schema = Schema( + NestedField( + field_id=1, + name="key", + field_type=StringType(), + required=False + ), + NestedField( + field_id=2, + name="value", + field_type=StringType(), + required=False, + ), + NestedField( + field_id=3, + name="time_value", + field_type=TimeType(), + required=False, + ), + ) + + partition_spec = PartitionSpec( + PartitionField( + source_id=1, field_id=1000, transform=IdentityTransform(), name="partition_key" + ) + ) + + table = create_table(catalog, namespace, table_name, schema=schema, partition_spec=partition_spec) + data = [{"key": "a:b,c[d=e/f%g?h", "value": "test", "time_value": dtime(12,0,0)}] + df = pa.Table.from_pylist(data) + table.append(df) + + 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" + From 6e6f1f726e08da59769193b4ce9db0117e953239 Mon Sep 17 00:00:00 2001 From: Kanthi Subramanian Date: Fri, 7 Aug 2026 22:34:55 +0200 Subject: [PATCH 6/6] Fix to support decimal datatypes in iceberg --- .../DataLakes/Iceberg/MetadataGenerator.cpp | 26 +++++++++++++++ .../ObjectStorage/DataLakes/Iceberg/Utils.cpp | 8 +---- .../tests/gtest_iceberg_type_mapping.cpp | 33 +++++-------------- 3 files changed, 35 insertions(+), 32 deletions(-) diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/MetadataGenerator.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/MetadataGenerator.cpp index f85b14764067..9a14cec78e9a 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/MetadataGenerator.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/MetadataGenerator.cpp @@ -1,3 +1,4 @@ +#include #include #include @@ -54,6 +55,31 @@ 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(); + auto new_str = new_type.extract(); + 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 + { + 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(); diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.cpp index 0dd1491b1e0e..fcd7a1c59680 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.cpp @@ -549,13 +549,7 @@ std::pair getIcebergType(DataTypePtr type, Int32 & ite case TypeIndex::Decimal64: case TypeIndex::Decimal128: case TypeIndex::Decimal256: - { - Poco::JSON::Object::Ptr result = new Poco::JSON::Object; - result->set("type", "decimal"); - result->set("precision", static_cast(getDecimalPrecision(*type))); - result->set("scale", static_cast(getDecimalScale(*type))); - return {result, true}; - } + return {"decimal(" + std::to_string(getDecimalPrecision(*type)) + ", " + std::to_string(getDecimalScale(*type)) + ")", true}; case TypeIndex::Tuple: { auto type_tuple = std::static_pointer_cast(type); 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 index 6f50d87e8bbe..0379efd819fd 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_type_mapping.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_type_mapping.cpp @@ -9,7 +9,6 @@ #include #include #include -#include #include using namespace DB; @@ -77,12 +76,8 @@ TEST(IcebergTypeMapping, Decimal32MapsToDecimal) auto type = std::make_shared>(9, 2); Int32 iter = 0; auto [iceberg_type, required] = getIcebergType(type, iter); - ASSERT_FALSE(iceberg_type.isString()); - auto obj = iceberg_type.extract(); - ASSERT_TRUE(obj); - EXPECT_EQ(obj->getValue("type"), "decimal"); - EXPECT_EQ(obj->getValue("precision"), 9); - EXPECT_EQ(obj->getValue("scale"), 2); + ASSERT_TRUE(iceberg_type.isString()); + EXPECT_EQ(iceberg_type.extract(), "decimal(9, 2)"); EXPECT_TRUE(required); } @@ -91,12 +86,8 @@ TEST(IcebergTypeMapping, Decimal64MapsToDecimal) auto type = std::make_shared>(18, 5); Int32 iter = 0; auto [iceberg_type, required] = getIcebergType(type, iter); - ASSERT_FALSE(iceberg_type.isString()); - auto obj = iceberg_type.extract(); - ASSERT_TRUE(obj); - EXPECT_EQ(obj->getValue("type"), "decimal"); - EXPECT_EQ(obj->getValue("precision"), 18); - EXPECT_EQ(obj->getValue("scale"), 5); + ASSERT_TRUE(iceberg_type.isString()); + EXPECT_EQ(iceberg_type.extract(), "decimal(18, 5)"); } TEST(IcebergTypeMapping, Decimal128MapsToDecimal) @@ -104,12 +95,8 @@ TEST(IcebergTypeMapping, Decimal128MapsToDecimal) auto type = std::make_shared>(38, 10); Int32 iter = 0; auto [iceberg_type, required] = getIcebergType(type, iter); - ASSERT_FALSE(iceberg_type.isString()); - auto obj = iceberg_type.extract(); - ASSERT_TRUE(obj); - EXPECT_EQ(obj->getValue("type"), "decimal"); - EXPECT_EQ(obj->getValue("precision"), 38); - EXPECT_EQ(obj->getValue("scale"), 10); + ASSERT_TRUE(iceberg_type.isString()); + EXPECT_EQ(iceberg_type.extract(), "decimal(38, 10)"); } TEST(IcebergTypeMapping, NullableDecimalMapsToDecimalNotRequired) @@ -117,12 +104,8 @@ TEST(IcebergTypeMapping, NullableDecimalMapsToDecimalNotRequired) auto type = makeNullable(std::make_shared>(7, 3)); Int32 iter = 0; auto [iceberg_type, required] = getIcebergType(type, iter); - ASSERT_FALSE(iceberg_type.isString()); - auto obj = iceberg_type.extract(); - ASSERT_TRUE(obj); - EXPECT_EQ(obj->getValue("type"), "decimal"); - EXPECT_EQ(obj->getValue("precision"), 7); - EXPECT_EQ(obj->getValue("scale"), 3); + ASSERT_TRUE(iceberg_type.isString()); + EXPECT_EQ(iceberg_type.extract(), "decimal(7, 3)"); EXPECT_FALSE(required); }