Fixed statelessness.

Author: khustup2

cpp/deeplake_pg/dl_catalog.cpp

@@ -37,10 +37,8 @@ std::string join_path(const std::string& root, const std::string& name)
     return root + "/" + k_catalog_dir + "/" + name;
 }
 
-std::shared_ptr<deeplake_api::catalog_table> open_or_create_table(
-    const std::string& path,
-    deeplake_api::catalog_table_schema schema,
-    icm::string_map<> creds)
+std::shared_ptr<deeplake_api::catalog_table>
+open_or_create_table(const std::string& path, deeplake_api::catalog_table_schema schema, icm::string_map<> creds)
 {
     return deeplake_api::open_or_create_catalog_table(path, schema, std::move(creds)).get_future().get();
 }
@@ -51,9 +49,8 @@ int64_t now_ms()
     return duration_cast<milliseconds>(system_clock::now().time_since_epoch()).count();
 }
 
-std::shared_ptr<deeplake_api::catalog_table> open_catalog_table(const std::string& root_path,
-                                                                const std::string& name,
-                                                                icm::string_map<> creds)
+std::shared_ptr<deeplake_api::catalog_table>
+open_catalog_table(const std::string& root_path, const std::string& name, icm::string_map<> creds)
 {
     const auto path = join_path(root_path, name);
     return deeplake_api::open_catalog_table(path, std::move(creds)).get_future().get();
@@ -213,8 +210,8 @@ std::vector<table_meta> load_tables(const std::string& root_path, icm::string_ma
             auto path_it = row.find("dataset_path");
             auto state_it = row.find("state");
             auto updated_it = row.find("updated_at");
-            if (table_id_it == row.end() || schema_it == row.end() || table_it == row.end() ||
-                path_it == row.end() || state_it == row.end() || updated_it == row.end()) {
+            if (table_id_it == row.end() || schema_it == row.end() || table_it == row.end() || path_it == row.end() ||
+                state_it == row.end() || updated_it == row.end()) {
                 continue;
             }
 

cpp/deeplake_pg/extension_init.cpp

@@ -55,7 +55,7 @@ bool treat_numeric_as_double = true; // Treat numeric types as double by default
 bool print_progress_during_seq_scan = false;
 bool use_shared_mem_for_refresh = false;
 bool enable_dataset_logging = false; // Enable dataset operation logging for debugging
-bool allow_custom_paths = true; // Allow dataset_path in CREATE TABLE options
+bool allow_custom_paths = true;      // Allow dataset_path in CREATE TABLE options
 
 } // namespace pg
 
@@ -133,7 +133,6 @@ void initialize_guc_parameters()
                              nullptr // check_hook, assign_hook, show_hook
     );
 
-
     DefineCustomBoolVariable("pg_deeplake.print_runtime_stats",
                              "Enable runtime statistics printing for pg_deeplake operations.",
                              nullptr,                  // optional long description

cpp/deeplake_pg/table_am.cpp

@@ -13,22 +13,22 @@ extern "C" {
 #include <catalog/namespace.h>
 #include <catalog/storage.h>
 #include <catalog/storage_xlog.h>
-#include <storage/smgr.h>
-#include <storage/bufmgr.h>
 #include <commands/vacuum.h> // For VacuumParams and VACOPT_* flags
 #include <miscadmin.h>
 #include <nodes/bitmapset.h>  // For bitmap operations
 #include <nodes/nodes.h>      // For node types
 #include <nodes/parsenodes.h> // For parse nodes
 #include <pgstat.h>           // For statistics collector integration
 #include <storage/block.h>
+#include <storage/bufmgr.h>
 #include <storage/relfilelocator.h>
+#include <storage/smgr.h>
 #include <utils/builtins.h> // For text conversion functions
 #include <utils/lsyscache.h>
 #include <utils/rel.h>
 #include <utils/relcache.h>
 #include <utils/snapshot.h> // For SNAPSHOT_DIRTY and snapshot types
-#include <utils/varlena.h> // For text functions
+#include <utils/varlena.h>  // For text functions
 
 #ifdef __cplusplus
 }
@@ -261,8 +261,7 @@ void deeplake_estimate_rel_size(
         }
         if (pages != nullptr) {
             constexpr uint32_t min_pages = 1;
-            const uint32_t num_blocks =
-                static_cast<uint32_t>(std::ceil(static_cast<double>(total_rows) / 65536.0));
+            const uint32_t num_blocks = static_cast<uint32_t>(std::ceil(static_cast<double>(total_rows) / 65536.0));
             *pages = std::max(min_pages, num_blocks);
         }
 
@@ -338,15 +337,15 @@ bool deeplake_scan_analyze_next_block(TableScanDesc scan, ReadStream* stream)
     Buffer buf = read_stream_next_buffer(stream, NULL);
 
     if (!BufferIsValid(buf)) {
-        return false;  // No more blocks to sample
+        return false; // No more blocks to sample
     }
 
     // Release the buffer immediately - we don't actually need the physical data
     // because our columnar storage provides the data via scan_analyze_next_tuple.
     // But we needed to consume the stream to increment bs.m.
     ReleaseBuffer(buf);
 
-    return true;  // Indicate we have data to process
+    return true; // Indicate we have data to process
 }
 #endif
 
@@ -754,7 +753,8 @@ void deeplake_table_am_routine::scan_rescan(TableScanDesc scan,
             std::string pre_msg = "Progress of sequential scan for table '" +
                                   scan_data->scan_state.get_table_data().get_table_name() + "'" + " (rescan " +
                                   std::to_string(scan_data->num_rescans) + ")";
-            scan_data->progress_bar.restart(scan_data->scan_state.get_table_data().num_total_rows(), std::move(pre_msg));
+            scan_data->progress_bar.restart(scan_data->scan_state.get_table_data().num_total_rows(),
+                                            std::move(pre_msg));
         }
     }
 }
@@ -1180,6 +1180,10 @@ void deeplake_table_am_routine::relation_set_new_node(
 
     convert_schema(tupdesc);
 
+    // Set DDL context to prevent auto-creation of tables from catalog during
+    // concurrent table creation (which causes race conditions).
+    table_storage::ddl_context_guard ddl_guard;
+
     try {
         table_storage::instance().create_table(table_name, RelationGetRelid(rel), tupdesc);
     } catch (const std::exception& e) {

cpp/deeplake_pg/table_storage.cpp

@@ -21,24 +21,24 @@ extern "C" {
 #include <utils/guc.h>
 #include <utils/lsyscache.h>
 #include <utils/rel.h>
-#include <utils/syscache.h>
 #include <utils/snapmgr.h>
+#include <utils/syscache.h>
 
 #ifdef __cplusplus
 }
 #endif
 
 #include "table_storage.hpp"
 
-#include "exceptions.hpp"
 #include "dl_catalog.hpp"
-#include <storage/exceptions.hpp>
+#include "exceptions.hpp"
 #include "logger.hpp"
 #include "memory_tracker.hpp"
 #include "nd_utils.hpp"
 #include "table_ddl_lock.hpp"
 #include "table_scan.hpp"
 #include "utils.hpp"
+#include <storage/exceptions.hpp>
 
 #include <icm/json.hpp>
 #include <icm/string_map.hpp>
@@ -276,7 +276,13 @@ void table_storage::load_table_metadata()
             auto* rel = makeRangeVar(pstrdup(meta.schema_name.c_str()), pstrdup(meta.table_name.c_str()), -1);
             Oid relid = RangeVarGetRelid(rel, NoLock, true);
             if (!OidIsValid(relid)) {
-                // Attempt to create missing table from catalog entry.
+                // Table exists in catalog but not in PostgreSQL.
+                if (in_ddl_context()) {
+                    // During DDL (CREATE TABLE), skip auto-creation to avoid races.
+                    // The table might be in the middle of being created by another backend.
+                    continue;
+                }
+                // Not in DDL context (e.g., SET root_path) - safe to auto-create.
                 pg::utils::memory_context_switcher context_switcher;
                 pg::utils::spi_connector connector;
                 bool pushed_snapshot = false;
@@ -293,14 +299,10 @@ void table_storage::load_table_metadata()
                 SPI_execute(buf.data, false, 0);
 
                 resetStringInfo(&buf);
-                appendStringInfo(&buf,
-                                 "SELECT create_deeplake_table(%s, %s)",
-                                 quote_literal_cstr(qualified_name.c_str()),
-                                 qpath);
+                appendStringInfo(
+                    &buf, "SELECT create_deeplake_table(%s, %s)", quote_literal_cstr(qualified_name.c_str()), qpath);
                 if (SPI_execute(buf.data, false, 0) != SPI_OK_SELECT) {
-                    elog(WARNING,
-                         "Failed to auto-create deeplake table %s from catalog",
-                         qualified_name.c_str());
+                    elog(WARNING, "Failed to auto-create deeplake table %s from catalog", qualified_name.c_str());
                 }
                 if (pushed_snapshot) {
                     PopActiveSnapshot();
@@ -334,7 +336,8 @@ void table_storage::load_table_metadata()
         return;
     }
 
-    struct snapshot_guard {
+    struct snapshot_guard
+    {
         bool active = false;
         snapshot_guard()
         {
@@ -355,22 +358,21 @@ void table_storage::load_table_metadata()
     // If not, drop and recreate the table with the correct schema
     if (!pg::utils::check_column_exists("pg_deeplake_tables", "table_oid")) {
         base::log_warning(base::log_channel::generic,
-            "Detected old schema for pg_deeplake_tables without table_oid column. "
-            "Dropping and recreating table to match current schema.");
+                          "Detected old schema for pg_deeplake_tables without table_oid column. "
+                          "Dropping and recreating table to match current schema.");
 
         pg::utils::spi_connector connector;
         const char* drop_query = "DROP TABLE IF EXISTS public.pg_deeplake_tables CASCADE";
         if (SPI_execute(drop_query, false, 0) != SPI_OK_UTILITY) {
             base::log_warning(base::log_channel::generic, "Failed to drop old pg_deeplake_tables table");
         }
 
-        const char* create_query =
-            "CREATE TABLE public.pg_deeplake_tables ("
-            "    id SERIAL PRIMARY KEY,"
-            "    table_oid OID NOT NULL UNIQUE,"
-            "    table_name NAME NOT NULL UNIQUE,"
-            "    ds_path TEXT NOT NULL UNIQUE"
-            ")";
+        const char* create_query = "CREATE TABLE public.pg_deeplake_tables ("
+                                   "    id SERIAL PRIMARY KEY,"
+                                   "    table_oid OID NOT NULL UNIQUE,"
+                                   "    table_name NAME NOT NULL UNIQUE,"
+                                   "    ds_path TEXT NOT NULL UNIQUE"
+                                   ")";
         if (SPI_execute(create_query, false, 0) != SPI_OK_UTILITY) {
             base::log_warning(base::log_channel::generic, "Failed to create new pg_deeplake_tables table");
         }
@@ -445,8 +447,10 @@ void table_storage::load_table_metadata()
                 // Use the actual relation name from PostgreSQL catalog, not the cached metadata name
                 // This ensures we have the current name even if the table was renamed
                 std::string actual_table_name = get_qualified_table_name(rel);
-                elog(DEBUG1, "Loading table from metadata: cached_name=%s, actual_name=%s",
-                     table_name, actual_table_name.c_str());
+                elog(DEBUG1,
+                     "Loading table from metadata: cached_name=%s, actual_name=%s",
+                     table_name,
+                     actual_table_name.c_str());
                 table_data td(
                     relid, actual_table_name, CreateTupleDescCopy(RelationGetDescr(rel)), std::string(ds_path), creds);
                 auto it2status = tables_.emplace(relid, std::move(td));
@@ -570,10 +574,11 @@ void table_storage::create_table(const std::string& table_name, Oid table_id, Tu
     // Use provided dataset path or construct default path
     if (!options.dataset_path().empty()) {
         if (!pg::allow_custom_paths) {
-            ereport(ERROR,
-                    (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
-                     errmsg("Custom dataset_path is disabled"),
-                     errhint("Set deeplake.allow_custom_paths=on or omit dataset_path and configure deeplake.root_path")));
+            ereport(
+                ERROR,
+                (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+                 errmsg("Custom dataset_path is disabled"),
+                 errhint("Set deeplake.allow_custom_paths=on or omit dataset_path and configure deeplake.root_path")));
         }
         // Explicit path provided via WITH clause
         dataset_path = options.dataset_path();

cpp/deeplake_pg/table_storage.hpp

@@ -7,7 +7,7 @@ namespace pg {
 // Session-level credentials management
 struct session_credentials
 {
-    static char* creds_guc_string; // GUC string variable for credentials
+    static char* creds_guc_string;     // GUC string variable for credentials
     static char* root_path_guc_string; // GUC string variable for root path
 
     // Get credentials from current session
@@ -228,7 +228,37 @@ class table_storage
         up_to_date_ = up_to_date;
     }
 
+    /**
+     * RAII guard to suppress auto-creation of tables during load_table_metadata().
+     *
+     * When creating a table concurrently, we don't want load_table_metadata() to
+     * auto-create tables from the catalog, as this causes race conditions with
+     * other backends that are also creating tables.
+     *
+     * Usage: Create this guard before calling table_storage::instance() during DDL.
+     */
+    class ddl_context_guard
+    {
+    public:
+        ddl_context_guard()
+        {
+            in_ddl_context_ = true;
+        }
+        ~ddl_context_guard()
+        {
+            in_ddl_context_ = false;
+        }
+        ddl_context_guard(const ddl_context_guard&) = delete;
+        ddl_context_guard& operator=(const ddl_context_guard&) = delete;
+    };
+
+    static bool in_ddl_context() noexcept
+    {
+        return in_ddl_context_;
+    }
+
 private:
+    static inline thread_local bool in_ddl_context_ = false;
     table_storage() = default;
 
     void save_table_metadata(const table_data& td);
@@ -242,4 +272,4 @@ class table_storage
     int64_t catalog_version_ = 0;
 };
 
-} // namespace pg 
+} // namespace pg