fix: concurrent mysql catalog loading access (1.4)

This is a backport of the PR #197 to `v1.4-andium` stable branch.

Based on postgresql extension entry/loading lock guards.

Author: ikrestov

src/include/storage/mysql_catalog_set.hpp

@@ -30,15 +30,18 @@ class MySQLCatalogSet {
 protected:
 	virtual void LoadEntries(ClientContext &context) = 0;
 
+	void TryLoadEntries(ClientContext &context);
+
 	void EraseEntryInternal(const string &name);
 
 protected:
 	Catalog &catalog;
 
 private:
 	mutex entry_lock;
+	mutex load_lock;
 	case_insensitive_map_t<unique_ptr<CatalogEntry>> entries;
-	bool is_loaded;
+	atomic<bool> is_loaded;
 };
 
 class MySQLInSchemaSet : public MySQLCatalogSet {

src/storage/mysql_catalog_set.cpp

@@ -9,10 +9,7 @@ MySQLCatalogSet::MySQLCatalogSet(Catalog &catalog) : catalog(catalog), is_loaded
 }
 
 optional_ptr<CatalogEntry> MySQLCatalogSet::GetEntry(ClientContext &context, const string &name) {
-	if (!is_loaded) {
-		is_loaded = true;
-		LoadEntries(context);
-	}
+	TryLoadEntries(context);
 	lock_guard<mutex> l(entry_lock);
 	auto entry = entries.find(name);
 	if (entry == entries.end()) {
@@ -21,6 +18,15 @@ optional_ptr<CatalogEntry> MySQLCatalogSet::GetEntry(ClientContext &context, con
 	return entry->second.get();
 }
 
+void MySQLCatalogSet::TryLoadEntries(ClientContext &context) {
+	lock_guard<mutex> l(load_lock);
+	if (is_loaded) {
+		return;
+	}
+	is_loaded = true;
+	LoadEntries(context);
+}
+
 void MySQLCatalogSet::DropEntry(ClientContext &context, DropInfo &info) {
 	string drop_query = "DROP ";
 	drop_query += CatalogTypeToString(info.type) + " ";
@@ -46,10 +52,7 @@ void MySQLCatalogSet::EraseEntryInternal(const string &name) {
 }
 
 void MySQLCatalogSet::Scan(ClientContext &context, const std::function<void(CatalogEntry &)> &callback) {
-	if (!is_loaded) {
-		is_loaded = true;
-		LoadEntries(context);
-	}
+	TryLoadEntries(context);
 	lock_guard<mutex> l(entry_lock);
 	for (auto &entry : entries) {
 		callback(*entry.second);
@@ -67,6 +70,7 @@ optional_ptr<CatalogEntry> MySQLCatalogSet::CreateEntry(unique_ptr<CatalogEntry>
 }
 
 void MySQLCatalogSet::ClearEntries() {
+	lock_guard<mutex> l(entry_lock);
 	entries.clear();
 	is_loaded = false;
 }