From 51b20062dbdcb119448e3ac9288175aeb0d2e0e0 Mon Sep 17 00:00:00 2001 From: khustup2 Date: Thu, 5 Feb 2026 01:02:54 +0000 Subject: [PATCH 1/5] Improved stateless. Switched to deeplake-api 4.5.1 --- cpp/deeplake_pg/dl_catalog.cpp | 292 +++++++++++++++++------ cpp/deeplake_pg/dl_catalog.hpp | 4 + cpp/deeplake_pg/duckdb_deeplake_scan.cpp | 24 +- cpp/deeplake_pg/extension_init.cpp | 16 +- cpp/deeplake_pg/hybrid_query_merge.hpp | 2 +- cpp/deeplake_pg/index_search.cpp | 6 +- cpp/deeplake_pg/sync_worker.cpp | 17 +- cpp/deeplake_pg/table_data.hpp | 13 +- cpp/deeplake_pg/table_storage.cpp | 17 +- cpp/deeplake_pg/table_storage.hpp | 1 + 10 files changed, 277 insertions(+), 115 deletions(-) diff --git a/cpp/deeplake_pg/dl_catalog.cpp b/cpp/deeplake_pg/dl_catalog.cpp index dc54fdcd6d..30a1b579fc 100644 --- a/cpp/deeplake_pg/dl_catalog.cpp +++ b/cpp/deeplake_pg/dl_catalog.cpp @@ -1,5 +1,6 @@ #include "dl_catalog.hpp" +#include #include #include #include @@ -17,8 +18,6 @@ extern "C" { #include } -#include - namespace pg::dl_catalog { namespace { @@ -37,6 +36,36 @@ std::string join_path(const std::string& root, const std::string& name) return root + "/" + k_catalog_dir + "/" + name; } +// Cache for catalog table handles to avoid repeated S3 opens +struct catalog_table_cache +{ + std::string root_path; + std::shared_ptr meta_table; + + static catalog_table_cache& instance() + { + static thread_local catalog_table_cache cache; + return cache; + } + + std::shared_ptr get_meta_table(const std::string& path, icm::string_map<> creds) + { + if (path != root_path || !meta_table) { + // Cache miss or path changed - open and cache + root_path = path; + const auto meta_path = join_path(path, k_meta_name); + meta_table = deeplake_api::open_catalog_table(meta_path, std::move(creds)).get_future().get(); + } + return meta_table; + } + + void invalidate() + { + root_path.clear(); + meta_table.reset(); + } +}; + std::shared_ptr open_or_create_table(const std::string& path, deeplake_api::catalog_table_schema schema, icm::string_map<> creds) { @@ -107,87 +136,63 @@ void ensure_catalog(const std::string& root_path, icm::string_map<> creds) const auto indexes_path = join_path(root_path, k_indexes_name); const auto meta_path = join_path(root_path, k_meta_name); - auto ensure_table = [&](const std::string& path, - deeplake_api::catalog_table_schema schema) -> std::shared_ptr { - bool exists = false; - try { - exists = deeplake_api::exists(path, icm::string_map<>(creds)).get_future().get(); - } catch (...) { - exists = false; - } - if (exists) { - bool is_catalog = false; - try { - is_catalog = deeplake_api::is_catalog_table(path, icm::string_map<>(creds)).get_future().get(); - } catch (...) { - is_catalog = false; - } - if (!is_catalog) { - elog(WARNING, - "Existing catalog path %s is not a catalog table. Recreating catalog table.", - path.c_str()); - try { - deeplake_api::delete_dataset(path, icm::string_map<>(creds)).get_future().get(); - } catch (...) { - elog(WARNING, "Failed to delete legacy dataset at %s", path.c_str()); - } - } - } - try { - return open_or_create_table(path, std::move(schema), icm::string_map<>(creds)); - } catch (const std::exception& e) { - elog(ERROR, "Failed to open or create catalog table at %s: %s", path.c_str(), e.what()); - } - return {}; - }; - - { - deeplake_api::catalog_table_schema schema; - schema.add("table_id", deeplake_core::type::text(codecs::compression::null)) - .add("schema_name", deeplake_core::type::text(codecs::compression::null)) - .add("table_name", deeplake_core::type::text(codecs::compression::null)) - .add("dataset_path", deeplake_core::type::text(codecs::compression::null)) - .add("state", deeplake_core::type::text(codecs::compression::null)) - .add("updated_at", deeplake_core::type::generic(nd::type::scalar(nd::dtype::int64))) - .set_primary_key("table_id"); - ensure_table(tables_path, std::move(schema)); - } - - { - deeplake_api::catalog_table_schema schema; - // Use column_id (table_id:column_name) as primary key to support multiple columns per table - schema.add("column_id", deeplake_core::type::text(codecs::compression::null)) - .add("table_id", deeplake_core::type::text(codecs::compression::null)) - .add("column_name", deeplake_core::type::text(codecs::compression::null)) - .add("pg_type", deeplake_core::type::text(codecs::compression::null)) - .add("dl_type_json", deeplake_core::type::text(codecs::compression::null)) - .add("nullable", deeplake_core::type::generic(nd::type::scalar(nd::dtype::boolean))) - .add("position", deeplake_core::type::generic(nd::type::scalar(nd::dtype::int32))) - .set_primary_key("column_id"); - ensure_table(columns_path, std::move(schema)); - } - - { - deeplake_api::catalog_table_schema schema; - schema.add("table_id", deeplake_core::type::text(codecs::compression::null)) - .add("column_names", deeplake_core::type::text(codecs::compression::null)) - .add("index_type", deeplake_core::type::text(codecs::compression::null)) - .add("order_type", deeplake_core::type::generic(nd::type::scalar(nd::dtype::int32))) - .set_primary_key("table_id"); - ensure_table(indexes_path, std::move(schema)); - } + // Build schemas for all catalog tables + deeplake_api::catalog_table_schema tables_schema; + tables_schema.add("table_id", deeplake_core::type::text(codecs::compression::null)) + .add("schema_name", deeplake_core::type::text(codecs::compression::null)) + .add("table_name", deeplake_core::type::text(codecs::compression::null)) + .add("dataset_path", deeplake_core::type::text(codecs::compression::null)) + .add("state", deeplake_core::type::text(codecs::compression::null)) + .add("updated_at", deeplake_core::type::generic(nd::type::scalar(nd::dtype::int64))) + .set_primary_key("table_id"); + + deeplake_api::catalog_table_schema columns_schema; + columns_schema.add("column_id", deeplake_core::type::text(codecs::compression::null)) + .add("table_id", deeplake_core::type::text(codecs::compression::null)) + .add("column_name", deeplake_core::type::text(codecs::compression::null)) + .add("pg_type", deeplake_core::type::text(codecs::compression::null)) + .add("dl_type_json", deeplake_core::type::text(codecs::compression::null)) + .add("nullable", deeplake_core::type::generic(nd::type::scalar(nd::dtype::boolean))) + .add("position", deeplake_core::type::generic(nd::type::scalar(nd::dtype::int32))) + .set_primary_key("column_id"); + + deeplake_api::catalog_table_schema indexes_schema; + indexes_schema.add("table_id", deeplake_core::type::text(codecs::compression::null)) + .add("column_names", deeplake_core::type::text(codecs::compression::null)) + .add("index_type", deeplake_core::type::text(codecs::compression::null)) + .add("order_type", deeplake_core::type::generic(nd::type::scalar(nd::dtype::int32))) + .set_primary_key("table_id"); deeplake_api::catalog_table_schema meta_schema; meta_schema.add("catalog_version", deeplake_core::type::generic(nd::type::scalar(nd::dtype::int64))) .add("updated_at", deeplake_core::type::generic(nd::type::scalar(nd::dtype::int64))) .set_primary_key("catalog_version"); - auto meta_table = ensure_table(meta_path, std::move(meta_schema)); - auto snapshot = meta_table->read().get_future().get(); - if (snapshot.row_count() == 0) { - icm::string_map row; - row["catalog_version"] = nd::adapt(static_cast(1)); - row["updated_at"] = nd::adapt(now_ms()); - meta_table->insert(std::move(row)).get_future().get(); + + // Launch all 4 open_or_create operations in parallel + std::vector>> promises; + promises.reserve(4); + promises.push_back( + deeplake_api::open_or_create_catalog_table(tables_path, std::move(tables_schema), icm::string_map<>(creds))); + promises.push_back( + deeplake_api::open_or_create_catalog_table(columns_path, std::move(columns_schema), icm::string_map<>(creds))); + promises.push_back( + deeplake_api::open_or_create_catalog_table(indexes_path, std::move(indexes_schema), icm::string_map<>(creds))); + promises.push_back( + deeplake_api::open_or_create_catalog_table(meta_path, std::move(meta_schema), icm::string_map<>(creds))); + + // Wait for all to complete + auto results = async::combine(std::move(promises)).get_future().get(); + + // Initialize meta table if empty (index 3 is meta) + auto& meta_table = results[3]; + if (meta_table) { + auto snapshot = meta_table->read().get_future().get(); + if (snapshot.row_count() == 0) { + icm::string_map row; + row["catalog_version"] = nd::adapt(static_cast(1)); + row["updated_at"] = nd::adapt(now_ms()); + meta_table->insert(std::move(row)).get_future().get(); + } } } @@ -313,6 +318,132 @@ std::vector load_indexes(const std::string&, icm::string_map<>) return {}; } +std::pair, std::vector> +load_tables_and_columns(const std::string& root_path, icm::string_map<> creds) +{ + std::vector tables_out; + std::vector columns_out; + + try { + // Open both catalog tables in parallel + auto tables_promise = deeplake_api::open_catalog_table(join_path(root_path, k_tables_name), icm::string_map<>(creds)); + auto columns_promise = deeplake_api::open_catalog_table(join_path(root_path, k_columns_name), icm::string_map<>(creds)); + + std::vector>> open_promises; + open_promises.push_back(std::move(tables_promise)); + open_promises.push_back(std::move(columns_promise)); + + auto catalog_tables = async::combine(std::move(open_promises)).get_future().get(); + + auto& tables_table = catalog_tables[0]; + auto& columns_table = catalog_tables[1]; + + if (!tables_table || !columns_table) { + return {tables_out, columns_out}; + } + + // Read both snapshots in parallel + auto tables_read_promise = tables_table->read(); + auto columns_read_promise = columns_table->read(); + + std::vector> read_promises; + read_promises.push_back(std::move(tables_read_promise)); + read_promises.push_back(std::move(columns_read_promise)); + + auto snapshots = async::combine(std::move(read_promises)).get_future().get(); + + auto& tables_snapshot = snapshots[0]; + auto& columns_snapshot = snapshots[1]; + + // Process tables + if (tables_snapshot.row_count() > 0) { + std::unordered_map latest; + for (const auto& row : tables_snapshot.rows()) { + auto table_id_it = row.find("table_id"); + auto schema_it = row.find("schema_name"); + auto table_it = row.find("table_name"); + 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()) { + continue; + } + + table_meta meta; + meta.table_id = deeplake_api::array_to_string(table_id_it->second); + meta.schema_name = deeplake_api::array_to_string(schema_it->second); + meta.table_name = deeplake_api::array_to_string(table_it->second); + meta.dataset_path = deeplake_api::array_to_string(path_it->second); + meta.state = deeplake_api::array_to_string(state_it->second); + auto updated_vec = load_int64_vector(updated_it->second); + meta.updated_at = updated_vec.empty() ? 0 : updated_vec.front(); + + auto it = latest.find(meta.table_id); + if (it == latest.end() || it->second.updated_at <= meta.updated_at) { + latest[meta.table_id] = std::move(meta); + } + } + + tables_out.reserve(latest.size()); + for (auto& [_, meta] : latest) { + if (meta.state == "ready") { + tables_out.push_back(std::move(meta)); + } + } + } + + // Process columns + if (columns_snapshot.row_count() > 0) { + for (const auto& row : columns_snapshot.rows()) { + auto table_id_it = row.find("table_id"); + auto column_name_it = row.find("column_name"); + auto pg_type_it = row.find("pg_type"); + auto dl_type_it = row.find("dl_type_json"); + auto nullable_it = row.find("nullable"); + auto position_it = row.find("position"); + + if (table_id_it == row.end() || column_name_it == row.end() || pg_type_it == row.end()) { + continue; + } + + column_meta meta; + meta.table_id = deeplake_api::array_to_string(table_id_it->second); + meta.column_name = deeplake_api::array_to_string(column_name_it->second); + meta.pg_type = deeplake_api::array_to_string(pg_type_it->second); + if (dl_type_it != row.end()) { + meta.dl_type_json = deeplake_api::array_to_string(dl_type_it->second); + } + if (nullable_it != row.end()) { + try { + meta.nullable = nullable_it->second.value(0); + } catch (...) { + meta.nullable = true; + } + } + if (position_it != row.end()) { + try { + meta.position = position_it->second.value(0); + } catch (...) { + auto pos_vec = load_int64_vector(position_it->second); + meta.position = pos_vec.empty() ? 0 : static_cast(pos_vec.front()); + } + } + + columns_out.push_back(std::move(meta)); + } + } + + return {tables_out, columns_out}; + } catch (const std::exception& e) { + elog(WARNING, "Failed to load catalog tables and columns: %s", e.what()); + return {tables_out, columns_out}; + } catch (...) { + elog(WARNING, "Failed to load catalog tables and columns: unknown error"); + return {tables_out, columns_out}; + } +} + void upsert_table(const std::string& root_path, icm::string_map<> creds, const table_meta& meta) { auto table = open_catalog_table(root_path, k_tables_name, std::move(creds)); @@ -352,7 +483,8 @@ void upsert_columns(const std::string& root_path, icm::string_map<> creds, const int64_t get_catalog_version(const std::string& root_path, icm::string_map<> creds) { try { - auto table = open_catalog_table(root_path, k_meta_name, std::move(creds)); + // Use cached meta table handle to avoid repeated S3 opens + auto table = catalog_table_cache::instance().get_meta_table(root_path, std::move(creds)); if (!table) { return 0; } @@ -361,9 +493,11 @@ int64_t get_catalog_version(const std::string& root_path, icm::string_map<> cred return static_cast(table->version().get_future().get()); } catch (const std::exception& e) { elog(WARNING, "Failed to read catalog version: %s", e.what()); + catalog_table_cache::instance().invalidate(); return 0; } catch (...) { elog(WARNING, "Failed to read catalog version: unknown error"); + catalog_table_cache::instance().invalidate(); return 0; } } diff --git a/cpp/deeplake_pg/dl_catalog.hpp b/cpp/deeplake_pg/dl_catalog.hpp index 11503fb30f..579cb66ff5 100644 --- a/cpp/deeplake_pg/dl_catalog.hpp +++ b/cpp/deeplake_pg/dl_catalog.hpp @@ -42,6 +42,10 @@ std::vector load_tables(const std::string& root_path, icm::string_ma std::vector load_columns(const std::string& root_path, icm::string_map<> creds); std::vector load_indexes(const std::string& root_path, icm::string_map<> creds); +// Load tables and columns in parallel for better performance +std::pair, std::vector> +load_tables_and_columns(const std::string& root_path, icm::string_map<> creds); + void upsert_table(const std::string& root_path, icm::string_map<> creds, const table_meta& meta); void upsert_columns(const std::string& root_path, icm::string_map<> creds, const std::vector& columns); diff --git a/cpp/deeplake_pg/duckdb_deeplake_scan.cpp b/cpp/deeplake_pg/duckdb_deeplake_scan.cpp index 3a27ff6959..19f00b02ed 100644 --- a/cpp/deeplake_pg/duckdb_deeplake_scan.cpp +++ b/cpp/deeplake_pg/duckdb_deeplake_scan.cpp @@ -43,7 +43,7 @@ struct deeplake_scan_bind_data final : public duckdb::TableFunctionData struct deeplake_scan_global_state final : public duckdb::GlobalTableFunctionState { duckdb::vector column_ids; - std::vector>()>> index_searchers; + icm::vector>()>> index_searchers; duckdb::unique_ptr filter_expr; std::mutex index_search_mutex; heimdall::dataset_view_ptr index_search_result; @@ -205,10 +205,10 @@ duckdb::unique_ptr deeplake_scan_bind(duckdb::ClientContex return duckdb::make_uniq(td, return_types); } -base::function>()> +base::function>()> try_get_index_searcher(heimdall::column_view_ptr column_view, const duckdb::ConstantFilter& filter) { - base::function>()> result; + base::function>()> result; auto index_holder = column_view->index_holder(); ASSERT(index_holder != nullptr); auto constant = pg::to_deeplake_value(filter.constant); @@ -250,7 +250,7 @@ try_get_index_searcher(heimdall::column_view_ptr column_view, const duckdb::Cons query_core::text_search_info info; info.column_name = column_view->name(); info.type = query_core::text_search_info::search_type::equals; - info.search_values.push_back(std::vector{filter.constant.ToString()}); + info.search_values.push_back(icm::vector{filter.constant.ToString()}); if (index_holder->can_run_query(info)) { result = [index_holder, si = std::move(info)]() { return index_holder->run_query(si); @@ -261,7 +261,7 @@ try_get_index_searcher(heimdall::column_view_ptr column_view, const duckdb::Cons return result; } -base::function>()> +base::function>()> try_get_index_searcher(heimdall::column_view_ptr column_view, const duckdb::InFilter& filter) { query_core::inverted_index_search_info info; @@ -277,10 +277,10 @@ try_get_index_searcher(heimdall::column_view_ptr column_view, const duckdb::InFi }; } -base::function>()> +base::function>()> try_get_index_searcher(heimdall::column_view_ptr column_view, const duckdb::TableFilter& filter) { - base::function>()> result; + base::function>()> result; ASSERT(column_view != nullptr); if (column_view->index_holder() == nullptr) { return result; @@ -887,14 +887,14 @@ class deeplake_scan_function_helper if (is_index_search_done()) { return; } - std::vector> promises; + icm::vector> promises; for (auto& is : global_state_.index_searchers) { - promises.push_back(is().then_any([](std::vector&& results) { + promises.push_back(is().then_any([](icm::vector&& results) { ASSERT(results.size() == 1); return std::move(results.front()); })); } - auto combined_promise = async::combine(std::move(promises)).then_any([](std::vector&& results) { + auto combined_promise = async::combine(std::move(promises)).then_any([](icm::vector&& results) { ASSERT(!results.empty()); icm::roaring& combined = results[0]; for (size_t i = 1; i < results.size(); ++i) { @@ -903,7 +903,7 @@ class deeplake_scan_function_helper return std::move(combined); }); auto indices = combined_promise.get_future().get(); - std::vector indices_vec; + icm::vector indices_vec; indices_vec.reserve(indices.cardinality()); for (auto x : indices) { indices_vec.push_back(x); @@ -955,7 +955,7 @@ class deeplake_scan_function_helper } ASSERT(output_.ColumnCount() == global_state_.column_ids.size()); - std::vector> column_promises; + icm::vector> column_promises; // Fill output vectors column by column using table_data streamers for (unsigned i = 0; i < global_state_.column_ids.size(); ++i) { const auto col_idx = global_state_.column_ids[i]; diff --git a/cpp/deeplake_pg/extension_init.cpp b/cpp/deeplake_pg/extension_init.cpp index f08205966e..5444a07499 100644 --- a/cpp/deeplake_pg/extension_init.cpp +++ b/cpp/deeplake_pg/extension_init.cpp @@ -1197,10 +1197,18 @@ static void process_utility(PlannedStmt* pstmt, } // When root_path is set, auto-discover tables from the deeplake catalog if (vstmt->name != nullptr && pg_strcasecmp(vstmt->name, "deeplake.root_path") == 0) { - // Reload table metadata from the catalog at the new root_path - // This enables stateless multi-instance support where tables are - // auto-discovered when pointing to a shared root_path - pg::table_storage::instance().force_load_table_metadata(); + // Track the previous root_path to detect actual changes + static thread_local std::string last_root_path; + auto current_root_path = pg::session_credentials::get_root_path(); + + if (current_root_path != last_root_path) { + // Path changed - force full reload + last_root_path = current_root_path; + pg::table_storage::instance().force_load_table_metadata(); + } else { + // Same path - just check for catalog updates (fast path) + pg::table_storage::instance().load_table_metadata(); + } } } } diff --git a/cpp/deeplake_pg/hybrid_query_merge.hpp b/cpp/deeplake_pg/hybrid_query_merge.hpp index b15d162990..4c713c4b9d 100644 --- a/cpp/deeplake_pg/hybrid_query_merge.hpp +++ b/cpp/deeplake_pg/hybrid_query_merge.hpp @@ -126,7 +126,7 @@ inline query_core::query_result merge_query_results( // Take top_k results size_t result_size = std::min(top_k, final_scores.size()); std::vector top_scores; - std::vector top_indices; + icm::vector top_indices; top_scores.reserve(result_size); top_indices.reserve(result_size); diff --git a/cpp/deeplake_pg/index_search.cpp b/cpp/deeplake_pg/index_search.cpp index f895164253..b267470449 100644 --- a/cpp/deeplake_pg/index_search.cpp +++ b/cpp/deeplake_pg/index_search.cpp @@ -108,7 +108,7 @@ struct scan_opaque query_core::query_result run_index_search(nd::array input_array, std::string func_name, const std::string& column_name, pg::index_info& idx_info) { - std::vector args; + icm::vector args; args.emplace_back(query_core::expr::make_column_ref(column_name, std::string{})); args.emplace_back(query_core::expr::make_literal_array(std::move(input_array))); const bool is_cosine_similarity = (func_name == "COSINE_SIMILARITY"); @@ -194,7 +194,7 @@ icm::roaring run_exact_text_search(std::string text_value, StrategyNumber strate return {}; } info.column_name = idx_info.column_name(); - info.search_values.emplace_back(std::vector{std::move(text_value)}); + info.search_values.emplace_back(icm::vector{std::move(text_value)}); return idx_info.run_query(std::move(info)); } @@ -563,7 +563,7 @@ void collect_index_data(IndexScanDesc scan, ScanKey keys, int32_t nkeys, ScanKey } } if (nkeys > 0) { - std::vector row_numbers; + icm::vector row_numbers; row_numbers.reserve(result.cardinality()); std::transform(result.begin(), result.end(), std::back_inserter(row_numbers), [](auto v) { return static_cast(v); diff --git a/cpp/deeplake_pg/sync_worker.cpp b/cpp/deeplake_pg/sync_worker.cpp index fc9b574dc9..66210ec1d6 100644 --- a/cpp/deeplake_pg/sync_worker.cpp +++ b/cpp/deeplake_pg/sync_worker.cpp @@ -67,8 +67,8 @@ void deeplake_sync_worker_sighup(SIGNAL_ARGS) */ void deeplake_sync_tables_from_catalog(const std::string& root_path, icm::string_map<> creds) { - auto catalog_tables = pg::dl_catalog::load_tables(root_path, creds); - auto catalog_columns = pg::dl_catalog::load_columns(root_path, creds); + // Load tables and columns in parallel for better performance + auto [catalog_tables, catalog_columns] = pg::dl_catalog::load_tables_and_columns(root_path, creds); for (const auto& meta : catalog_tables) { // Skip tables marked as dropping @@ -168,6 +168,8 @@ PGDLLEXPORT void deeplake_sync_worker_main(Datum main_arg) elog(LOG, "pg_deeplake sync worker started"); int64_t last_catalog_version = 0; + std::string last_root_path; // Track root_path to detect changes + bool catalog_ensured = false; while (!got_sigterm) { // Handle SIGHUP - reload configuration @@ -199,10 +201,15 @@ PGDLLEXPORT void deeplake_sync_worker_main(Datum main_arg) if (!root_path.empty()) { auto creds = pg::session_credentials::get_credentials(); - // Ensure catalog exists - pg::dl_catalog::ensure_catalog(root_path, creds); + // Only ensure catalog on first call or when root_path changes + if (!catalog_ensured || root_path != last_root_path) { + pg::dl_catalog::ensure_catalog(root_path, creds); + catalog_ensured = true; + last_root_path = root_path; + last_catalog_version = 0; // Reset version when path changes + } - // Use existing catalog version API to check for changes + // Use existing catalog version API to check for changes (now fast with cache) int64_t current_version = pg::dl_catalog::get_catalog_version(root_path, creds); if (current_version != last_catalog_version) { diff --git a/cpp/deeplake_pg/table_data.hpp b/cpp/deeplake_pg/table_data.hpp index 3acf6a1d3c..be7b419625 100644 --- a/cpp/deeplake_pg/table_data.hpp +++ b/cpp/deeplake_pg/table_data.hpp @@ -6,6 +6,7 @@ #include #include #include +#include #include #include @@ -165,16 +166,16 @@ struct table_data constexpr static int64_t batch_mask_ = batch_size_ - 1; streamer_info streamers_; - icm::string_map> insert_rows_; + icm::string_map> insert_rows_; std::deque> insert_promises_; - std::vector delete_rows_; - std::vector> update_rows_; + icm::vector delete_rows_; + icm::vector> update_rows_; std::shared_ptr dataset_; std::shared_ptr refreshing_dataset_; async::promise refresh_promise_; - std::vector requested_columns_; - std::vector base_typeids_; // Cached base type OIDs for performance - std::vector active_column_indices_; // Maps logical index to TupleDesc index (excludes dropped) + icm::vector requested_columns_; + icm::vector base_typeids_; // Cached base type OIDs for performance + icm::vector active_column_indices_; // Maps logical index to TupleDesc index (excludes dropped) icm::string_map<> creds_; TupleDesc tuple_descriptor_; http::uri dataset_path_ = http::uri(std::string()); diff --git a/cpp/deeplake_pg/table_storage.cpp b/cpp/deeplake_pg/table_storage.cpp index b07e2b3afa..ba28b16c4a 100644 --- a/cpp/deeplake_pg/table_storage.cpp +++ b/cpp/deeplake_pg/table_storage.cpp @@ -291,22 +291,29 @@ void table_storage::load_table_metadata() // Stateless catalog sync (only when enabled) if (pg::stateless_enabled) { - pg::dl_catalog::ensure_catalog(root_dir, creds); - + // Fast path: if already loaded, just check version without ensure_catalog if (tables_loaded_) { const auto current_version = pg::dl_catalog::get_catalog_version(root_dir, creds); if (current_version == catalog_version_) { return; } + // Version changed, need to reload tables_.clear(); views_.clear(); tables_loaded_ = false; + catalog_version_ = current_version; // Reuse the version we just fetched } + + // Only ensure catalog exists when we need to load/reload + pg::dl_catalog::ensure_catalog(root_dir, creds); tables_loaded_ = true; - catalog_version_ = pg::dl_catalog::get_catalog_version(root_dir, creds); + // Only fetch version if we don't already have it from the check above + if (catalog_version_ == 0) { + catalog_version_ = pg::dl_catalog::get_catalog_version(root_dir, creds); + } - auto catalog_tables = pg::dl_catalog::load_tables(root_dir, creds); - auto catalog_columns = pg::dl_catalog::load_columns(root_dir, creds); + // Load tables and columns in parallel for better performance + auto [catalog_tables, catalog_columns] = pg::dl_catalog::load_tables_and_columns(root_dir, creds); if (!catalog_tables.empty()) { for (const auto& meta : catalog_tables) { const std::string qualified_name = meta.schema_name + "." + meta.table_name; diff --git a/cpp/deeplake_pg/table_storage.hpp b/cpp/deeplake_pg/table_storage.hpp index d6b7e581af..78130cb789 100644 --- a/cpp/deeplake_pg/table_storage.hpp +++ b/cpp/deeplake_pg/table_storage.hpp @@ -183,6 +183,7 @@ class table_storage void force_load_table_metadata() { tables_loaded_ = false; + catalog_version_ = 0; // Reset so version gets re-fetched for new root_path load_table_metadata(); } void mark_metadata_stale() noexcept From 28b4990ac0aca23c83c7f34a04f08e25e42c7625 Mon Sep 17 00:00:00 2001 From: khustup2 Date: Fri, 6 Feb 2026 06:12:13 +0000 Subject: [PATCH 2/5] Switch to 4.5.1 deeplake. --- cpp/deeplake_pg/dl_catalog.cpp | 15 +++++++++------ cpp/deeplake_pg/duckdb_deeplake_convert.cpp | 5 +++-- cpp/deeplake_pg/nd_utils.hpp | 9 +++++---- cpp/deeplake_pg/table_data_impl.hpp | 2 +- 4 files changed, 18 insertions(+), 13 deletions(-) diff --git a/cpp/deeplake_pg/dl_catalog.cpp b/cpp/deeplake_pg/dl_catalog.cpp index 30a1b579fc..a64d7a6ffc 100644 --- a/cpp/deeplake_pg/dl_catalog.cpp +++ b/cpp/deeplake_pg/dl_catalog.cpp @@ -9,6 +9,7 @@ #include #include #include +#include #include #include @@ -169,7 +170,7 @@ void ensure_catalog(const std::string& root_path, icm::string_map<> creds) .set_primary_key("catalog_version"); // Launch all 4 open_or_create operations in parallel - std::vector>> promises; + icm::vector>> promises; promises.reserve(4); promises.push_back( deeplake_api::open_or_create_catalog_table(tables_path, std::move(tables_schema), icm::string_map<>(creds))); @@ -329,7 +330,7 @@ load_tables_and_columns(const std::string& root_path, icm::string_map<> creds) auto tables_promise = deeplake_api::open_catalog_table(join_path(root_path, k_tables_name), icm::string_map<>(creds)); auto columns_promise = deeplake_api::open_catalog_table(join_path(root_path, k_columns_name), icm::string_map<>(creds)); - std::vector>> open_promises; + icm::vector>> open_promises; open_promises.push_back(std::move(tables_promise)); open_promises.push_back(std::move(columns_promise)); @@ -346,7 +347,7 @@ load_tables_and_columns(const std::string& root_path, icm::string_map<> creds) auto tables_read_promise = tables_table->read(); auto columns_read_promise = columns_table->read(); - std::vector> read_promises; + icm::vector> read_promises; read_promises.push_back(std::move(tables_read_promise)); read_promises.push_back(std::move(columns_read_promise)); @@ -416,14 +417,16 @@ load_tables_and_columns(const std::string& root_path, icm::string_map<> creds) } if (nullable_it != row.end()) { try { - meta.nullable = nullable_it->second.value(0); + auto nullable_vec = load_vector(nullable_it->second); + meta.nullable = !nullable_vec.empty() && nullable_vec.front() != 0; } catch (...) { meta.nullable = true; } } if (position_it != row.end()) { try { - meta.position = position_it->second.value(0); + auto pos_vec = load_vector(position_it->second); + meta.position = pos_vec.empty() ? 0 : pos_vec.front(); } catch (...) { auto pos_vec = load_int64_vector(position_it->second); meta.position = pos_vec.empty() ? 0 : static_cast(pos_vec.front()); @@ -463,7 +466,7 @@ void upsert_columns(const std::string& root_path, icm::string_map<> creds, const return; } auto table = open_catalog_table(root_path, k_columns_name, std::move(creds)); - std::vector> rows; + icm::vector> rows; rows.reserve(columns.size()); for (const auto& col : columns) { icm::string_map row; diff --git a/cpp/deeplake_pg/duckdb_deeplake_convert.cpp b/cpp/deeplake_pg/duckdb_deeplake_convert.cpp index f6785d462b..3814fa51fc 100644 --- a/cpp/deeplake_pg/duckdb_deeplake_convert.cpp +++ b/cpp/deeplake_pg/duckdb_deeplake_convert.cpp @@ -4,6 +4,7 @@ #include "utils.hpp" #include +#include #include #include @@ -85,7 +86,7 @@ T to_cpp_value(const duckdb::Value& val) nd::array to_deeplake_value_as_array_list(const duckdb::vector& values) { - std::vector arr; + icm::vector arr; arr.reserve(values.size()); for (const auto& v : values) { arr.push_back(pg::to_deeplake_value(v)); @@ -105,7 +106,7 @@ nd::array to_deeplake_value(const duckdb::LogicalType& duckdb_type, const duckdb } return switch_duckdb_type(duckdb_type, [&values]() { if constexpr (std::is_same_v) { - std::vector arr; + icm::vector arr; arr.reserve(values.size()); for (const auto& val : values) { duckdb::string_t blob_data = duckdb::StringValue::Get(val); diff --git a/cpp/deeplake_pg/nd_utils.hpp b/cpp/deeplake_pg/nd_utils.hpp index 43074a0cf1..c132fbcbd8 100644 --- a/cpp/deeplake_pg/nd_utils.hpp +++ b/cpp/deeplake_pg/nd_utils.hpp @@ -3,6 +3,7 @@ #include "exceptions.hpp" #include +#include #include #include #include @@ -124,7 +125,7 @@ inline pg::array_type pg_to_nd_typed(ArrayType* array, bool copy_data = true) const auto nrows = dims[0]; const auto ncols = dims[1]; if (copy_data) { - std::vector data_vector; + icm::vector data_vector; data_vector.reserve(static_cast(nrows)); for (int i = 0; i < nrows; ++i) { data_vector.emplace_back(nd::adapt(std::vector(data + static_cast(i) * static_cast(ncols), @@ -564,7 +565,7 @@ inline nd::array datum_to_nd(Datum value, Oid attr_typeid, int32_t typmod) return nd::none(nd::dtype::byte, 0); } else { int nelems = ArrayGetNItems(ARR_NDIM(arr), ARR_DIMS(arr)); - std::vector elements; + icm::vector elements; elements.reserve(static_cast(nelems)); Datum* datums = nullptr; @@ -590,7 +591,7 @@ inline nd::array datum_to_nd(Datum value, Oid attr_typeid, int32_t typmod) return nd::none(nd::dtype::string, 1); } else { int nelems = ArrayGetNItems(ARR_NDIM(arr), ARR_DIMS(arr)); - std::vector elements; + icm::vector elements; elements.reserve(static_cast(nelems)); Datum* datums = nullptr; bool* nulls = nullptr; @@ -903,7 +904,7 @@ nd::array eval_with_nones(nd::array arr) return nd::eval(arr); } catch (const nd::invalid_dynamic_eval&) { } - std::vector result_elements; + icm::vector result_elements; result_elements.reserve(arr.size()); for (auto a : arr) { if (a.is_none()) { diff --git a/cpp/deeplake_pg/table_data_impl.hpp b/cpp/deeplake_pg/table_data_impl.hpp index b67c13f681..9f90517fe1 100644 --- a/cpp/deeplake_pg/table_data_impl.hpp +++ b/cpp/deeplake_pg/table_data_impl.hpp @@ -499,7 +499,7 @@ inline bool table_data::flush_updates() // Flush the update rows to the dataset try { streamers_.reset(); - std::vector> update_promises; + icm::vector> update_promises; update_promises.reserve(update_rows_.size()); for (const auto& [row_number, column_name, new_value] : update_rows_) { update_promises.emplace_back(get_dataset()->update_row(row_number, column_name, new_value)); From d01f8132e24f5283e678cad32fe739202fc7a92d Mon Sep 17 00:00:00 2001 From: khustup2 Date: Fri, 6 Feb 2026 06:51:30 +0000 Subject: [PATCH 3/5] Added migration for stateless. --- cpp/deeplake_pg/dl_catalog.cpp | 194 +++-- .../tests/py_tests/test_startup_latency.py | 676 ++++++++++++++++++ .../test_stateless_catalog_resilience.py | 75 ++ .../py_tests/test_stateless_multi_instance.py | 6 +- .../test_stateless_reserved_schema.py | 7 +- 5 files changed, 892 insertions(+), 66 deletions(-) create mode 100644 postgres/tests/py_tests/test_startup_latency.py create mode 100644 postgres/tests/py_tests/test_stateless_catalog_resilience.py diff --git a/cpp/deeplake_pg/dl_catalog.cpp b/cpp/deeplake_pg/dl_catalog.cpp index a64d7a6ffc..87dadf7b26 100644 --- a/cpp/deeplake_pg/dl_catalog.cpp +++ b/cpp/deeplake_pg/dl_catalog.cpp @@ -12,6 +12,8 @@ #include #include +#include +#include #include extern "C" { @@ -67,12 +69,6 @@ struct catalog_table_cache } }; -std::shared_ptr -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(); -} - int64_t now_ms() { using namespace std::chrono; @@ -137,63 +133,137 @@ void ensure_catalog(const std::string& root_path, icm::string_map<> creds) const auto indexes_path = join_path(root_path, k_indexes_name); const auto meta_path = join_path(root_path, k_meta_name); - // Build schemas for all catalog tables - deeplake_api::catalog_table_schema tables_schema; - tables_schema.add("table_id", deeplake_core::type::text(codecs::compression::null)) - .add("schema_name", deeplake_core::type::text(codecs::compression::null)) - .add("table_name", deeplake_core::type::text(codecs::compression::null)) - .add("dataset_path", deeplake_core::type::text(codecs::compression::null)) - .add("state", deeplake_core::type::text(codecs::compression::null)) - .add("updated_at", deeplake_core::type::generic(nd::type::scalar(nd::dtype::int64))) - .set_primary_key("table_id"); - - deeplake_api::catalog_table_schema columns_schema; - columns_schema.add("column_id", deeplake_core::type::text(codecs::compression::null)) - .add("table_id", deeplake_core::type::text(codecs::compression::null)) - .add("column_name", deeplake_core::type::text(codecs::compression::null)) - .add("pg_type", deeplake_core::type::text(codecs::compression::null)) - .add("dl_type_json", deeplake_core::type::text(codecs::compression::null)) - .add("nullable", deeplake_core::type::generic(nd::type::scalar(nd::dtype::boolean))) - .add("position", deeplake_core::type::generic(nd::type::scalar(nd::dtype::int32))) - .set_primary_key("column_id"); - - deeplake_api::catalog_table_schema indexes_schema; - indexes_schema.add("table_id", deeplake_core::type::text(codecs::compression::null)) - .add("column_names", deeplake_core::type::text(codecs::compression::null)) - .add("index_type", deeplake_core::type::text(codecs::compression::null)) - .add("order_type", deeplake_core::type::generic(nd::type::scalar(nd::dtype::int32))) - .set_primary_key("table_id"); - - deeplake_api::catalog_table_schema meta_schema; - meta_schema.add("catalog_version", deeplake_core::type::generic(nd::type::scalar(nd::dtype::int64))) - .add("updated_at", deeplake_core::type::generic(nd::type::scalar(nd::dtype::int64))) - .set_primary_key("catalog_version"); - - // Launch all 4 open_or_create operations in parallel - icm::vector>> promises; - promises.reserve(4); - promises.push_back( - deeplake_api::open_or_create_catalog_table(tables_path, std::move(tables_schema), icm::string_map<>(creds))); - promises.push_back( - deeplake_api::open_or_create_catalog_table(columns_path, std::move(columns_schema), icm::string_map<>(creds))); - promises.push_back( - deeplake_api::open_or_create_catalog_table(indexes_path, std::move(indexes_schema), icm::string_map<>(creds))); - promises.push_back( - deeplake_api::open_or_create_catalog_table(meta_path, std::move(meta_schema), icm::string_map<>(creds))); - - // Wait for all to complete - auto results = async::combine(std::move(promises)).get_future().get(); - - // Initialize meta table if empty (index 3 is meta) - auto& meta_table = results[3]; - if (meta_table) { - auto snapshot = meta_table->read().get_future().get(); - if (snapshot.row_count() == 0) { - icm::string_map row; - row["catalog_version"] = nd::adapt(static_cast(1)); - row["updated_at"] = nd::adapt(now_ms()); - meta_table->insert(std::move(row)).get_future().get(); + auto migrate_legacy_path_if_needed = [&](const std::string& path) { + const bool is_remote_path = path.find("://") != std::string::npos; + + if (!is_remote_path) { + // Local path migration: only remove clear path collisions (e.g. file at table path). + std::error_code ec; + if (!std::filesystem::exists(path, ec)) { + return; + } + if (std::filesystem::is_directory(path, ec)) { + // Keep directories intact; open_or_create handles valid local catalog dirs. + return; + } + + elog(WARNING, + "Catalog path %s is a non-directory filesystem artifact. Removing it before catalog initialization.", + path.c_str()); + if (!std::filesystem::remove(path, ec) && ec) { + elog(ERROR, "Failed to migrate local catalog path %s: %s", path.c_str(), ec.message().c_str()); + } + return; + } + + // Remote path migration for legacy non-catalog datasets. + bool exists = false; + try { + exists = deeplake_api::exists(path, icm::string_map<>(creds)).get_future().get(); + } catch (...) { + exists = false; } + if (!exists) { + return; + } + + bool is_catalog = false; + try { + is_catalog = deeplake_api::is_catalog_table(path, icm::string_map<>(creds)).get_future().get(); + } catch (...) { + is_catalog = false; + } + if (!is_catalog) { + elog(WARNING, + "Existing remote catalog path %s is not a catalog table. Recreating catalog table.", + path.c_str()); + try { + deeplake_api::delete_dataset(path, icm::string_map<>(creds)).get_future().get(); + } catch (const std::exception& e) { + elog(ERROR, "Failed to migrate remote catalog path %s: %s", path.c_str(), e.what()); + } catch (...) { + elog(ERROR, "Failed to migrate remote catalog path %s: unknown error", path.c_str()); + } + } + }; + + try { + // Handle legacy non-catalog artifacts before creating tables. + migrate_legacy_path_if_needed(tables_path); + migrate_legacy_path_if_needed(columns_path); + migrate_legacy_path_if_needed(indexes_path); + migrate_legacy_path_if_needed(meta_path); + + // Build schemas for all catalog tables + deeplake_api::catalog_table_schema tables_schema; + tables_schema.add("table_id", deeplake_core::type::text(codecs::compression::null)) + .add("schema_name", deeplake_core::type::text(codecs::compression::null)) + .add("table_name", deeplake_core::type::text(codecs::compression::null)) + .add("dataset_path", deeplake_core::type::text(codecs::compression::null)) + .add("state", deeplake_core::type::text(codecs::compression::null)) + .add("updated_at", deeplake_core::type::generic(nd::type::scalar(nd::dtype::int64))) + .set_primary_key("table_id"); + + deeplake_api::catalog_table_schema columns_schema; + columns_schema.add("column_id", deeplake_core::type::text(codecs::compression::null)) + .add("table_id", deeplake_core::type::text(codecs::compression::null)) + .add("column_name", deeplake_core::type::text(codecs::compression::null)) + .add("pg_type", deeplake_core::type::text(codecs::compression::null)) + .add("dl_type_json", deeplake_core::type::text(codecs::compression::null)) + .add("nullable", deeplake_core::type::generic(nd::type::scalar(nd::dtype::boolean))) + .add("position", deeplake_core::type::generic(nd::type::scalar(nd::dtype::int32))) + .set_primary_key("column_id"); + + deeplake_api::catalog_table_schema indexes_schema; + indexes_schema.add("table_id", deeplake_core::type::text(codecs::compression::null)) + .add("column_names", deeplake_core::type::text(codecs::compression::null)) + .add("index_type", deeplake_core::type::text(codecs::compression::null)) + .add("order_type", deeplake_core::type::generic(nd::type::scalar(nd::dtype::int32))) + .set_primary_key("table_id"); + + deeplake_api::catalog_table_schema meta_schema; + meta_schema.add("catalog_version", deeplake_core::type::generic(nd::type::scalar(nd::dtype::int64))) + .add("updated_at", deeplake_core::type::generic(nd::type::scalar(nd::dtype::int64))) + .set_primary_key("catalog_version"); + + // Launch all 4 open_or_create operations in parallel + icm::vector>> promises; + promises.reserve(4); + promises.push_back( + deeplake_api::open_or_create_catalog_table(tables_path, std::move(tables_schema), icm::string_map<>(creds))); + promises.push_back( + deeplake_api::open_or_create_catalog_table(columns_path, std::move(columns_schema), icm::string_map<>(creds))); + promises.push_back( + deeplake_api::open_or_create_catalog_table(indexes_path, std::move(indexes_schema), icm::string_map<>(creds))); + promises.push_back( + deeplake_api::open_or_create_catalog_table(meta_path, std::move(meta_schema), icm::string_map<>(creds))); + + // Wait for all to complete + auto results = async::combine(std::move(promises)).get_future().get(); + if (results.size() != 4) { + elog(ERROR, + "Failed to initialize catalog at %s: expected 4 catalog tables, got %zu", + root_path.c_str(), + static_cast(results.size())); + } + + // Initialize meta table if empty (index 3 is meta) + auto& meta_table = results[3]; + if (meta_table) { + auto snapshot = meta_table->read().get_future().get(); + if (snapshot.row_count() == 0) { + icm::string_map row; + row["catalog_version"] = nd::adapt(static_cast(1)); + row["updated_at"] = nd::adapt(now_ms()); + meta_table->insert(std::move(row)).get_future().get(); + } + } + } catch (const std::exception& e) { + catalog_table_cache::instance().invalidate(); + elog(ERROR, "Failed to ensure catalog at %s: %s", root_path.c_str(), e.what()); + } catch (...) { + catalog_table_cache::instance().invalidate(); + elog(ERROR, "Failed to ensure catalog at %s: unknown error", root_path.c_str()); } } diff --git a/postgres/tests/py_tests/test_startup_latency.py b/postgres/tests/py_tests/test_startup_latency.py new file mode 100644 index 0000000000..865d4fa20a --- /dev/null +++ b/postgres/tests/py_tests/test_startup_latency.py @@ -0,0 +1,676 @@ +""" +Test startup latency and catalog loading performance for pg_deeplake extension. + +This test measures: +1. Cold start latency (new PostgreSQL backend connecting) +2. Catalog loading time with stateless mode +3. Time to first query +4. Multi-table catalog discovery time +5. Comparison of stateless vs non-stateless modes + +Run with: pytest test_startup_latency.py -v -s +""" +import pytest +import asyncpg +import asyncio +import os +import shutil +import subprocess +import time +import tempfile +import statistics +from pathlib import Path +from typing import Dict, List, Tuple, Optional +from dataclasses import dataclass, field + + +@dataclass +class LatencyMetrics: + """Container for latency measurements.""" + connection_time_ms: float = 0.0 + extension_load_time_ms: float = 0.0 + root_path_set_time_ms: float = 0.0 + first_query_time_ms: float = 0.0 + table_create_time_ms: float = 0.0 + catalog_discovery_time_ms: float = 0.0 + total_ready_time_ms: float = 0.0 + + def __str__(self) -> str: + return ( + f" Connection: {self.connection_time_ms:8.2f} ms\n" + f" Extension load: {self.extension_load_time_ms:8.2f} ms\n" + f" Root path set: {self.root_path_set_time_ms:8.2f} ms\n" + f" First query: {self.first_query_time_ms:8.2f} ms\n" + f" Table create: {self.table_create_time_ms:8.2f} ms\n" + f" Catalog discovery: {self.catalog_discovery_time_ms:8.2f} ms\n" + f" Total ready time: {self.total_ready_time_ms:8.2f} ms" + ) + + +@dataclass +class LatencyReport: + """Aggregated latency report across multiple runs.""" + metrics: List[LatencyMetrics] = field(default_factory=list) + + def add(self, m: LatencyMetrics): + self.metrics.append(m) + + def summary(self, name: str) -> str: + if not self.metrics: + return f"{name}: No data" + + def stats(values: List[float]) -> Tuple[float, float, float, float]: + if not values: + return (0.0, 0.0, 0.0, 0.0) + return ( + min(values), + max(values), + statistics.mean(values), + statistics.median(values) + ) + + conn = stats([m.connection_time_ms for m in self.metrics]) + ext = stats([m.extension_load_time_ms for m in self.metrics]) + root = stats([m.root_path_set_time_ms for m in self.metrics]) + query = stats([m.first_query_time_ms for m in self.metrics]) + create = stats([m.table_create_time_ms for m in self.metrics]) + disc = stats([m.catalog_discovery_time_ms for m in self.metrics]) + total = stats([m.total_ready_time_ms for m in self.metrics]) + + return ( + f"\n{'='*60}\n" + f"{name} ({len(self.metrics)} runs)\n" + f"{'='*60}\n" + f"{'Metric':<22} {'Min':>10} {'Max':>10} {'Mean':>10} {'Median':>10}\n" + f"{'-'*60}\n" + f"{'Connection':<22} {conn[0]:>10.2f} {conn[1]:>10.2f} {conn[2]:>10.2f} {conn[3]:>10.2f}\n" + f"{'Extension load':<22} {ext[0]:>10.2f} {ext[1]:>10.2f} {ext[2]:>10.2f} {ext[3]:>10.2f}\n" + f"{'Root path set':<22} {root[0]:>10.2f} {root[1]:>10.2f} {root[2]:>10.2f} {root[3]:>10.2f}\n" + f"{'First query':<22} {query[0]:>10.2f} {query[1]:>10.2f} {query[2]:>10.2f} {query[3]:>10.2f}\n" + f"{'Table create':<22} {create[0]:>10.2f} {create[1]:>10.2f} {create[2]:>10.2f} {create[3]:>10.2f}\n" + f"{'Catalog discovery':<22} {disc[0]:>10.2f} {disc[1]:>10.2f} {disc[2]:>10.2f} {disc[3]:>10.2f}\n" + f"{'-'*60}\n" + f"{'TOTAL READY TIME':<22} {total[0]:>10.2f} {total[1]:>10.2f} {total[2]:>10.2f} {total[3]:>10.2f}\n" + f"{'='*60}\n" + ) + + +async def measure_connection_latency( + port: int = 5432, + database: str = "postgres", + with_extension: bool = True, + root_path: Optional[str] = None, + stateless_enabled: bool = False, + run_first_query: bool = True, + create_table: bool = False, + table_name: str = "latency_test", +) -> LatencyMetrics: + """ + Measure various latency components of connecting to PostgreSQL. + + Args: + port: PostgreSQL port + database: Database to connect to + with_extension: Whether to load pg_deeplake extension + root_path: If set, configure deeplake.root_path + stateless_enabled: Whether to enable stateless mode + run_first_query: Whether to measure first query time + create_table: Whether to measure table creation time + table_name: Name for test table + + Returns: + LatencyMetrics with all measurements + """ + user = os.environ.get("USER", "postgres") + metrics = LatencyMetrics() + total_start = time.perf_counter() + + # 1. Measure connection time + conn_start = time.perf_counter() + conn = await asyncpg.connect( + database=database, + user=user, + host="localhost", + port=port, + statement_cache_size=0 + ) + metrics.connection_time_ms = (time.perf_counter() - conn_start) * 1000 + + try: + # 2. Measure extension load time + if with_extension: + ext_start = time.perf_counter() + await conn.execute("DROP EXTENSION IF EXISTS pg_deeplake CASCADE") + await conn.execute("CREATE EXTENSION pg_deeplake") + metrics.extension_load_time_ms = (time.perf_counter() - ext_start) * 1000 + + # Set stateless mode if requested + if stateless_enabled: + await conn.execute("SET deeplake.stateless_enabled = true") + + # 3. Measure root_path set time (triggers catalog loading in stateless mode) + if root_path: + root_start = time.perf_counter() + await conn.execute(f"SET deeplake.root_path = '{root_path}'") + metrics.root_path_set_time_ms = (time.perf_counter() - root_start) * 1000 + + # 4. Measure first query time + if run_first_query and with_extension: + query_start = time.perf_counter() + await conn.execute("SELECT 1") + metrics.first_query_time_ms = (time.perf_counter() - query_start) * 1000 + + # 5. Measure table creation time + if create_table and with_extension: + create_start = time.perf_counter() + await conn.execute(f"DROP TABLE IF EXISTS {table_name} CASCADE") + await conn.execute(f""" + CREATE TABLE {table_name} ( + id INT, + name TEXT, + value FLOAT + ) USING deeplake + """) + metrics.table_create_time_ms = (time.perf_counter() - create_start) * 1000 + + # Cleanup + await conn.execute(f"DROP TABLE IF EXISTS {table_name} CASCADE") + + metrics.total_ready_time_ms = (time.perf_counter() - total_start) * 1000 + + finally: + await conn.close() + + return metrics + + +async def measure_catalog_discovery_latency( + port: int, + root_path: str, + num_tables: int, + stateless_enabled: bool = True, +) -> LatencyMetrics: + """ + Measure time to discover existing tables from catalog. + + This simulates a second instance discovering tables created by another instance. + """ + user = os.environ.get("USER", "postgres") + metrics = LatencyMetrics() + total_start = time.perf_counter() + + conn_start = time.perf_counter() + conn = await asyncpg.connect( + database="postgres", + user=user, + host="localhost", + port=port, + statement_cache_size=0 + ) + metrics.connection_time_ms = (time.perf_counter() - conn_start) * 1000 + + try: + # Load extension + ext_start = time.perf_counter() + await conn.execute("DROP EXTENSION IF EXISTS pg_deeplake CASCADE") + await conn.execute("CREATE EXTENSION pg_deeplake") + if stateless_enabled: + await conn.execute("SET deeplake.stateless_enabled = true") + metrics.extension_load_time_ms = (time.perf_counter() - ext_start) * 1000 + + # Set root_path - this triggers catalog discovery + disc_start = time.perf_counter() + await conn.execute(f"SET deeplake.root_path = '{root_path}'") + metrics.root_path_set_time_ms = (time.perf_counter() - disc_start) * 1000 + + # Verify tables were discovered + query_start = time.perf_counter() + count = await conn.fetchval("SELECT COUNT(*) FROM pg_deeplake_tables") + metrics.first_query_time_ms = (time.perf_counter() - query_start) * 1000 + + metrics.catalog_discovery_time_ms = metrics.root_path_set_time_ms + metrics.total_ready_time_ms = (time.perf_counter() - total_start) * 1000 + + finally: + await conn.close() + + return metrics + + +@pytest.fixture +def temp_root_path(temp_dir_for_postgres): + """Create a temporary root path for deeplake datasets.""" + return temp_dir_for_postgres + + +@pytest.mark.asyncio +async def test_baseline_connection_latency(pg_server): + """ + Measure baseline connection latency without extension. + + This establishes the baseline PostgreSQL connection overhead. + """ + print("\n" + "="*60) + print("BASELINE CONNECTION LATENCY (no extension)") + print("="*60) + + report = LatencyReport() + num_runs = 5 + + for i in range(num_runs): + metrics = await measure_connection_latency( + with_extension=False, + run_first_query=False, + ) + report.add(metrics) + print(f"Run {i+1}: Connection = {metrics.connection_time_ms:.2f} ms") + + print(report.summary("Baseline (no extension)")) + + +@pytest.mark.asyncio +async def test_extension_load_latency(pg_server): + """ + Measure latency of loading pg_deeplake extension. + + This measures the overhead of CREATE EXTENSION pg_deeplake. + """ + print("\n" + "="*60) + print("EXTENSION LOAD LATENCY") + print("="*60) + + report = LatencyReport() + num_runs = 5 + + for i in range(num_runs): + metrics = await measure_connection_latency( + with_extension=True, + run_first_query=True, + create_table=False, + ) + report.add(metrics) + print(f"Run {i+1}:") + print(metrics) + print() + + print(report.summary("Extension Load")) + + +@pytest.mark.asyncio +async def test_table_creation_latency(pg_server, temp_root_path): + """ + Measure latency of creating a deeplake table. + """ + print("\n" + "="*60) + print("TABLE CREATION LATENCY") + print("="*60) + + report = LatencyReport() + num_runs = 5 + + for i in range(num_runs): + metrics = await measure_connection_latency( + with_extension=True, + root_path=temp_root_path, + run_first_query=True, + create_table=True, + table_name=f"latency_test_{i}", + ) + report.add(metrics) + print(f"Run {i+1}:") + print(metrics) + print() + + print(report.summary("Table Creation")) + + +@pytest.mark.asyncio +async def test_stateless_catalog_loading_latency(pg_server, temp_root_path): + """ + Measure catalog loading latency with stateless mode enabled. + + This is the critical test for the parallelization improvement. + """ + print("\n" + "="*60) + print("STATELESS CATALOG LOADING LATENCY") + print("="*60) + + user = os.environ.get("USER", "postgres") + + # First, create some tables in the catalog + print("\nSetup: Creating tables in catalog...") + setup_conn = await asyncpg.connect( + database="postgres", + user=user, + host="localhost", + statement_cache_size=0 + ) + + try: + await setup_conn.execute("DROP EXTENSION IF EXISTS pg_deeplake CASCADE") + await setup_conn.execute("CREATE EXTENSION pg_deeplake") + await setup_conn.execute("SET deeplake.stateless_enabled = true") + await setup_conn.execute(f"SET deeplake.root_path = '{temp_root_path}'") + + # Create multiple tables to populate the catalog + num_tables = 5 + for i in range(num_tables): + await setup_conn.execute(f""" + CREATE TABLE catalog_test_{i} ( + id INT, + name TEXT, + data FLOAT + ) USING deeplake + """) + await setup_conn.execute(f"INSERT INTO catalog_test_{i} VALUES ({i}, 'test', {i}.5)") + + print(f"Created {num_tables} tables in catalog") + + # Verify tables in catalog + count = await setup_conn.fetchval("SELECT COUNT(*) FROM pg_deeplake_tables") + print(f"Tables in catalog: {count}") + + finally: + await setup_conn.close() + + # Now measure the catalog discovery time from a fresh connection + print("\nMeasuring catalog discovery latency...") + report = LatencyReport() + num_runs = 5 + + for i in range(num_runs): + metrics = await measure_catalog_discovery_latency( + port=5432, + root_path=temp_root_path, + num_tables=num_tables, + stateless_enabled=True, + ) + report.add(metrics) + print(f"Run {i+1}:") + print(metrics) + print() + + print(report.summary("Stateless Catalog Loading")) + + # Cleanup + cleanup_conn = await asyncpg.connect( + database="postgres", + user=user, + host="localhost", + statement_cache_size=0 + ) + try: + await cleanup_conn.execute("DROP EXTENSION IF EXISTS pg_deeplake CASCADE") + await cleanup_conn.execute("CREATE EXTENSION pg_deeplake") + await cleanup_conn.execute(f"SET deeplake.root_path = '{temp_root_path}'") + for i in range(num_tables): + await cleanup_conn.execute(f"DROP TABLE IF EXISTS catalog_test_{i} CASCADE") + finally: + await cleanup_conn.close() + + +@pytest.mark.asyncio +async def test_stateless_vs_nonstateless_comparison(pg_server, temp_root_path): + """ + Compare latency between stateless and non-stateless modes. + """ + print("\n" + "="*60) + print("STATELESS vs NON-STATELESS COMPARISON") + print("="*60) + + # Non-stateless (local catalog) + print("\n--- Non-Stateless Mode ---") + non_stateless_report = LatencyReport() + for i in range(3): + metrics = await measure_connection_latency( + with_extension=True, + root_path=temp_root_path, + stateless_enabled=False, + run_first_query=True, + create_table=True, + table_name=f"nonstateless_test_{i}", + ) + non_stateless_report.add(metrics) + + print(non_stateless_report.summary("Non-Stateless Mode")) + + # Stateless (shared catalog) + print("\n--- Stateless Mode ---") + stateless_report = LatencyReport() + for i in range(3): + metrics = await measure_connection_latency( + with_extension=True, + root_path=temp_root_path, + stateless_enabled=True, + run_first_query=True, + create_table=True, + table_name=f"stateless_test_{i}", + ) + stateless_report.add(metrics) + + print(stateless_report.summary("Stateless Mode")) + + # Calculate overhead + non_stateless_avg = statistics.mean([m.total_ready_time_ms for m in non_stateless_report.metrics]) + stateless_avg = statistics.mean([m.total_ready_time_ms for m in stateless_report.metrics]) + overhead = stateless_avg - non_stateless_avg + overhead_pct = (overhead / non_stateless_avg) * 100 if non_stateless_avg > 0 else 0 + + print(f"\n{'='*60}") + print(f"OVERHEAD ANALYSIS") + print(f"{'='*60}") + print(f"Non-stateless avg total: {non_stateless_avg:.2f} ms") + print(f"Stateless avg total: {stateless_avg:.2f} ms") + print(f"Stateless overhead: {overhead:.2f} ms ({overhead_pct:.1f}%)") + print(f"{'='*60}") + + +@pytest.mark.asyncio +async def test_multi_table_catalog_scaling(pg_server, temp_root_path): + """ + Test how catalog loading time scales with number of tables. + """ + print("\n" + "="*60) + print("CATALOG LOADING SCALING TEST") + print("="*60) + + user = os.environ.get("USER", "postgres") + table_counts = [1, 5, 10, 20] + results = [] + + for num_tables in table_counts: + print(f"\n--- Testing with {num_tables} tables ---") + + # Setup: Create tables + setup_conn = await asyncpg.connect( + database="postgres", + user=user, + host="localhost", + statement_cache_size=0 + ) + + try: + await setup_conn.execute("DROP EXTENSION IF EXISTS pg_deeplake CASCADE") + await setup_conn.execute("CREATE EXTENSION pg_deeplake") + await setup_conn.execute("SET deeplake.stateless_enabled = true") + await setup_conn.execute(f"SET deeplake.root_path = '{temp_root_path}'") + + # Create tables + for i in range(num_tables): + await setup_conn.execute(f""" + CREATE TABLE scale_test_{num_tables}_{i} ( + id INT, name TEXT + ) USING deeplake + """) + finally: + await setup_conn.close() + + # Measure catalog discovery + report = LatencyReport() + for _ in range(3): + metrics = await measure_catalog_discovery_latency( + port=5432, + root_path=temp_root_path, + num_tables=num_tables, + stateless_enabled=True, + ) + report.add(metrics) + + avg_time = statistics.mean([m.catalog_discovery_time_ms for m in report.metrics]) + results.append((num_tables, avg_time)) + print(f" Average catalog discovery time: {avg_time:.2f} ms") + + # Cleanup + cleanup_conn = await asyncpg.connect( + database="postgres", + user=user, + host="localhost", + statement_cache_size=0 + ) + try: + await cleanup_conn.execute("DROP EXTENSION IF EXISTS pg_deeplake CASCADE") + await cleanup_conn.execute("CREATE EXTENSION pg_deeplake") + await cleanup_conn.execute(f"SET deeplake.root_path = '{temp_root_path}'") + for i in range(num_tables): + await cleanup_conn.execute(f"DROP TABLE IF EXISTS scale_test_{num_tables}_{i} CASCADE") + finally: + await cleanup_conn.close() + + # Print scaling summary + print(f"\n{'='*60}") + print(f"SCALING SUMMARY") + print(f"{'='*60}") + print(f"{'Tables':<10} {'Avg Discovery Time (ms)':<25}") + print(f"{'-'*35}") + for num_tables, avg_time in results: + print(f"{num_tables:<10} {avg_time:<25.2f}") + + # Check if scaling is sub-linear (good) or linear/super-linear (bad) + if len(results) >= 2: + first_time = results[0][1] + last_time = results[-1][1] + first_count = results[0][0] + last_count = results[-1][0] + + table_ratio = last_count / first_count + time_ratio = last_time / first_time if first_time > 0 else 0 + + print(f"\nTable count increased {table_ratio:.1f}x") + print(f"Discovery time increased {time_ratio:.1f}x") + + if time_ratio < table_ratio: + print("Result: SUB-LINEAR scaling (good!)") + elif time_ratio > table_ratio * 1.5: + print("Result: SUPER-LINEAR scaling (needs optimization)") + else: + print("Result: APPROXIMATELY LINEAR scaling") + + print(f"{'='*60}") + + +@pytest.mark.asyncio +async def test_cold_start_simulation(pg_server, temp_root_path): + """ + Simulate a cold start scenario where a new backend connects. + + This measures the total time from connection to being ready to serve queries. + """ + print("\n" + "="*60) + print("COLD START SIMULATION") + print("="*60) + + user = os.environ.get("USER", "postgres") + + # Setup: Create some existing data + print("\nSetup: Creating existing data...") + setup_conn = await asyncpg.connect( + database="postgres", + user=user, + host="localhost", + statement_cache_size=0 + ) + + try: + await setup_conn.execute("DROP EXTENSION IF EXISTS pg_deeplake CASCADE") + await setup_conn.execute("CREATE EXTENSION pg_deeplake") + await setup_conn.execute("SET deeplake.stateless_enabled = true") + await setup_conn.execute(f"SET deeplake.root_path = '{temp_root_path}'") + + await setup_conn.execute(""" + CREATE TABLE existing_data ( + id INT, + name TEXT, + embedding FLOAT[] + ) USING deeplake + """) + + # Insert some data + for i in range(100): + await setup_conn.execute(f""" + INSERT INTO existing_data VALUES + ({i}, 'item_{i}', ARRAY[{','.join(str(float(j)) for j in range(128))}]) + """) + + print("Created table with 100 rows") + finally: + await setup_conn.close() + + # Measure cold start - simulating new backend connections to existing data + # We use separate connections without dropping the extension to preserve the table + print("\nMeasuring cold start latency...") + + for run in range(3): + total_start = time.perf_counter() + + conn = await asyncpg.connect( + database="postgres", + user=user, + host="localhost", + statement_cache_size=0 + ) + conn_time = (time.perf_counter() - total_start) * 1000 + + try: + # Extension is already loaded via shared_preload_libraries + # Just configure the session (simulating a new backend) + ext_start = time.perf_counter() + await conn.execute("SET deeplake.stateless_enabled = true") + ext_time = (time.perf_counter() - ext_start) * 1000 + + root_start = time.perf_counter() + await conn.execute(f"SET deeplake.root_path = '{temp_root_path}'") + root_time = (time.perf_counter() - root_start) * 1000 + + # First real query against the existing data + query_start = time.perf_counter() + count = await conn.fetchval("SELECT COUNT(*) FROM existing_data") + query_time = (time.perf_counter() - query_start) * 1000 + + total_time = (time.perf_counter() - total_start) * 1000 + + print(f"\nRun {run + 1}:") + print(f" Connection: {conn_time:8.2f} ms") + print(f" Session config: {ext_time:8.2f} ms") + print(f" Root path set: {root_time:8.2f} ms") + print(f" First query: {query_time:8.2f} ms (count={count})") + print(f" TOTAL COLD START: {total_time:8.2f} ms") + + finally: + await conn.close() + + # Cleanup + cleanup_conn = await asyncpg.connect( + database="postgres", + user=user, + host="localhost", + statement_cache_size=0 + ) + try: + await cleanup_conn.execute("DROP EXTENSION IF EXISTS pg_deeplake CASCADE") + await cleanup_conn.execute("CREATE EXTENSION pg_deeplake") + await cleanup_conn.execute(f"SET deeplake.root_path = '{temp_root_path}'") + await cleanup_conn.execute("DROP TABLE IF EXISTS existing_data CASCADE") + finally: + await cleanup_conn.close() + + print(f"\n{'='*60}") diff --git a/postgres/tests/py_tests/test_stateless_catalog_resilience.py b/postgres/tests/py_tests/test_stateless_catalog_resilience.py new file mode 100644 index 0000000000..453207aeef --- /dev/null +++ b/postgres/tests/py_tests/test_stateless_catalog_resilience.py @@ -0,0 +1,75 @@ +""" +Release-risk repro tests for pg_deeplake stateless mode. + +These tests are part of the default suite and should stay enabled. +Run with: + pytest postgres/tests/py_tests/test_stateless_release_risks.py +""" + +from pathlib import Path +import os + +import asyncpg +import pytest + + +def _sql_literal(value: str) -> str: + """Return a single-quoted SQL literal with escaped quotes.""" + return "'" + value.replace("'", "''") + "'" + + +@pytest.mark.asyncio +async def test_stateless_catalog_recovers_from_legacy_non_catalog_path(db_conn: asyncpg.Connection, temp_dir_for_postgres: str): + """ + Risk #1 repro: + A pre-existing non-catalog object at __deeplake_catalog/tables should be migrated/recovered. + + Expected release behavior: + - SET deeplake.root_path succeeds + - Catalog is usable after recovery + """ + await db_conn.execute("SET deeplake.stateless_enabled = true") + + root_path = Path(temp_dir_for_postgres) / "legacy_non_catalog_root" + poisoned_path = root_path / "__deeplake_catalog" / "tables" + poisoned_path.parent.mkdir(parents=True, exist_ok=True) + poisoned_path.write_text("not a deeplake catalog table", encoding="utf-8") + + await db_conn.execute(f"SET deeplake.root_path = {_sql_literal(str(root_path))}") + + # If recovery worked, catalog-backed table registration should still work. + await db_conn.execute("DROP TABLE IF EXISTS stateless_legacy_recovery") + await db_conn.execute("CREATE TABLE stateless_legacy_recovery (id INTEGER) USING deeplake") + + count = await db_conn.fetchval( + "SELECT COUNT(*) FROM pg_deeplake_tables WHERE table_name = 'public.stateless_legacy_recovery'" + ) + assert count == 1 + + +@pytest.mark.asyncio +async def test_stateless_bootstrap_permission_error_keeps_backend_alive(db_conn: asyncpg.Connection, temp_dir_for_postgres: str): + """ + Risk #2 repro: + Catalog bootstrap failure (permission denied) must not kill backend/session. + + Expected release behavior: + - SET deeplake.root_path fails with a PostgreSQL error + - Same connection remains usable afterwards + """ + await db_conn.execute("SET deeplake.stateless_enabled = true") + + readonly_root = Path(temp_dir_for_postgres) / "readonly_root" + readonly_root.mkdir(parents=True, exist_ok=True) + os.chmod(readonly_root, 0o555) + + try: + with pytest.raises(asyncpg.PostgresError): + await db_conn.execute(f"SET deeplake.root_path = {_sql_literal(str(readonly_root))}") + + # Critical assertion: backend/session is still alive. + health = await db_conn.fetchval("SELECT 1") + assert health == 1 + finally: + # Allow temp fixture cleanup. + os.chmod(readonly_root, 0o755) diff --git a/postgres/tests/py_tests/test_stateless_multi_instance.py b/postgres/tests/py_tests/test_stateless_multi_instance.py index 8aa9722a3f..2bbf86645f 100644 --- a/postgres/tests/py_tests/test_stateless_multi_instance.py +++ b/postgres/tests/py_tests/test_stateless_multi_instance.py @@ -10,8 +10,6 @@ """ import pytest -# Skip all tests in this module - stateless is disabled by default (deeplake.stateless_enabled=false) -pytestmark = pytest.mark.skip(reason="Stateless mode disabled by default") import asyncpg import asyncio import os @@ -215,6 +213,7 @@ async def primary_conn(pg_server): # Setup: Clean extension state await conn.execute("DROP EXTENSION IF EXISTS pg_deeplake CASCADE") await conn.execute("CREATE EXTENSION pg_deeplake") + await conn.execute("SET deeplake.stateless_enabled = true") yield conn finally: await conn.close() @@ -324,6 +323,7 @@ async def test_stateless_data_sync_between_instances( try: # Setup extension (create if not exists for session-scoped instance reuse) await conn_b.execute("CREATE EXTENSION IF NOT EXISTS pg_deeplake") + await conn_b.execute("SET deeplake.stateless_enabled = true") # Setting root_path should automatically discover and register tables from catalog await conn_b.execute(f"SET deeplake.root_path = '{shared_root_path}'") @@ -412,6 +412,7 @@ async def test_stateless_concurrent_writes( conn_b = await second_instance.connect() try: await conn_b.execute("CREATE EXTENSION IF NOT EXISTS pg_deeplake") + await conn_b.execute("SET deeplake.stateless_enabled = true") # Setting root_path should auto-discover tables from deeplake catalog await conn_b.execute(f"SET deeplake.root_path = '{shared_root_path}'") @@ -514,6 +515,7 @@ async def test_stateless_multiple_tables_discovery( conn_b = await second_instance.connect() try: await conn_b.execute("CREATE EXTENSION IF NOT EXISTS pg_deeplake") + await conn_b.execute("SET deeplake.stateless_enabled = true") # Setting root_path should auto-discover ALL tables from deeplake catalog await conn_b.execute(f"SET deeplake.root_path = '{shared_root_path}'") diff --git a/postgres/tests/py_tests/test_stateless_reserved_schema.py b/postgres/tests/py_tests/test_stateless_reserved_schema.py index 8b619db98a..0694d14404 100644 --- a/postgres/tests/py_tests/test_stateless_reserved_schema.py +++ b/postgres/tests/py_tests/test_stateless_reserved_schema.py @@ -14,8 +14,6 @@ """ import pytest -# Skip all tests in this module - stateless is disabled by default (deeplake.stateless_enabled=false) -pytestmark = pytest.mark.skip(reason="Stateless mode disabled by default") import asyncpg import os import shutil @@ -77,6 +75,7 @@ async def primary_conn(pg_server): try: await conn.execute("DROP EXTENSION IF EXISTS pg_deeplake CASCADE") await conn.execute("CREATE EXTENSION pg_deeplake") + await conn.execute("SET deeplake.stateless_enabled = true") yield conn finally: await conn.close() @@ -145,6 +144,7 @@ async def test_catalog_sync_default_schema( try: await conn_b.execute("CREATE EXTENSION IF NOT EXISTS pg_deeplake") + await conn_b.execute("SET deeplake.stateless_enabled = true") # This is the critical part - setting root_path triggers catalog sync # which should properly quote "default" schema name in generated DDL @@ -236,6 +236,7 @@ async def test_catalog_sync_multiple_reserved_schemas( try: await conn_b.execute("CREATE EXTENSION IF NOT EXISTS pg_deeplake") + await conn_b.execute("SET deeplake.stateless_enabled = true") await conn_b.execute(f"SET deeplake.root_path = '{shared_root_path}'") # Verify all tables discovered @@ -321,6 +322,7 @@ async def test_catalog_sync_default_schema_with_indexes( try: await conn_b.execute("CREATE EXTENSION IF NOT EXISTS pg_deeplake") + await conn_b.execute("SET deeplake.stateless_enabled = true") await conn_b.execute(f"SET deeplake.root_path = '{shared_root_path}'") # Verify table discovered @@ -376,6 +378,7 @@ async def test_catalog_sync_default_schema_write_from_secondary( try: await conn_b.execute("CREATE EXTENSION IF NOT EXISTS pg_deeplake") + await conn_b.execute("SET deeplake.stateless_enabled = true") await conn_b.execute(f"SET deeplake.root_path = '{shared_root_path}'") # Insert from Instance B From 28d7bd28d049b3a1911877a947b7c27158fac6ad Mon Sep 17 00:00:00 2001 From: khustup2 Date: Sat, 7 Feb 2026 08:04:20 +0000 Subject: [PATCH 4/5] Switch to deeplake 4.5.1 --- DEEPLAKE_API_VERSION | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/DEEPLAKE_API_VERSION b/DEEPLAKE_API_VERSION index a84947d6ff..4404a17bae 100644 --- a/DEEPLAKE_API_VERSION +++ b/DEEPLAKE_API_VERSION @@ -1 +1 @@ -4.5.0 +4.5.1 From 46b5cfd65a1adb22d424e172f25a39841f378173 Mon Sep 17 00:00:00 2001 From: khustup2 Date: Sat, 7 Feb 2026 15:40:26 +0000 Subject: [PATCH 5/5] Update test. --- .../tests/py_tests/test_stateless_catalog_resilience.py | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/postgres/tests/py_tests/test_stateless_catalog_resilience.py b/postgres/tests/py_tests/test_stateless_catalog_resilience.py index 453207aeef..17d1e6a0ee 100644 --- a/postgres/tests/py_tests/test_stateless_catalog_resilience.py +++ b/postgres/tests/py_tests/test_stateless_catalog_resilience.py @@ -8,6 +8,7 @@ from pathlib import Path import os +import shutil import asyncpg import pytest @@ -35,6 +36,13 @@ async def test_stateless_catalog_recovers_from_legacy_non_catalog_path(db_conn: poisoned_path.parent.mkdir(parents=True, exist_ok=True) poisoned_path.write_text("not a deeplake catalog table", encoding="utf-8") + # When running as root (CI), ensure postgres user can delete the file + if os.geteuid() == 0: + user = os.environ.get("USER", "postgres") + for p in [root_path, root_path / "__deeplake_catalog"]: + shutil.chown(p, user=user, group=user) + shutil.chown(poisoned_path, user=user, group=user) + await db_conn.execute(f"SET deeplake.root_path = {_sql_literal(str(root_path))}") # If recovery worked, catalog-backed table registration should still work.