Fix DataLakeCatalog table creation issues

src/Databases/DataLake/DatabaseDataLake.cpp

@@ -187,6 +187,14 @@ String getLocationScheme(const std::shared_ptr<DataLake::ICatalog> & catalog)
         catalog->getCatalogType());
 }
 
+/// Storage settings that a catalog table is opened with.
+std::shared_ptr<DataLakeStorageSettings> getTableStorageSettings(const DatabaseDataLakeSettings & settings)
+{
+    auto storage_settings = std::make_shared<DataLakeStorageSettings>();
+    storage_settings->loadFromSettingsChanges(settings.allChanged());
+    return storage_settings;
+}
+
 /// Translate the database-layer `TablesFilter` into the catalog-layer
 /// `TableNameFilter` so the catalog can restrict which namespaces it lists.
 DataLake::TableNameFilter toCatalogTableNameFilter(const TablesFilter & tables_filter)
@@ -1024,8 +1032,7 @@ StoragePtr DatabaseDataLake::tryGetTableImpl(const String & name, ContextPtr con
 
     LOG_TEST(log, "Using table endpoint: {}", args[0]->as<ASTLiteral>()->value.safeGet<String>());
 
-    auto storage_settings = std::make_shared<DataLakeStorageSettings>();
-    storage_settings->loadFromSettingsChanges(settings.allChanged());
+    auto storage_settings = getTableStorageSettings(settings);
 
     if (auto table_specific_properties = table_metadata.getDataLakeSpecificProperties();
         table_specific_properties.has_value())
@@ -1177,6 +1184,39 @@ void DatabaseDataLake::validateCreateTableEngine(const ASTStorage & storage, Con
             "This DataLakeCatalog assigns table locations itself, so an explicit table `ENGINE` "
             "is not supported; omit the `ENGINE` clause");
 
+    if (storage.settings)
+    {
+        /// Reject table-level settings that are lost when the table is reopened.
+#define DATA_LAKE_SETTING_NAME(TYPE, NAME, ...) #NAME,
+        static const std::unordered_set<std::string_view> settings_from_database = {
+            DATA_LAKE_STORAGE_RELATED_SETTINGS(DATA_LAKE_SETTING_NAME, DATA_LAKE_SETTING_NAME)};
+#undef DATA_LAKE_SETTING_NAME
+
+        const auto database_storage_settings = getTableStorageSettings(*database_settings.get());
+        DataLakeStorageSettings table_storage_settings;
+        table_storage_settings.loadFromSettingsChanges(storage.settings->changes);
+
+        for (const auto & change : storage.settings->changes)
+        {
+            /// An unknown setting is rejected when the storage is created.
+            const auto tier = DataLakeStorageSettings::tryGetTierOfBuiltin(change.name);
+            if (!tier || *tier == SettingsTierType::OBSOLETE || change.name == "iceberg_format_version")
+                continue;
+
+            if (!settings_from_database.contains(change.name))
+                throw Exception(ErrorCodes::BAD_ARGUMENTS,
+                    "Table setting `{}` is not kept by a DataLakeCatalog table: "
+                    "it is taken from the query settings when the table is reopened",
+                    change.name);
+
+            if (table_storage_settings.get(change.name) != database_storage_settings->get(change.name))
+                throw Exception(ErrorCodes::BAD_ARGUMENTS,
+                    "Table setting `{}` differs from its value in this DataLakeCatalog. "
+                    "The table is reopened with the database settings, so the table-level value would be lost",
+                    change.name);
+        }
+    }
+
     catalog->loadConfigIfNeeded(DataLake::getForwardedAuthToken(context_));
     const auto catalog_storage_type = catalog->getStorageType();
 
@@ -1901,7 +1941,7 @@ void registerDatabaseDataLake(DatabaseFactory & factory)
 
         DatabaseDataLakeSettings database_settings;
         if (database_engine_define->settings)
-            database_settings.loadFromQuery(*database_engine_define, args.create_query.attach);
+            database_settings.loadFromQuery(*database_engine_define);
 
         const auto & auth_header_str = database_settings[DatabaseDataLakeSetting::auth_header].value;
         /// Validate `auth_header` on CREATE only (matches the `allow_experimental_database_*`

src/Databases/DataLake/DatabaseDataLakeSettings.cpp

@@ -12,7 +12,6 @@ namespace DB
 
 namespace ErrorCodes
 {
-    extern const int BAD_ARGUMENTS;
     extern const int UNKNOWN_SETTING;
     extern const int LOGICAL_ERROR;
 }
@@ -92,17 +91,12 @@ void DatabaseDataLakeSettings::applyChanges(const SettingsChanges & changes)
     impl->applyChanges(changes);
 }
 
-void DatabaseDataLakeSettings::loadFromQuery(const ASTStorage & storage_def, bool is_attach)
+void DatabaseDataLakeSettings::loadFromQuery(const ASTStorage & storage_def)
 {
     if (storage_def.settings)
     {
         try
         {
-            for (const auto & change : storage_def.settings->changes)
-            {
-                if (!is_attach && change.name.starts_with("iceberg_"))
-                    throw Exception(ErrorCodes::BAD_ARGUMENTS, "The setting {} is used for storage. Please use the setting without `iceberg_` prefix", change.name);
-            }
             impl->applyChanges(storage_def.settings->changes);
         }
         catch (Exception & e)

src/Databases/DataLake/DatabaseDataLakeSettings.h

@@ -66,7 +66,7 @@ struct DatabaseDataLakeSettings
 
     DATABASE_ICEBERG_SETTINGS_SUPPORTED_TYPES(DatabaseDataLakeSettings, DECLARE_SETTING_SUBSCRIPT_OPERATOR)
 
-    void loadFromQuery(const ASTStorage & storage_def, bool is_attach);
+    void loadFromQuery(const ASTStorage & storage_def);
 
     void applyChanges(const SettingsChanges & changes);
 

src/Interpreters/InterpreterCreateQuery.cpp

@@ -1476,7 +1476,8 @@ void InterpreterCreateQuery::setEngine(ASTCreateQuery & create) const
         /// holds credentials
         getContext()->checkAccess(AccessType::SHOW_COLUMNS, as_database_name, as_table_name);
 
-        ASTPtr as_create_ptr = DatabaseCatalog::instance().getDatabase(as_database_name)->getCreateTableQuery(as_table_name, getContext());
+        const auto as_database = DatabaseCatalog::instance().getDatabase(as_database_name);
+        ASTPtr as_create_ptr = as_database->getCreateTableQuery(as_table_name, getContext());
 
         const auto & as_create = as_create_ptr->as<ASTCreateQuery &>();
 
@@ -1538,9 +1539,10 @@ void InterpreterCreateQuery::setEngine(ASTCreateQuery & create) const
                     if (target.inner_engine && !create.getTargetInnerEngine(target.kind))
                         check_access_to_inherited_definition(*target.inner_engine);
         }
-        else if (as_create.storage)
+        else if (as_create.storage || as_database->isDatalakeCatalog())
         {
-            /// A `DataLakeCatalog` that assigns table locations itself shows its tables without `ENGINE`.
+            /// A `DataLakeCatalog` that assigns table locations itself shows its tables without `ENGINE`,
+            /// and a table without keys has no storage clause at all.
             /// Only a `DataLakeCatalog` target can be created from such a table: it builds its own storage
             /// and copies the keys from the source metadata.
             const auto target_database = DatabaseCatalog::instance().tryGetDatabase(getContext()->resolveDatabase(create.getDatabase()));

src/Storages/ObjectStorage/DataLakes/DataLakeStorageSettings.cpp

@@ -50,6 +50,11 @@ bool DataLakeStorageSettings::hasBuiltin(std::string_view name)
     return DataLakeStorageSettingsImpl::hasBuiltin(name);
 }
 
+std::optional<SettingsTierType> DataLakeStorageSettings::tryGetTierOfBuiltin(std::string_view name)
+{
+    return DataLakeStorageSettingsImpl::tryGetTierOfBuiltin(name);
+}
+
 void DataLakeStorageSettings::loadFromSettingsChanges(const SettingsChanges & changes)
 {
     for (const auto & [name, value, _] : changes)

src/Storages/ObjectStorage/DataLakes/DataLakeStorageSettings.h

@@ -4,9 +4,11 @@
 #include <Core/FormatFactorySettings.h>
 #include <Core/SettingsEnums.h>
 #include <Core/SettingsFields.h>
+#include <Core/SettingsTierType.h>
 #include <base/unit.h>
 
 #include <limits>
+#include <optional>
 
 
 namespace DB
@@ -190,6 +192,8 @@ struct DataLakeStorageSettings
 
     static bool hasBuiltin(std::string_view name);
 
+    static std::optional<SettingsTierType> tryGetTierOfBuiltin(std::string_view name);
+
     void serialize(WriteBuffer & out) const;
     static DataLakeStorageSettings deserialize(ReadBuffer & in);
 

src/Storages/ObjectStorage/DataLakes/Iceberg/Compaction.cpp

@@ -490,6 +490,7 @@ static void writeDataFiles(
 
 static bool writeConsolidatedManifestFile(
     int metadata_version,
+    const String & metadata_file_path,
     Poco::JSON::Object::Ptr metadata_object,
     const PersistentTableComponents & persistent_table_components,
     ObjectStoragePtr object_storage, ContextPtr context,
@@ -818,7 +819,7 @@ static bool writeConsolidatedManifestFile(
     // Manifest-only rewrite: use a snapshot type that carries all total-* counters forward unchanged, since passing deltas would inflate the totals.
     auto new_snapshot = metadata_generator.generateManifestOnlySnapshot(
         generator,
-        generated_metadata_info.path,
+        path_resolver.reverseResolve(metadata_file_path),
         current_snapshot_id);
 
     // Write one manifest file per (partition spec, partition value) group.
@@ -1579,6 +1580,7 @@ void compactIcebergManifests(
 
         if (writeConsolidatedManifestFile(
                 metadata_version,
+                metadata_file_path,
                 metadata_object,
                 persistent_table_components,
                 object_storage_,

src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.cpp

@@ -897,7 +897,7 @@ void IcebergMetadata::truncate(ContextPtr context, std::shared_ptr<DataLake::ICa
     auto metadata_info = filename_generator.generateMetadataPathWithInfo();
 
     auto [new_snapshot, manifest_list_path] = MetadataGenerator(metadata_object).generateNextMetadata(
-        filename_generator, metadata_info.path, parent_snapshot_id,
+        filename_generator, resolver.reverseResolve(actual_table_state_snapshot.metadata_file_path), parent_snapshot_id,
         /* added_files */ 0, /* added_records */ 0, /* added_files_size */ 0,
         /* num_partitions */ 0, /* added_delete_files */ 0, /* num_deleted_rows */ 0,
         std::nullopt, std::nullopt, MetadataGenerator::SnapshotOperation::Append, /*is_truncate=*/true);
@@ -2309,6 +2309,7 @@ std::vector<Field> recomputeExportPartitionValues(
 std::optional<IStorage::ExportPartitionCommitInfo> IcebergMetadata::commitImportPartitionTransactionImpl(
     FileNamesGenerator & filename_generator,
     Poco::JSON::Object::Ptr & metadata,
+    String & metadata_file_path,
     Poco::JSON::Object::Ptr & partition_spec,
     const String & transaction_id,
     Int64 original_schema_id,
@@ -2353,7 +2354,7 @@ std::optional<IStorage::ExportPartitionCommitInfo> IcebergMetadata::commitImport
         parent_snapshot = metadata->getValue<Int64>(Iceberg::f_current_snapshot_id);
 
     auto [new_snapshot, manifest_list_path] = MetadataGenerator(metadata).generateNextMetadata(
-        filename_generator, metadata_info.path, parent_snapshot, total_data_files, total_rows, total_chunks_size, total_data_files, /* added_delete_files */0, /* num_deleted_rows */0);
+        filename_generator, resolver.reverseResolve(metadata_file_path), parent_snapshot, total_data_files, total_rows, total_chunks_size, total_data_files, /* added_delete_files */0, /* num_deleted_rows */0);
     const auto storage_manifest_list_name = resolver.resolve(manifest_list_path);
 
     /// Embed the stable transaction identifier in the snapshot summary so that a retry after crash
@@ -2430,6 +2431,7 @@ std::optional<IStorage::ExportPartitionCommitInfo> IcebergMetadata::commitImport
                 getLogger("IcebergMetadata"),
                 compression_method,
                 persistent_components.table_uuid);
+            metadata_file_path = metadata_path;
 
             /// For the export path the schema and partition spec are fixed at the start of the
             /// operation (saved in ZooKeeper). If either changed we must fail immediately —
@@ -2740,12 +2742,14 @@ IStorage::ExportPartitionCommitInfo IcebergMetadata::commitExportPartitionTransa
         per_file_stats.push_back(std::move(stats));
     }
 
+    String metadata_file_path = updated_metadata_file_info.path;
     size_t attempt = 0;
     while (attempt < MAX_TRANSACTION_RETRIES)
     {
         auto commit_info = commitImportPartitionTransactionImpl(
                 filename_generator,
                 metadata,
+                metadata_file_path,
                 partition_spec,
                 transaction_id,
                 original_schema_id,

src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.h

@@ -295,6 +295,7 @@ class IcebergMetadata : public IDataLakeMetadata
     std::optional<IStorage::ExportPartitionCommitInfo> commitImportPartitionTransactionImpl(
         FileNamesGenerator & filename_generator,
         Poco::JSON::Object::Ptr & metadata,
+        String & metadata_file_path,
         Poco::JSON::Object::Ptr & partition_spec,
         const String & transaction_id,
         Int64 original_schema_id,

src/Storages/ObjectStorage/DataLakes/Iceberg/MetadataGenerator.cpp

@@ -450,6 +450,14 @@ Poco::JSON::Object::Ptr MetadataGenerator::getParentSnapshot(Int64 parent_snapsh
     return nullptr;
 }
 
+void MetadataGenerator::appendMetadataLog(const Iceberg::IcebergPathFromMetadata & previous_metadata_file_path, Int64 timestamp_ms)
+{
+    Poco::JSON::Object::Ptr new_metadata_item = new Poco::JSON::Object;
+    new_metadata_item->set(Iceberg::f_metadata_file, previous_metadata_file_path.serialize());
+    new_metadata_item->set(Iceberg::f_timestamp_ms, timestamp_ms);
+    getOrCreateArray(metadata_object, Iceberg::f_metadata_log)->add(new_metadata_item);
+}
+
 MetadataGenerator::NextMetadataResult MetadataGenerator::generateNextMetadata(
     FileNamesGenerator & generator,
     const Iceberg::IcebergPathFromMetadata & previous_metadata_file_path,
@@ -588,12 +596,7 @@ MetadataGenerator::NextMetadataResult MetadataGenerator::generateNextMetadata(
         metadata_object->getObject(Iceberg::f_refs)->getObject(Iceberg::f_main)->set(Iceberg::f_metadata_snapshot_id, snapshot_id);
 
     if (!previous_metadata_file_path.empty())
-    {
-        Poco::JSON::Object::Ptr new_metadata_item = new Poco::JSON::Object;
-        new_metadata_item->set(Iceberg::f_metadata_file, previous_metadata_file_path.serialize());
-        new_metadata_item->set(Iceberg::f_timestamp_ms, timestamp);
-        getOrCreateArray(metadata_object, Iceberg::f_metadata_log)->add(new_metadata_item);
-    }
+        appendMetadataLog(previous_metadata_file_path, timestamp);
     {
         Poco::JSON::Object::Ptr new_snapshot_item = new Poco::JSON::Object;
         new_snapshot_item->set(Iceberg::f_metadata_snapshot_id, snapshot_id);
@@ -696,12 +699,7 @@ MetadataGenerator::NextMetadataResult MetadataGenerator::appendReplaceSnapshot(
         metadata_object->getObject(Iceberg::f_refs)->getObject(Iceberg::f_main)->set(Iceberg::f_metadata_snapshot_id, snapshot_id);
     }
 
-    {
-        Poco::JSON::Object::Ptr new_metadata_item = new Poco::JSON::Object;
-        new_metadata_item->set(Iceberg::f_metadata_file, metadata_file_path.serialize());
-        new_metadata_item->set(Iceberg::f_timestamp_ms, timestamp);
-        metadata_object->getArray(Iceberg::f_metadata_log)->add(new_metadata_item);
-    }
+    appendMetadataLog(metadata_file_path, timestamp);
     {
         Poco::JSON::Object::Ptr new_snapshot_item = new Poco::JSON::Object;
         new_snapshot_item->set(Iceberg::f_metadata_snapshot_id, snapshot_id);

src/Storages/ObjectStorage/DataLakes/Iceberg/MetadataGenerator.h

@@ -68,6 +68,9 @@ class MetadataGenerator
         Int64 removed_files_size,
         Int64 num_partitions);
 
+    /// Records the metadata file that the generated one replaces.
+    void appendMetadataLog(const Iceberg::IcebergPathFromMetadata & previous_metadata_file_path, Int64 timestamp_ms);
+
     void generateDropColumnMetadata(const String & column_name);
     /// Returns false when neither the type nor the position changed (true no-op).
     /// Throws when the requested type differs from the current one but maps to the same Iceberg type.

src/Storages/ObjectStorage/DataLakes/Iceberg/Mutations.cpp

@@ -981,6 +981,10 @@ void alter(
         if (succeeded)
             break;
 
+        metadata_json_generator.appendMetadataLog(
+            persistent_table_components.path_resolver.reverseResolve(metadata_path),
+            std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::system_clock::now().time_since_epoch()).count());
+
         const auto new_schema_id = metadata->getValue<Int32>(Iceberg::f_current_schema_id);
         Poco::JSON::Object::Ptr new_schema;
         auto schemas = metadata->getArray(Iceberg::f_schemas);

tests/integration/test_database_iceberg/test.py

@@ -2036,6 +2036,44 @@ def test_create_table_with_engine_unsupported_clauses(started_cluster):
     assert "PRIMARY KEY, SAMPLE BY, TTL, and UNIQUE KEY are not supported" in err
 
 
+def test_create_table_with_engine_settings(started_cluster):
+    # A table-level setting that would be lost when the table is reopened must be rejected.
+    node = started_cluster.instances["node1"]
+    namespace = f"engine_settings_{uuid.uuid4().hex[:8]}"
+
+    def create_table(name, settings):
+        return (
+            f"CREATE TABLE {CATALOG_NAME}.`{namespace}.{name}` (id Int64) "
+            f"ENGINE = IcebergS3('http://minio1:9001/warehouse-rest/{namespace}/{name}/', "
+            f"'{minio_access_key}', '{minio_secret_key}') SETTINGS {settings}"
+        )
+
+    query_settings = {"allow_database_iceberg": 1, "write_full_path_in_iceberg_metadata": 1}
+
+    create_clickhouse_iceberg_database(
+        started_cluster, node, CATALOG_NAME, {"iceberg_recent_metadata_file_by_last_updated_ms_field": 1}
+    )
+
+    err = node.query_and_get_error(
+        create_table("by_last_updated", "iceberg_recent_metadata_file_by_last_updated_ms_field = 0"), settings=query_settings
+    )
+    assert "Table setting `iceberg_recent_metadata_file_by_last_updated_ms_field` differs from its value in this DataLakeCatalog" in err
+
+    err = node.query_and_get_error(create_table("block_size", "input_format_parquet_max_block_size = 1234"), settings=query_settings)
+    assert "Table setting `input_format_parquet_max_block_size` is not kept by a DataLakeCatalog table" in err
+
+    # A value equal to the database one is kept, the format version is written into the table metadata,
+    # and an obsolete setting does nothing.
+    node.query(
+        create_table(
+            "created",
+            "iceberg_recent_metadata_file_by_last_updated_ms_field = 1, iceberg_format_version = 1, "
+            "allow_dynamic_metadata_for_data_lakes = 0",
+        ),
+        settings=query_settings,
+    )
+
+
 @pytest.mark.parametrize("explicit_engine", [False, True])
 def test_create_table_ignore_unsupported_properties(started_cluster, explicit_engine):
     node = started_cluster.instances["node1"]

tests/integration/test_export_partition_to_iceberg/test_lifecycle.py

@@ -1,4 +1,10 @@
+import json
+import os
+import re
+
 from helpers.export_partition_helpers import (
+    MINIO_PASS,
+    MINIO_USER,
     make_iceberg_s3,
     make_source,
     unique_suffix,
@@ -47,6 +53,17 @@ def test_export_partition_to_iceberg(cluster, source_engine):
         f"Unexpected data in Iceberg table:\n{result}"
     )
 
+    # The commit must append the metadata file it replaces to `metadata-log`.
+    rows = node.query(
+        f"SELECT _file, raw FROM s3('http://minio1:9001/root/data/{iceberg_table}/metadata/*.metadata.json', "
+        f"'{MINIO_USER}', '{MINIO_PASS}', 'RawBLOB', 'raw String') FORMAT JSONEachRow"
+    ).strip().split("\n")
+    metadata_files = sorted(
+        (json.loads(row) for row in rows), key=lambda row: int(re.match(r"v(\d+)", row["_file"]).group(1))
+    )
+    metadata_log = json.loads(metadata_files[-1]["raw"])["metadata-log"]
+    assert os.path.basename(metadata_log[-1]["metadata-file"]) == metadata_files[-2]["_file"], metadata_log
+
 
 def test_export_two_partitions_to_iceberg(cluster, source_engine):
     """

tests/queries/0_stateless/05428_iceberg_metadata_log_previous_file.reference

@@ -0,0 +1,3 @@
+v3.metadata.json True
+v6.metadata.json True
+v7.metadata.json True