diff --git a/src/duckdb/src/catalog/catalog.cpp b/src/duckdb/src/catalog/catalog.cpp
index 9ba5b5a4f..a6a3e1f3d 100644
--- a/src/duckdb/src/catalog/catalog.cpp
+++ b/src/duckdb/src/catalog/catalog.cpp
@@ -661,7 +661,7 @@ CatalogException Catalog::CreateMissingEntryException(CatalogEntryRetriever &ret
 		if (unseen_schemas.size() >= max_schema_count) {
 			break;
 		}
-		auto &catalog = database.get().GetCatalog();
+		auto &catalog = database->GetCatalog();
 		auto current_schemas = catalog.GetSchemas(context);
 		for (auto &current_schema : current_schemas) {
 			if (unseen_schemas.size() >= max_schema_count) {
@@ -1119,8 +1119,8 @@ vector<reference<SchemaCatalogEntry>> Catalog::GetAllSchemas(ClientContext &cont
 
 	auto &db_manager = DatabaseManager::Get(context);
 	auto databases = db_manager.GetDatabases(context);
-	for (auto database : databases) {
-		auto &catalog = database.get().GetCatalog();
+	for (auto &database : databases) {
+		auto &catalog = database->GetCatalog();
 		auto new_schemas = catalog.GetSchemas(context);
 		result.insert(result.end(), new_schemas.begin(), new_schemas.end());
 	}
diff --git a/src/duckdb/src/common/enum_util.cpp b/src/duckdb/src/common/enum_util.cpp
index 3b0578f81..df460d3dc 100644
--- a/src/duckdb/src/common/enum_util.cpp
+++ b/src/duckdb/src/common/enum_util.cpp
@@ -4392,19 +4392,20 @@ const StringUtil::EnumStringLiteral *GetUndoFlagsValues() {
 		{ static_cast<uint32_t>(UndoFlags::INSERT_TUPLE), "INSERT_TUPLE" },
 		{ static_cast<uint32_t>(UndoFlags::DELETE_TUPLE), "DELETE_TUPLE" },
 		{ static_cast<uint32_t>(UndoFlags::UPDATE_TUPLE), "UPDATE_TUPLE" },
-		{ static_cast<uint32_t>(UndoFlags::SEQUENCE_VALUE), "SEQUENCE_VALUE" }
+		{ static_cast<uint32_t>(UndoFlags::SEQUENCE_VALUE), "SEQUENCE_VALUE" },
+		{ static_cast<uint32_t>(UndoFlags::ATTACHED_DATABASE), "ATTACHED_DATABASE" }
 	};
 	return values;
 }
 
 template<>
 const char* EnumUtil::ToChars<UndoFlags>(UndoFlags value) {
-	return StringUtil::EnumToString(GetUndoFlagsValues(), 6, "UndoFlags", static_cast<uint32_t>(value));
+	return StringUtil::EnumToString(GetUndoFlagsValues(), 7, "UndoFlags", static_cast<uint32_t>(value));
 }
 
 template<>
 UndoFlags EnumUtil::FromString<UndoFlags>(const char *value) {
-	return static_cast<UndoFlags>(StringUtil::StringToEnum(GetUndoFlagsValues(), 6, "UndoFlags", value));
+	return static_cast<UndoFlags>(StringUtil::StringToEnum(GetUndoFlagsValues(), 7, "UndoFlags", value));
 }
 
 const StringUtil::EnumStringLiteral *GetUnionInvalidReasonValues() {
diff --git a/src/duckdb/src/execution/operator/helper/physical_transaction.cpp b/src/duckdb/src/execution/operator/helper/physical_transaction.cpp
index a8ad410db..336154f48 100644
--- a/src/duckdb/src/execution/operator/helper/physical_transaction.cpp
+++ b/src/duckdb/src/execution/operator/helper/physical_transaction.cpp
@@ -35,8 +35,8 @@ SourceResultType PhysicalTransaction::GetData(ExecutionContext &context, DataChu
 			if (config.options.immediate_transaction_mode) {
 				// if immediate transaction mode is enabled then start all transactions immediately
 				auto databases = DatabaseManager::Get(client).GetDatabases(client);
-				for (auto db : databases) {
-					context.client.transaction.ActiveTransaction().GetTransaction(db.get());
+				for (auto &db : databases) {
+					context.client.transaction.ActiveTransaction().GetTransaction(*db);
 				}
 			}
 		} else {
diff --git a/src/duckdb/src/execution/operator/schema/physical_attach.cpp b/src/duckdb/src/execution/operator/schema/physical_attach.cpp
index d871f9ef5..cc3c6ff5c 100644
--- a/src/duckdb/src/execution/operator/schema/physical_attach.cpp
+++ b/src/duckdb/src/execution/operator/schema/physical_attach.cpp
@@ -73,6 +73,8 @@ SourceResultType PhysicalAttach::GetData(ExecutionContext &context, DataChunk &c
 		attached_db->GetCatalog().SetDefaultTable(options.default_table.schema, options.default_table.name);
 	}
 	attached_db->FinalizeLoad(context.client);
+
+	db_manager.FinalizeAttach(context.client, *info, std::move(attached_db));
 	return SourceResultType::FINISHED;
 }
 
diff --git a/src/duckdb/src/function/table/system/duckdb_databases.cpp b/src/duckdb/src/function/table/system/duckdb_databases.cpp
index 2c7c9c913..0e2031f79 100644
--- a/src/duckdb/src/function/table/system/duckdb_databases.cpp
+++ b/src/duckdb/src/function/table/system/duckdb_databases.cpp
@@ -8,7 +8,7 @@ struct DuckDBDatabasesData : public GlobalTableFunctionState {
 	DuckDBDatabasesData() : offset(0) {
 	}
 
-	vector<reference<AttachedDatabase>> entries;
+	vector<shared_ptr<AttachedDatabase>> entries;
 	idx_t offset;
 };
 
@@ -62,7 +62,7 @@ void DuckDBDatabasesFunction(ClientContext &context, TableFunctionInput &data_p,
 	while (data.offset < data.entries.size() && count < STANDARD_VECTOR_SIZE) {
 		auto &entry = data.entries[data.offset++];
 
-		auto &attached = entry.get().Cast<AttachedDatabase>();
+		auto &attached = *entry;
 		// return values:
 
 		idx_t col = 0;
diff --git a/src/duckdb/src/function/table/system/pragma_database_size.cpp b/src/duckdb/src/function/table/system/pragma_database_size.cpp
index ba8f2020e..dfe816f0d 100644
--- a/src/duckdb/src/function/table/system/pragma_database_size.cpp
+++ b/src/duckdb/src/function/table/system/pragma_database_size.cpp
@@ -15,7 +15,7 @@ struct PragmaDatabaseSizeData : public GlobalTableFunctionState {
 	}
 
 	idx_t index;
-	vector<reference<AttachedDatabase>> databases;
+	vector<shared_ptr<AttachedDatabase>> databases;
 	Value memory_usage;
 	Value memory_limit;
 };
@@ -68,7 +68,7 @@ void PragmaDatabaseSizeFunction(ClientContext &context, TableFunctionInput &data
 	auto &data = data_p.global_state->Cast<PragmaDatabaseSizeData>();
 	idx_t row = 0;
 	for (; data.index < data.databases.size() && row < STANDARD_VECTOR_SIZE; data.index++) {
-		auto &db = data.databases[data.index].get();
+		auto &db = *data.databases[data.index];
 		if (db.IsSystem() || db.IsTemporary()) {
 			continue;
 		}
diff --git a/src/duckdb/src/function/table/version/pragma_version.cpp b/src/duckdb/src/function/table/version/pragma_version.cpp
index 0a47fd2b3..cf31766cd 100644
--- a/src/duckdb/src/function/table/version/pragma_version.cpp
+++ b/src/duckdb/src/function/table/version/pragma_version.cpp
@@ -1,5 +1,5 @@
 #ifndef DUCKDB_PATCH_VERSION
-#define DUCKDB_PATCH_VERSION "3-dev200"
+#define DUCKDB_PATCH_VERSION "3-dev227"
 #endif
 #ifndef DUCKDB_MINOR_VERSION
 #define DUCKDB_MINOR_VERSION 3
@@ -8,10 +8,10 @@
 #define DUCKDB_MAJOR_VERSION 1
 #endif
 #ifndef DUCKDB_VERSION
-#define DUCKDB_VERSION "v1.3.3-dev200"
+#define DUCKDB_VERSION "v1.3.3-dev227"
 #endif
 #ifndef DUCKDB_SOURCE_ID
-#define DUCKDB_SOURCE_ID "c5310ec83b"
+#define DUCKDB_SOURCE_ID "df0a3de744"
 #endif
 #include "duckdb/function/table/system_functions.hpp"
 #include "duckdb/main/database.hpp"
diff --git a/src/duckdb/src/include/duckdb/common/enums/undo_flags.hpp b/src/duckdb/src/include/duckdb/common/enums/undo_flags.hpp
index a91548bda..8492ffff8 100644
--- a/src/duckdb/src/include/duckdb/common/enums/undo_flags.hpp
+++ b/src/duckdb/src/include/duckdb/common/enums/undo_flags.hpp
@@ -18,7 +18,8 @@ enum class UndoFlags : uint32_t { // far too big but aligned (TM)
 	INSERT_TUPLE = 2,
 	DELETE_TUPLE = 3,
 	UPDATE_TUPLE = 4,
-	SEQUENCE_VALUE = 5
+	SEQUENCE_VALUE = 5,
+	ATTACHED_DATABASE = 6
 };
 
 } // namespace duckdb
diff --git a/src/duckdb/src/include/duckdb/main/attached_database.hpp b/src/duckdb/src/include/duckdb/main/attached_database.hpp
index b92e0dbfd..632d84a86 100644
--- a/src/duckdb/src/include/duckdb/main/attached_database.hpp
+++ b/src/duckdb/src/include/duckdb/main/attached_database.hpp
@@ -32,6 +32,14 @@ enum class AttachedDatabaseType {
 	TEMP_DATABASE,
 };
 
+struct StoredDatabasePath {
+	StoredDatabasePath(DatabaseManager &manager, string path, const string &name);
+	~StoredDatabasePath();
+
+	DatabaseManager &manager;
+	string path;
+};
+
 //! AttachOptions holds information about a database we plan to attach. These options are generalized, i.e.,
 //! they have to apply to any database file type (duckdb, sqlite, etc.).
 struct AttachOptions {
@@ -51,7 +59,7 @@ struct AttachOptions {
 };
 
 //! The AttachedDatabase represents an attached database instance.
-class AttachedDatabase : public CatalogEntry {
+class AttachedDatabase : public CatalogEntry, public enable_shared_from_this<AttachedDatabase> {
 public:
 	//! Create the built-in system database (without storage).
 	explicit AttachedDatabase(DatabaseInstance &db, AttachedDatabaseType type = AttachedDatabaseType::SYSTEM_DATABASE);
@@ -94,8 +102,12 @@ class AttachedDatabase : public CatalogEntry {
 	static bool NameIsReserved(const string &name);
 	static string ExtractDatabaseName(const string &dbpath, FileSystem &fs);
 
+private:
+	void InsertDatabasePath(const string &path);
+
 private:
 	DatabaseInstance &db;
+	unique_ptr<StoredDatabasePath> stored_database_path;
 	unique_ptr<StorageManager> storage;
 	unique_ptr<Catalog> catalog;
 	unique_ptr<TransactionManager> transaction_manager;
diff --git a/src/duckdb/src/include/duckdb/main/database.hpp b/src/duckdb/src/include/duckdb/main/database.hpp
index d3c5fb9bd..fa573b5da 100644
--- a/src/duckdb/src/include/duckdb/main/database.hpp
+++ b/src/duckdb/src/include/duckdb/main/database.hpp
@@ -75,7 +75,7 @@ class DatabaseInstance : public enable_shared_from_this<DatabaseInstance> {
 
 	DUCKDB_API SettingLookupResult TryGetCurrentSetting(const string &key, Value &result) const;
 
-	unique_ptr<AttachedDatabase> CreateAttachedDatabase(ClientContext &context, AttachInfo &info,
+	shared_ptr<AttachedDatabase> CreateAttachedDatabase(ClientContext &context, AttachInfo &info,
 	                                                    AttachOptions &options);
 
 	void AddExtensionInfo(const string &name, const ExtensionLoadedInfo &info);
diff --git a/src/duckdb/src/include/duckdb/main/database_manager.hpp b/src/duckdb/src/include/duckdb/main/database_manager.hpp
index 4fb8690d9..34a933d67 100644
--- a/src/duckdb/src/include/duckdb/main/database_manager.hpp
+++ b/src/duckdb/src/include/duckdb/main/database_manager.hpp
@@ -48,9 +48,14 @@ class DatabaseManager {
 	//! Get an attached database by its name
 	optional_ptr<AttachedDatabase> GetDatabase(ClientContext &context, const string &name);
 	//! Attach a new database
-	optional_ptr<AttachedDatabase> AttachDatabase(ClientContext &context, AttachInfo &info, AttachOptions &options);
+	shared_ptr<AttachedDatabase> AttachDatabase(ClientContext &context, AttachInfo &info, AttachOptions &options);
+
+	optional_ptr<AttachedDatabase> FinalizeAttach(ClientContext &context, AttachInfo &info,
+	                                              shared_ptr<AttachedDatabase> database);
 	//! Detach an existing database
 	void DetachDatabase(ClientContext &context, const string &name, OnEntryNotFound if_not_found);
+	//! Rollback the attach of a database
+	shared_ptr<AttachedDatabase> DetachInternal(const string &name);
 	//! Returns a reference to the system catalog
 	Catalog &GetSystemCatalog();
 
@@ -58,9 +63,11 @@ class DatabaseManager {
 	void SetDefaultDatabase(ClientContext &context, const string &new_value);
 
 	//! Inserts a path to name mapping to the database paths map
-	void InsertDatabasePath(ClientContext &context, const string &path, const string &name);
+	void InsertDatabasePath(const string &path, const string &name);
 	//! Erases a path from the database paths map
 	void EraseDatabasePath(const string &path);
+	//! Check if a path has a conflict
+	void CheckPathConflict(const string &path, const string &name);
 
 	//! Returns the database type. This might require checking the header of the file, in which case the file handle is
 	//! necessary. We can only grab the file handle, if it is not yet held, even for uncommitted changes. Thus, we have
@@ -68,10 +75,10 @@ class DatabaseManager {
 	void GetDatabaseType(ClientContext &context, AttachInfo &info, const DBConfig &config, AttachOptions &options);
 	//! Scans the catalog set and adds each committed database entry, and each database entry of the current
 	//! transaction, to a vector holding AttachedDatabase references
-	vector<reference<AttachedDatabase>> GetDatabases(ClientContext &context,
-	                                                 const optional_idx max_db_count = optional_idx());
+	vector<shared_ptr<AttachedDatabase>> GetDatabases(ClientContext &context,
+	                                                  const optional_idx max_db_count = optional_idx());
 	//! Scans the catalog set and returns each committed database entry
-	vector<reference<AttachedDatabase>> GetDatabases();
+	vector<shared_ptr<AttachedDatabase>> GetDatabases();
 	//! Removes all databases from the catalog set. This is necessary for the database instance's destructor,
 	//! as the database manager has to be alive when destroying the catalog set objects.
 	void ResetDatabases(unique_ptr<TaskScheduler> &scheduler);
@@ -97,14 +104,13 @@ class DatabaseManager {
 	//! Gets a list of all attached database paths
 	vector<string> GetAttachedDatabasePaths();
 
-private:
-	void CheckPathConflict(ClientContext &context, const string &path);
-
 private:
 	//! The system database is a special database that holds system entries (e.g. functions)
-	unique_ptr<AttachedDatabase> system;
+	shared_ptr<AttachedDatabase> system;
+	//! Lock for databases
+	mutex databases_lock;
 	//! The set of attached databases
-	unique_ptr<CatalogSet> databases;
+	case_insensitive_map_t<shared_ptr<AttachedDatabase>> databases;
 	//! The next object id handed out by the NextOid method
 	atomic<idx_t> next_oid;
 	//! The current query number
diff --git a/src/duckdb/src/include/duckdb/transaction/duck_transaction.hpp b/src/duckdb/src/include/duckdb/transaction/duck_transaction.hpp
index 5d911e7d9..e6e494b10 100644
--- a/src/duckdb/src/include/duckdb/transaction/duck_transaction.hpp
+++ b/src/duckdb/src/include/duckdb/transaction/duck_transaction.hpp
@@ -51,6 +51,7 @@ class DuckTransaction : public Transaction {
 	LocalStorage &GetLocalStorage();
 
 	void PushCatalogEntry(CatalogEntry &entry, data_ptr_t extra_data, idx_t extra_data_size);
+	void PushAttach(AttachedDatabase &db);
 
 	void SetReadWrite() override;
 
diff --git a/src/duckdb/src/include/duckdb/transaction/duck_transaction_manager.hpp b/src/duckdb/src/include/duckdb/transaction/duck_transaction_manager.hpp
index de1249b89..63531ae7d 100644
--- a/src/duckdb/src/include/duckdb/transaction/duck_transaction_manager.hpp
+++ b/src/duckdb/src/include/duckdb/transaction/duck_transaction_manager.hpp
@@ -70,6 +70,7 @@ class DuckTransactionManager : public TransactionManager {
 
 	void PushCatalogEntry(Transaction &transaction_p, CatalogEntry &entry, data_ptr_t extra_data = nullptr,
 	                      idx_t extra_data_size = 0);
+	void PushAttach(Transaction &transaction_p, AttachedDatabase &db);
 
 protected:
 	struct CheckpointDecision {
diff --git a/src/duckdb/src/include/duckdb/transaction/meta_transaction.hpp b/src/duckdb/src/include/duckdb/transaction/meta_transaction.hpp
index ca6b2c920..42a412fcd 100644
--- a/src/duckdb/src/include/duckdb/transaction/meta_transaction.hpp
+++ b/src/duckdb/src/include/duckdb/transaction/meta_transaction.hpp
@@ -63,6 +63,7 @@ class MetaTransaction {
 	const vector<reference<AttachedDatabase>> &OpenedTransactions() const {
 		return all_transactions;
 	}
+	AttachedDatabase &UseDatabase(shared_ptr<AttachedDatabase> &database);
 
 private:
 	//! Lock to prevent all_transactions and transactions from getting out of sync
@@ -75,6 +76,8 @@ class MetaTransaction {
 	optional_ptr<AttachedDatabase> modified_database;
 	//! Whether or not the meta transaction is marked as read only
 	bool is_read_only;
+	//! The set of used / referenced databases
+	reference_map_t<AttachedDatabase, shared_ptr<AttachedDatabase>> referenced_databases;
 };
 
 } // namespace duckdb
diff --git a/src/duckdb/src/main/attached_database.cpp b/src/duckdb/src/main/attached_database.cpp
index 51b03744a..fca4a051a 100644
--- a/src/duckdb/src/main/attached_database.cpp
+++ b/src/duckdb/src/main/attached_database.cpp
@@ -13,6 +13,14 @@
 
 namespace duckdb {
 
+StoredDatabasePath::StoredDatabasePath(DatabaseManager &manager, string path_p, const string &name)
+    : manager(manager), path(std::move(path_p)) {
+	manager.InsertDatabasePath(path, name);
+}
+
+StoredDatabasePath::~StoredDatabasePath() {
+	manager.EraseDatabasePath(path);
+}
 //===--------------------------------------------------------------------===//
 // Attach Options
 //===--------------------------------------------------------------------===//
@@ -66,7 +74,6 @@ AttachOptions::AttachOptions(const unique_ptr<AttachInfo> &info, const AccessMod
 //===--------------------------------------------------------------------===//
 // Attached Database
 //===--------------------------------------------------------------------===//
-
 AttachedDatabase::AttachedDatabase(DatabaseInstance &db, AttachedDatabaseType type)
     : CatalogEntry(CatalogType::DATABASE_ENTRY,
                    type == AttachedDatabaseType::SYSTEM_DATABASE ? SYSTEM_CATALOG : TEMP_CATALOG, 0),
@@ -86,7 +93,6 @@ AttachedDatabase::AttachedDatabase(DatabaseInstance &db, AttachedDatabaseType ty
 AttachedDatabase::AttachedDatabase(DatabaseInstance &db, Catalog &catalog_p, string name_p, string file_path_p,
                                    AttachOptions &options)
     : CatalogEntry(CatalogType::DATABASE_ENTRY, catalog_p, std::move(name_p)), db(db), parent_catalog(&catalog_p) {
-
 	if (options.access_mode == AccessMode::READ_ONLY) {
 		type = AttachedDatabaseType::READ_ONLY_DATABASE;
 	} else {
@@ -107,8 +113,10 @@ AttachedDatabase::AttachedDatabase(DatabaseInstance &db, Catalog &catalog_p, str
 		}
 		throw BinderException("Unrecognized option for attach \"%s\"", entry.first);
 	}
+
 	// We create the storage after the catalog to guarantee we allow extensions to instantiate the DuckCatalog.
 	catalog = make_uniq<DuckCatalog>(*this);
+	InsertDatabasePath(file_path_p);
 	auto read_only = options.access_mode == AccessMode::READ_ONLY;
 	storage = make_uniq<SingleFileStorageManager>(*this, std::move(file_path_p), read_only);
 	transaction_manager = make_uniq<DuckTransactionManager>(*this);
@@ -131,6 +139,7 @@ AttachedDatabase::AttachedDatabase(DatabaseInstance &db, Catalog &catalog_p, Sto
 		throw InternalException("AttachedDatabase - attach function did not return a catalog");
 	}
 	if (catalog->IsDuckCatalog()) {
+		InsertDatabasePath(info.path);
 		// The attached database uses the DuckCatalog.
 		auto read_only = options.access_mode == AccessMode::READ_ONLY;
 		storage = make_uniq<SingleFileStorageManager>(*this, info.path, read_only);
@@ -159,6 +168,13 @@ bool AttachedDatabase::IsReadOnly() const {
 	return type == AttachedDatabaseType::READ_ONLY_DATABASE;
 }
 
+void AttachedDatabase::InsertDatabasePath(const string &path) {
+	if (path.empty() || path == IN_MEMORY_PATH) {
+		return;
+	}
+	stored_database_path = make_uniq<StoredDatabasePath>(db.GetDatabaseManager(), path, name);
+}
+
 bool AttachedDatabase::NameIsReserved(const string &name) {
 	return name == DEFAULT_SCHEMA || name == TEMP_CATALOG || name == SYSTEM_CATALOG;
 }
@@ -232,38 +248,31 @@ void AttachedDatabase::OnDetach(ClientContext &context) {
 }
 
 void AttachedDatabase::Close() {
-	D_ASSERT(catalog);
 	if (is_closed) {
 		return;
 	}
+	D_ASSERT(catalog);
 	is_closed = true;
 
-	if (!IsSystem() && !catalog->InMemory()) {
-		db.GetDatabaseManager().EraseDatabasePath(catalog->GetDBPath());
-	}
-
-	if (Exception::UncaughtException()) {
-		return;
-	}
-	if (!storage) {
-		return;
-	}
-
 	// shutting down: attempt to checkpoint the database
 	// but only if we are not cleaning up as part of an exception unwind
-	try {
-		if (!storage->InMemory()) {
+	if (!Exception::UncaughtException() && storage && !storage->InMemory()) {
+		try {
 			auto &config = DBConfig::GetConfig(db);
-			if (!config.options.checkpoint_on_shutdown) {
-				return;
+			if (config.options.checkpoint_on_shutdown) {
+				CheckpointOptions options;
+				options.wal_action = CheckpointWALAction::DELETE_WAL;
+				storage->CreateCheckpoint(nullptr, options);
 			}
-			CheckpointOptions options;
-			options.wal_action = CheckpointWALAction::DELETE_WAL;
-			storage->CreateCheckpoint(nullptr, options);
+		} catch (...) { // NOLINT
 		}
-	} catch (...) { // NOLINT
 	}
 
+	transaction_manager.reset();
+	catalog.reset();
+	storage.reset();
+	stored_database_path.reset();
+
 	if (Allocator::SupportsFlush()) {
 		Allocator::FlushAll();
 	}
diff --git a/src/duckdb/src/main/client_context.cpp b/src/duckdb/src/main/client_context.cpp
index 9b561ecc1..ca053937a 100644
--- a/src/duckdb/src/main/client_context.cpp
+++ b/src/duckdb/src/main/client_context.cpp
@@ -496,7 +496,8 @@ void ClientContext::CheckIfPreparedStatementIsExecutable(PreparedStatementData &
 		auto &modified_database = it.first;
 		auto entry = manager.GetDatabase(*this, modified_database);
 		if (!entry) {
-			throw InternalException("Database \"%s\" not found", modified_database);
+			// database has been detached
+			throw InvalidInputException("Database \"%s\" not found", modified_database);
 		}
 		if (entry->IsReadOnly()) {
 			throw InvalidInputException(StringUtil::Format(
diff --git a/src/duckdb/src/main/database.cpp b/src/duckdb/src/main/database.cpp
index 596f98bd9..b2a6b9530 100644
--- a/src/duckdb/src/main/database.cpp
+++ b/src/duckdb/src/main/database.cpp
@@ -172,9 +172,9 @@ ConnectionManager &ConnectionManager::Get(ClientContext &context) {
 	return ConnectionManager::Get(DatabaseInstance::GetDatabase(context));
 }
 
-unique_ptr<AttachedDatabase> DatabaseInstance::CreateAttachedDatabase(ClientContext &context, AttachInfo &info,
+shared_ptr<AttachedDatabase> DatabaseInstance::CreateAttachedDatabase(ClientContext &context, AttachInfo &info,
                                                                       AttachOptions &options) {
-	unique_ptr<AttachedDatabase> attached_database;
+	shared_ptr<AttachedDatabase> attached_database;
 	auto &catalog = Catalog::GetSystemCatalog(*this);
 
 	if (!options.db_type.empty()) {
@@ -188,16 +188,16 @@ unique_ptr<AttachedDatabase> DatabaseInstance::CreateAttachedDatabase(ClientCont
 		if (entry->second->attach != nullptr && entry->second->create_transaction_manager != nullptr) {
 			// Use the storage extension to create the initial database.
 			attached_database =
-			    make_uniq<AttachedDatabase>(*this, catalog, *entry->second, context, info.name, info, options);
+			    make_shared_ptr<AttachedDatabase>(*this, catalog, *entry->second, context, info.name, info, options);
 			return attached_database;
 		}
 
-		attached_database = make_uniq<AttachedDatabase>(*this, catalog, info.name, info.path, options);
+		attached_database = make_shared_ptr<AttachedDatabase>(*this, catalog, info.name, info.path, options);
 		return attached_database;
 	}
 
 	// An empty db_type defaults to a duckdb database file.
-	attached_database = make_uniq<AttachedDatabase>(*this, catalog, info.name, info.path, options);
+	attached_database = make_shared_ptr<AttachedDatabase>(*this, catalog, info.name, info.path, options);
 	return attached_database;
 }
 
@@ -206,14 +206,13 @@ void DatabaseInstance::CreateMainDatabase() {
 	info.name = AttachedDatabase::ExtractDatabaseName(config.options.database_path, GetFileSystem());
 	info.path = config.options.database_path;
 
-	optional_ptr<AttachedDatabase> initial_database;
 	Connection con(*this);
 	con.BeginTransaction();
 	AttachOptions options(config.options);
-	initial_database = db_manager->AttachDatabase(*con.context, info, options);
-
+	auto initial_database = db_manager->AttachDatabase(*con.context, info, options);
 	initial_database->SetInitialDatabase();
 	initial_database->Initialize(*con.context);
+	db_manager->FinalizeAttach(*con.context, info, std::move(initial_database));
 	con.Commit();
 }
 
diff --git a/src/duckdb/src/main/database_manager.cpp b/src/duckdb/src/main/database_manager.cpp
index f3b1599a2..5dc833d9d 100644
--- a/src/duckdb/src/main/database_manager.cpp
+++ b/src/duckdb/src/main/database_manager.cpp
@@ -8,13 +8,14 @@
 #include "duckdb/main/database_path_and_type.hpp"
 #include "duckdb/main/extension_helper.hpp"
 #include "duckdb/storage/storage_manager.hpp"
+#include "duckdb/transaction/duck_transaction.hpp"
+#include "duckdb/transaction/duck_transaction_manager.hpp"
 
 namespace duckdb {
 
 DatabaseManager::DatabaseManager(DatabaseInstance &db)
     : next_oid(0), current_query_number(1), current_transaction_id(0) {
-	system = make_uniq<AttachedDatabase>(db);
-	databases = make_uniq<CatalogSet>(system->GetCatalog());
+	system = make_shared_ptr<AttachedDatabase>(db);
 }
 
 DatabaseManager::~DatabaseManager() {
@@ -32,22 +33,29 @@ void DatabaseManager::InitializeSystemCatalog() {
 void DatabaseManager::FinalizeStartup() {
 	auto dbs = GetDatabases();
 	for (auto &db : dbs) {
-		db.get().FinalizeLoad(nullptr);
+		db->FinalizeLoad(nullptr);
 	}
 }
 
 optional_ptr<AttachedDatabase> DatabaseManager::GetDatabase(ClientContext &context, const string &name) {
+	auto &meta_transaction = MetaTransaction::Get(context);
+	lock_guard<mutex> guard(databases_lock);
 	if (StringUtil::Lower(name) == TEMP_CATALOG) {
-		return context.client_data->temporary_objects.get();
+		return meta_transaction.UseDatabase(context.client_data->temporary_objects);
 	}
 	if (StringUtil::Lower(name) == SYSTEM_CATALOG) {
-		return system;
+		return meta_transaction.UseDatabase(system);
 	}
-	return reinterpret_cast<AttachedDatabase *>(databases->GetEntry(context, name).get());
+	auto entry = databases.find(name);
+	if (entry == databases.end()) {
+		// not found
+		return nullptr;
+	}
+	return meta_transaction.UseDatabase(entry->second);
 }
 
-optional_ptr<AttachedDatabase> DatabaseManager::AttachDatabase(ClientContext &context, AttachInfo &info,
-                                                               AttachOptions &options) {
+shared_ptr<AttachedDatabase> DatabaseManager::AttachDatabase(ClientContext &context, AttachInfo &info,
+                                                             AttachOptions &options) {
 	if (AttachedDatabase::NameIsReserved(info.name)) {
 		throw BinderException("Attached database name \"%s\" cannot be used because it is a reserved name", info.name);
 	}
@@ -68,27 +76,32 @@ optional_ptr<AttachedDatabase> DatabaseManager::AttachDatabase(ClientContext &co
 	// now create the attached database
 	auto &db = DatabaseInstance::GetDatabase(context);
 	auto attached_db = db.CreateAttachedDatabase(context, info, options);
+	return attached_db;
+}
 
-	if (options.db_type.empty()) {
-		InsertDatabasePath(context, info.path, attached_db->name);
-	}
-
+optional_ptr<AttachedDatabase> DatabaseManager::FinalizeAttach(ClientContext &context, AttachInfo &info,
+                                                               shared_ptr<AttachedDatabase> attached_db) {
 	const auto name = attached_db->GetName();
 	attached_db->oid = NextOid();
-	LogicalDependencyList dependencies;
 	if (default_database.empty()) {
 		default_database = name;
 	}
-
-	// and add it to the databases catalog set
 	if (info.on_conflict == OnCreateConflict::REPLACE_ON_CONFLICT) {
 		DetachDatabase(context, name, OnEntryNotFound::RETURN_NULL);
 	}
-	if (!databases->CreateEntry(context, name, std::move(attached_db), dependencies)) {
-		throw BinderException("Failed to attach database: database with name \"%s\" already exists", name);
+	{
+		lock_guard<mutex> guard(databases_lock);
+		auto entry = databases.emplace(name, attached_db);
+		if (!entry.second) {
+			throw BinderException("Failed to attach database: database with name \"%s\" already exists", name);
+		}
 	}
-
-	return GetDatabase(context, name);
+	auto &meta_transaction = MetaTransaction::Get(context);
+	auto &db_ref = meta_transaction.UseDatabase(attached_db);
+	auto &transaction = DuckTransaction::Get(context, *system);
+	auto &transaction_manager = DuckTransactionManager::Get(*system);
+	transaction_manager.PushAttach(transaction, db_ref);
+	return db_ref;
 }
 
 void DatabaseManager::DetachDatabase(ClientContext &context, const string &name, OnEntryNotFound if_not_found) {
@@ -98,59 +111,57 @@ void DatabaseManager::DetachDatabase(ClientContext &context, const string &name,
 		                      name);
 	}
 
-	auto entry = databases->GetEntry(context, name);
-	if (!entry) {
+	auto attached_db = DetachInternal(name);
+	if (!attached_db) {
 		if (if_not_found == OnEntryNotFound::THROW_EXCEPTION) {
 			throw BinderException("Failed to detach database with name \"%s\": database not found", name);
 		}
 		return;
 	}
-	auto &db = entry->Cast<AttachedDatabase>();
-	db.OnDetach(context);
 
-	if (!databases->DropEntry(context, name, false, true)) {
-		throw InternalException("Failed to drop attached database");
-	}
+	attached_db->OnDetach(context);
 }
 
-void DatabaseManager::CheckPathConflict(ClientContext &context, const string &path) {
-	// Ensure that we did not already attach a database with the same path.
-	string db_name = "";
+shared_ptr<AttachedDatabase> DatabaseManager::DetachInternal(const string &name) {
+	shared_ptr<AttachedDatabase> attached_db;
 	{
-		lock_guard<mutex> path_lock(db_paths_lock);
-		auto it = db_paths_to_name.find(path);
-		if (it != db_paths_to_name.end()) {
-			db_name = it->second;
+		lock_guard<mutex> guard(databases_lock);
+		auto entry = databases.find(name);
+		if (entry == databases.end()) {
+			return nullptr;
 		}
+		attached_db = std::move(entry->second);
+		databases.erase(entry);
 	}
-	if (db_name.empty()) {
-		return;
-	}
+	return attached_db;
+}
 
-	// Check against the catalog set.
-	auto entry = GetDatabase(context, db_name);
-	if (!entry) {
-		return;
-	}
-	if (entry->IsSystem()) {
+void DatabaseManager::CheckPathConflict(const string &path, const string &name) {
+	if (path.empty() || path == IN_MEMORY_PATH) {
 		return;
 	}
-	auto &catalog = Catalog::GetCatalog(*entry);
-	if (catalog.InMemory()) {
-		return;
+
+	lock_guard<mutex> path_lock(db_paths_lock);
+	auto entry = db_paths_to_name.find(path);
+	if (entry != db_paths_to_name.end()) {
+		throw BinderException("Unique file handle conflict: Cannot attach \"%s\" - the database file \"%s\" is already "
+		                      "attached by database \"%s\"",
+		                      name, path, entry->second);
 	}
-	throw BinderException("Unique file handle conflict: Database \"%s\" is already attached with path \"%s\", ",
-	                      db_name, path);
 }
 
-void DatabaseManager::InsertDatabasePath(ClientContext &context, const string &path, const string &name) {
+void DatabaseManager::InsertDatabasePath(const string &path, const string &name) {
 	if (path.empty() || path == IN_MEMORY_PATH) {
 		return;
 	}
 
-	CheckPathConflict(context, path);
 	lock_guard<mutex> path_lock(db_paths_lock);
-	db_paths_to_name[path] = name;
+	auto entry = db_paths_to_name.emplace(path, name);
+	if (!entry.second) {
+		throw BinderException("Unique file handle conflict: Cannot attach \"%s\" - the database file \"%s\" is already "
+		                      "attached by database \"%s\"",
+		                      name, path, entry.first->second);
+	}
 }
 
 void DatabaseManager::EraseDatabasePath(const string &path) {
@@ -158,10 +169,7 @@ void DatabaseManager::EraseDatabasePath(const string &path) {
 		return;
 	}
 	lock_guard<mutex> path_lock(db_paths_lock);
-	auto path_it = db_paths_to_name.find(path);
-	if (path_it != db_paths_to_name.end()) {
-		db_paths_to_name.erase(path_it);
-	}
+	db_paths_to_name.erase(path);
 }
 
 vector<string> DatabaseManager::GetAttachedDatabasePaths() {
@@ -184,9 +192,8 @@ void DatabaseManager::GetDatabaseType(ClientContext &context, AttachInfo &info,
 
 	// Try to extract the database type from the path.
 	if (options.db_type.empty()) {
-		CheckPathConflict(context, info.path);
-
 		auto &fs = FileSystem::GetFileSystem(context);
+		CheckPathConflict(info.path, info.name);
 		DBPathAndType::CheckMagicBytes(fs, info.path, options.db_type);
 	}
 
@@ -237,44 +244,46 @@ void DatabaseManager::SetDefaultDatabase(ClientContext &context, const string &n
 }
 // LCOV_EXCL_STOP
 
-vector<reference<AttachedDatabase>> DatabaseManager::GetDatabases(ClientContext &context,
-                                                                  const optional_idx max_db_count) {
-	vector<reference<AttachedDatabase>> result;
+vector<shared_ptr<AttachedDatabase>> DatabaseManager::GetDatabases(ClientContext &context,
+                                                                   const optional_idx max_db_count) {
+	vector<shared_ptr<AttachedDatabase>> result;
+
+	lock_guard<mutex> guard(databases_lock);
 	idx_t count = 2;
-	databases->ScanWithReturn(context, [&](CatalogEntry &entry) {
+	for (auto &entry : databases) {
 		if (max_db_count.IsValid() && count >= max_db_count.GetIndex()) {
-			return false;
+			break;
 		}
-		result.push_back(entry.Cast<AttachedDatabase>());
+		result.push_back(entry.second);
 		count++;
-		return true;
-	});
-
+	}
 	if (!max_db_count.IsValid() || max_db_count.GetIndex() >= 1) {
-		result.push_back(*system);
+		result.push_back(system);
 	}
 	if (!max_db_count.IsValid() || max_db_count.GetIndex() >= 2) {
-		result.push_back(*context.client_data->temporary_objects);
+		result.push_back(context.client_data->temporary_objects);
 	}
 
 	return result;
 }
 
-vector<reference<AttachedDatabase>> DatabaseManager::GetDatabases() {
-	vector<reference<AttachedDatabase>> result;
-	databases->Scan([&](CatalogEntry &entry) { result.push_back(entry.Cast<AttachedDatabase>()); });
-	result.push_back(*system);
+vector<shared_ptr<AttachedDatabase>> DatabaseManager::GetDatabases() {
+	vector<shared_ptr<AttachedDatabase>> result;
+
+	lock_guard<mutex> guard(databases_lock);
+	for (auto &entry : databases) {
+		result.push_back(entry.second);
+	}
+	result.push_back(system);
 	return result;
 }
 
 void DatabaseManager::ResetDatabases(unique_ptr<TaskScheduler> &scheduler) {
-	vector<reference<AttachedDatabase>> result;
-	databases->Scan([&](CatalogEntry &entry) { result.push_back(entry.Cast<AttachedDatabase>()); });
-	for (auto &database : result) {
-		database.get().Close();
+	auto databases = GetDatabases();
+	for (auto &entry : databases) {
+		entry->Close();
+		entry.reset();
 	}
-	scheduler.reset();
-	databases.reset();
 }
 
 Catalog &DatabaseManager::GetSystemCatalog() {
diff --git a/src/duckdb/src/transaction/commit_state.cpp b/src/duckdb/src/transaction/commit_state.cpp
index d4b934d58..0f5d75bd2 100644
--- a/src/duckdb/src/transaction/commit_state.cpp
+++ b/src/duckdb/src/transaction/commit_state.cpp
@@ -182,6 +182,7 @@ void CommitState::CommitEntry(UndoFlags type, data_ptr_t data) {
 		info->version_number = commit_id;
 		break;
 	}
+	case UndoFlags::ATTACHED_DATABASE:
 	case UndoFlags::SEQUENCE_VALUE: {
 		break;
 	}
@@ -222,6 +223,7 @@ void CommitState::RevertCommit(UndoFlags type, data_ptr_t data) {
 		info->version_number = transaction_id;
 		break;
 	}
+	case UndoFlags::ATTACHED_DATABASE:
 	case UndoFlags::SEQUENCE_VALUE: {
 		break;
 	}
diff --git a/src/duckdb/src/transaction/duck_transaction.cpp b/src/duckdb/src/transaction/duck_transaction.cpp
index 959c62ea2..cf051ab6e 100644
--- a/src/duckdb/src/transaction/duck_transaction.cpp
+++ b/src/duckdb/src/transaction/duck_transaction.cpp
@@ -76,6 +76,13 @@ void DuckTransaction::PushCatalogEntry(CatalogEntry &entry, data_ptr_t extra_dat
 	}
 }
 
+void DuckTransaction::PushAttach(AttachedDatabase &db) {
+	auto undo_entry = undo_buffer.CreateEntry(UndoFlags::ATTACHED_DATABASE, sizeof(AttachedDatabase *));
+	auto ptr = undo_entry.Ptr();
+	// store the pointer to the database
+	Store<CatalogEntry *>(&db, ptr);
+}
+
 void DuckTransaction::PushDelete(DataTable &table, RowVersionManager &info, idx_t vector_idx, row_t rows[], idx_t count,
                                  idx_t base_row) {
 	ModifyTable(table);
diff --git a/src/duckdb/src/transaction/duck_transaction_manager.cpp b/src/duckdb/src/transaction/duck_transaction_manager.cpp
index 2a1718e48..6465aa83b 100644
--- a/src/duckdb/src/transaction/duck_transaction_manager.cpp
+++ b/src/duckdb/src/transaction/duck_transaction_manager.cpp
@@ -518,4 +518,13 @@ void DuckTransactionManager::PushCatalogEntry(Transaction &transaction_p, duckdb
 	transaction.PushCatalogEntry(entry, extra_data, extra_data_size);
 }
 
+void DuckTransactionManager::PushAttach(Transaction &transaction_p, AttachedDatabase &attached_db) {
+	auto &transaction = transaction_p.Cast<DuckTransaction>();
+	if (!db.IsSystem()) {
+		throw InternalException("Can only ATTACH in the system catalog");
+	}
+	transaction.catalog_version = ++last_uncommitted_catalog_version;
+	transaction.PushAttach(attached_db);
+}
+
 } // namespace duckdb
diff --git a/src/duckdb/src/transaction/meta_transaction.cpp b/src/duckdb/src/transaction/meta_transaction.cpp
index 1a4749273..7a2205ffc 100644
--- a/src/duckdb/src/transaction/meta_transaction.cpp
+++ b/src/duckdb/src/transaction/meta_transaction.cpp
@@ -64,6 +64,7 @@ Transaction &MetaTransaction::GetTransaction(AttachedDatabase &db) {
 #endif
 		all_transactions.push_back(db);
 		transactions.insert(make_pair(reference<AttachedDatabase>(db), reference<Transaction>(new_transaction)));
+		referenced_databases.insert(make_pair(reference<AttachedDatabase>(db), db.shared_from_this()));
 
 		return new_transaction;
 	} else {
@@ -171,6 +172,15 @@ void MetaTransaction::SetActiveQuery(transaction_t query_number) {
 	}
 }
 
+AttachedDatabase &MetaTransaction::UseDatabase(shared_ptr<AttachedDatabase> &database) {
+	auto &db_ref = *database;
+	auto entry = referenced_databases.find(db_ref);
+	if (entry == referenced_databases.end()) {
+		referenced_databases.emplace(reference<AttachedDatabase>(db_ref), database);
+	}
+	return db_ref;
+}
+
 void MetaTransaction::ModifyDatabase(AttachedDatabase &db) {
 	if (db.IsSystem() || db.IsTemporary()) {
 		// we can always modify the system and temp databases
diff --git a/src/duckdb/src/transaction/rollback_state.cpp b/src/duckdb/src/transaction/rollback_state.cpp
index 335c35eff..19fd74d08 100644
--- a/src/duckdb/src/transaction/rollback_state.cpp
+++ b/src/duckdb/src/transaction/rollback_state.cpp
@@ -10,6 +10,7 @@
 #include "duckdb/storage/data_table.hpp"
 #include "duckdb/storage/table/update_segment.hpp"
 #include "duckdb/storage/table/row_version_manager.hpp"
+#include "duckdb/main/attached_database.hpp"
 
 namespace duckdb {
 
@@ -42,6 +43,12 @@ void RollbackState::RollbackEntry(UndoFlags type, data_ptr_t data) {
 		info->segment->RollbackUpdate(*info);
 		break;
 	}
+	case UndoFlags::ATTACHED_DATABASE: {
+		auto db = Load<AttachedDatabase *>(data);
+		auto &db_manager = DatabaseManager::Get(db->GetDatabase());
+		db_manager.DetachInternal(db->name);
+		break;
+	}
 	case UndoFlags::SEQUENCE_VALUE:
 		break;
 	default: // LCOV_EXCL_START
diff --git a/src/duckdb/src/transaction/wal_write_state.cpp b/src/duckdb/src/transaction/wal_write_state.cpp
index 275a12571..5fe17e050 100644
--- a/src/duckdb/src/transaction/wal_write_state.cpp
+++ b/src/duckdb/src/transaction/wal_write_state.cpp
@@ -281,6 +281,8 @@ void WALWriteState::CommitEntry(UndoFlags type, data_ptr_t data) {
 		}
 		break;
 	}
+	case UndoFlags::ATTACHED_DATABASE:
+		break;
 	case UndoFlags::SEQUENCE_VALUE: {
 		auto info = reinterpret_cast<SequenceValue *>(data);
 		log.WriteSequenceValue(*info);