diff --git a/src/Common/FailPoint.cpp b/src/Common/FailPoint.cpp index 60b516e377e0..7887bf1a77d3 100644 --- a/src/Common/FailPoint.cpp +++ b/src/Common/FailPoint.cpp @@ -164,6 +164,8 @@ 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_schema_fail) \ + ONCE(iceberg_alter_catalog_commit_reported_as_failed) \ REGULAR(storage_cluster_read_sleep) \ ONCE(iceberg_writes_non_retry_cleanup) \ ONCE(iceberg_writes_post_publish_throw) \ diff --git a/src/Databases/DataLake/GlueCatalog.cpp b/src/Databases/DataLake/GlueCatalog.cpp index 781a3bb19e22..a8dda240e7ce 100644 --- a/src/Databases/DataLake/GlueCatalog.cpp +++ b/src/Databases/DataLake/GlueCatalog.cpp @@ -746,7 +746,9 @@ 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*/, + Int32 /*new_last_column_id*/, + Poco::JSON::Object::Ptr /*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 4b2a6f0d570c..8d10ba0c8667 100644 --- a/src/Databases/DataLake/GlueCatalog.h +++ b/src/Databases/DataLake/GlueCatalog.h @@ -75,7 +75,9 @@ 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, + Int32 new_last_column_id, + Poco::JSON::Object::Ptr 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 50200b3d3cc6..62cf44930225 100644 --- a/src/Databases/DataLake/ICatalog.cpp +++ b/src/Databases/DataLake/ICatalog.cpp @@ -350,7 +350,9 @@ 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*/, + Int32 /*new_last_column_id*/, + Poco::JSON::Object::Ptr /*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 57aff577c675..22de3accbae1 100644 --- a/src/Databases/DataLake/ICatalog.h +++ b/src/Databases/DataLake/ICatalog.h @@ -206,7 +206,9 @@ 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, + Int32 new_last_column_id, + Poco::JSON::Object::Ptr 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 1af7b42f1951..77ec93c55d47 100644 --- a/src/Databases/DataLake/RestCatalog.cpp +++ b/src/Databases/DataLake/RestCatalog.cpp @@ -9,6 +9,8 @@ #include #include #include +#include +#include #include #include #include "config.h" @@ -35,6 +37,7 @@ #include #include +#include #include #include #include @@ -57,6 +60,7 @@ namespace DB::ErrorCodes extern const int LOGICAL_ERROR; extern const int BAD_ARGUMENTS; extern const int FAULT_INJECTED; + extern const int NOT_IMPLEMENTED; extern const int CATALOG_NAMESPACE_DISABLED; } @@ -68,6 +72,8 @@ namespace DB::Setting namespace DB::FailPoints { extern const char check_database_datalake_negative[]; + extern const char iceberg_alter_catalog_update_schema_fail[]; + extern const char iceberg_alter_catalog_commit_reported_as_failed[]; } namespace ProfileEvents @@ -173,6 +179,217 @@ 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); +} + +} + +Poco::JSON::Object::Ptr buildUpdateSchemaRequestBody( + const String & namespace_name, + const String & table_name, + Poco::JSON::Object::Ptr metadata, + Poco::JSON::Object::Ptr new_schema, + Int32 previous_schema_id, + Int32 new_last_column_id) +{ + 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 (previous_schema_id >= 0) + { + 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); + } + + Poco::JSON::Object::Ptr schema_for_rest = cloneJsonObject(new_schema); + 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); + } + + std::optional existing_equivalent_schema_id; + if (metadata && metadata->has(DB::Iceberg::f_schemas)) + { + auto schemas = metadata->getArray(DB::Iceberg::f_schemas); + auto new_schema_id = new_schema->getValue(DB::Iceberg::f_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)) + { + 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); + add_schema->set("last-column-id", new_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); + } + } + + request_body->set("updates", updates); + return request_body; +} + +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("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; @@ -1357,57 +1574,15 @@ 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); - - 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); - } + if (!new_snapshot) + throw DB::Exception( + DB::ErrorCodes::NOT_IMPLEMENTED, + "REST catalog does not support metadata-only updates without a snapshot " + "(required for EXPIRE SNAPSHOTS)"); - { - 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); try { @@ -1417,8 +1592,14 @@ bool RestCatalog::updateMetadata(const String & namespace_name, const String & t } catch (const DB::HTTPException & ex) { - LOG_TRACE(log, "Unsucceeded request {}", ex.what()); - return false; + const auto status = static_cast(ex.getHTTPStatus()); + if (status == 409 || status == 429 || status >= 500) + { + LOG_WARNING(log, "Iceberg REST updateMetadata for {}.{} got retryable HTTP {}: {}", + namespace_name, table_name, status, ex.displayText()); + return false; + } + throw; } return true; } @@ -1428,50 +1609,16 @@ 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, + Int32 new_last_column_id, + Poco::JSON::Object::Ptr 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 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); - } - - { - 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); - } + fiu_do_on(DB::FailPoints::iceberg_alter_catalog_update_schema_fail, { return false; }); - { - 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); - } + const std::string endpoint = (base_url / config.prefix / NAMESPACES_ENDPOINT / encodeNamespaceForURI(namespace_name) / "tables" / table_name).generic_string(); - request_body->set("updates", updates); - } + auto request_body = buildUpdateSchemaRequestBody( + namespace_name, table_name, metadata, new_schema, previous_schema_id, new_last_column_id); try { @@ -1479,9 +1626,20 @@ bool RestCatalog::updateSchema( } catch (const DB::HTTPException & ex) { - LOG_TRACE(log, "Unsucceeded request {}", ex.what()); - return false; + const auto status = static_cast(ex.getHTTPStatus()); + if (status == 409 || status == 429 || status >= 500) + { + LOG_WARNING(log, "Iceberg REST updateSchema for {}.{} got retryable HTTP {}: {}", + namespace_name, table_name, status, ex.displayText()); + return false; + } + throw; } + + /// Simulates the Iceberg "commit state unknown" case: the catalog applied the update but the + /// client observes a failure, e.g. because a proxy turned the response into a 5xx. + fiu_do_on(DB::FailPoints::iceberg_alter_catalog_commit_reported_as_failed, { return false; }); + return true; } diff --git a/src/Databases/DataLake/RestCatalog.h b/src/Databases/DataLake/RestCatalog.h index 21ed59b64b33..8fdd516b4105 100644 --- a/src/Databases/DataLake/RestCatalog.h +++ b/src/Databases/DataLake/RestCatalog.h @@ -82,7 +82,9 @@ 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, + Int32 new_last_column_id, + Poco::JSON::Object::Ptr metadata = nullptr) const override; bool isTransactional() const override { return true; } @@ -288,6 +290,23 @@ class BigLakeCatalog : public RestCatalog AccessToken retrieveGoogleCloudAccessTokenFromRefreshToken() const; }; +/// Builds the JSON body for a schema-update commit via the Iceberg REST catalog. +/// Includes an assert-current-schema-id requirement (when previous_schema_id >= 0), +/// schema deduplication against existing schemas in metadata, and last-column-id +/// propagation when adding a new schema. +Poco::JSON::Object::Ptr buildUpdateSchemaRequestBody( + const String & namespace_name, + const String & table_name, + Poco::JSON::Object::Ptr metadata, + Poco::JSON::Object::Ptr new_schema, + Int32 previous_schema_id, + Int32 new_last_column_id); + +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..06dbc4591da1 --- /dev/null +++ b/src/Databases/DataLake/tests/gtest_rest_catalog_update_metadata.cpp @@ -0,0 +1,206 @@ +#include "config.h" + +#if USE_AVRO + +#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, 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")); +} + +TEST(RestCatalogUpdateSchemaBody, EquivalentSchemaDeduplicates) +{ + Poco::JSON::Object::Ptr metadata = new Poco::JSON::Object; + + Poco::JSON::Array::Ptr schemas = new Poco::JSON::Array; + Poco::JSON::Object::Ptr schema0 = new Poco::JSON::Object; + schema0->set(Iceberg::f_schema_id, 0); + schema0->set(Iceberg::f_type, "struct"); + Poco::JSON::Array::Ptr fields = new Poco::JSON::Array; + Poco::JSON::Object::Ptr field1 = new Poco::JSON::Object; + field1->set(Iceberg::f_id, 1); + field1->set(Iceberg::f_name, "a"); + field1->set(Iceberg::f_required, false); + field1->set(Iceberg::f_type, "int"); + fields->add(field1); + schema0->set(Iceberg::f_fields, fields); + schemas->add(schema0); + + Poco::JSON::Object::Ptr schema1 = new Poco::JSON::Object; + schema1->set(Iceberg::f_schema_id, 1); + schema1->set(Iceberg::f_type, "struct"); + Poco::JSON::Array::Ptr fields1 = new Poco::JSON::Array; + Poco::JSON::Object::Ptr field1b = new Poco::JSON::Object; + field1b->set(Iceberg::f_id, 1); + field1b->set(Iceberg::f_name, "a"); + field1b->set(Iceberg::f_required, false); + field1b->set(Iceberg::f_type, "int"); + Poco::JSON::Object::Ptr field2b = new Poco::JSON::Object; + field2b->set(Iceberg::f_id, 2); + field2b->set(Iceberg::f_name, "b"); + field2b->set(Iceberg::f_required, false); + field2b->set(Iceberg::f_type, "int"); + fields1->add(field1b); + fields1->add(field2b); + schema1->set(Iceberg::f_fields, fields1); + schemas->add(schema1); + metadata->set(Iceberg::f_schemas, schemas); + + Poco::JSON::Object::Ptr new_schema = new Poco::JSON::Object; + new_schema->set(Iceberg::f_schema_id, 2); + new_schema->set(Iceberg::f_type, "struct"); + Poco::JSON::Array::Ptr new_fields = new Poco::JSON::Array; + Poco::JSON::Object::Ptr nf1 = new Poco::JSON::Object; + nf1->set(Iceberg::f_id, 1); + nf1->set(Iceberg::f_name, "a"); + nf1->set(Iceberg::f_required, false); + nf1->set(Iceberg::f_type, "int"); + new_fields->add(nf1); + new_schema->set(Iceberg::f_fields, new_fields); + + auto body = DataLake::buildUpdateSchemaRequestBody("ns", "t", metadata, new_schema, 1, 2); + ASSERT_TRUE(body); + + auto updates = body->getArray("updates"); + EXPECT_FALSE(findUpdateByAction(updates, "add-schema")); + + auto set_schema = findUpdateByAction(updates, "set-current-schema"); + ASSERT_TRUE(set_schema); + EXPECT_EQ(set_schema->getValue("schema-id"), 0); +} + +TEST(RestCatalogUpdateSchemaBody, NormalPathEmitsAddSchema) +{ + Poco::JSON::Object::Ptr metadata = new Poco::JSON::Object; + + Poco::JSON::Array::Ptr schemas = new Poco::JSON::Array; + Poco::JSON::Object::Ptr schema0 = new Poco::JSON::Object; + schema0->set(Iceberg::f_schema_id, 0); + schema0->set(Iceberg::f_type, "struct"); + Poco::JSON::Array::Ptr fields = new Poco::JSON::Array; + Poco::JSON::Object::Ptr field1 = new Poco::JSON::Object; + field1->set(Iceberg::f_id, 1); + field1->set(Iceberg::f_name, "a"); + field1->set(Iceberg::f_required, false); + field1->set(Iceberg::f_type, "int"); + fields->add(field1); + schema0->set(Iceberg::f_fields, fields); + schemas->add(schema0); + metadata->set(Iceberg::f_schemas, schemas); + + Poco::JSON::Object::Ptr new_schema = new Poco::JSON::Object; + new_schema->set(Iceberg::f_schema_id, 1); + new_schema->set(Iceberg::f_type, "struct"); + Poco::JSON::Array::Ptr new_fields = new Poco::JSON::Array; + Poco::JSON::Object::Ptr nf1 = new Poco::JSON::Object; + nf1->set(Iceberg::f_id, 1); + nf1->set(Iceberg::f_name, "a"); + nf1->set(Iceberg::f_required, false); + nf1->set(Iceberg::f_type, "int"); + Poco::JSON::Object::Ptr nf2 = new Poco::JSON::Object; + nf2->set(Iceberg::f_id, 2); + nf2->set(Iceberg::f_name, "b"); + nf2->set(Iceberg::f_required, false); + nf2->set(Iceberg::f_type, "string"); + new_fields->add(nf1); + new_fields->add(nf2); + new_schema->set(Iceberg::f_fields, new_fields); + + auto body = DataLake::buildUpdateSchemaRequestBody("ns", "t", metadata, new_schema, 0, 5); + ASSERT_TRUE(body); + + auto updates = body->getArray("updates"); + + auto add_schema = findUpdateByAction(updates, "add-schema"); + ASSERT_TRUE(add_schema); + EXPECT_EQ(add_schema->getValue("last-column-id"), 5); + EXPECT_TRUE(add_schema->has("schema")); + auto schema_obj = add_schema->getObject("schema"); + EXPECT_TRUE(schema_obj->has("identifier-field-ids")); + + auto set_schema = findUpdateByAction(updates, "set-current-schema"); + ASSERT_TRUE(set_schema); + EXPECT_EQ(set_schema->getValue("schema-id"), -1); + + 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); +} + +#endif diff --git a/src/Storages/ObjectStorage/DataLakes/DataLakeConfiguration.h b/src/Storages/ObjectStorage/DataLakes/DataLakeConfiguration.h index 20ca0d8b93ec..253412aefdde 100644 --- a/src/Storages/ObjectStorage/DataLakes/DataLakeConfiguration.h +++ b/src/Storages/ObjectStorage/DataLakes/DataLakeConfiguration.h @@ -184,12 +184,14 @@ class DataLakeConfiguration : public BaseStorageConfiguration, public std::enabl void checkMutationIsPossible(ObjectStoragePtr object_storage, ContextPtr context, const MutationCommands & commands) override { lazyInitializeIfNeeded(object_storage, context); + assertInitialized(); current_metadata->checkMutationIsPossible(commands); } void checkAlterIsPossible(ObjectStoragePtr object_storage, ContextPtr context, const AlterCommands & commands) override { lazyInitializeIfNeeded(object_storage, context); + assertInitialized(); current_metadata->checkAlterIsPossible(commands); } @@ -201,6 +203,7 @@ class DataLakeConfiguration : public BaseStorageConfiguration, public std::enabl std::shared_ptr catalog) override { lazyInitializeIfNeeded(object_storage, context); + assertInitialized(); current_metadata->alter(params, context, storage_id, catalog); } @@ -355,6 +358,7 @@ class DataLakeConfiguration : public BaseStorageConfiguration, public std::enabl std::shared_ptr catalog) override { lazyInitializeIfNeeded(object_storage, context); + assertInitialized(); return current_metadata->write( sample_block, table_id, @@ -390,11 +394,13 @@ class DataLakeConfiguration : public BaseStorageConfiguration, public std::enabl bool optimize(ObjectStoragePtr object_storage, const StorageMetadataPtr & metadata_snapshot, ContextPtr context, const std::optional & format_settings) override { lazyInitializeIfNeeded(object_storage, context); + assertInitialized(); return current_metadata->optimize(metadata_snapshot, context, format_settings); } void addDeleteTransformers(ObjectInfoPtr object_info, QueryPipelineBuilder & builder, const std::optional & format_settings, FormatParserSharedResourcesPtr parser_shared_resources, ContextPtr local_context) const override { + assertInitialized(); current_metadata->addDeleteTransformers(object_info, builder, format_settings, parser_shared_resources, local_context); } 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 459a78056e0d..e9b42a603606 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/MetadataGenerator.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/MetadataGenerator.cpp @@ -1,3 +1,5 @@ +#include +#include #include #include @@ -10,6 +12,7 @@ #include #include +#include #include #include @@ -54,13 +57,39 @@ 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(); auto new_complex_type = new_type.extract(); if (old_complex_type && new_complex_type && old_complex_type->has("precision") && new_complex_type->has("precision") && (old_complex_type->getValue("precision") <= new_complex_type->getValue("precision") && - old_complex_type->getValue("scale") <= new_complex_type->getValue("scale"))) + old_complex_type->getValue("scale") == new_complex_type->getValue("scale"))) { return true; } @@ -69,10 +98,39 @@ 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; +} + +/// Allocate the next schema id as max(existing schema ids) + 1 to avoid +/// collisions when current-schema-id is not the highest in the list. +Int32 getNextSchemaId(Poco::JSON::Object::Ptr metadata_object) +{ + Int32 max_id = 0; + auto schemas = metadata_object->getArray(Iceberg::f_schemas); + for (UInt32 i = 0; i < schemas->size(); ++i) + max_id = std::max(max_id, schemas->getObject(i)->getValue(Iceberg::f_schema_id)); + return max_id + 1; +} + } -MetadataGenerator::MetadataGenerator(Poco::JSON::Object::Ptr metadata_object_) +MetadataGenerator::MetadataGenerator(Poco::JSON::Object::Ptr metadata_object_, bool allow_geo_parser_) : metadata_object(metadata_object_) + , allow_geo_parser(allow_geo_parser_) , gen(randomSeed()) , dis(1, std::numeric_limits::max()) { @@ -97,6 +155,84 @@ Int64 MetadataGenerator::getMaxSequenceNumber() return max_seq_number; } +Poco::JSON::Object::Ptr MetadataGenerator::findCurrentSchema() const +{ + auto current_schema_id = metadata_object->getValue(Iceberg::f_current_schema_id); + auto schemas = metadata_object->getArray(Iceberg::f_schemas); + for (UInt32 i = 0; i < schemas->size(); ++i) + { + if (schemas->getObject(i)->getValue(Iceberg::f_schema_id) == current_schema_id) + return schemas->getObject(i); + } + return nullptr; +} + +Poco::JSON::Object::Ptr MetadataGenerator::getCurrentSchema() const +{ + auto current_schema = findCurrentSchema(); + if (!current_schema) + throw Exception( + ErrorCodes::BAD_ARGUMENTS, + "Not found schema with id {}", + metadata_object->getValue(Iceberg::f_current_schema_id)); + return current_schema; +} + +bool MetadataGenerator::isAddColumnApplied(const String & column_name, DataTypePtr type) const +{ + auto current_schema = findCurrentSchema(); + if (!current_schema) + return false; + + Int32 unused_field_id = metadata_object->getValue(Iceberg::f_last_column_id); + auto expected_type = Iceberg::getIcebergType(type, unused_field_id); + + auto fields = current_schema->getArray(Iceberg::f_fields); + for (UInt32 i = 0; i < fields->size(); ++i) + { + auto field = fields->getObject(i); + if (field->getValue(Iceberg::f_name) != column_name) + continue; + return field->getValue(Iceberg::f_required) == expected_type.second + && icebergTypesEqual(field->get(Iceberg::f_type), expected_type.first); + } + return false; +} + +bool MetadataGenerator::isDropColumnApplied(const String & column_name) const +{ + auto current_schema = findCurrentSchema(); + if (!current_schema) + return false; + + auto fields = current_schema->getArray(Iceberg::f_fields); + for (UInt32 i = 0; i < fields->size(); ++i) + { + if (fields->getObject(i)->getValue(Iceberg::f_name) == column_name) + return false; + } + return true; +} + +bool MetadataGenerator::isRenameColumnApplied(const String & column_name, const String & new_column_name) const +{ + auto current_schema = findCurrentSchema(); + if (!current_schema) + return false; + + bool found_new_name = false; + auto fields = current_schema->getArray(Iceberg::f_fields); + for (UInt32 i = 0; i < fields->size(); ++i) + { + auto name = fields->getObject(i)->getValue(Iceberg::f_name); + if (name == column_name) + return false; + if (name == new_column_name) + found_new_name = true; + } + return found_new_name; +} + Poco::JSON::Object::Ptr MetadataGenerator::getParentSnapshot(Int64 parent_snapshot_id) { auto snapshots = metadata_object->get(Iceberg::f_snapshots).extract(); @@ -253,38 +389,83 @@ MetadataGenerator::NextMetadataResult MetadataGenerator::generateNextMetadata( void MetadataGenerator::generateDropColumnMetadata(const String & column_name) { - auto current_schema_id = metadata_object->getValue(Iceberg::f_current_schema_id); - metadata_object->set(Iceberg::f_current_schema_id, current_schema_id + 1); + const auto next_schema_id = getNextSchemaId(metadata_object); - Poco::JSON::Object::Ptr current_schema; - auto schemas = metadata_object->getArray(Iceberg::f_schemas); - for (UInt32 i = 0; i < schemas->size(); ++i) - { - if (schemas->getObject(i)->getValue(Iceberg::f_schema_id) == current_schema_id) - { - current_schema = schemas->getObject(i); - break; - } - } - - if (!current_schema) - throw Exception(ErrorCodes::BAD_ARGUMENTS, "Not found schema with id {}", current_schema_id); - current_schema = deepCopy(current_schema); + auto current_schema = deepCopy(getCurrentSchema()); auto fields = current_schema->getArray(Iceberg::f_fields); UInt32 index_to_drop = static_cast(fields->size()); + Int32 dropped_field_id = -1; for (UInt32 i = 0; i < fields->size(); ++i) { if (fields->getObject(i)->getValue(Iceberg::f_name) == column_name) { index_to_drop = i; + dropped_field_id = fields->getObject(i)->getValue(Iceberg::f_id); break; } } if (index_to_drop == fields->size()) throw Exception(ErrorCodes::BAD_ARGUMENTS, "Not found column {}", column_name); + + /// Reject the drop if the column is referenced by the active sort order. + if (metadata_object->has(Iceberg::f_sort_orders) && metadata_object->has(Iceberg::f_default_sort_order_id)) + { + auto default_sort_order_id = metadata_object->getValue(Iceberg::f_default_sort_order_id); + if (default_sort_order_id != 0) + { + auto sort_orders = metadata_object->getArray(Iceberg::f_sort_orders); + for (UInt32 i = 0; i < sort_orders->size(); ++i) + { + auto sort_order = sort_orders->getObject(i); + if (sort_order->getValue(Iceberg::f_order_id) != default_sort_order_id) + continue; + if (!sort_order->has(Iceberg::f_fields)) + break; + auto sort_fields = sort_order->getArray(Iceberg::f_fields); + for (UInt32 j = 0; j < sort_fields->size(); ++j) + { + auto sf = sort_fields->getObject(j); + if (sf->has(Iceberg::f_source_id) && sf->getValue(Iceberg::f_source_id) == dropped_field_id) + throw Exception( + ErrorCodes::BAD_ARGUMENTS, + "Cannot drop column '{}' (field id {}): it is referenced by the active sort order", + column_name, dropped_field_id); + } + break; + } + } + } + + /// Reject the drop if the column is referenced by the active partition spec. + if (metadata_object->has(Iceberg::f_partition_specs) && metadata_object->has(Iceberg::f_default_spec_id)) + { + auto default_spec_id = metadata_object->getValue(Iceberg::f_default_spec_id); + auto partition_specs = metadata_object->getArray(Iceberg::f_partition_specs); + for (UInt32 i = 0; i < partition_specs->size(); ++i) + { + auto spec = partition_specs->getObject(i); + if (spec->getValue(Iceberg::f_spec_id) != default_spec_id) + continue; + if (!spec->has(Iceberg::f_fields)) + break; + auto spec_fields = spec->getArray(Iceberg::f_fields); + for (UInt32 j = 0; j < spec_fields->size(); ++j) + { + auto pf = spec_fields->getObject(j); + if (pf->has(Iceberg::f_source_id) && pf->getValue(Iceberg::f_source_id) == dropped_field_id) + throw Exception( + ErrorCodes::BAD_ARGUMENTS, + "Cannot drop column '{}' (field id {}): it is referenced by the active partition spec", + column_name, dropped_field_id); + } + break; + } + } + current_schema->getArray(Iceberg::f_fields)->remove(index_to_drop); - current_schema->set(Iceberg::f_schema_id, current_schema_id + 1); + current_schema->set(Iceberg::f_schema_id, next_schema_id); + metadata_object->set(Iceberg::f_current_schema_id, next_schema_id); metadata_object->getArray(Iceberg::f_schemas)->add(current_schema); } @@ -292,23 +473,9 @@ void MetadataGenerator::generateAddColumnMetadata(const String & column_name, Da { if (!type->isNullable()) throw Exception(ErrorCodes::BAD_ARGUMENTS, "Iceberg spec doesn't allow to add non-nullable columns"); - auto current_schema_id = metadata_object->getValue(Iceberg::f_current_schema_id); - metadata_object->set(Iceberg::f_current_schema_id, current_schema_id + 1); + const auto next_schema_id = getNextSchemaId(metadata_object); - Poco::JSON::Object::Ptr current_schema; - auto schemas = metadata_object->getArray(Iceberg::f_schemas); - for (UInt32 i = 0; i < schemas->size(); ++i) - { - if (schemas->getObject(i)->getValue(Iceberg::f_schema_id) == current_schema_id) - { - current_schema = schemas->getObject(i); - break; - } - } - - if (!current_schema) - throw Exception(ErrorCodes::BAD_ARGUMENTS, "Not found schema with id {}", current_schema_id); - current_schema = deepCopy(current_schema); + auto current_schema = deepCopy(getCurrentSchema()); auto existing_fields = current_schema->getArray(Iceberg::f_fields); for (UInt32 i = 0; i < existing_fields->size(); ++i) @@ -318,7 +485,6 @@ void MetadataGenerator::generateAddColumnMetadata(const String & column_name, Da } auto last_column_id = metadata_object->getValue(Iceberg::f_last_column_id); - metadata_object->set(Iceberg::f_last_column_id, last_column_id + 1); auto new_type = Iceberg::getIcebergType(type, last_column_id); Poco::JSON::Object::Ptr new_field = new Poco::JSON::Object; @@ -327,80 +493,88 @@ void MetadataGenerator::generateAddColumnMetadata(const String & column_name, Da new_field->set(Iceberg::f_required, new_type.second); new_field->set(Iceberg::f_type, new_type.first); + metadata_object->set(Iceberg::f_last_column_id, last_column_id + 1); + current_schema->getArray(Iceberg::f_fields)->add(new_field); - current_schema->set(Iceberg::f_schema_id, current_schema_id + 1); + current_schema->set(Iceberg::f_schema_id, next_schema_id); + metadata_object->set(Iceberg::f_current_schema_id, next_schema_id); 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); - for (UInt32 i = 0; i < schemas->size(); ++i) - { - if (schemas->getObject(i)->getValue(Iceberg::f_schema_id) == current_schema_id) - { - current_schema = schemas->getObject(i); - break; - } - } + auto current_schema = getCurrentSchema(); - 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 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)) + { + /// Iceberg types are identical. Reconstruct the ClickHouse type the + /// existing field maps back to and check whether it equals the + /// requested type. For simple string-typed fields we can use + /// IcebergSchemaProcessor::getSimpleType; for complex types (JSON + /// objects) reconstruction is lossy so we allow the no-op silently. + auto existing_iceberg_type = current_field->get(Iceberg::f_type); + if (existing_iceberg_type.isString()) + { + auto reconstructed_ch_type = Iceberg::IcebergSchemaProcessor::getSimpleType( + existing_iceberg_type.extract(), allow_geo_parser); + if (!current_field->getValue(Iceberg::f_required) && reconstructed_ch_type->canBeInsideNullable()) + reconstructed_ch_type = makeNullable(reconstructed_ch_type); + + auto requested_type_normalized = type; + if (reconstructed_ch_type->equals(*requested_type_normalized)) + return false; + + throw Exception( + ErrorCodes::BAD_ARGUMENTS, + "Cannot MODIFY COLUMN '{}' from {} to {}: both map to the same Iceberg type '{}' " + "so the change cannot be recorded in the Iceberg schema", + column_name, + reconstructed_ch_type->getName(), + requested_type_normalized->getName(), + existing_iceberg_type.extract()); + } + 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()); + const auto next_schema_id = getNextSchemaId(metadata_object); + + 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, next_schema_id); + current_schema->set(Iceberg::f_schema_id, next_schema_id); + metadata_object->getArray(Iceberg::f_schemas)->add(current_schema); + metadata_object->set(Iceberg::f_last_column_id, last_column_id); + 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) { - auto current_schema_id = metadata_object->getValue(Iceberg::f_current_schema_id); - - Poco::JSON::Object::Ptr current_schema; - auto schemas = metadata_object->getArray(Iceberg::f_schemas); - for (UInt32 i = 0; i < schemas->size(); ++i) - { - if (schemas->getObject(i)->getValue(Iceberg::f_schema_id) == current_schema_id) - { - current_schema = schemas->getObject(i); - break; - } - } - - if (!current_schema) - throw Exception(ErrorCodes::BAD_ARGUMENTS, "Not found schema with id {}", current_schema_id); - current_schema = deepCopy(current_schema); + auto current_schema = deepCopy(getCurrentSchema()); auto schema_fields = current_schema->getArray(Iceberg::f_fields); @@ -425,8 +599,9 @@ void MetadataGenerator::generateRenameColumnMetadata(const String & column_name, if (!found) throw Exception(ErrorCodes::BAD_ARGUMENTS, "Not found column {}", column_name); - metadata_object->set(Iceberg::f_current_schema_id, current_schema_id + 1); - current_schema->set(Iceberg::f_schema_id, current_schema_id + 1); + const auto next_schema_id = getNextSchemaId(metadata_object); + metadata_object->set(Iceberg::f_current_schema_id, next_schema_id); + current_schema->set(Iceberg::f_schema_id, next_schema_id); metadata_object->getArray(Iceberg::f_schemas)->add(current_schema); } diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/MetadataGenerator.h b/src/Storages/ObjectStorage/DataLakes/Iceberg/MetadataGenerator.h index 3576837f6f70..3559badd170b 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/MetadataGenerator.h +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/MetadataGenerator.h @@ -17,7 +17,7 @@ namespace DB class MetadataGenerator { public: - explicit MetadataGenerator(Poco::JSON::Object::Ptr metadata_object_); + explicit MetadataGenerator(Poco::JSON::Object::Ptr metadata_object_, bool allow_geo_parser_ = false); struct NextMetadataResult { @@ -44,17 +44,32 @@ 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); + /// A commit attempt can land in the catalog even when the client observes a failure + /// (the Iceberg "commit state unknown" case, e.g. a proxy returning 5xx after the catalog + /// applied the update). These predicates let a retry detect that the requested change is + /// already present instead of applying it a second time and failing. + bool isAddColumnApplied(const String & column_name, DataTypePtr type) const; + bool isDropColumnApplied(const String & column_name) const; + bool isRenameColumnApplied(const String & column_name, const String & new_column_name) const; + private: Poco::JSON::Object::Ptr metadata_object; + bool allow_geo_parser; pcg64_fast gen; std::uniform_int_distribution dis; Int64 getMaxSequenceNumber(); Poco::JSON::Object::Ptr getParentSnapshot(Int64 parent_snapshot_id); + + /// Returns the schema referenced by `current-schema-id`, or nullptr when it is absent. + Poco::JSON::Object::Ptr findCurrentSchema() const; + /// Returns the schema referenced by `current-schema-id`, throwing when it is absent. + Poco::JSON::Object::Ptr getCurrentSchema() const; }; #endif diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/Mutations.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/Mutations.cpp index 8fb45abb86aa..33bb729a1f6c 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/Mutations.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/Mutations.cpp @@ -12,6 +12,7 @@ #include #include #include +#include #include #include #include @@ -41,6 +42,7 @@ namespace DB::ErrorCodes extern const int BAD_ARGUMENTS; extern const int LOGICAL_ERROR; extern const int LIMIT_EXCEEDED; +extern const int QUERY_WAS_CANCELLED; } namespace DB::DataLakeStorageSetting @@ -49,6 +51,11 @@ extern const DataLakeStorageSettingsBool iceberg_use_version_hint; extern const DataLakeStorageSettingsString iceberg_metadata_file_path; } +namespace DB::Setting +{ +extern const SettingsBool allow_experimental_geo_types_in_iceberg; +} + namespace DB::FailPoints { extern const char iceberg_writes_cleanup[]; @@ -63,6 +70,76 @@ static constexpr const char * block_datafile_path = "_iceberg_metadata_file_path static constexpr const char * block_row_number = "_row_number"; static constexpr auto MAX_TRANSACTION_RETRIES = 100; +/// Walk an Iceberg type descriptor and return the highest field id found. +static Int32 getHighestFieldIdFromType(const Poco::Dynamic::Var & type_var) +{ + if (type_var.type() != typeid(Poco::JSON::Object::Ptr)) + return 0; + auto obj = type_var.extract(); + Int32 result = 0; + + auto type_str = obj->optValue(Iceberg::f_type, ""); + if (type_str == "struct") + { + auto fields = obj->getArray(Iceberg::f_fields); + for (UInt32 i = 0; i < fields->size(); ++i) + { + auto field = fields->getObject(i); + result = std::max(result, field->getValue(Iceberg::f_id)); + result = std::max(result, getHighestFieldIdFromType(field->get(Iceberg::f_type))); + } + } + else if (type_str == "list") + { + result = std::max(result, obj->getValue(Iceberg::f_element_id)); + result = std::max(result, getHighestFieldIdFromType(obj->get(Iceberg::f_element))); + } + else if (type_str == "map") + { + result = std::max(result, obj->getValue(Iceberg::f_key_id)); + result = std::max(result, obj->getValue(Iceberg::f_value_id)); + result = std::max(result, getHighestFieldIdFromType(obj->get(Iceberg::f_key))); + result = std::max(result, getHighestFieldIdFromType(obj->get(Iceberg::f_value))); + } + return result; +} + +/// Whether the schema already reflects `command`. Used after a commit attempt whose outcome is +/// unknown: the catalog may have applied the update and still reported a failure, in which case +/// re-applying the same command on the refreshed metadata would fail with "Column already exists" +/// or "Not found column" for an ALTER that actually succeeded. +static bool alterAlreadyApplied(const MetadataGenerator & generator, const AlterCommand & command) +{ + switch (command.type) + { + case AlterCommand::Type::ADD_COLUMN: + return generator.isAddColumnApplied(command.column_name, command.data_type); + case AlterCommand::Type::DROP_COLUMN: + return generator.isDropColumnApplied(command.column_name); + case AlterCommand::Type::RENAME_COLUMN: + return generator.isRenameColumnApplied(command.column_name, command.rename_to); + case AlterCommand::Type::MODIFY_COLUMN: + /// `generateModifyColumnMetadata` already reports an unchanged schema as a no-op. + return false; + default: + return false; + } +} + +/// Return the highest field id across all fields in an Iceberg schema object. +static Int32 getHighestFieldId(Poco::JSON::Object::Ptr schema) +{ + Int32 result = 0; + auto fields = schema->getArray(Iceberg::f_fields); + for (UInt32 i = 0; i < fields->size(); ++i) + { + auto field = fields->getObject(i); + result = std::max(result, field->getValue(Iceberg::f_id)); + result = std::max(result, getHighestFieldIdFromType(field->get(Iceberg::f_type))); + } + return result; +} + struct DeleteFileWriteResult { /// Metadata path (e.g. "wasb://container@account/table/data/uuid-deletes.parquet") @@ -724,8 +801,13 @@ void alter( size_t i = 0; bool succeeded = false; + /// Set once we hand a commit to storage or to the catalog, i.e. once its outcome can be unknown. + bool commit_attempted = false; while (i < MAX_TRANSACTION_RETRIES) { + if (auto elem = context->getProcessListElement(); elem && elem->isKilled()) + throw Exception(ErrorCodes::QUERY_WAS_CANCELLED, "ALTER TABLE cancelled during retry loop"); + auto log = getLogger("IcebergMutations"); int last_version = 0; @@ -790,7 +872,18 @@ void alter( const auto previous_schema_id = metadata->getValue(Iceberg::f_current_schema_id); - auto metadata_json_generator = MetadataGenerator(metadata); + auto metadata_json_generator = MetadataGenerator(metadata, context->getSettingsRef()[Setting::allow_experimental_geo_types_in_iceberg]); + + if (commit_attempted && alterAlreadyApplied(metadata_json_generator, params[0])) + { + LOG_WARNING( + log, + "A previous ALTER TABLE commit attempt for {} was reported as failed but is present in the " + "table metadata, treating the operation as succeeded", + storage_id.getNameForLogs()); + succeeded = true; + break; + } switch (params[0].type) { @@ -803,8 +896,13 @@ 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)) + { + succeeded = true; + } break; + } case AlterCommand::Type::RENAME_COLUMN: metadata_json_generator.generateRenameColumnMetadata(params[0].column_name, params[0].rename_to); break; @@ -812,14 +910,18 @@ void alter( throw Exception(ErrorCodes::LOGICAL_ERROR, "Unknown type of alter {}", params[0].type); } + if (succeeded) + break; + const auto new_schema_id = metadata->getValue(Iceberg::f_current_schema_id); Poco::JSON::Object::Ptr new_schema; auto schemas = metadata->getArray(Iceberg::f_schemas); - for (UInt32 schema_index = 0; schema_index < schemas->size(); ++schema_index) + for (auto schema_index = schemas->size(); schema_index > 0; --schema_index) { - if (schemas->getObject(schema_index)->getValue(Iceberg::f_schema_id) == new_schema_id) + auto candidate = schemas->getObject(static_cast(schema_index - 1)); + if (candidate->getValue(Iceberg::f_schema_id) == new_schema_id) { - new_schema = schemas->getObject(schema_index); + new_schema = candidate; break; } } @@ -833,25 +935,32 @@ void alter( auto hint_path = filename_generator.generateVersionHint(); const bool catalog_writes_metadata_file = catalog && catalog->isTransactional(); - if (!catalog_writes_metadata_file - && !writeMetadataFileAndVersionHint( - persistent_table_components.path_resolver, - metadata_info, - json_representation, - hint_path, - object_storage, - context, - data_lake_settings[DataLakeStorageSetting::iceberg_use_version_hint])) + if (!catalog_writes_metadata_file) { - ++i; - continue; + commit_attempted = true; + if (!writeMetadataFileAndVersionHint( + persistent_table_components.path_resolver, + metadata_info, + json_representation, + hint_path, + object_storage, + context, + data_lake_settings[DataLakeStorageSetting::iceberg_use_version_hint])) + { + ++i; + continue; + } } if (catalog) { 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)) + const auto new_last_column_id = std::max( + metadata->getValue(Iceberg::f_last_column_id), + getHighestFieldId(new_schema)); + commit_attempted = true; + if (!catalog->updateSchema(namespace_name, table_name, catalog_filename, new_schema, previous_schema_id, new_last_column_id, metadata)) { ++i; continue; @@ -863,7 +972,9 @@ void alter( } if (!succeeded) - throw Exception(ErrorCodes::LIMIT_EXCEEDED, "Too many unsuccessed retries to alter iceberg table"); + throw Exception(ErrorCodes::LIMIT_EXCEEDED, + "ALTER TABLE commit kept losing to concurrent modifications after {} retries", + MAX_TRANSACTION_RETRIES); /// Invalidate the metadata files cache so that subsequent operations on this table see the /// schema we just wrote. See `PersistentTableComponents::invalidateMetadataCache` for the diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.cpp index 3ac47f1d9a69..57f46c024557 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.cpp @@ -521,6 +521,13 @@ 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: @@ -549,6 +556,11 @@ 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: + return {"decimal(" + std::to_string(getDecimalPrecision(*type)) + ", " + std::to_string(getDecimalScale(*type)) + ")", true}; case TypeIndex::Tuple: { auto type_tuple = std::static_pointer_cast(type); @@ -1257,7 +1269,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) { @@ -1317,7 +1330,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 9629de20c643..8d87d3ef3638 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.h +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.h @@ -112,7 +112,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/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_metadata_generator.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_metadata_generator.cpp new file mode 100644 index 000000000000..44dc9a86b234 --- /dev/null +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_metadata_generator.cpp @@ -0,0 +1,244 @@ +#include "config.h" + +#if USE_AVRO + +#include + +#include +#include +#include +#include +#include +#include +#include + +using namespace DB; +using namespace DB::Iceberg; + +namespace +{ + +Poco::JSON::Object::Ptr makeMinimalMetadata(Int32 current_schema_id, Int32 last_column_id) +{ + auto metadata = Poco::JSON::Object::Ptr(new Poco::JSON::Object); + metadata->set(f_format_version, 2); + metadata->set(f_current_schema_id, current_schema_id); + metadata->set(f_last_column_id, last_column_id); + + auto schemas = Poco::JSON::Array::Ptr(new Poco::JSON::Array); + auto schema = Poco::JSON::Object::Ptr(new Poco::JSON::Object); + schema->set(f_schema_id, current_schema_id); + schema->set(f_type, "struct"); + + auto fields = Poco::JSON::Array::Ptr(new Poco::JSON::Array); + auto field = Poco::JSON::Object::Ptr(new Poco::JSON::Object); + field->set(f_id, 1); + field->set(f_name, "x"); + field->set(f_required, true); + field->set(f_type, "int"); + fields->add(field); + schema->set(f_fields, fields); + schemas->add(schema); + metadata->set(f_schemas, schemas); + + return metadata; +} + +Poco::JSON::Object::Ptr makeMetadataWithGap() +{ + auto metadata = Poco::JSON::Object::Ptr(new Poco::JSON::Object); + metadata->set(f_format_version, 2); + metadata->set(f_current_schema_id, 0); + metadata->set(f_last_column_id, 2); + + auto schemas = Poco::JSON::Array::Ptr(new Poco::JSON::Array); + + auto schema0 = Poco::JSON::Object::Ptr(new Poco::JSON::Object); + schema0->set(f_schema_id, 0); + schema0->set(f_type, "struct"); + auto fields0 = Poco::JSON::Array::Ptr(new Poco::JSON::Array); + auto field_x = Poco::JSON::Object::Ptr(new Poco::JSON::Object); + field_x->set(f_id, 1); + field_x->set(f_name, "x"); + field_x->set(f_required, true); + field_x->set(f_type, "int"); + fields0->add(field_x); + auto field_y = Poco::JSON::Object::Ptr(new Poco::JSON::Object); + field_y->set(f_id, 2); + field_y->set(f_name, "y"); + field_y->set(f_required, false); + field_y->set(f_type, "string"); + fields0->add(field_y); + schema0->set(f_fields, fields0); + schemas->add(schema0); + + // Simulate a historical schema with id=5 (higher than current-schema-id=0) + auto schema5 = Poco::JSON::Object::Ptr(new Poco::JSON::Object); + schema5->set(f_schema_id, 5); + schema5->set(f_type, "struct"); + auto fields5 = Poco::JSON::Array::Ptr(new Poco::JSON::Array); + fields5->add(field_x); + schema5->set(f_fields, fields5); + schemas->add(schema5); + + metadata->set(f_schemas, schemas); + return metadata; +} + +} + + +TEST(IcebergMetadataGenerator, AddColumnAllocatesSchemaIdAboveMax) +{ + auto metadata = makeMetadataWithGap(); + MetadataGenerator gen(metadata); + + gen.generateAddColumnMetadata("z", makeNullable(std::make_shared())); + + auto new_schema_id = metadata->getValue(f_current_schema_id); + EXPECT_EQ(new_schema_id, 6); + + auto schemas = metadata->getArray(f_schemas); + bool found = false; + for (UInt32 i = 0; i < schemas->size(); ++i) + { + if (schemas->getObject(i)->getValue(f_schema_id) == 6) + { + found = true; + break; + } + } + EXPECT_TRUE(found); +} + + +TEST(IcebergMetadataGenerator, DropColumnAllocatesSchemaIdAboveMax) +{ + auto metadata = makeMetadataWithGap(); + MetadataGenerator gen(metadata); + + gen.generateDropColumnMetadata("y"); + + EXPECT_EQ(metadata->getValue(f_current_schema_id), 6); +} + + +TEST(IcebergMetadataGenerator, DropColumnRejectsIfInSortOrder) +{ + auto metadata = makeMinimalMetadata(0, 1); + + auto sort_orders = Poco::JSON::Array::Ptr(new Poco::JSON::Array); + auto sort_order = Poco::JSON::Object::Ptr(new Poco::JSON::Object); + sort_order->set(f_order_id, static_cast(1)); + auto sort_fields = Poco::JSON::Array::Ptr(new Poco::JSON::Array); + auto sf = Poco::JSON::Object::Ptr(new Poco::JSON::Object); + sf->set(f_source_id, 1); + sf->set("transform", "identity"); + sf->set("direction", "asc"); + sf->set("null-order", "nulls-first"); + sort_fields->add(sf); + sort_order->set(f_fields, sort_fields); + sort_orders->add(sort_order); + metadata->set(f_sort_orders, sort_orders); + metadata->set(f_default_sort_order_id, static_cast(1)); + + MetadataGenerator gen(metadata); + EXPECT_THROW(gen.generateDropColumnMetadata("x"), DB::Exception); +} + + +TEST(IcebergMetadataGenerator, DropColumnRejectsIfInPartitionSpec) +{ + auto metadata = makeMinimalMetadata(0, 1); + + auto partition_specs = Poco::JSON::Array::Ptr(new Poco::JSON::Array); + auto spec = Poco::JSON::Object::Ptr(new Poco::JSON::Object); + spec->set(f_spec_id, static_cast(1)); + auto spec_fields = Poco::JSON::Array::Ptr(new Poco::JSON::Array); + auto pf = Poco::JSON::Object::Ptr(new Poco::JSON::Object); + pf->set(f_source_id, 1); + pf->set("transform", "identity"); + pf->set("name", "x_part"); + spec_fields->add(pf); + spec->set(f_fields, spec_fields); + partition_specs->add(spec); + metadata->set(f_partition_specs, partition_specs); + metadata->set(f_default_spec_id, static_cast(1)); + + MetadataGenerator gen(metadata); + EXPECT_THROW(gen.generateDropColumnMetadata("x"), DB::Exception); +} + + +TEST(IcebergMetadataGenerator, ModifyColumnNoopSameType) +{ + auto metadata = makeMinimalMetadata(0, 1); + MetadataGenerator gen(metadata); + + bool changed = gen.generateModifyColumnMetadata("x", std::make_shared()); + EXPECT_FALSE(changed); +} + + +TEST(IcebergMetadataGenerator, ModifyColumnRejectsIndistinguishableType) +{ + auto metadata = makeMinimalMetadata(0, 1); + MetadataGenerator gen(metadata); + + EXPECT_THROW(gen.generateModifyColumnMetadata("x", std::make_shared()), DB::Exception); +} + + +TEST(IcebergMetadataGenerator, AddColumnAppliedDetectsCommittedColumn) +{ + auto metadata = makeMetadataWithGap(); + MetadataGenerator gen(metadata); + + auto type = makeNullable(std::make_shared()); + EXPECT_FALSE(gen.isAddColumnApplied("z", type)); + + /// Emulate the commit that the catalog applied while reporting a failure. + gen.generateAddColumnMetadata("z", type); + EXPECT_TRUE(gen.isAddColumnApplied("z", type)); +} + + +TEST(IcebergMetadataGenerator, AddColumnAppliedRejectsTypeMismatch) +{ + auto metadata = makeMetadataWithGap(); + MetadataGenerator gen(metadata); + + /// `y` exists as an optional Iceberg `string`, so the same name with another type is not the + /// column this ALTER asked for and must still be applied. + EXPECT_TRUE(gen.isAddColumnApplied("y", makeNullable(std::make_shared()))); + EXPECT_FALSE(gen.isAddColumnApplied("y", makeNullable(std::make_shared()))); + EXPECT_FALSE(gen.isAddColumnApplied("y", std::make_shared())); +} + + +TEST(IcebergMetadataGenerator, DropColumnAppliedDetectsCommittedDrop) +{ + auto metadata = makeMetadataWithGap(); + MetadataGenerator gen(metadata); + + EXPECT_FALSE(gen.isDropColumnApplied("y")); + + gen.generateDropColumnMetadata("y"); + EXPECT_TRUE(gen.isDropColumnApplied("y")); +} + + +TEST(IcebergMetadataGenerator, RenameColumnAppliedDetectsCommittedRename) +{ + auto metadata = makeMetadataWithGap(); + MetadataGenerator gen(metadata); + + EXPECT_FALSE(gen.isRenameColumnApplied("y", "w")); + + gen.generateRenameColumnMetadata("y", "w"); + EXPECT_TRUE(gen.isRenameColumnApplied("y", "w")); + /// A rename to a different target name is not what this ALTER asked for. + EXPECT_FALSE(gen.isRenameColumnApplied("y", "v")); +} + +#endif diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_type_mapping.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_type_mapping.cpp new file mode 100644 index 000000000000..0379efd819fd --- /dev/null +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_iceberg_type_mapping.cpp @@ -0,0 +1,112 @@ +#include "config.h" + +#if USE_AVRO + +#include + +#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_TRUE(iceberg_type.isString()); + EXPECT_EQ(iceberg_type.extract(), "decimal(9, 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_TRUE(iceberg_type.isString()); + EXPECT_EQ(iceberg_type.extract(), "decimal(18, 5)"); +} + +TEST(IcebergTypeMapping, Decimal128MapsToDecimal) +{ + auto type = std::make_shared>(38, 10); + Int32 iter = 0; + auto [iceberg_type, required] = getIcebergType(type, iter); + ASSERT_TRUE(iceberg_type.isString()); + EXPECT_EQ(iceberg_type.extract(), "decimal(38, 10)"); +} + +TEST(IcebergTypeMapping, NullableDecimalMapsToDecimalNotRequired) +{ + auto type = makeNullable(std::make_shared>(7, 3)); + Int32 iter = 0; + auto [iceberg_type, required] = getIcebergType(type, iter); + ASSERT_TRUE(iceberg_type.isString()); + EXPECT_EQ(iceberg_type.extract(), "decimal(7, 3)"); + EXPECT_FALSE(required); +} + +#endif diff --git a/tests/integration/test_database_iceberg/test.py b/tests/integration/test_database_iceberg/test.py index 6c949cc73330..f8443a5b0bf2 100644 --- a/tests/integration/test_database_iceberg/test.py +++ b/tests/integration/test_database_iceberg/test.py @@ -815,6 +815,109 @@ def test_insert(started_cluster): assert node.query(f"SELECT * FROM {CATALOG_NAME}.`{root_namespace}.{table_name}` ORDER BY ALL") == "\\N\tAAPL\t193.24\t193.31\t('bot')\n\\N\tPavel Ivanov (pudge1000-7) pereezhai v amsterdam\t193.24\t193.31\t('bot')\n" +def test_optimize_manifest_with_catalog(started_cluster): + # OPTIMIZE TABLE ... MANIFEST on a catalog-managed table must consolidate the per-insert manifests + # and commit the new snapshot back through the catalog, without changing the data. + node = started_cluster.instances["node1"] + + test_ref = f"test_optimize_manifest_{uuid.uuid4()}" + table_name = f"{test_ref}_table" + root_namespace = f"{test_ref}_namespace" + + catalog = load_catalog_impl(started_cluster) + catalog.create_namespace(root_namespace) + # Unpartitioned table, so every per-insert data manifest can consolidate into a single one. + create_table(catalog, root_namespace, table_name, DEFAULT_SCHEMA, PartitionSpec(), DEFAULT_SORT_ORDER) + + create_clickhouse_iceberg_database(started_cluster, node, CATALOG_NAME) + + table_ref = f"{CATALOG_NAME}.`{root_namespace}.{table_name}`" + write_settings = {"allow_insert_into_iceberg": 1, "write_full_path_in_iceberg_metadata": 1} + + # Several separate inserts -> several snapshots, each adding its own data manifest. + num_inserts = 5 + for i in range(num_inserts): + node.query( + f"INSERT INTO {table_ref} VALUES (NULL, 'sym{i}', {100 + i}, {200 + i}, tuple('bot'));", + settings=write_settings, + ) + + def current_snapshot_id(): + # Read the current snapshot from the catalog's metadata.json (avoids parsing the manifest-list + # Avro, which pyiceberg rejects because ClickHouse omits field-ids there). + table = catalog.load_table(f"{root_namespace}.{table_name}") + assert table.current_snapshot() is not None, "expected a current snapshot after inserts" + return table.metadata.current_snapshot_id + + snapshot_id_before = current_snapshot_id() + rows_before = node.query(f"SELECT symbol, bid, ask FROM {table_ref} ORDER BY ALL") + + node.query( + f"OPTIMIZE TABLE {table_ref} MANIFEST", + settings={ + "allow_experimental_iceberg_compaction": 1, + "iceberg_manifest_min_count_to_compact": 2, + "allow_insert_into_iceberg": 1, + "write_full_path_in_iceberg_metadata": 1, + }, + ) + + # The compaction must commit a new (replace) snapshot back through the catalog. + assert current_snapshot_id() != snapshot_id_before, ( + "OPTIMIZE TABLE ... MANIFEST did not commit a new snapshot through the catalog" + ) + + # The metadata-only rewrite must not change the data. + rows_after = node.query(f"SELECT symbol, bid, ask FROM {table_ref} ORDER BY ALL") + assert rows_after == rows_before + + +@pytest.mark.parametrize( + "fields_to_remove", + [ + ["snapshots"], + ["metadata-log"], + ["snapshot-log"], + ["snapshots", "metadata-log", "snapshot-log"], + ], +) +def test_insert_into_table_without_optional_metadata_arrays(started_cluster, fields_to_remove): + # The Iceberg spec marks snapshots / metadata-log / snapshot-log as optional, so external + # engines may create empty-table metadata that omits any of them. Inserting into such a table + # must still succeed instead of aborting in the metadata write path. + node = started_cluster.instances["node1"] + + test_ref = f"test_insert_no_optional_arrays_{uuid.uuid4()}" + table_name = f"{test_ref}_table" + root_namespace = f"{test_ref}_namespace" + + catalog = load_catalog_impl(started_cluster) + catalog.create_namespace(root_namespace) + create_table(catalog, root_namespace, table_name, DEFAULT_SCHEMA, PartitionSpec(), DEFAULT_SORT_ORDER) + + iceberg_table = catalog.load_table(f"{root_namespace}.{table_name}") + assert iceberg_table.metadata_location.startswith("s3://") + metadata_bucket, metadata_key = iceberg_table.metadata_location[len("s3://"):].split("/", 1) + metadata = json.loads(get_file_contents(started_cluster.minio_client, metadata_bucket, metadata_key)) + for field in fields_to_remove: + metadata.pop(field, None) + metadata_bytes = json.dumps(metadata).encode() + started_cluster.minio_client.put_object( + metadata_bucket, + metadata_key, + io.BytesIO(metadata_bytes), + len(metadata_bytes), + content_type="application/json", + ) + + create_clickhouse_iceberg_database(started_cluster, node, CATALOG_NAME) + node.query( + f"INSERT INTO {CATALOG_NAME}.`{root_namespace}.{table_name}` VALUES (NULL, 'AAPL', 193.24, 193.31, tuple('bot'));", + 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}`") == "\\N\tAAPL\t193.24\t193.31\t('bot')\n" + + def test_create(started_cluster): node = started_cluster.instances["node1"] @@ -1201,6 +1304,85 @@ def test_writes_schema_evolution(started_cluster): ) +def test_writes_schema_evolution_drop_last_column(started_cluster): + """DROP COLUMN of the highest-id column must not be rejected by the catalog. + + Reproducer for the bug where the REST add-schema update omitted + last-column-id, causing the catalog to derive it from the schema's + highestFieldId which decreases after dropping the last-added column. + """ + node = started_cluster.instances["node1"] + + test_ref = f"test_writes_schema_evolution_drop_last_{uuid.uuid4()}" + table_name = f"{test_ref}_table" + root_namespace = f"{test_ref}_namespace" + table_ref = f"{CATALOG_NAME}.`{root_namespace}.{table_name}`" + write_settings = {"allow_insert_into_iceberg": 1, "write_full_path_in_iceberg_metadata": 1} + + create_clickhouse_iceberg_database(started_cluster, node, CATALOG_NAME) + create_clickhouse_iceberg_table(started_cluster, node, root_namespace, table_name, "(x String, y Int32)") + + node.query(f"INSERT INTO {table_ref} VALUES ('abc', 1);", settings=write_settings) + + node.query(f"ALTER TABLE {table_ref} ADD COLUMN z Nullable(String);", settings=write_settings) + assert "z" in node.query(f"DESCRIBE TABLE {table_ref}", settings=write_settings) + + node.query(f"ALTER TABLE {table_ref} DROP COLUMN z;", settings=write_settings) + desc = node.query(f"DESCRIBE TABLE {table_ref}", settings=write_settings) + assert "z" not in desc + + assert node.query(f"SELECT x, y FROM {table_ref} ORDER BY ALL", settings=write_settings) == "abc\t1\n" + + # Add another column after the drop to exercise schema-id allocation when + # current-schema-id is not the highest in the schemas list (Fix 1 reproducer). + node.query(f"ALTER TABLE {table_ref} ADD COLUMN w Nullable(Int64);", settings=write_settings) + desc = node.query(f"DESCRIBE TABLE {table_ref}", settings=write_settings) + assert "w" in desc + assert "z" not in desc + + node.query(f"INSERT INTO {table_ref} (x, y, w) VALUES ('def', 2, 42);", settings=write_settings) + assert node.query(f"SELECT x, y, w FROM {table_ref} ORDER BY x", settings=write_settings) == "abc\t1\t\\N\ndef\t2\t42\n" + + +def test_writes_alter_when_commit_is_reported_as_failed(started_cluster): + """An Iceberg commit can land in the catalog while the client observes a failure + (commit state unknown, e.g. a proxy rewriting the response to 5xx). The ALTER retry + must notice that the change is already present instead of applying it a second time + and failing with `Column already exists`. + """ + node = started_cluster.instances["node1"] + + test_ref = f"test_writes_alter_commit_unknown_{uuid.uuid4()}" + table_name = f"{test_ref}_table" + root_namespace = f"{test_ref}_namespace" + table_ref = f"{CATALOG_NAME}.`{root_namespace}.{table_name}`" + write_settings = {"allow_insert_into_iceberg": 1, "write_full_path_in_iceberg_metadata": 1} + + create_clickhouse_iceberg_database(started_cluster, node, CATALOG_NAME) + create_clickhouse_iceberg_table(started_cluster, node, root_namespace, table_name, "(x String, y Int32)") + + node.query(f"INSERT INTO {table_ref} VALUES ('abc', 1);", settings=write_settings) + + failpoint = "iceberg_alter_catalog_commit_reported_as_failed" + node.query(f"SYSTEM ENABLE FAILPOINT {failpoint}") + try: + node.query(f"ALTER TABLE {table_ref} ADD COLUMN z Nullable(String);", settings=write_settings) + finally: + node.query(f"SYSTEM DISABLE FAILPOINT {failpoint}") + + description = node.query(f"DESCRIBE TABLE {table_ref}", settings=write_settings) + columns = [line.split("\t")[0] for line in description.strip().split("\n")] + assert columns.count("z") == 1, f"expected exactly one `z` column in:\n{description}" + assert sorted(columns) == sorted(["x", "y", "z"]) + + node.query(f"INSERT INTO {table_ref} VALUES ('def', 2, 'zz');", settings=write_settings) + assert ( + node.query(f"SELECT x, y, z FROM {table_ref} ORDER BY x", settings=write_settings) + == "abc\t1\t\\N\ndef\t2\tzz\n" + ) + + + def test_writes_schema_evolution_concurrent_add_columns(started_cluster): node = started_cluster.instances["node1"] diff --git a/tests/integration/test_storage_iceberg_no_spark/test_writes_add_column.py b/tests/integration/test_storage_iceberg_no_spark/test_writes_add_column.py new file mode 100644 index 000000000000..b1c8a08407cc --- /dev/null +++ b/tests/integration/test_storage_iceberg_no_spark/test_writes_add_column.py @@ -0,0 +1,106 @@ +import pytest + +from helpers.iceberg_utils import ( + create_iceberg_table, + get_uuid_str, +) + +INSERT_SETTINGS = {"allow_insert_into_iceberg": 1} + + +@pytest.mark.parametrize("format_version", [1, 2]) +@pytest.mark.parametrize("storage_type", ["local", "s3"]) +def test_add_column_basic(started_cluster_iceberg_no_spark, format_version, storage_type): + """ADD COLUMN (nullable): existing rows read with NULL in the new column; new inserts can set it.""" + instance = started_cluster_iceberg_no_spark.instances["node1"] + TABLE_NAME = "test_add_column_basic_" + storage_type + "_" + get_uuid_str() + + create_iceberg_table( + storage_type, + instance, + TABLE_NAME, + started_cluster_iceberg_no_spark, + "(id Int32, value Nullable(String))", + format_version, + ) + + instance.query(f"INSERT INTO {TABLE_NAME} VALUES (1, 'hello'), (2, 'world');", settings=INSERT_SETTINGS) + assert instance.query(f"SELECT id, value FROM {TABLE_NAME} ORDER BY id") == "1\thello\n2\tworld\n" + + instance.query(f"ALTER TABLE {TABLE_NAME} ADD COLUMN extra Nullable(Int32);", settings=INSERT_SETTINGS) + + assert instance.query(f"SELECT id, value, extra FROM {TABLE_NAME} ORDER BY id") == ( + "1\thello\t\\N\n2\tworld\t\\N\n" + ) + + instance.query(f"INSERT INTO {TABLE_NAME} VALUES (3, 'foo', 7);", settings=INSERT_SETTINGS) + assert instance.query(f"SELECT id, value, extra FROM {TABLE_NAME} ORDER BY id") == ( + "1\thello\t\\N\n2\tworld\t\\N\n3\tfoo\t7\n" + ) + + +@pytest.mark.parametrize("format_version", [1, 2]) +@pytest.mark.parametrize("storage_type", ["local", "s3"]) +def test_add_column_errors(started_cluster_iceberg_no_spark, format_version, storage_type): + """Non-nullable ADD COLUMN and duplicate name must fail; schema unchanged.""" + instance = started_cluster_iceberg_no_spark.instances["node1"] + TABLE_NAME = "test_add_column_errors_" + storage_type + "_" + get_uuid_str() + + create_iceberg_table( + storage_type, + instance, + TABLE_NAME, + started_cluster_iceberg_no_spark, + "(id Int32, value Nullable(String))", + format_version, + ) + + error = instance.query_and_get_error( + f"ALTER TABLE {TABLE_NAME} ADD COLUMN bad Int32;", + settings=INSERT_SETTINGS, + ) + assert "non-nullable" in error.lower() or "doesn't allow" in error.lower() + + error = instance.query_and_get_error( + f"ALTER TABLE {TABLE_NAME} ADD COLUMN value Nullable(Int32);", + settings=INSERT_SETTINGS, + ) + assert "DUPLICATE_COLUMN" in error or "already exists" in error + + assert instance.query( + f"SELECT name FROM system.columns WHERE database = currentDatabase() AND table = '{TABLE_NAME}' ORDER BY name" + ) == "id\nvalue\n" + + +@pytest.mark.parametrize("format_version", [1, 2]) +@pytest.mark.parametrize("storage_type", ["local", "s3"]) +def test_add_column_bool_and_decimal(started_cluster_iceberg_no_spark, format_version, storage_type): + """ADD COLUMN with Bool (Iceberg boolean) and Decimal (Iceberg decimal) types.""" + instance = started_cluster_iceberg_no_spark.instances["node1"] + TABLE_NAME = "test_add_column_bool_dec_" + storage_type + "_" + get_uuid_str() + + create_iceberg_table( + storage_type, + instance, + TABLE_NAME, + started_cluster_iceberg_no_spark, + "(id Int32, value Nullable(String))", + format_version, + ) + + instance.query(f"INSERT INTO {TABLE_NAME} VALUES (1, 'a'), (2, 'b');", settings=INSERT_SETTINGS) + + instance.query(f"ALTER TABLE {TABLE_NAME} ADD COLUMN flag Nullable(Bool);", settings=INSERT_SETTINGS) + instance.query(f"ALTER TABLE {TABLE_NAME} ADD COLUMN price Nullable(Decimal(10, 2));", settings=INSERT_SETTINGS) + + assert instance.query(f"SELECT id, value, flag, price FROM {TABLE_NAME} ORDER BY id") == ( + "1\ta\t\\N\t\\N\n2\tb\t\\N\t\\N\n" + ) + + instance.query( + f"INSERT INTO {TABLE_NAME} VALUES (3, 'c', true, 99.95), (4, 'd', false, 123.40);", + settings=INSERT_SETTINGS, + ) + assert instance.query(f"SELECT id, flag, price FROM {TABLE_NAME} ORDER BY id") == ( + "1\t\\N\t\\N\n2\t\\N\t\\N\n3\ttrue\t99.95\n4\tfalse\t123.40\n" + ) diff --git a/tests/integration/test_storage_iceberg_no_spark/test_writes_drop_column.py b/tests/integration/test_storage_iceberg_no_spark/test_writes_drop_column.py new file mode 100644 index 000000000000..f29f8903e2c6 --- /dev/null +++ b/tests/integration/test_storage_iceberg_no_spark/test_writes_drop_column.py @@ -0,0 +1,62 @@ +import pytest + +from helpers.iceberg_utils import ( + create_iceberg_table, + get_uuid_str, +) + +INSERT_SETTINGS = {"allow_insert_into_iceberg": 1} + + +@pytest.mark.parametrize("format_version", [1, 2]) +@pytest.mark.parametrize("storage_type", ["local", "s3"]) +def test_drop_column_basic(started_cluster_iceberg_no_spark, format_version, storage_type): + """DROP COLUMN removes the column from reads and inserts; remaining columns unchanged.""" + instance = started_cluster_iceberg_no_spark.instances["node1"] + TABLE_NAME = "test_drop_column_basic_" + storage_type + "_" + get_uuid_str() + + create_iceberg_table( + storage_type, + instance, + TABLE_NAME, + started_cluster_iceberg_no_spark, + "(id Int32, value Nullable(String))", + format_version, + ) + + instance.query(f"INSERT INTO {TABLE_NAME} VALUES (1, 'hello'), (2, 'world');", settings=INSERT_SETTINGS) + assert instance.query(f"SELECT id, value FROM {TABLE_NAME} ORDER BY id") == "1\thello\n2\tworld\n" + + instance.query(f"ALTER TABLE {TABLE_NAME} DROP COLUMN value;", settings=INSERT_SETTINGS) + + assert instance.query(f"SELECT id FROM {TABLE_NAME} ORDER BY id") == "1\n2\n" + + instance.query(f"INSERT INTO {TABLE_NAME} VALUES (3);", settings=INSERT_SETTINGS) + assert instance.query(f"SELECT id FROM {TABLE_NAME} ORDER BY id") == "1\n2\n3\n" + + +@pytest.mark.parametrize("format_version", [1, 2]) +@pytest.mark.parametrize("storage_type", ["local", "s3"]) +def test_drop_column_errors(started_cluster_iceberg_no_spark, format_version, storage_type): + """Dropping a non-existent column must fail; table structure unchanged.""" + instance = started_cluster_iceberg_no_spark.instances["node1"] + TABLE_NAME = "test_drop_column_errors_" + storage_type + "_" + get_uuid_str() + + create_iceberg_table( + storage_type, + instance, + TABLE_NAME, + started_cluster_iceberg_no_spark, + "(id Int32, value Nullable(String))", + format_version, + ) + + error = instance.query_and_get_error( + f"ALTER TABLE {TABLE_NAME} DROP COLUMN nonexistent;", + settings=INSERT_SETTINGS, + ) + assert "nonexistent" in error + + assert instance.query( + f"SELECT name FROM system.columns WHERE database = currentDatabase() AND table = '{TABLE_NAME}' ORDER BY name" + ) == "id\nvalue\n" diff --git a/tests/integration/test_storage_iceberg_no_spark/test_writes_modify_column.py b/tests/integration/test_storage_iceberg_no_spark/test_writes_modify_column.py new file mode 100644 index 000000000000..ac03af1bd624 --- /dev/null +++ b/tests/integration/test_storage_iceberg_no_spark/test_writes_modify_column.py @@ -0,0 +1,114 @@ +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→Int64 is not a valid Iceberg schema evolution; must get BAD_ARGUMENTS. + assert "doesn't allow schema evolution" in el + + 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_modify_column_noop_same_type(started_cluster_iceberg_no_spark, format_version, storage_type): + """MODIFY COLUMN to the same type (Int32→Int32) is a no-op and must succeed silently.""" + instance = started_cluster_iceberg_no_spark.instances["node1"] + TABLE_NAME = "test_modify_noop_" + 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');", settings=INSERT_SETTINGS) + + # MODIFY to the exact same type should be a silent no-op (no schema change). + instance.query(f"ALTER TABLE {TABLE_NAME} MODIFY COLUMN id Int32;", settings=INSERT_SETTINGS) + + assert instance.query(f"SELECT id, value FROM {TABLE_NAME} ORDER BY id") == "1\ta\n" + + +@pytest.mark.parametrize("format_version", [1, 2]) +@pytest.mark.parametrize("storage_type", ["local", "s3"]) +def test_modify_column_rejects_indistinguishable_type(started_cluster_iceberg_no_spark, format_version, storage_type): + """MODIFY COLUMN id UInt32 on an Iceberg 'int' column (Int32) must fail because + Iceberg represents both as 'int' and the change cannot be recorded.""" + instance = started_cluster_iceberg_no_spark.instances["node1"] + TABLE_NAME = "test_modify_reject_" + 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, 'x');", settings=INSERT_SETTINGS) + + error = instance.query_and_get_error( + f"ALTER TABLE {TABLE_NAME} MODIFY COLUMN id UInt32;", + settings=INSERT_SETTINGS, + ) + assert "same iceberg type" in error.lower() or "cannot modify" in error.lower() or "bad_arguments" in error.lower()