Merge pull request #2607 from Altinity/backports/unity_v2_catalog

Antalya 26.8: Backport of ClickHouse#115879 Unity catalog(v2)

Author: zvonand

src/Databases/DataLake/DatabaseDataLake.cpp

@@ -30,6 +30,7 @@
 #include <Databases/DatabaseFactory.h>
 #include <Databases/DataLake/UnityCatalog.h>
 #include <Databases/DataLake/RestCatalog.h>
+#include <Databases/DataLake/UnityV2Catalog.h>
 #include <Databases/DataLake/GlueCatalog.h>
 #include <Databases/DataLake/PaimonRestCatalog.h>
 #if USE_AWS_S3 && USE_SSL
@@ -71,6 +72,7 @@ namespace DatabaseDataLakeSetting
     extern const DatabaseDataLakeSettingsDatabaseDataLakeCatalogType catalog_type;
     extern const DatabaseDataLakeSettingsString warehouse;
     extern const DatabaseDataLakeSettingsString catalog_credential;
+    extern const DatabaseDataLakeSettingsBool use_unity_catalog_v2;
     extern const DatabaseDataLakeSettingsString auth_header;
     extern const DatabaseDataLakeSettingsString auth_scope;
     extern const DatabaseDataLakeSettingsString storage_endpoint;
@@ -506,11 +508,28 @@ void DatabaseDataLake::initialize() const
         }
         case DB::DatabaseDataLakeCatalogType::UNITY:
         {
-            catalog_impl = std::make_shared<DataLake::UnityCatalog>(
+            if (!settings[DatabaseDataLakeSetting::use_unity_catalog_v2])
+            {
+                catalog_impl = std::make_shared<DataLake::UnityCatalog>(
+                    settings[DatabaseDataLakeSetting::warehouse].value,
+                    url,
+                    settings[DatabaseDataLakeSetting::catalog_credential].value,
+                    settings[DatabaseDataLakeSetting::namespaces].value,
+                    Context::getGlobalContextInstance());
+                break;
+            }
+
+            /// Databricks OIDC expects `all-apis`; the default `auth_scope` value targets Iceberg REST catalogs.
+            const std::string unity_auth_scope = settings[DatabaseDataLakeSetting::auth_scope].changed
+                ? settings[DatabaseDataLakeSetting::auth_scope].value
+                : "all-apis";
+            catalog_impl = std::make_shared<DataLake::UnityV2Catalog>(
                 settings[DatabaseDataLakeSetting::warehouse].value,
                 url,
                 settings[DatabaseDataLakeSetting::catalog_credential].value,
-                settings[DatabaseDataLakeSetting::namespaces].value,
+                unity_auth_scope,
+                settings[DatabaseDataLakeSetting::auth_header].value,
+                settings[DatabaseDataLakeSetting::oauth_server_uri].value,
                 Context::getGlobalContextInstance());
             break;
         }
@@ -658,15 +677,23 @@ void DatabaseDataLake::resetCatalog(String reason) const
 
 StorageObjectStorageConfigurationPtr DatabaseDataLake::getConfiguration(
     DatabaseDataLakeStorageType type,
-    DataLakeStorageSettingsPtr storage_settings) const
+    DataLakeStorageSettingsPtr storage_settings,
+    DataLake::DataLakeTableFormat table_format) const
 {
     /// TODO: add tests for azure, local storage types.
 
     const auto settings_version = database_settings.get();
     const DatabaseDataLakeSettings & settings = *settings_version;
 
     auto catalog = getCatalog();
-    switch (catalog->getCatalogType())
+    auto catalog_type = catalog->getCatalogType();
+
+    /// `UnityV2Catalog` serves both formats. Its Iceberg tables need the same configuration as an
+    /// Iceberg REST catalog, so route them to that arm; the legacy catalog never reports Iceberg.
+    if (catalog_type == DatabaseDataLakeCatalogType::UNITY && table_format == DataLake::DataLakeTableFormat::ICEBERG)
+        catalog_type = DatabaseDataLakeCatalogType::ICEBERG_REST;
+
+    switch (catalog_type)
     {
         case DatabaseDataLakeCatalogType::ICEBERG_ONELAKE:
         {
@@ -1011,7 +1038,7 @@ StoragePtr DatabaseDataLake::tryGetTableImpl(const String & name, ContextPtr con
         }
     }
 
-    const auto configuration = getConfiguration(storage_type, storage_settings);
+    const auto configuration = getConfiguration(storage_type, storage_settings, table_metadata.getTableFormat());
 
     /// HACK: Hacky-hack to enable lazy load
     ContextMutablePtr context_copy = Context::createCopy(context_);
@@ -1093,7 +1120,7 @@ StoragePtr DatabaseDataLake::tryGetTableImpl(const String & name, ContextPtr con
             return std::nullopt;
         if (!with_vended_credentials && !catalog_manages_provider_chain)
             return std::nullopt;
-        return catalog->getCredentialsConfigurationCallback(storage_id, DataLake::getForwardedAuthToken(context_));
+        return catalog->getCredentialsConfigurationCallback(storage_id, table_metadata, DataLake::getForwardedAuthToken(context_));
     };
 
     const auto catalog_uuid = table_metadata.getTableUUID();
@@ -1421,7 +1448,7 @@ DatabaseTablesIteratorPtr DatabaseDataLake::getTablesIteratorWithHint(
 DatabaseTablesIteratorPtr DatabaseDataLake::getTablesIteratorImpl(
     ContextPtr context_,
     const FilterByNameFunction & filter_by_table_name,
-    bool skip_not_loaded,
+    bool /*skip_not_loaded*/,
     const TablesFilter & tables_filter,
     bool keep_unresolved_tables) const
 {
@@ -1470,13 +1497,15 @@ DatabaseTablesIteratorPtr DatabaseDataLake::getTablesIteratorImpl(
             futures.emplace_back(promises.back()->get_future());
 
             pool.scheduleOrThrow(
-                [this, table_name, skip_not_loaded, context_, keep_unresolved_tables, promise=promises.back()]() mutable
+                [this, table_name, context_, keep_unresolved_tables, promise=promises.back()]() mutable
                 {
                     StoragePtr storage = nullptr;
                     try
                     {
                         LOG_INFO(log, "Get table information for table {}", table_name);
-                        storage = tryGetTableImpl(table_name, context_, false, skip_not_loaded);
+                        /// Tables marked unreadable at metadata time are skipped, not an error;
+                        /// direct access still reports the reason.
+                        storage = tryGetTableImpl(table_name, context_, /*lightweight*/ false, /*ignore_if_not_iceberg*/ true);
                     }
                     catch (...)
                     {
@@ -1668,6 +1697,10 @@ void DatabaseDataLake::applySettingsChanges(const SettingsChanges & settings_cha
     auto new_settings = std::make_unique<DatabaseDataLakeSettings>(*current_settings);
     new_settings->applyChanges(settings_changes);
 
+    /// Switching the Unity implementation replaces the catalog object rather than altering it in place.
+    const bool implementation_changed = (*new_settings)[DatabaseDataLakeSetting::use_unity_catalog_v2].value
+        != (*current_settings)[DatabaseDataLakeSetting::use_unity_catalog_v2].value;
+
     ASTPtr new_engine_definition;
     {
         std::lock_guard lock(mutex);
@@ -1695,7 +1728,17 @@ void DatabaseDataLake::applySettingsChanges(const SettingsChanges & settings_cha
         storage->set(storage->settings, storage_settings_ast);
     }
 
+    /// `use_unity_catalog_v2` is consumed by the database, not by the catalog object, so it is
+    /// never handed to the catalog. Re-applying the current value is a metadata-only no-op.
+    SettingsChanges catalog_settings_changes;
+    for (const auto & change : settings_changes)
+    {
+        if (change.name != DatabaseDataLakeSettings::getSettingName(DatabaseDataLakeSetting::use_unity_catalog_v2))
+            catalog_settings_changes.push_back(change);
+    }
+
     std::shared_ptr<DataLake::ICatalog> local_catalog_snapshot;
+    if (!implementation_changed)
     {
         std::lock_guard lock(catalog_mutex);
         local_catalog_snapshot = catalog_impl;
@@ -1704,8 +1747,9 @@ void DatabaseDataLake::applySettingsChanges(const SettingsChanges & settings_cha
     /// Prepare the new catalog state without publishing it: validation, the eager token
     /// fetch and the config reload may throw, and then nothing has changed yet.
     DataLake::ICatalog::PreparedSettingsChangesPtr prepared_catalog_changes;
-    if (local_catalog_snapshot)
-        prepared_catalog_changes = local_catalog_snapshot->prepareSettingsChanges(settings_changes, DataLake::getForwardedAuthToken(query_context));
+    const bool alter_catalog = local_catalog_snapshot && !catalog_settings_changes.empty();
+    if (alter_catalog)
+        prepared_catalog_changes = local_catalog_snapshot->prepareSettingsChanges(catalog_settings_changes, DataLake::getForwardedAuthToken(query_context));
 
     /// Persist the new metadata before publishing anything: if the write fails, the live
     /// state is untouched and matches the old metadata on disk. The create query is built
@@ -1717,7 +1761,7 @@ void DatabaseDataLake::applySettingsChanges(const SettingsChanges & settings_cha
     DatabaseCatalog::instance().updateMetadataFile(getDatabaseName(), new_create_query);
 
     /// Publish. Nothing below throws.
-    if (local_catalog_snapshot)
+    if (alter_catalog)
         local_catalog_snapshot->commitSettingsChanges(std::move(prepared_catalog_changes));
     database_settings.set(std::move(new_settings));
     {
@@ -1726,9 +1770,9 @@ void DatabaseDataLake::applySettingsChanges(const SettingsChanges & settings_cha
     }
     if (!local_catalog_snapshot)
     {
-        /// The catalog was not built when the ALTER started. If a concurrent query
-        /// built it meanwhile, it used the old settings: drop it so the next access
-        /// rebuilds it with the new ones. Also clear a recorded construction failure
+        /// The catalog was not built when the ALTER started, or the implementation was switched.
+        /// If a concurrent query built it meanwhile, it used the old settings: drop it so the next
+        /// access rebuilds it with the new ones. Also clear a recorded construction failure
         /// (e.g. credentials lost on RESTORE) for the same reason.
         std::lock_guard lock(catalog_mutex);
         resetCatalog(/* reason */ "");
@@ -1813,6 +1857,10 @@ ASTPtr DatabaseDataLake::getCreateTableQueryImpl(
     storage->engine->setKind(ASTFunction::Kind::TABLE_ENGINE);
     if (!table_metadata.isDefaultReadableTable())
         storage->engine->name = DataLake::FAKE_TABLE_ENGINE_NAME_FOR_UNREADABLE_TABLES;
+    else if (table_metadata.getTableFormat() != DataLake::DataLakeTableFormat::UNKNOWN)
+        /// Mixed-format catalogs (e.g. `UnityV2Catalog`) set the format per table; single-format
+        /// catalogs leave it UNKNOWN and keep the database-level table engine name.
+        storage->engine->name = String(catalog->getTableEngineName(table_metadata));
 
     storage->settings = {};
     if (partition_by)
@@ -2197,9 +2245,10 @@ The following settings are supported:
 
 | Setting                 | Description                                                                             |
 |-------------------------|-----------------------------------------------------------------------------------------|
-| `catalog_type`          | Type of catalog: `glue`, `unity` (Delta), `rest` (Iceberg), `hive`, `onelake` (Iceberg), `delta_sharing` (Iceberg, flat namespaces), `horizon` (Snowflake Horizon Iceberg REST) |
+| `catalog_type`          | Type of catalog: `glue`, `unity` (Delta, or Delta and Iceberg with `use_unity_catalog_v2`), `rest` (Iceberg), `hive`, `onelake` (Iceberg), `delta_sharing` (Iceberg, flat namespaces), `horizon` (Snowflake Horizon Iceberg REST) |
 | `warehouse`             | The warehouse/database name to use in the catalog.                                      |
 | `catalog_credential`    | Authentication credential for the catalog (e.g., API key or token)                      |
+| `use_unity_catalog_v2`  | For `catalog_type = 'unity'`: use the new implementation, which serves both Delta Lake and Iceberg tables. Default: `false`. Set it on `CREATE`, or change it for an existing database with `ALTER DATABASE ... MODIFY SETTING`. |
 | `auth_header`           | Custom HTTP header for authentication with the catalog service                          |
 | `auth_scope`            | OAuth2 scope for authentication (if using OAuth)                                        |
 | `storage_endpoint`      | Endpoint URL for the underlying storage                                                 |
@@ -2325,6 +2374,22 @@ Some catalogs support only one of the two modes:
 See below sections for examples of using the `DataLakeCatalog` engine:
 
 * [Unity Catalog](/guides/use-cases/data-warehousing/unity-catalog)
+* Unity Catalog with Delta Lake and Iceberg tables
+    With `use_unity_catalog_v2 = 1`, the `unity` catalog serves both Delta Lake and Iceberg tables,
+    detecting the format of each table.
+```sql
+CREATE DATABASE database_name
+ENGINE = DataLakeCatalog('https://<workspace>.cloud.databricks.com/api/2.1/unity-catalog')
+SETTINGS
+    catalog_type = 'unity',
+    use_unity_catalog_v2 = 1,
+    warehouse = 'my_catalog',
+    catalog_credential = '<client_id>:<client_secret>';
+SHOW TABLES FROM database_name;
+SELECT count() FROM database_name.`schema_name.table_name`;
+```
+    `catalog_credential` accepts an OAuth service principal as `<client_id>:<client_secret>`,
+    a personal access token (`dapi...`), or a pre-obtained OAuth token.
 * [Glue Catalog](/guides/use-cases/data-warehousing/glue-catalog)
 * OneLake Catalog
     Can be used by enabling `allow_experimental_database_iceberg` or `allow_database_iceberg`.

src/Databases/DataLake/DatabaseDataLake.h

@@ -147,7 +147,8 @@ class DatabaseDataLake final : public IDatabase, WithContext
 
     StorageObjectStorageConfigurationPtr getConfiguration(
         DatabaseDataLakeStorageType type,
-        DataLakeStorageSettingsPtr storage_settings) const;
+        DataLakeStorageSettingsPtr storage_settings,
+        DataLake::DataLakeTableFormat table_format) const;
 
     std::string getStorageEndpointForTable(const DataLake::TableMetadata & table_metadata) const;
 

src/Databases/DataLake/DatabaseDataLakeSettings.cpp

@@ -20,6 +20,7 @@ namespace ErrorCodes
 #define DATABASE_ICEBERG_RELATED_SETTINGS(DECLARE, ALIAS) \
     DECLARE(DatabaseDataLakeCatalogType, catalog_type, DatabaseDataLakeCatalogType::NONE, "Catalog type", 0) \
     DECLARE(String, catalog_credential, "", "", 0) \
+    DECLARE(Bool, use_unity_catalog_v2, false, "Use the new Unity catalog implementation, which serves both Delta Lake and Iceberg tables. Only for catalog_type = 'unity'", 0) \
     DECLARE(Bool, vended_credentials, true, "Use vended credentials (storage credentials) from catalog", 0) \
     DECLARE(UInt64, vended_credentials_cache_ttl, 300, "Maximum cache entry lifetime (in seconds) for vended credentials. '0' disables caching.", 0) \
     DECLARE(String, auth_scope, "PRINCIPAL_ROLE:ALL", "Authorization scope for client credentials or token exchange", 0) \
@@ -121,13 +122,23 @@ SettingsChanges DatabaseDataLakeSettings::allChanged() const
     return changes;
 }
 
-const String & DatabaseDataLakeSettings::getSettingName(DatabaseDataLakeSettingsString setting)
+static const String & getSettingNameByOffset(size_t offset)
 {
     const auto & accessor = DatabaseDataLakeSettingsTraits::Accessor::instance();
-    const size_t index = accessor.findByOffset(setting.offset);
+    const size_t index = accessor.findByOffset(offset);
     if (index == static_cast<size_t>(-1))
         throw Exception(ErrorCodes::LOGICAL_ERROR, "Unknown database DataLake setting");
     return accessor.getName(index);
 }
 
+const String & DatabaseDataLakeSettings::getSettingName(DatabaseDataLakeSettingsString setting)
+{
+    return getSettingNameByOffset(setting.offset);
+}
+
+const String & DatabaseDataLakeSettings::getSettingName(DatabaseDataLakeSettingsBool setting)
+{
+    return getSettingNameByOffset(setting.offset);
+}
+
 }

src/Databases/DataLake/DatabaseDataLakeSettings.h

@@ -75,6 +75,7 @@ struct DatabaseDataLakeSettings
     /// Name of the setting referenced by its `DatabaseDataLakeSetting::*` index,
     /// so catalog code can match `SettingsChanges` entries without magic strings.
     static const String & getSettingName(DatabaseDataLakeSettingsString setting);
+    static const String & getSettingName(DatabaseDataLakeSettingsBool setting);
 
 private:
     std::unique_ptr<DatabaseDataLakeSettingsImpl> impl;

src/Databases/DataLake/GlueCatalog.cpp

@@ -786,7 +786,7 @@ void GlueCatalog::setCredentials(const AuthenticatedClient & client, TableMetada
 }
 
 ICatalog::CredentialsRefreshCallback GlueCatalog::getCredentialsConfigurationCallback(
-    const DB::StorageID & storage_id, const DB::ForwardedAuthTokenPtr & auth_token)
+    const DB::StorageID & storage_id, const TableMetadata & /*table_metadata*/, const DB::ForwardedAuthTokenPtr & auth_token)
 {
     /// The AWS SDK credentials provider chain (instance profile, STS assume-role,
     /// web-identity, etc.) refreshes its cached credentials internally before

src/Databases/DataLake/GlueCatalog.h

@@ -109,7 +109,7 @@ class GlueCatalog final : public ICatalog, private DB::WithContext
     /// fails with `ExpiredToken`, so that a long-running read can recover without
     /// the user having to restart the query.
     ICatalog::CredentialsRefreshCallback getCredentialsConfigurationCallback(
-        const DB::StorageID & storage_id, const DB::ForwardedAuthTokenPtr & auth_token) override;
+        const DB::StorageID & storage_id, const TableMetadata & table_metadata, const DB::ForwardedAuthTokenPtr & auth_token) override;
 
     /// Resolves the precise Iceberg timestamp type for `column_name` by searching the current schema
     /// in the Iceberg `metadata_object`. Falls back to `"timestamp_ns"` when `glue_column_type` is

src/Databases/DataLake/HTTPBasedCatalogUtils.cpp

@@ -5,6 +5,10 @@
 #include <Core/Types.h>
 #include <Common/FailPoint.h>
 #include <Common/HTTPHeaderFilter.h>
+#include <Common/RemoteHostFilter.h>
+#include <IO/ConnectionTimeouts.h>
+#include <Poco/StreamCopier.h>
+#include <Poco/Net/HTTPRequest.h>
 
 namespace DB::ErrorCodes
 {
@@ -101,5 +105,54 @@ std::pair<Poco::Dynamic::Var, std::string> makeHTTPRequestAndReadJSON(
     }
 }
 
+AccessToken requestOAuthToken(const DB::ContextPtr & context, const Poco::URI & url, const std::string & body)
+{
+    context->getRemoteHostFilter().checkHostAndPort(url.getHost(), std::to_string(url.getPort()));
+    auto timeouts = DB::ConnectionTimeouts::getHTTPTimeouts(context->getSettingsRef(), context->getServerSettings());
+    auto session = makeHTTPSession(DB::HTTPConnectionGroupType::HTTP, url, timeouts, {});
+
+    Poco::Net::HTTPRequest request(Poco::Net::HTTPRequest::HTTP_POST, url.getPathAndQuery(), Poco::Net::HTTPMessage::HTTP_1_1);
+    request.setContentType("application/x-www-form-urlencoded");
+    request.setContentLength(body.size());
+    request.set("Accept", "application/json");
+
+    session->sendRequest(request) << body;
+
+    Poco::Net::HTTPResponse response;
+    std::istream & rs = session->receiveResponse(response);
+
+    std::string json_str;
+    Poco::StreamCopier::copyToString(rs, json_str);
+
+    /// The body of a failed response is an OAuth error object, safe to show.
+    if (response.getStatus() != Poco::Net::HTTPResponse::HTTP_OK)
+        throw DB::Exception(
+            DB::ErrorCodes::DATALAKE_DATABASE_ERROR,
+            "OAuth token request failed with status {} ({}): {}",
+            static_cast<int>(response.getStatus()), response.getReason(), json_str);
+
+    Poco::JSON::Parser parser;
+    Poco::Dynamic::Var res_json = parser.parse(json_str);
+    const Poco::JSON::Object::Ptr & object = res_json.extract<Poco::JSON::Object::Ptr>();
+
+    if (!object->has("access_token"))
+        throw DB::Exception(
+            DB::ErrorCodes::DATALAKE_DATABASE_ERROR,
+            "OAuth token response has no `access_token` field: {}",
+            json_str);
+
+    AccessToken token;
+    token.token = object->getValue<String>("access_token");
+
+    if (object->has("expires_in"))
+    {
+        Int64 expires_in = object->getValue<Int64>("expires_in");
+        /// Use 90% of the token lifetime as the validity window so that short-lived tokens
+        /// (e.g. expires_in=300) still get a sensible buffer instead of going non-positive.
+        token.expires_at = std::chrono::system_clock::now() + std::chrono::seconds(expires_in * 9 / 10);
+    }
+
+    return token;
+}
 
 }