Native CREATE TABLE / DROP TABLE for DataLakeCatalog; optionally prune on DROP

src/Client/BuzzHouse/Generator/SessionSettings.cpp

@@ -873,7 +873,7 @@ std::unordered_map<String, CHSetting> serverSettings = {
          [](RandomGenerator & rg, FuzzConfig &) { return std::to_string(rg.thresholdGenerator<uint64_t>(0.3, 0.2, 0, 10800)); },
          {},
          false)},
-    {"iceberg_delete_data_on_drop", trueOrFalseSettingNoOracle},
+    {"data_lake_delete_data_on_drop", trueOrFalseSettingNoOracle},
     {"iceberg_expire_default_min_snapshots_to_keep",
      CHSetting(
          [](RandomGenerator & rg, FuzzConfig &) { return std::to_string(rg.thresholdGenerator<uint64_t>(0.2, 0.2, 0, 10)); }, {}, false)},

src/Core/Settings.cpp

@@ -5551,9 +5551,9 @@ Possible values:
 - manifest_file_entry - Everything above + traversed avro manifest files entries.
 )", 0) \
     \
-    DECLARE(Bool, iceberg_delete_data_on_drop, false, R"(
-Whether to delete all iceberg files on drop or not.
-)", 0) \
+    DECLARE_WITH_ALIAS(Bool, data_lake_delete_data_on_drop, false, R"(
+Whether to delete the underlying data files when dropping a data lake table. For catalog databases the catalog is asked to purge the data (`purgeRequested=true`); for self-managed tables ClickHouse removes the files directly.
+)", 0, iceberg_delete_data_on_drop) \
     DECLARE(Int64, iceberg_expire_default_min_snapshots_to_keep, 1, R"(
 Default value for Iceberg table property `history.expire.min-snapshots-to-keep` used by `expire_snapshots` when that property is absent.
 )", 0) \

src/Core/SettingsChangesHistory.cpp

@@ -42,6 +42,7 @@ const VersionToSettingsChangesMap & getSettingsChangesHistory()
         addSettingsChanges(settings_changes_history, "26.6.2.20001.altinityantalya",
         {
             {"use_puffin_files_cache", false, true, "Enables cache of parsed Puffin file content such as deletion vectors."},
+            {"data_lake_delete_data_on_drop", false, false, "New setting that unifies dropping of data lake data; the released `iceberg_delete_data_on_drop` is kept as an alias for it."},
         });
 
         addSettingsChanges(settings_changes_history, "26.6",

src/Databases/DataLake/Common.cpp

@@ -16,6 +16,9 @@
 
 #include <Storages/ObjectStorage/DataLakes/Iceberg/SchemaProcessor.h>
 
+#include <fmt/format.h>
+#include <Poco/URI.h>
+
 namespace DB::ErrorCodes
 {
 extern const int BAD_ARGUMENTS;
@@ -121,4 +124,91 @@ std::pair<std::string, std::string> parseTableName(const std::string & name)
     return {namespace_name, table_name};
 }
 
+String constructTableLocation(
+    const String & location_scheme,
+    const String & storage_endpoint,
+    const String & namespace_name,
+    const String & table_name,
+    DB::S3UriStyle uri_style)
+{
+    Poco::URI uri(storage_endpoint);
+    auto path = uri.getPath();
+    while (path.starts_with('/'))
+        path.erase(0, 1);
+    while (path.ends_with('/'))
+        path.pop_back();
+
+    if (location_scheme == "abfss")
+    {
+        /// Azure: `abfss://<container>@<host>/<path>`. `storage_endpoint` is
+        /// `https://<host>/<container>/<extra>` or `abfss://<container>@<host>/<extra>`
+        String container = uri.getUserInfo();
+        String account_host = uri.getHost();
+        String extra_path = path;
+
+        if (container.empty())
+        {
+            auto first_slash = extra_path.find('/');
+            if (first_slash == String::npos)
+            {
+                container = std::move(extra_path);
+                extra_path.clear();
+            }
+            else
+            {
+                container = extra_path.substr(0, first_slash);
+                extra_path = extra_path.substr(first_slash + 1);
+            }
+        }
+
+        if (account_host.empty() || container.empty())
+            throw DB::Exception(
+                DB::ErrorCodes::BAD_ARGUMENTS,
+                "`storage_endpoint` ({}) for Azure must include both account host and container "
+                "(expected https://<account>.dfs.core.windows.net/<container>[/<sub-path>] or "
+                "abfss://<container>@<account>.dfs.core.windows.net[/<sub-path>])",
+                storage_endpoint);
+
+        if (extra_path.empty())
+            return fmt::format("abfss://{}@{}/{}/{}", container, account_host, namespace_name, table_name);
+        return fmt::format("abfss://{}@{}/{}/{}/{}", container, account_host, extra_path, namespace_name, table_name);
+    }
+
+    if (location_scheme == "s3")
+    {
+        if (uri_style == DB::S3UriStyle::VIRTUAL_HOSTED)
+            /// A virtual-hosted host cannot be split into bucket and service unambiguously
+            /// (`s3.us-east-1.amazonaws.com`, an IP host and a dotted bucket name all look alike).
+            throw DB::Exception(
+                DB::ErrorCodes::BAD_ARGUMENTS,
+                "CREATE TABLE with `storage_uri_style = 'virtual_hosted'` cannot derive the bucket from "
+                "`storage_endpoint` ({}); set `default_base_location` (a full s3://<bucket>/<prefix>) instead.",
+                storage_endpoint);
+
+        if (path.empty())
+            throw DB::Exception(
+                DB::ErrorCodes::BAD_ARGUMENTS,
+                "`storage_endpoint` ({}) does not contain a bucket; "
+                "CREATE TABLE in DataLakeCatalog requires `storage_endpoint` to include a non-empty bucket path.",
+                storage_endpoint);
+        return fmt::format("s3://{}/{}/{}", path, namespace_name, table_name);
+    }
+
+    /// HDFS / file / other schemes that may have `authority`.
+    String authority = uri.getAuthority();
+    if (authority.empty())
+    {
+        if (path.empty())
+            throw DB::Exception(
+                DB::ErrorCodes::BAD_ARGUMENTS,
+                "`storage_endpoint` ({}) does not contain a path",
+                storage_endpoint);
+        return fmt::format("{}:///{}/{}/{}", location_scheme, path, namespace_name, table_name);
+    }
+
+    if (path.empty())
+        return fmt::format("{}://{}/{}/{}", location_scheme, authority, namespace_name, table_name);
+    return fmt::format("{}://{}/{}/{}/{}", location_scheme, authority, path, namespace_name, table_name);
+}
+
 }

src/Databases/DataLake/Common.h

@@ -1,12 +1,39 @@
 #pragma once
 
 #include <Core/NamesAndTypes.h>
+#include <Core/SettingsEnums.h>
 #include <Core/Types.h>
 #include <Interpreters/Context_fwd.h>
+#include <Common/Exception.h>
+
+namespace DB::ErrorCodes
+{
+extern const int TABLE_ALREADY_EXISTS;
+}
 
 namespace DataLake
 {
 
+/// Thrown by a `CREATE TABLE` that registered nothing in the data lake catalog: the table name, or the
+/// location the table would use, is already taken. `InterpreterCreateQuery` matches on this type rather
+/// than on its `TABLE_ALREADY_EXISTS` code, which would also swallow unrelated exceptions from below.
+class TableAlreadyExistsInCatalogException : public DB::Exception
+{
+public:
+    template <typename... Args>
+    explicit TableAlreadyExistsInCatalogException(FormatStringHelper<Args...> fmt, Args &&... args)
+        : DB::Exception(DB::ErrorCodes::TABLE_ALREADY_EXISTS, std::move(fmt), std::forward<Args>(args)...)
+    {
+    }
+
+    TableAlreadyExistsInCatalogException * clone() const override { return new TableAlreadyExistsInCatalogException(*this); }
+    void rethrow() const override { throw *this; } /// NOLINT(bugprone-exception-copy-constructor-throws,cert-err60-cpp)
+
+private:
+    const char * name() const noexcept override { return "DataLake::TableAlreadyExistsInCatalogException"; }
+    const char * className() const noexcept override { return "DataLake::TableAlreadyExistsInCatalogException"; }
+};
+
 String trim(const String & str);
 
 std::vector<String> splitTypeArguments(const String & type_str);
@@ -19,4 +46,11 @@ DB::DataTypePtr getType(const String & type_name, bool nullable, DB::ContextPtr
 /// `E` is a table name.
 std::pair<std::string, std::string> parseTableName(const std::string & name);
 
+String constructTableLocation(
+    const String & location_scheme,
+    const String & storage_endpoint,
+    const String & namespace_name,
+    const String & table_name,
+    DB::S3UriStyle uri_style = DB::S3UriStyle::AUTO);
+
 }