diff --git a/src/Common/ProfileEvents.cpp b/src/Common/ProfileEvents.cpp
index 7f35ef869e92..6032f1d7fda9 100644
--- a/src/Common/ProfileEvents.cpp
+++ b/src/Common/ProfileEvents.cpp
@@ -1503,6 +1503,51 @@ The server successfully detected this situation and will download merged part fr
     M(AIRowsProcessed, "Number of rows that received an AI result.", ValueType::Number) \
     M(AIRowsSkipped, "Number of rows that received a default value due to quota or error.", ValueType::Number) \
     \
+    M(DataLakeRestCatalogLoadConfig, "Number of 'load config' requests to Iceberg REST catalog.", ValueType::Number) \
+    M(DataLakeRestCatalogLoadConfigMicroseconds, "Total time of 'load config' requests to Iceberg REST catalog.", ValueType::Microseconds) \
+    M(DataLakeRestCatalogGetNamespaces, "Number of 'get namespaces' requests to Iceberg REST catalog.", ValueType::Number) \
+    M(DataLakeRestCatalogGetNamespacesMicroseconds, "Total time of 'get namespaces' requests to Iceberg REST catalog.", ValueType::Microseconds) \
+    M(DataLakeRestCatalogGetTables, "Number of 'get tables' requests to Iceberg REST catalog.", ValueType::Number) \
+    M(DataLakeRestCatalogGetTablesMicroseconds, "Total time of 'get tables' requests to Iceberg REST catalog.", ValueType::Microseconds) \
+    M(DataLakeRestCatalogGetTableMetadata, "Number of 'get table metadata' requests to Iceberg REST catalog.", ValueType::Number) \
+    M(DataLakeRestCatalogGetTableMetadataMicroseconds, "Total time of 'get table metadata' requests to Iceberg REST catalog.", ValueType::Microseconds) \
+    M(DataLakeRestCatalogGetCredentials, "Number of 'get credentials' requests to Iceberg REST catalog.", ValueType::Number) \
+    M(DataLakeRestCatalogGetCredentialsMicroseconds, "Total time of 'get credentials' requests to Iceberg REST catalog.", ValueType::Microseconds) \
+    M(DataLakeRestCatalogCreateNamespace, "Number of 'create namespace' requests to Iceberg REST catalog.", ValueType::Number) \
+    M(DataLakeRestCatalogCreateNamespaceMicroseconds, "Total time of 'create namespace' requests to Iceberg REST catalog.", ValueType::Microseconds) \
+    M(DataLakeRestCatalogCreateTable, "Number of 'create table' requests to Iceberg REST catalog.", ValueType::Number) \
+    M(DataLakeRestCatalogCreateTableMicroseconds, "Total time of 'create table' requests to Iceberg REST catalog.", ValueType::Microseconds) \
+    M(DataLakeRestCatalogUpdateTable, "Number of 'update table' requests to Iceberg REST catalog.", ValueType::Number) \
+    M(DataLakeRestCatalogUpdateTableMicroseconds, "Total time of 'update table' requests to Iceberg REST catalog.", ValueType::Microseconds) \
+    M(DataLakeRestCatalogDropTable, "Number of 'drop table' requests to Iceberg REST catalog.", ValueType::Number) \
+    M(DataLakeRestCatalogDropTableMicroseconds, "Total time of 'drop table' requests to Iceberg REST catalog.", ValueType::Microseconds) \
+    \
+    M(DataLakeGlueCatalogGetDatabases, "Number of 'get databases' requests to Iceberg Glue catalog.", ValueType::Number) \
+    M(DataLakeGlueCatalogGetDatabasesMicroseconds, "Total time of 'get databases' requests to Iceberg Glue catalog.", ValueType::Microseconds) \
+    M(DataLakeGlueCatalogGetTables, "Number of 'get tables' requests to Iceberg Glue catalog.", ValueType::Number) \
+    M(DataLakeGlueCatalogGetTablesMicroseconds, "Total time of 'get tables' requests to Iceberg Glue catalog.", ValueType::Microseconds) \
+    M(DataLakeGlueCatalogGetTable, "Number of 'get table' requests to Iceberg Glue catalog.", ValueType::Number) \
+    M(DataLakeGlueCatalogGetTableMicroseconds, "Total time of 'get table' requests to Iceberg Glue catalog.", ValueType::Microseconds) \
+    M(DataLakeGlueCatalogCreateDatabase, "Number of 'create database' requests to Iceberg Glue catalog.", ValueType::Number) \
+    M(DataLakeGlueCatalogCreateDatabaseMicroseconds, "Total time of 'create database' requests to Iceberg Glue catalog.", ValueType::Microseconds) \
+    M(DataLakeGlueCatalogCreateTable, "Number of 'create table' requests to Iceberg Glue catalog.", ValueType::Number) \
+    M(DataLakeGlueCatalogCreateTableMicroseconds, "Total time of 'create table' requests to Iceberg Glue catalog.", ValueType::Microseconds) \
+    M(DataLakeGlueCatalogUpdateTable, "Number of 'update table' requests to Iceberg Glue catalog.", ValueType::Number) \
+    M(DataLakeGlueCatalogUpdateTableMicroseconds, "Total time of 'update table' requests to Iceberg Glue catalog.", ValueType::Microseconds) \
+    M(DataLakeGlueCatalogDropTable, "Number of 'drop table' requests to Iceberg Glue catalog.", ValueType::Number) \
+    M(DataLakeGlueCatalogDropTableMicroseconds, "Total time of 'drop table' requests to Iceberg Glue catalog.", ValueType::Microseconds) \
+    \
+    M(DataLakeUnityCatalogGetTables, "Number of 'get tables' requests to Iceberg Unity catalog.", ValueType::Number) \
+    M(DataLakeUnityCatalogGetTablesMicroseconds, "Total time of 'get tables' requests to Iceberg Unity catalog.", ValueType::Microseconds) \
+    M(DataLakeUnityCatalogGetTable, "Number of 'get table' requests to Iceberg Unity catalog.", ValueType::Number) \
+    M(DataLakeUnityCatalogGetTableMicroseconds, "Total time of 'get table' requests to Iceberg Unity catalog.", ValueType::Microseconds) \
+    M(DataLakeUnityCatalogGetTableMetadata, "Number of 'get table metadata' requests to Iceberg Unity catalog.", ValueType::Number) \
+    M(DataLakeUnityCatalogGetTableMetadataMicroseconds, "Total time of 'get table metadata' requests to Iceberg Unity catalog.", ValueType::Microseconds) \
+    M(DataLakeUnityCatalogGetSchemas, "Number of 'get schemas' requests to Iceberg Unity catalog.", ValueType::Number) \
+    M(DataLakeUnityCatalogGetSchemasMicroseconds, "Total time of 'get schemas' requests to Iceberg Unity catalog.", ValueType::Microseconds) \
+    M(DataLakeUnityCatalogGetCredentials, "Number of 'get credentials' requests to Iceberg Unity catalog.", ValueType::Number) \
+    M(DataLakeUnityCatalogGetCredentialsMicroseconds, "Total time of 'get credentials' requests to Iceberg Unity catalog.", ValueType::Microseconds) \
+
 
 #ifdef APPLY_FOR_EXTERNAL_EVENTS
     #define APPLY_FOR_EVENTS(M) APPLY_FOR_BUILTIN_EVENTS(M) APPLY_FOR_EXTERNAL_EVENTS(M)
diff --git a/src/Core/Settings.cpp b/src/Core/Settings.cpp
index 7f5f7a2a00ff..503215f90e13 100644
--- a/src/Core/Settings.cpp
+++ b/src/Core/Settings.cpp
@@ -8321,6 +8321,9 @@ Multiple algorithms can be specified as a comma-separated list, e.g. `dphyp,gree
     )", EXPERIMENTAL) \
     DECLARE(Bool, allow_experimental_database_paimon_rest_catalog, false, R"(
 Allow experimental database engine DataLakeCatalog with catalog_type = 'paimon_rest'
+)", EXPERIMENTAL) \
+    DECLARE(Bool, allow_experimental_database_s3_tables, false, R"(
+Allow experimental database engine DataLakeCatalog with catalog_type = 's3tables' (Amazon S3 Tables Iceberg REST with SigV4)
 )", EXPERIMENTAL) \
     DECLARE(UInt64, webassembly_udf_max_fuel, 100'000, R"(
 Fuel limit per WebAssembly UDF instance execution. Each WebAssembly instruction consumes some amount of fuel. The value is scaled by 1024 before being passed to the runtime, so `webassembly_udf_max_fuel = 1` corresponds to approximately 1024 fuel units. Set to 0 for no finite limit. Applies only to functions whose per-function setting `webassembly_udf_enable_fuel` is true, which is the default.
diff --git a/src/Core/SettingsChangesHistory.cpp b/src/Core/SettingsChangesHistory.cpp
index c2b62c209863..e6ab2a70b36a 100644
--- a/src/Core/SettingsChangesHistory.cpp
+++ b/src/Core/SettingsChangesHistory.cpp
@@ -86,6 +86,7 @@ const VersionToSettingsChangesMap & getSettingsChangesHistory()
             {"allow_experimental_query_deduplication", false, false, "The setting is obsolete, the feature has been removed."},
             {"query_plan_min_columns_for_join_lazy_indexing", 0, 3, "Control the minimum number of payload columns from the left side required for enabling lazy indexing optimization in JOIN"},
             {"query_plan_max_limit_for_join_lazy_indexing", 1000, 1000, "Added new setting to control maximum limit value that allows to use query plan for lazy join indexing optimization. If zero, there is no limit"},
+            {"allow_experimental_database_s3_tables", false, false, "New setting to enable experimental database S3 tables (AWS Iceberg REST catalog)."},
         });
 
         addSettingsChanges(settings_changes_history, "26.5",
diff --git a/src/Core/SettingsEnums.cpp b/src/Core/SettingsEnums.cpp
index 0ff8daf7faa0..d498bd5a9db6 100644
--- a/src/Core/SettingsEnums.cpp
+++ b/src/Core/SettingsEnums.cpp
@@ -359,7 +359,8 @@ IMPLEMENT_SETTING_ENUM(
      {"hive", DatabaseDataLakeCatalogType::ICEBERG_HIVE},
      {"onelake", DatabaseDataLakeCatalogType::ICEBERG_ONELAKE},
      {"biglake", DatabaseDataLakeCatalogType::ICEBERG_BIGLAKE},
-     {"paimon_rest", DatabaseDataLakeCatalogType::PAIMON_REST}})
+     {"paimon_rest", DatabaseDataLakeCatalogType::PAIMON_REST},
+     {"s3tables", DatabaseDataLakeCatalogType::S3_TABLES}})
 
 IMPLEMENT_SETTING_ENUM(
     FileCachePolicy,
diff --git a/src/Core/SettingsEnums.h b/src/Core/SettingsEnums.h
index 09d175e72680..e6a9b2b6a53b 100644
--- a/src/Core/SettingsEnums.h
+++ b/src/Core/SettingsEnums.h
@@ -442,6 +442,7 @@ enum class DatabaseDataLakeCatalogType : uint8_t
     ICEBERG_ONELAKE,
     ICEBERG_BIGLAKE,
     PAIMON_REST,
+    S3_TABLES,
 };
 
 DECLARE_SETTING_ENUM(DatabaseDataLakeCatalogType)
diff --git a/src/Databases/DataLake/AWSV4Signer.cpp b/src/Databases/DataLake/AWSV4Signer.cpp
new file mode 100644
index 000000000000..f5bd8c4bc6ed
--- /dev/null
+++ b/src/Databases/DataLake/AWSV4Signer.cpp
@@ -0,0 +1,110 @@
+#include "config.h"
+
+#if USE_AVRO && USE_SSL && USE_AWS_S3
+
+#include <Databases/DataLake/AWSV4Signer.h>
+
+#include <Common/Exception.h>
+#include <Poco/Net/HTTPRequest.h>
+#include <Poco/String.h>
+
+#include <aws/core/auth/signer/AWSAuthV4Signer.h>
+#include <aws/core/http/standard/StandardHttpRequest.h>
+#include <aws/core/http/URI.h>
+#include <aws/core/utils/memory/AWSMemory.h>
+
+#include <sstream>
+#include <utility>
+
+namespace DB
+{
+namespace ErrorCodes
+{
+    extern const int BAD_ARGUMENTS;
+    extern const int S3_ERROR;
+}
+}
+
+namespace DataLake
+{
+namespace
+{
+
+Aws::Http::HttpMethod mapPocoMethodToAws(const String & method)
+{
+    using Aws::Http::HttpMethod;
+    using Poco::Net::HTTPRequest;
+
+    static const std::pair<String, HttpMethod> supported_methods[] = {
+        {HTTPRequest::HTTP_GET, HttpMethod::HTTP_GET},
+        {HTTPRequest::HTTP_POST, HttpMethod::HTTP_POST},
+        {HTTPRequest::HTTP_PUT, HttpMethod::HTTP_PUT},
+        {HTTPRequest::HTTP_DELETE, HttpMethod::HTTP_DELETE},
+        {HTTPRequest::HTTP_HEAD, HttpMethod::HTTP_HEAD},
+        {HTTPRequest::HTTP_PATCH, HttpMethod::HTTP_PATCH},
+    };
+
+    for (const auto & [poco_method, aws_method] : supported_methods)
+        if (method == poco_method)
+            return aws_method;
+
+    throw DB::Exception(DB::ErrorCodes::BAD_ARGUMENTS, "Unsupported HTTP method for AWS SigV4 signing: {}", method);
+}
+
+}
+
+void signRequestWithAWSV4(
+    const String & method,
+    const Poco::URI & uri,
+    const DB::HTTPHeaderEntries & extra_headers,
+    const String & payload,
+    Aws::Client::AWSAuthV4Signer & signer,
+    const String & region,
+    const String & service,
+    DB::HTTPHeaderEntries & out_headers)
+{
+    const Aws::Http::URI aws_uri(uri.toString().c_str());
+    Aws::Http::Standard::StandardHttpRequest request(aws_uri, mapPocoMethodToAws(method));
+
+    for (const auto & h : extra_headers)
+    {
+        if (Poco::icompare(h.name, "authorization") == 0)
+            continue;
+        request.SetHeaderValue(Aws::String(h.name.c_str(), h.name.size()), Aws::String(h.value.c_str(), h.value.size()));
+    }
+
+    if (!payload.empty())
+    {
+        auto body_stream = Aws::MakeShared<std::stringstream>("AWSV4Signer");
+        body_stream->write(payload.data(), static_cast<std::streamsize>(payload.size()));
+        body_stream->seekg(0);
+        request.AddContentBody(body_stream);
+    }
+
+    static constexpr bool sign_body = true;
+    if (!signer.SignRequest(request, region.c_str(), service.c_str(), sign_body))
+        throw DB::Exception(DB::ErrorCodes::S3_ERROR, "AWS SigV4 signing failed");
+
+    bool has_authorization = false;
+    for (const auto & [key, value] : request.GetHeaders())
+    {
+        if (Poco::icompare(key, "authorization") == 0 && !value.empty())
+            has_authorization = true;
+    }
+    if (!has_authorization)
+        throw DB::Exception(
+            DB::ErrorCodes::BAD_ARGUMENTS,
+            "AWS credentials are missing or incomplete; cannot sign S3 Tables REST request");
+
+    out_headers.clear();
+    for (const auto & [key, value] : request.GetHeaders())
+    {
+        if (Poco::icompare(key, "host") == 0)
+            continue;
+        out_headers.emplace_back(String(key.c_str(), key.size()), String(value.c_str(), value.size()));
+    }
+}
+
+}
+
+#endif
diff --git a/src/Databases/DataLake/AWSV4Signer.h b/src/Databases/DataLake/AWSV4Signer.h
new file mode 100644
index 000000000000..cdc42adaca5f
--- /dev/null
+++ b/src/Databases/DataLake/AWSV4Signer.h
@@ -0,0 +1,34 @@
+#pragma once
+
+#include "config.h"
+
+#if USE_AVRO && USE_SSL && USE_AWS_S3
+
+#include <Core/Types.h>
+#include <IO/HTTPHeaderEntries.h>
+#include <Poco/URI.h>
+
+namespace Aws::Client
+{
+class AWSAuthV4Signer;
+}
+
+namespace DataLake
+{
+
+/// Sign a Poco-style HTTP request using the AWS SDK's AWSAuthV4Signer.
+/// Builds a temporary Aws::Http::StandardHttpRequest, signs it, then extracts
+/// the resulting headers into out_headers (excluding Host; ReadWriteBufferFromHTTP sets it from the URI).
+void signRequestWithAWSV4(
+    const String & method,
+    const Poco::URI & uri,
+    const DB::HTTPHeaderEntries & extra_headers,
+    const String & payload,
+    Aws::Client::AWSAuthV4Signer & signer,
+    const String & region,
+    const String & service,
+    DB::HTTPHeaderEntries & out_headers);
+
+}
+
+#endif
diff --git a/src/Databases/DataLake/DatabaseDataLake.cpp b/src/Databases/DataLake/DatabaseDataLake.cpp
index 12fbb051ba4a..37b9bd2cf8f0 100644
--- a/src/Databases/DataLake/DatabaseDataLake.cpp
+++ b/src/Databases/DataLake/DatabaseDataLake.cpp
@@ -27,6 +27,9 @@
 #include <Databases/DataLake/RestCatalog.h>
 #include <Databases/DataLake/GlueCatalog.h>
 #include <Databases/DataLake/PaimonRestCatalog.h>
+#if USE_AWS_S3 && USE_SSL
+#include <Databases/DataLake/S3TablesCatalog.h>
+#endif
 #include <DataTypes/DataTypeString.h>
 
 #include <Storages/ObjectStorage/S3/Configuration.h>
@@ -93,6 +96,7 @@ namespace Setting
     extern const SettingsBool allow_experimental_database_glue_catalog;
     extern const SettingsBool allow_experimental_database_hms_catalog;
     extern const SettingsBool allow_experimental_database_paimon_rest_catalog;
+    extern const SettingsBool allow_experimental_database_s3_tables;
     extern const SettingsBool use_hive_partitioning;
     extern const SettingsBool log_queries;
     extern const SettingsBool parallel_replicas_for_cluster_engines;
@@ -155,9 +159,21 @@ void DatabaseDataLake::validateSettings()
     {
         if (settings[DatabaseDataLakeSetting::region].value.empty())
             throw Exception(
-                ErrorCodes::BAD_ARGUMENTS, "`region` setting cannot be empty for Glue Catalog. "
+                ErrorCodes::BAD_ARGUMENTS, "`region` setting cannot be empty for Glue catalog. "
                 "Please specify 'SETTINGS region=<region_name>' in the CREATE DATABASE query");
     }
+    else if (settings[DatabaseDataLakeSetting::catalog_type].value == DB::DatabaseDataLakeCatalogType::S3_TABLES)
+    {
+        if (settings[DatabaseDataLakeSetting::region].value.empty())
+            throw Exception(
+                ErrorCodes::BAD_ARGUMENTS, "`region` setting cannot be empty for S3 Tables catalog. "
+                "Please specify 'SETTINGS region=<region_name>' in the CREATE DATABASE query");
+
+        if (settings[DatabaseDataLakeSetting::warehouse].value.empty())
+            throw Exception(
+                ErrorCodes::BAD_ARGUMENTS, "`warehouse` setting cannot be empty for S3 Tables catalog. "
+                "Please specify 'SETTINGS warehouse=<table_bucket_arn>' in the CREATE DATABASE query");
+    }
     else if (settings[DatabaseDataLakeSetting::warehouse].value.empty())
     {
         throw Exception(
@@ -310,6 +326,22 @@ void DatabaseDataLake::initialize() const
             }
             break;
         }
+        case DB::DatabaseDataLakeCatalogType::S3_TABLES:
+        {
+#if USE_AWS_S3 && USE_SSL
+            catalog_impl = std::make_shared<DataLake::S3TablesCatalog>(
+                settings[DatabaseDataLakeSetting::warehouse].value,
+                url,
+                settings[DatabaseDataLakeSetting::region].value,
+                catalog_parameters,
+                Context::getGlobalContextInstance());
+#else
+            throw Exception(
+                ErrorCodes::SUPPORT_IS_DISABLED,
+                "Amazon S3 Tables catalog requires ClickHouse built with USE_AWS_S3 and USE_SSL");
+#endif
+            break;
+        }
     }
 }
 
@@ -350,6 +382,7 @@ std::shared_ptr<StorageObjectStorageConfiguration> DatabaseDataLake::getConfigur
         case DatabaseDataLakeCatalogType::ICEBERG_HIVE:
         case DatabaseDataLakeCatalogType::ICEBERG_REST:
         case DatabaseDataLakeCatalogType::ICEBERG_BIGLAKE:
+        case DatabaseDataLakeCatalogType::S3_TABLES:
         {
             switch (type)
             {
@@ -1138,6 +1171,19 @@ void registerDatabaseDataLake(DatabaseFactory & factory)
                 engine_func->name = "Paimon";
                 break;
             }
+            case DatabaseDataLakeCatalogType::S3_TABLES:
+            {
+                if (!args.create_query.attach
+                    && !args.context->getSettingsRef()[Setting::allow_experimental_database_s3_tables])
+                {
+                    throw Exception(ErrorCodes::SUPPORT_IS_DISABLED,
+                                    "DatabaseDataLake with S3 Tables catalog is experimental. "
+                                    "To allow its usage, enable setting allow_experimental_database_s3_tables");
+                }
+
+                engine_func->name = "Iceberg";
+                break;
+            }
             case DatabaseDataLakeCatalogType::NONE:
                 break;
         }
diff --git a/src/Databases/DataLake/GlueCatalog.cpp b/src/Databases/DataLake/GlueCatalog.cpp
index 53ac171c79ff..ca6555aa6f39 100644
--- a/src/Databases/DataLake/GlueCatalog.cpp
+++ b/src/Databases/DataLake/GlueCatalog.cpp
@@ -20,6 +20,7 @@
 
 #include <Common/Exception.h>
 #include <Common/CurrentMetrics.h>
+#include <Common/CurrentThread.h>
 #include <Core/Settings.h>
 #include <Interpreters/Context.h>
 
@@ -83,6 +84,24 @@ namespace DB::ServerSetting
     extern const ServerSettingsUInt64 s3_retry_attempts;
 }
 
+namespace ProfileEvents
+{
+    extern const Event DataLakeGlueCatalogGetDatabases;
+    extern const Event DataLakeGlueCatalogGetDatabasesMicroseconds;
+    extern const Event DataLakeGlueCatalogGetTables;
+    extern const Event DataLakeGlueCatalogGetTablesMicroseconds;
+    extern const Event DataLakeGlueCatalogGetTable;
+    extern const Event DataLakeGlueCatalogGetTableMicroseconds;
+    extern const Event DataLakeGlueCatalogCreateDatabase;
+    extern const Event DataLakeGlueCatalogCreateDatabaseMicroseconds;
+    extern const Event DataLakeGlueCatalogCreateTable;
+    extern const Event DataLakeGlueCatalogCreateTableMicroseconds;
+    extern const Event DataLakeGlueCatalogUpdateTable;
+    extern const Event DataLakeGlueCatalogUpdateTableMicroseconds;
+    extern const Event DataLakeGlueCatalogDropTable;
+    extern const Event DataLakeGlueCatalogDropTableMicroseconds;
+}
+
 namespace CurrentMetrics
 {
     extern const Metric MarkCacheBytes;
@@ -191,7 +210,14 @@ DataLake::ICatalog::Namespaces GlueCatalog::getDatabases(const std::string & pre
     do
     {
         request.SetNextToken(next_token);
-        auto outcome = glue_client->GetDatabases(request);
+
+        Aws::Glue::Model::GetDatabasesOutcome outcome;
+        {
+            ProfileEvents::increment(ProfileEvents::DataLakeGlueCatalogGetDatabases);
+            auto timer = DB::CurrentThread::getProfileEvents().timer(ProfileEvents::DataLakeGlueCatalogGetDatabasesMicroseconds);
+            outcome = glue_client->GetDatabases(request);
+        }
+
         if (outcome.IsSuccess())
         {
             const auto & databases_result = outcome.GetResult();
@@ -240,7 +266,12 @@ DB::Names GlueCatalog::getTablesForDatabase(const std::string & db_name, size_t
     do
     {
         request.SetNextToken(next_token);
-        auto outcome = glue_client->GetTables(request);
+        Aws::Glue::Model::GetTablesOutcome outcome;
+        {
+            ProfileEvents::increment(ProfileEvents::DataLakeGlueCatalogGetTables);
+            auto timer = DB::CurrentThread::getProfileEvents().timer(ProfileEvents::DataLakeGlueCatalogGetTablesMicroseconds);
+            outcome = glue_client->GetTables(request);
+        }
         if (outcome.IsSuccess())
         {
             const auto & tables_result = outcome.GetResult();
@@ -290,6 +321,8 @@ bool GlueCatalog::existsTable(const std::string & database_name, const std::stri
     request.SetDatabaseName(database_name);
     request.SetName(table_name);
 
+    ProfileEvents::increment(ProfileEvents::DataLakeGlueCatalogGetTable);
+    auto timer = DB::CurrentThread::getProfileEvents().timer(ProfileEvents::DataLakeGlueCatalogGetTableMicroseconds);
     auto outcome = glue_client->GetTable(request);
     return outcome.IsSuccess();
 }
@@ -303,7 +336,12 @@ bool GlueCatalog::tryGetTableMetadata(
     request.SetDatabaseName(database_name);
     request.SetName(table_name);
 
-    auto outcome = glue_client->GetTable(request);
+    Aws::Glue::Model::GetTableOutcome outcome;
+    {
+        ProfileEvents::increment(ProfileEvents::DataLakeGlueCatalogGetTable);
+        auto timer = DB::CurrentThread::getProfileEvents().timer(ProfileEvents::DataLakeGlueCatalogGetTableMicroseconds);
+        outcome = glue_client->GetTable(request);
+    }
     if (outcome.IsSuccess())
     {
         const auto & table_outcome = outcome.GetResult().GetTable();
@@ -599,6 +637,8 @@ void GlueCatalog::createNamespaceIfNotExists(const String & namespace_name) cons
     db_input.SetName(namespace_name);
     create_request.SetDatabaseInput(db_input);
 
+    ProfileEvents::increment(ProfileEvents::DataLakeGlueCatalogCreateDatabase);
+    auto timer = DB::CurrentThread::getProfileEvents().timer(ProfileEvents::DataLakeGlueCatalogCreateDatabaseMicroseconds);
     glue_client->CreateDatabase(create_request);
 }
 
@@ -631,7 +671,13 @@ void GlueCatalog::createTable(const String & namespace_name, const String & tabl
 
     request.SetTableInput(table_input);
 
-    auto response = glue_client->CreateTable(request);
+    Aws::Glue::Model::CreateTableOutcome response;
+
+    {
+        ProfileEvents::increment(ProfileEvents::DataLakeGlueCatalogCreateTable);
+        auto timer = DB::CurrentThread::getProfileEvents().timer(ProfileEvents::DataLakeGlueCatalogCreateTableMicroseconds);
+        response = glue_client->CreateTable(request);
+    }
 
     if (!response.IsSuccess())
         throw DB::Exception(DB::ErrorCodes::DATALAKE_DATABASE_ERROR, "Can not create metadata in glue catalog: {}", response.GetError().GetMessage());
@@ -666,7 +712,13 @@ bool GlueCatalog::updateMetadata(const String & namespace_name, const String & t
 
     request.SetTableInput(table_input);
 
-    auto response = glue_client->UpdateTable(request);
+    Aws::Glue::Model::UpdateTableOutcome response;
+
+    {
+        ProfileEvents::increment(ProfileEvents::DataLakeGlueCatalogUpdateTable);
+        auto timer = DB::CurrentThread::getProfileEvents().timer(ProfileEvents::DataLakeGlueCatalogUpdateTableMicroseconds);
+        response = glue_client->UpdateTable(request);
+    }
 
     if (!response.IsSuccess())
         throw DB::Exception(DB::ErrorCodes::DATALAKE_DATABASE_ERROR, "Can not update metadata in glue catalog {}", response.GetError().GetMessage());
@@ -690,7 +742,13 @@ void GlueCatalog::dropTable(const String & namespace_name, const String & table_
     request.SetDatabaseName(namespace_name);
     request.SetName(table_name);
 
-    auto response = glue_client->DeleteTable(request);
+    Aws::Glue::Model::DeleteTableOutcome response;
+
+    {
+        ProfileEvents::increment(ProfileEvents::DataLakeGlueCatalogDropTable);
+        auto timer = DB::CurrentThread::getProfileEvents().timer(ProfileEvents::DataLakeGlueCatalogDropTableMicroseconds);
+        response = glue_client->DeleteTable(request);
+    }
 
     if (!response.IsSuccess())
         throw DB::Exception(
diff --git a/src/Databases/DataLake/ICatalog.cpp b/src/Databases/DataLake/ICatalog.cpp
index 432b4d8b61c5..50200b3d3cc6 100644
--- a/src/Databases/DataLake/ICatalog.cpp
+++ b/src/Databases/DataLake/ICatalog.cpp
@@ -103,7 +103,18 @@ void TableMetadata::setLocation(const std::string & location_)
     auto pos_to_path = location_.substr(pos_to_bucket).find('/');
 
     if (pos_to_path == std::string::npos)
+    {
+        /// An empty path is allowed for AWS S3 Tables: the table location is just `s3://<bucket>`.
+        if (storage_type_str == "s3://")
+        {
+            location_without_path = location_;
+            path.clear();
+            bucket = location_.substr(pos_to_bucket);
+            return;
+        }
+
         throw DB::Exception(DB::ErrorCodes::NOT_IMPLEMENTED, "Unexpected location format: {}", location_);
+    }
 
     pos_to_path = pos_to_bucket + pos_to_path;
 
@@ -284,8 +295,22 @@ std::string TableMetadata::getMetadataLocation(const std::string & iceberg_metad
             metadata_location = metadata_location.substr(storage_type_str.size());
         if (data_location.starts_with(storage_type_str))
             data_location = data_location.substr(storage_type_str.size());
-        else if (!endpoint.empty() && data_location.starts_with(endpoint))
-            data_location = data_location.substr(endpoint.size());
+        else if (!endpoint.empty())
+        {
+            std::string normalized_endpoint = endpoint;
+            if (normalized_endpoint.ends_with('/'))
+                normalized_endpoint.pop_back();
+
+            if (data_location.starts_with(normalized_endpoint))
+            {
+                data_location = data_location.substr(normalized_endpoint.size());
+                /// `metadata_location` is relative to the bucket (the `s3://` prefix is stripped above),
+                /// while `data_location` still has the leading slash left over from the endpoint,
+                /// e.g. "/bucket/table-uuid/". Drop it so that the prefix comparison below works.
+                if (azure_account_with_suffix.empty() && data_location.starts_with('/'))
+                    data_location = data_location.substr(1);
+            }
+        }
 
         if (metadata_location.starts_with(data_location))
         {
diff --git a/src/Databases/DataLake/ICatalog.h b/src/Databases/DataLake/ICatalog.h
index e14b00ac3732..8211ba7aa39c 100644
--- a/src/Databases/DataLake/ICatalog.h
+++ b/src/Databases/DataLake/ICatalog.h
@@ -38,7 +38,6 @@ class TableMetadata
     bool hasLocation() const;
     bool hasSchema() const;
     bool hasStorageCredentials() const;
-    bool hasDataLakeSpecificProperties() const;
 
     void setLocation(const std::string & location_);
     std::string getLocation() const;
diff --git a/src/Databases/DataLake/RestCatalog.cpp b/src/Databases/DataLake/RestCatalog.cpp
index 9f85e9d80d6d..8e2ff1d3f694 100644
--- a/src/Databases/DataLake/RestCatalog.cpp
+++ b/src/Databases/DataLake/RestCatalog.cpp
@@ -5,6 +5,7 @@
 #include <Common/RemoteHostFilter.h>
 #include <Common/logger_useful.h>
 #include <Common/setThreadName.h>
+#include <Common/CurrentThread.h>
 #include <Storages/ObjectStorage/DataLakes/Iceberg/IcebergWrites.h>
 #include <mutex>
 #include <chrono>
@@ -68,6 +69,28 @@ namespace DB::FailPoints
     extern const char check_database_datalake_negative[];
 }
 
+namespace ProfileEvents
+{
+    extern const Event DataLakeRestCatalogLoadConfig;
+    extern const Event DataLakeRestCatalogLoadConfigMicroseconds;
+    extern const Event DataLakeRestCatalogGetNamespaces;
+    extern const Event DataLakeRestCatalogGetNamespacesMicroseconds;
+    extern const Event DataLakeRestCatalogGetTables;
+    extern const Event DataLakeRestCatalogGetTablesMicroseconds;
+    extern const Event DataLakeRestCatalogGetTableMetadata;
+    extern const Event DataLakeRestCatalogGetTableMetadataMicroseconds;
+    extern const Event DataLakeRestCatalogGetCredentials;
+    extern const Event DataLakeRestCatalogGetCredentialsMicroseconds;
+    extern const Event DataLakeRestCatalogCreateNamespace;
+    extern const Event DataLakeRestCatalogCreateNamespaceMicroseconds;
+    extern const Event DataLakeRestCatalogCreateTable;
+    extern const Event DataLakeRestCatalogCreateTableMicroseconds;
+    extern const Event DataLakeRestCatalogUpdateTable;
+    extern const Event DataLakeRestCatalogUpdateTableMicroseconds;
+    extern const Event DataLakeRestCatalogDropTable;
+    extern const Event DataLakeRestCatalogDropTableMicroseconds;
+}
+
 namespace DataLake
 {
 
@@ -220,10 +243,15 @@ RestCatalog::RestCatalog(
 RestCatalog::Config RestCatalog::loadConfig()
 {
     Poco::URI::QueryParameters params = {{"warehouse", warehouse}};
-    auto buf = createReadBuffer(CONFIG_ENDPOINT, params);
 
     std::string json_str;
-    readJSONObjectPossiblyInvalid(json_str, *buf);
+
+    {
+        ProfileEvents::increment(ProfileEvents::DataLakeRestCatalogLoadConfig);
+        auto timer = DB::CurrentThread::getProfileEvents().timer(ProfileEvents::DataLakeRestCatalogLoadConfigMicroseconds);
+        auto buf = createReadBuffer(CONFIG_ENDPOINT, params);
+        readJSONObjectPossiblyInvalid(json_str, *buf);
+    }
 
     LOG_DEBUG(log, "Received catalog configuration settings: {}", json_str);
 
@@ -255,7 +283,12 @@ void RestCatalog::parseCatalogConfigurationSettings(const Poco::JSON::Object::Pt
         result.default_base_location = object->get("default-base-location").extract<String>();
 }
 
-DB::HTTPHeaderEntries RestCatalog::getAuthHeaders(bool update_token) const
+DB::HTTPHeaderEntries RestCatalog::getAuthHeaders(
+    bool update_token,
+    const String & /*method*/,
+    const Poco::URI & /*url*/,
+    const DB::HTTPHeaderEntries & /*extra_headers*/,
+    const String & /*body*/) const
 {
     fiu_do_on(DB::FailPoints::check_database_datalake_negative,
     {
@@ -428,7 +461,12 @@ BigLakeCatalog::BigLakeCatalog(
     config = loadConfig();
 }
 
-DB::HTTPHeaderEntries BigLakeCatalog::getAuthHeaders(bool update_token) const
+DB::HTTPHeaderEntries BigLakeCatalog::getAuthHeaders(
+    bool update_token,
+    const String & /*method*/,
+    const Poco::URI & /*url*/,
+    const DB::HTTPHeaderEntries & /*extra_headers*/,
+    const String & /*body*/) const
 {
     /// Google Cloud OAuth2 for BigLake.
     /// Uses GCP metadata service or Application Default Credentials to get access token.
@@ -602,7 +640,7 @@ DB::ReadWriteBufferFromHTTPPtr RestCatalog::createReadBuffer(
 
     auto create_buffer = [&](bool update_token)
     {
-        auto result_headers = getAuthHeaders(update_token);
+        auto result_headers = getAuthHeaders(update_token, Poco::Net::HTTPRequest::HTTP_GET, url, headers, {});
         std::move(headers.begin(), headers.end(), std::back_inserter(result_headers));
 
         return DB::BuilderRWBufferFromHTTP(url)
@@ -765,6 +803,8 @@ RestCatalog::Namespaces RestCatalog::getNamespaces(const std::string & base_name
             if (!page_token.empty())
                 params.push_back({"pageToken", page_token});
 
+            ProfileEvents::increment(ProfileEvents::DataLakeRestCatalogGetNamespaces);
+            auto timer = DB::CurrentThread::getProfileEvents().timer(ProfileEvents::DataLakeRestCatalogGetNamespacesMicroseconds);
             auto buf = createReadBuffer(config.prefix / NAMESPACES_ENDPOINT, params);
             String next_page_token;
             auto page_namespaces = parseNamespaces(*buf, base_namespace, next_page_token);
@@ -915,6 +955,8 @@ DB::Names RestCatalog::getTables(const std::string & base_namespace, size_t limi
         if (!page_token.empty())
             params.push_back({"pageToken", page_token});
 
+        ProfileEvents::increment(ProfileEvents::DataLakeRestCatalogGetTables);
+        auto timer = DB::CurrentThread::getProfileEvents().timer(ProfileEvents::DataLakeRestCatalogGetTablesMicroseconds);
         auto buf = createReadBuffer(config.prefix / endpoint, params);
 
         /// Pass through the remaining limit so that single-page short-circuiting still works
@@ -1051,16 +1093,21 @@ bool RestCatalog::getTableMetadataImpl(
     }
 
     const std::string endpoint = std::filesystem::path(NAMESPACES_ENDPOINT) / encodeNamespaceForURI(namespace_name) / "tables" / table_name;
-    auto buf = createReadBuffer(config.prefix / endpoint, /* params */{}, headers);
+    String json_str;
 
-    if (buf->eof())
     {
-        LOG_DEBUG(log, "Table doesn't exist (endpoint: {})", endpoint);
-        return false;
-    }
+        ProfileEvents::increment(ProfileEvents::DataLakeRestCatalogGetTableMetadata);
+        auto timer = DB::CurrentThread::getProfileEvents().timer(ProfileEvents::DataLakeRestCatalogGetTableMetadataMicroseconds);
+        auto buf = createReadBuffer(config.prefix / endpoint, /* params */{}, headers);
 
-    String json_str;
-    readJSONObjectPossiblyInvalid(json_str, *buf);
+        if (buf->eof())
+        {
+            LOG_DEBUG(log, "Table doesn't exist (endpoint: {})", endpoint);
+            return false;
+        }
+
+        readJSONObjectPossiblyInvalid(json_str, *buf);
+    }
 
 #ifdef DEBUG_OR_SANITIZER_BUILD
     /// This log message might contain credentials,
@@ -1135,9 +1182,6 @@ void RestCatalog::sendRequest(const String & endpoint, Poco::JSON::Object::Ptr r
         request_body->stringify(oss);
     const std::string body_str = DB::removeEscapedSlashes(oss.str());
 
-    DB::HTTPHeaderEntries headers = getAuthHeaders(/* update_token = */ true);
-    headers.emplace_back("Content-Type", "application/json");
-
     const auto & context = getContext();
 
     DB::ReadWriteBufferFromHTTP::OutStreamCallback out_stream_callback;
@@ -1151,6 +1195,12 @@ void RestCatalog::sendRequest(const String & endpoint, Poco::JSON::Object::Ptr r
 
     /// enable_url_encoding=false to allow use tables with encoded sequences in names like 'foo%2Fbar'
     Poco::URI url(endpoint, /* enable_url_encoding */ false);
+
+    DB::HTTPHeaderEntries extra_headers;
+    extra_headers.emplace_back("Content-Type", "application/json");
+
+    DB::HTTPHeaderEntries headers = getAuthHeaders(/* update_token = */ true, method, url, extra_headers, body_str);
+    headers.emplace_back("Content-Type", "application/json");
     auto wb = DB::BuilderRWBufferFromHTTP(url)
         .withConnectionGroup(DB::HTTPConnectionGroupType::HTTP)
         .withMethod(method)
@@ -1187,6 +1237,8 @@ void RestCatalog::createNamespaceIfNotExists(const String & namespace_name, cons
 
     try
     {
+        ProfileEvents::increment(ProfileEvents::DataLakeRestCatalogCreateNamespace);
+        auto timer = DB::CurrentThread::getProfileEvents().timer(ProfileEvents::DataLakeRestCatalogCreateNamespaceMicroseconds);
         sendRequest(endpoint, request_body);
     }
     catch (...)
@@ -1229,6 +1281,8 @@ void RestCatalog::createTable(const String & namespace_name, const String & tabl
 
     try
     {
+        ProfileEvents::increment(ProfileEvents::DataLakeRestCatalogCreateTable);
+        auto timer = DB::CurrentThread::getProfileEvents().timer(ProfileEvents::DataLakeRestCatalogCreateTableMicroseconds);
         sendRequest(endpoint, request_body);
     }
     catch (const DB::HTTPException & ex)
@@ -1294,6 +1348,8 @@ bool RestCatalog::updateMetadata(const String & namespace_name, const String & t
 
     try
     {
+        ProfileEvents::increment(ProfileEvents::DataLakeRestCatalogUpdateTable);
+        auto timer = DB::CurrentThread::getProfileEvents().timer(ProfileEvents::DataLakeRestCatalogUpdateTableMicroseconds);
         sendRequest(endpoint, request_body);
     }
     catch (const DB::HTTPException & ex)
@@ -1368,11 +1424,15 @@ bool RestCatalog::updateSchema(
 
 void RestCatalog::dropTable(const String & namespace_name, const String & table_name) const
 {
-    const std::string endpoint = fmt::format("{}/namespaces/{}/tables/{}?purgeRequested=False", base_url, namespace_name, table_name);
+    const std::string endpoint
+        = (base_url / config.prefix / "namespaces" / namespace_name / "tables" / table_name).string()
+        + "?purgeRequested=False";
 
     Poco::JSON::Object::Ptr request_body = nullptr;
     try
     {
+        ProfileEvents::increment(ProfileEvents::DataLakeRestCatalogDropTable);
+        auto timer = DB::CurrentThread::getProfileEvents().timer(ProfileEvents::DataLakeRestCatalogDropTableMicroseconds);
         sendRequest(endpoint, request_body, Poco::Net::HTTPRequest::HTTP_DELETE, true);
     }
     catch (const DB::HTTPException & ex)
@@ -1460,16 +1520,21 @@ ICatalog::CredentialsRefreshCallback RestCatalog::getCredentialsConfigurationCal
         const auto & table = storage_id.getTableName();
         auto [namespace_name, table_name] = DataLake::parseTableName(table);
         const std::string endpoint = std::filesystem::path(NAMESPACES_ENDPOINT) / encodeNamespaceForURI(namespace_name) / "tables" / table_name;
-        auto buf = createReadBuffer(config.prefix / endpoint, /* params */{}, headers);
+        String json_str;
 
-        if (buf->eof())
         {
-            LOG_DEBUG(log, "Table doesn't exist (endpoint: {})", endpoint);
-            return nullptr;
-        }
+            ProfileEvents::increment(ProfileEvents::DataLakeRestCatalogGetCredentials);
+            auto timer = DB::CurrentThread::getProfileEvents().timer(ProfileEvents::DataLakeRestCatalogGetCredentialsMicroseconds);
+            auto buf = createReadBuffer(config.prefix / endpoint, /* params */{}, headers);
 
-        String json_str;
-        readJSONObjectPossiblyInvalid(json_str, *buf);
+            if (buf->eof())
+            {
+                LOG_DEBUG(log, "Table doesn't exist (endpoint: {})", endpoint);
+                return nullptr;
+            }
+
+            readJSONObjectPossiblyInvalid(json_str, *buf);
+        }
 
         Poco::JSON::Parser parser;
         Poco::Dynamic::Var json = parser.parse(json_str);
diff --git a/src/Databases/DataLake/RestCatalog.h b/src/Databases/DataLake/RestCatalog.h
index 982475ee2c96..90c481a899b1 100644
--- a/src/Databases/DataLake/RestCatalog.h
+++ b/src/Databases/DataLake/RestCatalog.h
@@ -163,7 +163,12 @@ class RestCatalog : public ICatalog, public DB::WithContext
         TableMetadata & result) const;
 
     Config loadConfig();
-    virtual DB::HTTPHeaderEntries getAuthHeaders(bool update_token) const;
+    virtual DB::HTTPHeaderEntries getAuthHeaders(
+        bool update_token,
+        const String & method = {},
+        const Poco::URI & url = {},
+        const DB::HTTPHeaderEntries & extra_headers = {},
+        const String & body = {}) const;
     static void parseCatalogConfigurationSettings(const Poco::JSON::Object::Ptr & object, Config & result);
 
     void sendRequest(
@@ -223,7 +228,12 @@ class BigLakeCatalog : public RestCatalog
         return DB::DatabaseDataLakeCatalogType::ICEBERG_BIGLAKE;
     }
 
-    DB::HTTPHeaderEntries getAuthHeaders(bool update_token) const override;
+    DB::HTTPHeaderEntries getAuthHeaders(
+        bool update_token,
+        const String & method = {},
+        const Poco::URI & url = {},
+        const DB::HTTPHeaderEntries & extra_headers = {},
+        const String & body = {}) const override;
 
     const std::string & getGoogleADCClientId() const { return google_adc_client_id; }
     const std::string & getGoogleADCClientSecret() const { return google_adc_client_secret; }
diff --git a/src/Databases/DataLake/S3TablesCatalog.cpp b/src/Databases/DataLake/S3TablesCatalog.cpp
new file mode 100644
index 000000000000..5b9e8f93fae8
--- /dev/null
+++ b/src/Databases/DataLake/S3TablesCatalog.cpp
@@ -0,0 +1,259 @@
+#include "config.h"
+
+#if USE_AVRO && USE_SSL && USE_AWS_S3
+
+#include <Databases/DataLake/S3TablesCatalog.h>
+#include <Databases/DataLake/AWSV4Signer.h>
+#include <Databases/DataLake/S3TablesCredentialRefresh.h>
+
+#include <Common/Exception.h>
+#include <Common/logger_useful.h>
+#include <Common/setThreadName.h>
+#include <Common/CurrentThread.h>
+#include <Common/threadPoolCallbackRunner.h>
+#include <Core/ServerSettings.h>
+#include <Core/Settings.h>
+#include <Interpreters/Context.h>
+#include <IO/ConnectionTimeouts.h>
+#include <IO/ReadWriteBufferFromHTTP.h>
+#include <IO/S3/Client.h>
+#include <IO/S3/URI.h>
+#include <IO/ReadHelpers.h>
+#include <Poco/JSON/Object.h>
+#include <Poco/JSON/Stringifier.h>
+#include <Poco/Net/HTTPRequest.h>
+#include <Poco/URI.h>
+
+#include <aws/core/auth/AWSCredentials.h>
+#include <aws/core/auth/signer/AWSAuthV4Signer.h>
+
+#include <mutex>
+
+namespace DB::ErrorCodes
+{
+    extern const int BAD_ARGUMENTS;
+    extern const int DATALAKE_DATABASE_ERROR;
+}
+
+namespace DB::Setting
+{
+    extern const SettingsUInt64 s3_max_redirects;
+    extern const SettingsUInt64 s3_retry_attempts;
+    extern const SettingsBool s3_slow_all_threads_after_network_error;
+    extern const SettingsBool enable_s3_requests_logging;
+}
+
+namespace DB::ServerSetting
+{
+    extern const ServerSettingsUInt64 s3_max_redirects;
+    extern const ServerSettingsUInt64 s3_retry_attempts;
+}
+
+namespace ProfileEvents
+{
+    extern const Event DataLakeRestCatalogDropTable;
+    extern const Event DataLakeRestCatalogDropTableMicroseconds;
+}
+
+namespace DataLake
+{
+
+S3TablesCatalog::S3TablesCatalog(
+    const String & warehouse_,
+    const String & base_url_,
+    const String & region_,
+    const CatalogSettings & catalog_settings_,
+    DB::ContextPtr context_)
+    : RestCatalog(warehouse_, base_url_, "", "", false, context_)
+    , region(region_)
+    , storage_endpoint(catalog_settings_.storage_endpoint)
+{
+    if (region.empty())
+        throw DB::Exception(DB::ErrorCodes::BAD_ARGUMENTS, "S3 Tables catalog requires non-empty `region` setting");
+
+    DB::S3::CredentialsConfiguration creds_config;
+    creds_config.use_environment_credentials = true;
+    creds_config.role_arn = catalog_settings_.aws_role_arn;
+    creds_config.role_session_name = catalog_settings_.aws_role_session_name;
+
+    const auto & server_settings = getContext()->getGlobalContext()->getServerSettings();
+    const DB::Settings & global_settings = getContext()->getGlobalContext()->getSettingsRef();
+
+    int s3_max_redirects = static_cast<int>(server_settings[DB::ServerSetting::s3_max_redirects]);
+    if (global_settings.isChanged("s3_max_redirects"))
+        s3_max_redirects = static_cast<int>(global_settings[DB::Setting::s3_max_redirects]);
+
+    int s3_retry_attempts = static_cast<int>(server_settings[DB::ServerSetting::s3_retry_attempts]);
+    if (global_settings.isChanged("s3_retry_attempts"))
+        s3_retry_attempts = static_cast<int>(global_settings[DB::Setting::s3_retry_attempts]);
+
+    bool s3_slow_all_threads_after_network_error = global_settings[DB::Setting::s3_slow_all_threads_after_network_error];
+    bool s3_slow_all_threads_after_retryable_error = false;
+    bool enable_s3_requests_logging = global_settings[DB::Setting::enable_s3_requests_logging];
+
+    DB::S3::PocoHTTPClientConfiguration poco_config = DB::S3::ClientFactory::instance().createClientConfiguration(
+        region,
+        getContext()->getRemoteHostFilter(),
+        s3_max_redirects,
+        DB::S3::PocoHTTPClientConfiguration::RetryStrategy{.max_retries = static_cast<unsigned>(s3_retry_attempts)},
+        s3_slow_all_threads_after_network_error,
+        s3_slow_all_threads_after_retryable_error,
+        enable_s3_requests_logging,
+        /* for_disk_s3 = */ false,
+        /* opt_disk_name = */ {},
+        /* request_throttler = */ {});
+
+    Aws::Auth::AWSCredentials credentials(catalog_settings_.aws_access_key_id, catalog_settings_.aws_secret_access_key);
+    credentials_provider = DB::S3::getCredentialsProvider(poco_config, credentials, creds_config);
+
+    signer = std::make_unique<Aws::Client::AWSAuthV4Signer>(
+        credentials_provider,
+        "s3tables",
+        Aws::String(region.data(), region.size()),
+        Aws::Client::AWSAuthV4Signer::PayloadSigningPolicy::Always,
+        /* urlEscapePath = */ false);
+
+    config = loadConfig();
+
+    if (config.prefix.empty())
+    {
+        String encoded_warehouse;
+        Poco::URI::encode(warehouse_, "", encoded_warehouse);
+        config.prefix = encoded_warehouse;
+    }
+}
+
+/// S3 Tables only supports a single level of namespaces (no nesting),
+/// so we use flat getNamespaces() instead of the base class's getNamespacesRecursive().
+DB::Names S3TablesCatalog::getTables() const
+{
+    auto namespaces = getNamespaces("");
+
+    auto & pool = getContext()->getIcebergCatalogThreadpool();
+    DB::ThreadPoolCallbackRunnerLocal<void> runner(pool, DB::ThreadName::DATALAKE_REST_CATALOG);
+
+    DB::Names tables;
+    std::mutex mutex;
+    for (const auto & ns : namespaces)
+    {
+        runner.enqueueAndKeepTrack(
+            [&, ns]
+            {
+                auto tables_in_ns = RestCatalog::getTables(ns);
+                std::lock_guard lock(mutex);
+                std::move(tables_in_ns.begin(), tables_in_ns.end(), std::back_inserter(tables));
+            });
+    }
+    runner.waitForAllToFinishAndRethrowFirstError();
+    return tables;
+}
+
+bool S3TablesCatalog::tryGetTableMetadata(
+    const std::string & namespace_name,
+    const std::string & table_name,
+    TableMetadata & result) const
+{
+    if (!RestCatalog::tryGetTableMetadata(namespace_name, table_name, result))
+        return false;
+
+    if (!result.requiresCredentials())
+        return true;
+
+    bool need_credentials = true;
+    if (const auto storage_credentials = result.getStorageCredentials())
+    {
+        auto creds = std::dynamic_pointer_cast<S3Credentials>(storage_credentials);
+        if (creds && !creds->isEmpty())
+            need_credentials = false;
+    }
+
+    if (need_credentials)
+    {
+        LOG_DEBUG(log, "S3 Tables: no vended credentials for {}.{}, injecting catalog IAM credentials", namespace_name, table_name);
+        auto aws_creds = credentials_provider->GetAWSCredentials();
+        if (aws_creds.GetAWSAccessKeyId().empty() || aws_creds.GetAWSSecretKey().empty())
+            throw DB::Exception(
+                DB::ErrorCodes::BAD_ARGUMENTS,
+                "S3 Tables: catalog IAM credentials are empty for {}.{}, "
+                "check AWS credentials configuration",
+                namespace_name, table_name);
+        result.setStorageCredentials(std::make_shared<S3Credentials>(
+            aws_creds.GetAWSAccessKeyId(), aws_creds.GetAWSSecretKey(), aws_creds.GetSessionToken()));
+    }
+
+    if (result.getEndpoint().empty())
+    {
+        String endpoint = storage_endpoint.empty()
+            ? DB::S3::resolveS3Endpoint(region)
+            : storage_endpoint;
+        LOG_DEBUG(log, "S3 Tables: no endpoint for {}.{}, injecting: {}", namespace_name, table_name, endpoint);
+        result.setEndpoint(endpoint);
+    }
+
+    return true;
+}
+
+ICatalog::CredentialsRefreshCallback S3TablesCatalog::getCredentialsConfigurationCallback(const DB::StorageID & storage_id)
+{
+    auto base_cb = RestCatalog::getCredentialsConfigurationCallback(storage_id);
+    return [this, base_callback = std::move(base_cb)] () -> std::shared_ptr<IStorageCredentials>
+    {
+        if (base_callback)
+        {
+            if (auto creds = (*base_callback)())
+            {
+                auto s3_creds = std::dynamic_pointer_cast<S3Credentials>(creds);
+                if (s3_creds && !s3_creds->isEmpty())
+                    return creds;
+            }
+            LOG_DEBUG(log, "S3 Tables: vended credentials unavailable on refresh, falling back to catalog IAM credentials");
+        }
+
+        return resolveS3TablesRefreshCredentials(std::nullopt, *credentials_provider);
+    };
+}
+
+void S3TablesCatalog::dropTable(const String & namespace_name, const String & table_name) const
+{
+    const std::string endpoint
+        = (base_url / config.prefix / "namespaces" / namespace_name / "tables" / table_name).string()
+        + "?purgeRequested=True";
+
+    Poco::JSON::Object::Ptr request_body = nullptr;
+    try
+    {
+        ProfileEvents::increment(ProfileEvents::DataLakeRestCatalogDropTable);
+        auto timer = DB::CurrentThread::getProfileEvents().timer(ProfileEvents::DataLakeRestCatalogDropTableMicroseconds);
+        sendRequest(endpoint, request_body, Poco::Net::HTTPRequest::HTTP_DELETE, true);
+    }
+    catch (const DB::HTTPException & ex)
+    {
+        if (ex.getHTTPStatus() == Poco::Net::HTTPResponse::HTTP_NOT_FOUND)
+            LOG_DEBUG(log, "S3 Tables: table {}.{} already does not exist (404 on purge-delete)", namespace_name, table_name);
+        else
+            throw DB::Exception(DB::ErrorCodes::DATALAKE_DATABASE_ERROR, "Failed to drop table {}", ex.displayText());
+    }
+}
+
+DB::HTTPHeaderEntries S3TablesCatalog::getAuthHeaders(
+    bool /*update_token*/,
+    const String & method,
+    const Poco::URI & url,
+    const DB::HTTPHeaderEntries & extra_headers,
+    const String & body) const
+{
+    DB::HTTPHeaderEntries all_signed;
+    signRequestWithAWSV4(method, url, extra_headers, body, *signer, region, "s3tables", all_signed);
+
+    DB::HTTPHeaderEntries auth_headers;
+    for (auto & h : all_signed)
+    {
+        if (h.name == "authorization" || h.name.starts_with("x-amz-"))
+            auth_headers.push_back(std::move(h));
+    }
+    return auth_headers;
+}
+
+}
+
+#endif
diff --git a/src/Databases/DataLake/S3TablesCatalog.h b/src/Databases/DataLake/S3TablesCatalog.h
new file mode 100644
index 000000000000..4ac24c03747e
--- /dev/null
+++ b/src/Databases/DataLake/S3TablesCatalog.h
@@ -0,0 +1,64 @@
+#pragma once
+
+#include "config.h"
+
+#if USE_AVRO && USE_SSL && USE_AWS_S3
+
+#include <Databases/DataLake/RestCatalog.h>
+#include <IO/S3/Credentials.h>
+
+#include <aws/core/auth/signer/AWSAuthV4Signer.h>
+
+#include <memory>
+
+namespace Aws::Auth
+{
+class AWSCredentialsProvider;
+}
+
+namespace DataLake
+{
+
+/// Iceberg REST catalog for Amazon S3 Tables (SigV4, signing name `s3tables`).
+/// https://docs.aws.amazon.com/AmazonS3/latest/userguide/s3-tables-integrating-open-source.html
+class S3TablesCatalog final : public RestCatalog
+{
+public:
+    S3TablesCatalog(
+        const String & warehouse_,
+        const String & base_url_,
+        const String & region_,
+        const DataLake::CatalogSettings & catalog_settings_,
+        DB::ContextPtr context_);
+
+    DB::DatabaseDataLakeCatalogType getCatalogType() const override { return DB::DatabaseDataLakeCatalogType::S3_TABLES; }
+
+    DB::Names getTables() const override;
+
+    bool tryGetTableMetadata(
+        const std::string & namespace_name,
+        const std::string & table_name,
+        TableMetadata & result) const override;
+
+    void dropTable(const String & namespace_name, const String & table_name) const override;
+
+    ICatalog::CredentialsRefreshCallback getCredentialsConfigurationCallback(const DB::StorageID & storage_id) override;
+
+protected:
+    DB::HTTPHeaderEntries getAuthHeaders(
+        bool update_token,
+        const String & method = {},
+        const Poco::URI & url = {},
+        const DB::HTTPHeaderEntries & extra_headers = {},
+        const String & body = {}) const override;
+
+private:
+    const String region;
+    const String storage_endpoint;
+    std::shared_ptr<Aws::Auth::AWSCredentialsProvider> credentials_provider;
+    std::unique_ptr<Aws::Client::AWSAuthV4Signer> signer;
+};
+
+}
+
+#endif
diff --git a/src/Databases/DataLake/S3TablesCredentialRefresh.cpp b/src/Databases/DataLake/S3TablesCredentialRefresh.cpp
new file mode 100644
index 000000000000..8de09e4d553e
--- /dev/null
+++ b/src/Databases/DataLake/S3TablesCredentialRefresh.cpp
@@ -0,0 +1,43 @@
+#include "config.h"
+
+#if USE_AVRO && USE_SSL && USE_AWS_S3
+
+#include <Databases/DataLake/S3TablesCredentialRefresh.h>
+
+namespace DataLake
+{
+
+namespace
+{
+
+std::shared_ptr<IStorageCredentials> getCatalogIAMCredentials(Aws::Auth::AWSCredentialsProvider & provider)
+{
+    auto aws_creds = provider.GetAWSCredentials();
+    if (aws_creds.GetAWSAccessKeyId().empty() || aws_creds.GetAWSSecretKey().empty())
+        return nullptr;
+    return std::make_shared<S3Credentials>(
+        aws_creds.GetAWSAccessKeyId(), aws_creds.GetAWSSecretKey(), aws_creds.GetSessionToken());
+}
+
+}
+
+std::shared_ptr<IStorageCredentials> resolveS3TablesRefreshCredentials(
+    const ICatalog::CredentialsRefreshCallback & base_callback,
+    Aws::Auth::AWSCredentialsProvider & credentials_provider)
+{
+    if (base_callback)
+    {
+        if (auto creds = (*base_callback)())
+        {
+            auto s3_creds = std::dynamic_pointer_cast<S3Credentials>(creds);
+            if (s3_creds && !s3_creds->isEmpty())
+                return creds;
+        }
+    }
+
+    return getCatalogIAMCredentials(credentials_provider);
+}
+
+}
+
+#endif
diff --git a/src/Databases/DataLake/S3TablesCredentialRefresh.h b/src/Databases/DataLake/S3TablesCredentialRefresh.h
new file mode 100644
index 000000000000..e5b959f88860
--- /dev/null
+++ b/src/Databases/DataLake/S3TablesCredentialRefresh.h
@@ -0,0 +1,23 @@
+#pragma once
+
+#include "config.h"
+
+#if USE_AVRO && USE_SSL && USE_AWS_S3
+
+#include <Databases/DataLake/ICatalog.h>
+#include <Databases/DataLake/StorageCredentials.h>
+
+#include <aws/core/auth/AWSCredentialsProvider.h>
+
+#include <memory>
+
+namespace DataLake
+{
+
+std::shared_ptr<IStorageCredentials> resolveS3TablesRefreshCredentials(
+    const ICatalog::CredentialsRefreshCallback & base_callback,
+    Aws::Auth::AWSCredentialsProvider & credentials_provider);
+
+}
+
+#endif
diff --git a/src/Databases/DataLake/StorageCredentials.h b/src/Databases/DataLake/StorageCredentials.h
index 3a2f6f793e89..7e73d66ff673 100644
--- a/src/Databases/DataLake/StorageCredentials.h
+++ b/src/Databases/DataLake/StorageCredentials.h
@@ -32,6 +32,8 @@ class S3Credentials final : public IStorageCredentials
         , session_token(session_token_)
     {}
 
+    bool isEmpty() const { return access_key_id.empty() || secret_access_key.empty(); }
+
     void addCredentialsToEngineArgs(DB::ASTs & engine_args) const override
     {
         if (engine_args.size() != 1)
diff --git a/src/Databases/DataLake/UnityCatalog.cpp b/src/Databases/DataLake/UnityCatalog.cpp
index 414b7e439ecf..f3d7077b7923 100644
--- a/src/Databases/DataLake/UnityCatalog.cpp
+++ b/src/Databases/DataLake/UnityCatalog.cpp
@@ -8,6 +8,7 @@
 #include <Poco/JSON/Array.h>
 #include <Poco/JSON/Parser.h>
 #include <Common/checkStackSize.h>
+#include <Common/CurrentThread.h>
 #include <IO/ReadHelpers.h>
 #include <IO/Operators.h>
 #include <Core/NamesAndTypes.h>
@@ -15,6 +16,20 @@
 #include <Databases/DataLake/StorageCredentials.h>
 #include <fmt/ranges.h>
 
+namespace ProfileEvents
+{
+    extern const Event DataLakeUnityCatalogGetTables;
+    extern const Event DataLakeUnityCatalogGetTablesMicroseconds;
+    extern const Event DataLakeUnityCatalogGetTable;
+    extern const Event DataLakeUnityCatalogGetTableMicroseconds;
+    extern const Event DataLakeUnityCatalogGetTableMetadata;
+    extern const Event DataLakeUnityCatalogGetTableMetadataMicroseconds;
+    extern const Event DataLakeUnityCatalogGetCredentials;
+    extern const Event DataLakeUnityCatalogGetCredentialsMicroseconds;
+    extern const Event DataLakeUnityCatalogGetSchemas;
+    extern const Event DataLakeUnityCatalogGetSchemasMicroseconds;
+}
+
 namespace DB::ErrorCodes
 {
     extern const int DATALAKE_DATABASE_ERROR;
@@ -106,7 +121,14 @@ Poco::JSON::Object::Ptr UnityCatalog::requestReadCredentials(const String & tabl
     request_body.set("operation", "READ");
 
     auto callback = [&request_body] (std::ostream & os) { request_body.stringify(os); };
-    auto [json, _] = postJSONRequest(TEMPORARY_CREDENTIALS_ENDPOINT, callback);
+
+    Poco::Dynamic::Var json;
+    {
+        ProfileEvents::increment(ProfileEvents::DataLakeUnityCatalogGetCredentials);
+        auto timer = DB::CurrentThread::getProfileEvents().timer(ProfileEvents::DataLakeUnityCatalogGetCredentialsMicroseconds);
+        std::string _;
+        std::tie(json, _) = postJSONRequest(TEMPORARY_CREDENTIALS_ENDPOINT, callback);
+    }
     return json.extract<Poco::JSON::Object::Ptr>();
 }
 
@@ -167,7 +189,11 @@ bool UnityCatalog::tryGetTableMetadata(
     std::string json_str;
     try
     {
-        std::tie(json, json_str) = getJSONRequest(std::filesystem::path{TABLES_ENDPOINT} / full_table_name);
+        {
+            ProfileEvents::increment(ProfileEvents::DataLakeUnityCatalogGetTableMetadata);
+            auto timer = DB::CurrentThread::getProfileEvents().timer(ProfileEvents::DataLakeUnityCatalogGetTableMetadataMicroseconds);
+            std::tie(json, json_str) = getJSONRequest(std::filesystem::path{TABLES_ENDPOINT} / full_table_name);
+        }
         const Poco::JSON::Object::Ptr & object = json.extract<Poco::JSON::Object::Ptr>();
         if (hasValueAndItsNotNone("name", object) && object->get("name").extract<String>() == table_name)
         {
@@ -288,7 +314,11 @@ bool UnityCatalog::existsTable(const std::string & schema_name, const std::strin
     Poco::Dynamic::Var json;
     try
     {
-        std::tie(json, json_str) = getJSONRequest(std::filesystem::path{TABLES_ENDPOINT} / (warehouse + "." + schema_name + "." + table_name));
+        {
+            ProfileEvents::increment(ProfileEvents::DataLakeUnityCatalogGetTable);
+            auto timer = DB::CurrentThread::getProfileEvents().timer(ProfileEvents::DataLakeUnityCatalogGetTableMicroseconds);
+            std::tie(json, json_str) = getJSONRequest(std::filesystem::path{TABLES_ENDPOINT} / (warehouse + "." + schema_name + "." + table_name));
+        }
         const Poco::JSON::Object::Ptr & object = json.extract<Poco::JSON::Object::Ptr>();
         if (hasValueAndItsNotNone("name", object) && object->get("name").extract<String>() == table_name)
             return true;
@@ -316,7 +346,11 @@ DB::Names UnityCatalog::getTablesForSchema(const std::string & schema, size_t li
 
         try
         {
-            std::tie(json, json_str) = getJSONRequest(TABLES_ENDPOINT, params);
+            {
+                ProfileEvents::increment(ProfileEvents::DataLakeUnityCatalogGetTables);
+                auto timer = DB::CurrentThread::getProfileEvents().timer(ProfileEvents::DataLakeUnityCatalogGetTablesMicroseconds);
+                std::tie(json, json_str) = getJSONRequest(TABLES_ENDPOINT, params);
+            }
             const Poco::JSON::Object::Ptr & object = json.extract<Poco::JSON::Object::Ptr>();
 
             if (!hasValueAndItsNotNone("tables", object))
@@ -380,7 +414,11 @@ DataLake::ICatalog::Namespaces UnityCatalog::getSchemas(const std::string & base
 
         try
         {
-            std::tie(json, json_str) = getJSONRequest(SCHEMAS_ENDPOINT, params);
+            {
+                ProfileEvents::increment(ProfileEvents::DataLakeUnityCatalogGetSchemas);
+                auto timer = DB::CurrentThread::getProfileEvents().timer(ProfileEvents::DataLakeUnityCatalogGetSchemasMicroseconds);
+                std::tie(json, json_str) = getJSONRequest(SCHEMAS_ENDPOINT, params);
+            }
             const Poco::JSON::Object::Ptr & object = json.extract<Poco::JSON::Object::Ptr>();
 
             auto schemas_object = object->get("schemas").extract<Poco::JSON::Array::Ptr>();
diff --git a/src/Databases/DataLake/tests/gtest_azure_abfss_parsing.cpp b/src/Databases/DataLake/tests/gtest_azure_abfss_parsing.cpp
index f1c93eee6dc6..1d710b01a7b4 100644
--- a/src/Databases/DataLake/tests/gtest_azure_abfss_parsing.cpp
+++ b/src/Databases/DataLake/tests/gtest_azure_abfss_parsing.cpp
@@ -240,4 +240,31 @@ TEST_F(AzureAbfssParsingTest, TableMetadataGetLocationWithEndpointVirtualHostedD
     EXPECT_EQ(location, "https://my.dotted.bucket.s3.mycompany.com/path/to/table/");
 }
 
+TEST_F(AzureAbfssParsingTest, TableMetadataGetMetadataLocationS3TablesWithAwsEndpoint)
+{
+    TableMetadata metadata;
+    metadata.withLocation();
+    metadata.setLocation("s3://bucket/table-uuid/");
+    metadata.setEndpoint("https://s3.us-east-2.amazonaws.com");
+
+    EXPECT_EQ(metadata.getLocation(), "https://s3.us-east-2.amazonaws.com/bucket/table-uuid/");
+
+    const std::string metadata_file =
+        "s3://bucket/table-uuid/metadata/v1.metadata.json";
+    EXPECT_EQ(metadata.getMetadataLocation(metadata_file), "metadata/v1.metadata.json");
+}
+
+TEST_F(AzureAbfssParsingTest, TableMetadataGetMetadataLocationS3TablesEmptyPathWithAwsEndpoint)
+{
+    TableMetadata metadata;
+    metadata.withLocation();
+    metadata.setLocation("s3://bucket");
+    metadata.setEndpoint("https://s3.us-east-2.amazonaws.com");
+
+    EXPECT_EQ(metadata.getLocation(), "https://s3.us-east-2.amazonaws.com/bucket/");
+
+    const std::string metadata_file = "s3://bucket/metadata/v1.metadata.json";
+    EXPECT_EQ(metadata.getMetadataLocation(metadata_file), "metadata/v1.metadata.json");
+}
+
 }
diff --git a/src/Databases/DataLake/tests/gtest_s3tables_credential_refresh.cpp b/src/Databases/DataLake/tests/gtest_s3tables_credential_refresh.cpp
new file mode 100644
index 000000000000..6c5bc705fea5
--- /dev/null
+++ b/src/Databases/DataLake/tests/gtest_s3tables_credential_refresh.cpp
@@ -0,0 +1,112 @@
+#include "config.h"
+
+#if USE_AVRO && USE_SSL && USE_AWS_S3
+
+#include <gtest/gtest.h>
+
+#include <Databases/DataLake/S3TablesCredentialRefresh.h>
+#include <Databases/DataLake/StorageCredentials.h>
+
+#include <aws/core/auth/AWSCredentials.h>
+#include <aws/core/auth/AWSCredentialsProvider.h>
+
+#include <mutex>
+
+namespace
+{
+
+class RotatingAWSCredentialsProvider : public Aws::Auth::AWSCredentialsProvider
+{
+public:
+    explicit RotatingAWSCredentialsProvider(std::vector<Aws::Auth::AWSCredentials> credentials_sets_)
+        : credentials_sets(std::move(credentials_sets_))
+    {
+    }
+
+    Aws::Auth::AWSCredentials GetAWSCredentials() override
+    {
+        std::lock_guard lock(mutex);
+        const size_t index = call_count++;
+        if (index >= credentials_sets.size())
+            return credentials_sets.back();
+        return credentials_sets[index];
+    }
+
+private:
+    std::vector<Aws::Auth::AWSCredentials> credentials_sets;
+    std::mutex mutex;
+    size_t call_count = 0;
+};
+
+}
+
+TEST(S3TablesCredentialRefresh, FallsBackToCatalogIAMWhenVendedCredentialsMissing)
+{
+    RotatingAWSCredentialsProvider provider({
+        Aws::Auth::AWSCredentials("access_key_1", "secret_key_1", "session_token_1"),
+        Aws::Auth::AWSCredentials("access_key_2", "secret_key_2", "session_token_2"),
+    });
+
+    DataLake::ICatalog::CredentialsRefreshCallback base_callback = []() -> std::shared_ptr<DataLake::IStorageCredentials>
+    {
+        return nullptr;
+    };
+
+    auto first = DataLake::resolveS3TablesRefreshCredentials(base_callback, provider);
+    ASSERT_NE(first, nullptr);
+    auto first_s3 = std::dynamic_pointer_cast<DataLake::S3Credentials>(first);
+    ASSERT_NE(first_s3, nullptr);
+    EXPECT_EQ(first_s3->getAccessKeyId(), "access_key_1");
+    EXPECT_EQ(first_s3->getSecretAccessKey(), "secret_key_1");
+    EXPECT_EQ(first_s3->getSessionToken(), "session_token_1");
+
+    auto second = DataLake::resolveS3TablesRefreshCredentials(base_callback, provider);
+    ASSERT_NE(second, nullptr);
+    auto second_s3 = std::dynamic_pointer_cast<DataLake::S3Credentials>(second);
+    ASSERT_NE(second_s3, nullptr);
+    EXPECT_EQ(second_s3->getAccessKeyId(), "access_key_2");
+    EXPECT_EQ(second_s3->getSecretAccessKey(), "secret_key_2");
+    EXPECT_EQ(second_s3->getSessionToken(), "session_token_2");
+}
+
+TEST(S3TablesCredentialRefresh, PrefersVendedCredentialsWhenPresent)
+{
+    RotatingAWSCredentialsProvider provider({
+        Aws::Auth::AWSCredentials("catalog_access", "catalog_secret", "catalog_token"),
+    });
+
+    DataLake::ICatalog::CredentialsRefreshCallback base_callback = []() -> std::shared_ptr<DataLake::IStorageCredentials>
+    {
+        return std::make_shared<DataLake::S3Credentials>("vended_access", "vended_secret", "vended_token");
+    };
+
+    auto creds = DataLake::resolveS3TablesRefreshCredentials(base_callback, provider);
+    ASSERT_NE(creds, nullptr);
+    auto s3_creds = std::dynamic_pointer_cast<DataLake::S3Credentials>(creds);
+    ASSERT_NE(s3_creds, nullptr);
+    EXPECT_EQ(s3_creds->getAccessKeyId(), "vended_access");
+    EXPECT_EQ(s3_creds->getSecretAccessKey(), "vended_secret");
+    EXPECT_EQ(s3_creds->getSessionToken(), "vended_token");
+}
+
+TEST(S3TablesCredentialRefresh, FallsBackWhenVendedCredentialsEmpty)
+{
+    RotatingAWSCredentialsProvider provider({
+        Aws::Auth::AWSCredentials("catalog_access", "catalog_secret", "catalog_token"),
+    });
+
+    DataLake::ICatalog::CredentialsRefreshCallback base_callback = []() -> std::shared_ptr<DataLake::IStorageCredentials>
+    {
+        return std::make_shared<DataLake::S3Credentials>("", "", "");
+    };
+
+    auto creds = DataLake::resolveS3TablesRefreshCredentials(base_callback, provider);
+    ASSERT_NE(creds, nullptr);
+    auto s3_creds = std::dynamic_pointer_cast<DataLake::S3Credentials>(creds);
+    ASSERT_NE(s3_creds, nullptr);
+    EXPECT_EQ(s3_creds->getAccessKeyId(), "catalog_access");
+    EXPECT_EQ(s3_creds->getSecretAccessKey(), "catalog_secret");
+    EXPECT_EQ(s3_creds->getSessionToken(), "catalog_token");
+}
+
+#endif
diff --git a/src/Databases/enableAllExperimentalSettings.cpp b/src/Databases/enableAllExperimentalSettings.cpp
index ac29261994ee..d20e34d5f9ae 100644
--- a/src/Databases/enableAllExperimentalSettings.cpp
+++ b/src/Databases/enableAllExperimentalSettings.cpp
@@ -69,6 +69,7 @@ void enableAllExperimentalSettings(ContextMutablePtr context)
     context->setSetting("allow_dynamic_type_in_join_keys", 1);
     context->setSetting("allow_experimental_alias_table_engine", 1);
     context->setSetting("allow_experimental_database_paimon_rest_catalog", 1);
+    context->setSetting("allow_experimental_database_s3_tables", 1);
     context->setSetting("allow_experimental_object_storage_queue_hive_partitioning", 1);
     context->setSetting("allow_experimental_json_lazy_type_hints", 1);
     context->setSetting("allow_experimental_full_text_index", 1);
diff --git a/src/IO/S3/URI.cpp b/src/IO/S3/URI.cpp
index dd5429dcf55f..8e88c6257876 100644
--- a/src/IO/S3/URI.cpp
+++ b/src/IO/S3/URI.cpp
@@ -8,6 +8,8 @@
 #include <Common/re2.h>
 #include <IO/Archives/ArchiveUtils.h>
 
+#include <aws/s3/S3EndpointProvider.h>
+
 #include <boost/algorithm/string/case_conv.hpp>
 #include <Poco/Util/AbstractConfiguration.h>
 
@@ -262,6 +264,19 @@ void URI::validateKey(const String & key, const Poco::URI & uri)
     }
 }
 
+std::string resolveS3Endpoint(const std::string & region)
+{
+    Aws::S3::Endpoint::S3EndpointProvider provider;
+    provider.AccessBuiltInParameters().SetStringParameter("Region", Aws::String(region));
+    auto outcome = provider.ResolveEndpoint({});
+    if (outcome.IsSuccess())
+    {
+        auto uri = outcome.GetResult().GetURI();
+        return uri.GetURIString();
+    }
+    return "https://s3." + region + ".amazonaws.com";
+}
+
 }
 
 }
diff --git a/src/IO/S3/URI.h b/src/IO/S3/URI.h
index 64b4def76744..28504e39cb7f 100644
--- a/src/IO/S3/URI.h
+++ b/src/IO/S3/URI.h
@@ -52,6 +52,10 @@ struct URI
     bool tryInitVirtualHostedStyle(bool is_using_aws_private_link_interface, bool use_strict_pattern);
 };
 
+/// Resolve the S3 endpoint URL for a given AWS region using the SDK's
+/// Smithy endpoint rules (handles all partitions: standard, China, GovCloud, etc.).
+std::string resolveS3Endpoint(const std::string & region);
+
 }
 
 #endif
diff --git a/src/IO/S3/tests/gtest_s3_uri.cpp b/src/IO/S3/tests/gtest_s3_uri.cpp
index a429c645fca3..5c0d3debc0c0 100644
--- a/src/IO/S3/tests/gtest_s3_uri.cpp
+++ b/src/IO/S3/tests/gtest_s3_uri.cpp
@@ -38,4 +38,23 @@ TEST(IOTestS3URI, PathStyleWithKey)
     ASSERT_EQ(uri_with_no_key_and_with_slash.key, "key/key/key/key");
 }
 
+TEST(IOTestS3URI, ResolveS3Endpoint)
+{
+    using namespace DB;
+
+    ASSERT_EQ(S3::resolveS3Endpoint("us-east-1"),
+              "https://s3.us-east-1.amazonaws.com");
+    ASSERT_EQ(S3::resolveS3Endpoint("eu-west-1"),
+              "https://s3.eu-west-1.amazonaws.com");
+
+    auto cn_north = S3::resolveS3Endpoint("cn-north-1");
+    ASSERT_TRUE(cn_north.ends_with(".amazonaws.com.cn"))
+        << "China region should resolve to .amazonaws.com.cn suffix, got: " << cn_north;
+    ASSERT_TRUE(cn_north.find("cn-north-1") != std::string::npos)
+        << "Got: " << cn_north;
+
+    ASSERT_EQ(S3::resolveS3Endpoint("us-gov-west-1"),
+              "https://s3.us-gov-west-1.amazonaws.com");
+}
+
 #endif