Fixed indexes propagation.

Author: khustup2

cpp/deeplake_pg/dl_catalog.cpp

@@ -457,9 +457,52 @@ std::vector<column_meta> load_columns(const std::string& root_path, const std::s
     }
 }
 
-std::vector<index_meta> load_indexes(const std::string&, const std::string&, icm::string_map<>)
+std::vector<index_meta> load_indexes(const std::string& root_path, const std::string& db_name, icm::string_map<> creds)
 {
-    return {};
+    std::vector<index_meta> out;
+    try {
+        auto table = open_db_catalog_table(root_path, db_name, k_indexes_name, std::move(creds));
+        if (!table) {
+            return out;
+        }
+        auto snapshot = table->read().get_future().get();
+        if (snapshot.row_count() == 0) {
+            return out;
+        }
+
+        for (const auto& row : snapshot.rows()) {
+            auto table_id_it = row.find("table_id");
+            auto column_names_it = row.find("column_names");
+            auto index_type_it = row.find("index_type");
+            auto order_type_it = row.find("order_type");
+
+            if (table_id_it == row.end() || column_names_it == row.end() || index_type_it == row.end()) {
+                continue;
+            }
+
+            index_meta meta;
+            meta.table_id = deeplake_api::array_to_string(table_id_it->second);
+            meta.column_names = deeplake_api::array_to_string(column_names_it->second);
+            meta.index_type = deeplake_api::array_to_string(index_type_it->second);
+            if (order_type_it != row.end()) {
+                try {
+                    auto order_vec = load_vector<int32_t>(order_type_it->second);
+                    meta.order_type = order_vec.empty() ? 0 : order_vec.front();
+                } catch (...) {
+                    meta.order_type = 0;
+                }
+            }
+
+            out.push_back(std::move(meta));
+        }
+        return out;
+    } catch (const std::exception& e) {
+        elog(DEBUG1, "Failed to load catalog indexes for db '%s': %s", db_name.c_str(), e.what());
+        return out;
+    } catch (...) {
+        elog(DEBUG1, "Failed to load catalog indexes for db '%s': unknown error", db_name.c_str());
+        return out;
+    }
 }
 
 std::vector<schema_meta> load_schemas(const std::string& root_path, const std::string& db_name, icm::string_map<> creds)
@@ -699,6 +742,25 @@ void upsert_columns(const std::string& root_path, const std::string& db_name, ic
     table->upsert_many(std::move(rows)).get_future().get();
 }
 
+void upsert_indexes(const std::string& root_path, const std::string& db_name, icm::string_map<> creds, const std::vector<index_meta>& indexes)
+{
+    if (indexes.empty()) {
+        return;
+    }
+    auto table = open_db_catalog_table(root_path, db_name, k_indexes_name, std::move(creds));
+    icm::vector<icm::string_map<nd::array>> rows;
+    rows.reserve(indexes.size());
+    for (const auto& idx : indexes) {
+        icm::string_map<nd::array> row;
+        row["table_id"] = nd::adapt(idx.table_id);
+        row["column_names"] = nd::adapt(idx.column_names);
+        row["index_type"] = nd::adapt(idx.index_type);
+        row["order_type"] = nd::adapt(idx.order_type);
+        rows.push_back(std::move(row));
+    }
+    table->upsert_many(std::move(rows)).get_future().get();
+}
+
 std::vector<database_meta> load_databases(const std::string& root_path, icm::string_map<> creds)
 {
     std::vector<database_meta> out;

cpp/deeplake_pg/dl_catalog.hpp

@@ -82,6 +82,7 @@ void upsert_schema(const std::string& root_path, const std::string& db_name, icm
 // Per-database upserts (write to {root}/{db_name}/__deeplake_catalog/)
 void upsert_table(const std::string& root_path, const std::string& db_name, icm::string_map<> creds, const table_meta& meta);
 void upsert_columns(const std::string& root_path, const std::string& db_name, icm::string_map<> creds, const std::vector<column_meta>& columns);
+void upsert_indexes(const std::string& root_path, const std::string& db_name, icm::string_map<> creds, const std::vector<index_meta>& indexes);
 
 // Shared (cluster-wide) database catalog
 std::vector<database_meta> load_databases(const std::string& root_path, icm::string_map<> creds);

cpp/deeplake_pg/pg_deeplake.cpp

@@ -1,6 +1,8 @@
 #include "pg_deeplake.hpp"
+#include "dl_catalog.hpp"
 #include "logger.hpp"
 #include "table_storage.hpp"
+#include "utils.hpp"
 
 #include <deeplake_api/deeplake_api.hpp>
 #include <deeplake_core/deeplake_index_type.hpp>
@@ -9,6 +11,8 @@
 extern "C" {
 #endif
 
+#include <commands/dbcommands.h>
+#include <miscadmin.h>
 #include <storage/ipc.h>
 
 #ifdef __cplusplus
@@ -260,6 +264,39 @@ void save_index_metadata(Oid oid)
     if (SPI_execute(buf.data, false, 0) != SPI_OK_INSERT) {
         ereport(ERROR, (errcode(ERRCODE_INTERNAL_ERROR), errmsg("Failed to save metadata")));
     }
+
+    // Persist index to shared catalog for stateless multi-instance sync
+    if (pg::stateless_enabled) {
+        try {
+            auto root_dir = pg::session_credentials::get_root_path();
+            if (root_dir.empty()) {
+                root_dir = pg::utils::get_deeplake_root_directory();
+            }
+            if (!root_dir.empty()) {
+                auto creds = pg::session_credentials::get_credentials();
+                const char* dbname = get_database_name(MyDatabaseId);
+                std::string db_name = dbname ? dbname : "postgres";
+                if (dbname) pfree(const_cast<char*>(dbname));
+
+                const std::string& table_id = idx_info.table_name(); // already schema-qualified
+
+                pg::dl_catalog::index_meta idx_meta;
+                idx_meta.table_id = table_id;
+                idx_meta.column_names = idx_info.get_column_names_string();
+                idx_meta.index_type = std::string(deeplake_core::deeplake_index_type::to_string(idx_info.index_type()));
+                idx_meta.order_type = static_cast<int32_t>(idx_info.order_type());
+
+                std::vector<pg::dl_catalog::index_meta> indexes = {idx_meta};
+                pg::dl_catalog::upsert_indexes(root_dir, db_name, creds, indexes);
+                pg::dl_catalog::bump_db_catalog_version(root_dir, db_name, creds);
+                pg::dl_catalog::bump_catalog_version(root_dir, creds);
+            }
+        } catch (const std::exception& e) {
+            elog(DEBUG1, "pg_deeplake: failed to persist index to shared catalog: %s", e.what());
+        } catch (...) {
+            elog(DEBUG1, "pg_deeplake: failed to persist index to shared catalog: unknown error");
+        }
+    }
 }
 
 void load_index_metadata()

cpp/deeplake_pg/sync_worker.cpp

@@ -346,9 +346,34 @@ void deeplake_sync_schemas_for_db(const std::string& db_name,
  * Sync tables for a specific database from pre-loaded catalog data via libpq.
  * Creates missing tables in the target database.
  */
+/**
+ * Parse comma-separated column names string into a vector.
+ * The column_names string uses trailing comma format: "col1,col2,"
+ */
+std::vector<std::string> parse_column_names(const std::string& column_names)
+{
+    std::vector<std::string> result;
+    std::string current;
+    for (char c : column_names) {
+        if (c == ',') {
+            if (!current.empty()) {
+                result.push_back(current);
+                current.clear();
+            }
+        } else {
+            current += c;
+        }
+    }
+    if (!current.empty()) {
+        result.push_back(current);
+    }
+    return result;
+}
+
 void deeplake_sync_tables_for_db(const std::string& db_name,
     const std::vector<pg::dl_catalog::table_meta>& tables,
-    const std::vector<pg::dl_catalog::column_meta>& columns)
+    const std::vector<pg::dl_catalog::column_meta>& columns,
+    const std::vector<pg::dl_catalog::index_meta>& indexes)
 {
     for (const auto& meta : tables) {
         if (meta.state == "dropping") {
@@ -373,6 +398,24 @@ void deeplake_sync_tables_for_db(const std::string& db_name,
             continue;
         }
 
+        // Find indexes for this table
+        std::vector<pg::dl_catalog::index_meta> table_indexes;
+        for (const auto& idx : indexes) {
+            if (idx.table_id == meta.table_id) {
+                table_indexes.push_back(idx);
+            }
+        }
+
+        // Determine which columns are part of a primary key (inverted_index on non-nullable columns)
+        // The primary key columns are stored as comma-separated names in column_names
+        std::vector<std::string> pk_columns;
+        for (const auto& idx : table_indexes) {
+            if (idx.index_type == "inverted_index") {
+                pk_columns = parse_column_names(idx.column_names);
+                break;
+            }
+        }
+
         const char* qschema = quote_identifier(meta.schema_name.c_str());
         const char* qtable = quote_identifier(meta.table_name.c_str());
 
@@ -390,6 +433,19 @@ void deeplake_sync_tables_for_db(const std::string& db_name,
             first = false;
             appendStringInfo(&buf, "%s %s", quote_identifier(col.column_name.c_str()), col.pg_type.c_str());
         }
+
+        // Add PRIMARY KEY table constraint if we have PK columns
+        if (!pk_columns.empty()) {
+            appendStringInfoString(&buf, ", PRIMARY KEY (");
+            for (size_t i = 0; i < pk_columns.size(); ++i) {
+                if (i > 0) {
+                    appendStringInfoString(&buf, ", ");
+                }
+                appendStringInfoString(&buf, quote_identifier(pk_columns[i].c_str()));
+            }
+            appendStringInfoChar(&buf, ')');
+        }
+
         appendStringInfo(&buf, ") USING deeplake");
 
         if (execute_via_libpq(db_name.c_str(), buf.data)) {
@@ -487,9 +543,10 @@ void sync_all_databases(
             }
 
             auto [tables, columns] = pg::dl_catalog::load_tables_and_columns(root_path, db_name, creds);
-            deeplake_sync_tables_for_db(db_name, tables, columns);
-            elog(LOG, "pg_deeplake sync: synced %zu schemas, %zu tables for database '%s'",
-                 schemas.size(), tables.size(), db_name.c_str());
+            auto indexes = pg::dl_catalog::load_indexes(root_path, db_name, creds);
+            deeplake_sync_tables_for_db(db_name, tables, columns, indexes);
+            elog(LOG, "pg_deeplake sync: synced %zu schemas, %zu tables, %zu indexes for database '%s'",
+                 schemas.size(), tables.size(), indexes.size(), db_name.c_str());
         } catch (const std::exception& e) {
             elog(WARNING, "pg_deeplake sync: failed to sync database '%s': %s", db_name.c_str(), e.what());
         } catch (...) {

cpp/deeplake_pg/table_storage.cpp

@@ -353,8 +353,9 @@ void table_storage::load_table_metadata()
         }
         tables_loaded_ = true;
 
-        // Load tables and columns in parallel from per-database path
+        // Load tables, columns, and indexes from per-database path
         auto [catalog_tables, catalog_columns] = pg::dl_catalog::load_tables_and_columns(root_dir, db_name, creds);
+        auto catalog_indexes = pg::dl_catalog::load_indexes(root_dir, db_name, creds);
 
         if (!catalog_tables.empty()) {
             for (const auto& meta : catalog_tables) {
@@ -387,6 +388,29 @@ void table_storage::load_table_metadata()
                         continue;
                     }
 
+                    // Find primary key columns from indexes
+                    std::vector<std::string> pk_columns;
+                    for (const auto& idx : catalog_indexes) {
+                        if (idx.table_id == meta.table_id && idx.index_type == "inverted_index") {
+                            // Parse comma-separated column names
+                            std::string current;
+                            for (char c : idx.column_names) {
+                                if (c == ',') {
+                                    if (!current.empty()) {
+                                        pk_columns.push_back(current);
+                                        current.clear();
+                                    }
+                                } else {
+                                    current += c;
+                                }
+                            }
+                            if (!current.empty()) {
+                                pk_columns.push_back(current);
+                            }
+                            break;
+                        }
+                    }
+
                     // Build CREATE TABLE IF NOT EXISTS from catalog metadata.
                     // Wrap in a subtransaction so that if another backend concurrently
                     // creates the same table (race on composite type), the error is
@@ -406,6 +430,19 @@ void table_storage::load_table_metadata()
                         first = false;
                         appendStringInfo(&buf, "%s %s", quote_identifier(col.column_name.c_str()), col.pg_type.c_str());
                     }
+
+                    // Add PRIMARY KEY table constraint if found in catalog indexes
+                    if (!pk_columns.empty()) {
+                        appendStringInfoString(&buf, ", PRIMARY KEY (");
+                        for (size_t i = 0; i < pk_columns.size(); ++i) {
+                            if (i > 0) {
+                                appendStringInfoString(&buf, ", ");
+                            }
+                            appendStringInfoString(&buf, quote_identifier(pk_columns[i].c_str()));
+                        }
+                        appendStringInfoChar(&buf, ')');
+                    }
+
                     appendStringInfo(&buf, ") USING deeplake");
 
                     MemoryContext saved_context = CurrentMemoryContext;