Repository: ClickHouse/ClickHouse
File: src/Databases/DataLake/DatabaseDataLake.cpp (C++)

Code before:
            configuration,
            configuration->createObjectStorage(context_copy, /* is_readonly */ false),
            StorageID(getDatabaseName(), name),
            columns,
            ConstraintsDescription{},
            nullptr,
            context_);

        storage_cluster->startup();
        return storage_cluster;
    }

    return std::make_shared<StorageObjectStorage>(
        configuration,
        configuration->createObjectStorage(context_copy, /* is_readonly */ false),
        context_copy,
        StorageID(getDatabaseName(), name),
        /* columns */columns,
        /* constraints */ConstraintsDescription{},
        /* comment */"",
        getFormatSettings(context_copy),
        LoadingStrictnessLevel::CREATE,
        getCatalog(),
        /* if_not_exists*/true,
        /* is_datalake_query*/true,
        /* distributed_processing */can_use_parallel_replicas,
        /* partition_by */nullptr,
        /* is_table_function */false,
        /* lazy_init */true);
}

void DatabaseDataLake::dropTable( /// NOLINT
    ContextPtr context_,
    const String & name,
    bool /*sync*/)
{
    auto table = tryGetTable(name, context_);
    if (table)
        table->drop();
    else
        throw Exception(ErrorCodes::BAD_ARGUMENTS, "Cannot drop table {} because it does not exist", name);
}

DatabaseTablesIteratorPtr DatabaseDataLake::getTablesIterator(
    ContextPtr context_,
    const FilterByNameFunction & filter_by_table_name,
    bool skip_not_loaded) const
{
    Tables tables;
    auto catalog = getCatalog();
    DB::Names iceberg_tables;

Review comment (suggestion):
Shouldn't we also check conditions like it is done here in case of secondary query: https://github.com/ClickHouse/ClickHouse/blob/e50ba3ad2360f0508c72ab398d5afbe8e8bec319/src/TableFunctions/TableFunctionObjectStorageCluster.cpp#L39-L42

Code after:
            StorageID(getDatabaseName(), name),
            columns,
            ConstraintsDescription{},
            nullptr,
            context_);

        storage_cluster->startup();
        return storage_cluster;
    }

    bool can_use_distributed_iterator =
        context_->getClientInfo().collaborate_with_initiator &&
        can_use_parallel_replicas;

    return std::make_shared<StorageObjectStorage>(
        configuration,
        configuration->createObjectStorage(context_copy, /* is_readonly */ false),
        context_copy,
        StorageID(getDatabaseName(), name),
        /* columns */columns,
        /* constraints */ConstraintsDescription{},
        /* comment */"",
        getFormatSettings(context_copy),
        LoadingStrictnessLevel::CREATE,
        getCatalog(),
        /* if_not_exists*/true,
        /* is_datalake_query*/true,
        /* distributed_processing */can_use_distributed_iterator,
        /* partition_by */nullptr,
        /* is_table_function */false,
        /* lazy_init */true);
}

void DatabaseDataLake::dropTable( /// NOLINT
    ContextPtr context_,
    const String & name,
    bool /*sync*/)
{
    auto table = tryGetTable(name, context_);
    if (table)
        table->drop();
    else
        throw Exception(ErrorCodes::BAD_ARGUMENTS, "Cannot drop table {} because it does not exist", name);
}

DatabaseTablesIteratorPtr DatabaseDataLake::getTablesIterator(
    ContextPtr context_,
    const FilterByNameFunction & filter_by_table_name,
    bool skip_not_loaded) const
{
    Tables tables;