fix(pybind): load each import-cache entry once, under a lock

src_cpp/cached_import/py_cached_import.cpp

@@ -11,6 +11,13 @@ PythonCachedImport::~PythonCachedImport() {
     allObjects.clear();
 }
 
+std::unique_lock<std::mutex> PythonCachedImport::lockForLoading() {
+    // Wait with the GIL released (detached on free-threaded builds): the holder needs the GIL to
+    // finish its import, and a detached waiter does not stall stop-the-world pauses.
+    py::gil_scoped_release release;
+    return std::unique_lock<std::mutex>(loadMutex);
+}
+
 py::handle PythonCachedImport::addToCache(py::object obj) {
     auto ptr = obj.ptr();
     allObjects.push_back(obj);

src_cpp/cached_import/py_cached_item.cpp

@@ -7,16 +7,18 @@ namespace lbug {
 
 py::handle PythonCachedItem::operator()() {
     assert((bool)PyGILState_Check());
-    // load if unloaded, return cached object if already loaded
-    if (loaded) {
+    if (isLoaded()) {
         return object;
     }
-    if (parent == nullptr) {
-        object = importCache->addToCache(std::move(py::module::import(name.c_str())));
-    } else {
-        object = importCache->addToCache(std::move((*parent)().attr(name.c_str())));
+    // The parent takes the same lock, so load it before locking.
+    py::handle parentObject = parent == nullptr ? py::handle() : (*parent)();
+    auto lock = importCache->lockForLoading();
+    if (!isLoaded()) {
+        auto obj = parent == nullptr ? py::object(py::module::import(name.c_str())) :
+                                       py::object(parentObject.attr(name.c_str()));
+        object = importCache->addToCache(std::move(obj));
+        loaded.store(true, std::memory_order_release);
     }
-    loaded = true;
     return object;
 }
 

src_cpp/include/cached_import/py_cached_import.h

@@ -1,5 +1,6 @@
 #pragma once
 
+#include <mutex>
 #include <string>
 #include <vector>
 
@@ -15,7 +16,9 @@ class PythonCachedImport {
     PythonCachedImport() = default;
     ~PythonCachedImport();
 
+    // Call with the lock from lockForLoading() held.
     py::handle addToCache(py::object obj);
+    std::unique_lock<std::mutex> lockForLoading();
 
     DateTimeCachedItem datetime;
     DecimalCachedItem decimal;
@@ -28,6 +31,7 @@ class PythonCachedImport {
     UUIDCachedItem uuid;
 
 private:
+    std::mutex loadMutex;
     std::vector<py::object> allObjects;
 };
 

src_cpp/include/cached_import/py_cached_item.h

@@ -1,5 +1,6 @@
 #pragma once
 
+#include <atomic>
 #include <memory>
 #include <string>
 
@@ -10,16 +11,16 @@ namespace lbug {
 class PythonCachedItem {
 public:
     explicit PythonCachedItem(std::string name, PythonCachedItem* parent = nullptr)
-        : name(std::move(name)), parent(parent), loaded(false) {}
+        : name(std::move(name)), parent(parent) {}
     virtual ~PythonCachedItem() = default;
 
-    bool isLoaded() const { return loaded; }
+    bool isLoaded() const { return loaded.load(std::memory_order_acquire); }
     py::handle operator()();
 
 protected:
     std::string name;
     PythonCachedItem* parent;
-    bool loaded;
+    std::atomic<bool> loaded{false};
     py::handle object;
 };
 

src_cpp/pandas/pandas_analyzer.cpp

@@ -193,7 +193,7 @@ static py::object findFirstNonNull(const py::handle& row, uint64_t numRows) {
 
 common::LogicalType PandasAnalyzer::innerAnalyze(py::object column, bool& canConvert) {
     auto numRows = py::len(column);
-    auto pandasModule = importCache->pandas;
+    auto& pandasModule = importCache->pandas;
     auto pandasSeries = pandasModule.core.series.Series();
 
     if (py::isinstance(column, pandasSeries)) {

src_cpp/py_database.cpp

@@ -1,6 +1,7 @@
 #include "include/py_database.h"
 
 #include <memory>
+#include <mutex>
 
 #include "common/exception/runtime.h"
 #include "extension/extension.h"
@@ -65,6 +66,9 @@ PyDatabase::PyDatabase(const std::string& databasePath, uint64_t bufferPoolSize,
     lbug::extension::ExtensionUtils::addTableFunc<lbug::PandasScanFunction>(*state->database);
     state->storageDriver = std::make_unique<StorageDriver>(state->database.get());
     py::gil_scoped_acquire acquire;
+    // Taken with the GIL held: nothing inside may call into Python.
+    static std::mutex importCacheMutex;
+    std::lock_guard<std::mutex> lock(importCacheMutex);
     if (lbug::importCache.get() == nullptr) {
         lbug::importCache = std::make_shared<lbug::PythonCachedImport>();
     }

test/test_import_cache_threading.py

@@ -0,0 +1,60 @@
+from __future__ import annotations
+
+import os
+import subprocess
+import sys
+import sysconfig
+from textwrap import dedent
+from typing import TYPE_CHECKING
+
+import pytest
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+# The driver fills its import cache on first use, and the cache cannot be reset within a
+# process, so every attempt runs in a fresh interpreter. Only a free-threaded build with the GIL
+# disabled can expose a race; on other builds this is a smoke test.
+FREE_THREADED = bool(sysconfig.get_config_var("Py_GIL_DISABLED"))
+
+
+@pytest.mark.parametrize("attempt", range(10 if FREE_THREADED else 1))
+def test_concurrent_first_use_of_the_import_cache(
+    attempt: int, build_dir: Path
+) -> None:
+    code = dedent(f"""
+        import sys
+        import threading
+
+        sys.path.append(r"{build_dir!s}")
+
+        import ladybug as lb
+
+        db = lb.Database(":memory:")
+        conns = [lb.Connection(db) for _ in range(16)]
+        start = threading.Barrier(len(conns))
+        errors = []
+
+        def run(conn, n):
+            start.wait()
+            try:
+                assert conn.execute("RETURN $n", {{"n": n}}).get_next() == [n]
+            except BaseException as e:
+                errors.append(e)
+
+        threads = [threading.Thread(target=run, args=(c, i)) for i, c in enumerate(conns)]
+        for t in threads:
+            t.start()
+        for t in threads:
+            t.join()
+        assert not errors, errors
+    """)
+    env = {**os.environ, "PYTHON_GIL": "0"} if FREE_THREADED else None
+    result = subprocess.run(
+        [sys.executable, "-c", code],
+        env=env,
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    assert result.returncode == 0, (result.returncode, result.stderr)