From f614995f7b21021b7fe751c769c4df866b961b67 Mon Sep 17 00:00:00 2001 From: Grace Li Date: Fri, 25 Sep 2026 18:00:03 -0700 Subject: [PATCH 1/4] Fix bug where lambda_min_edge was being incorrectly counted in sum_lambda_edges_added for the root chain clusters, giving incorrect stability calculation --- .../distributed_clustering_build_hierarchy.hpp | 15 +++++++++------ 1 file changed, 9 insertions(+), 6 deletions(-) diff --git a/src/clustering/distributed_clustering_build_hierarchy.hpp b/src/clustering/distributed_clustering_build_hierarchy.hpp index 45a80bf..6fa5503 100644 --- a/src/clustering/distributed_clustering_build_hierarchy.hpp +++ b/src/clustering/distributed_clustering_build_hierarchy.hpp @@ -710,10 +710,11 @@ supernode_t fill_missing_root_chain_cluster_info( full_root_chain_cluster_array_ptr) { // Calculate num points added and sum lambda edges added for this // cluster - id_t num_points_added = edges.size() - 1; - distance_t sum_lambda_edges_added = 0.0; + id_t num_points_added = edges.size() - 1; + distance_t sum_lambda_edges_added_including_min_edge = 0.0; for (auto &edge : edges) { - sum_lambda_edges_added += lambda_from_dist(edge.second); + sum_lambda_edges_added_including_min_edge += + lambda_from_dist(edge.second); } // Set the values in the cluster array @@ -722,12 +723,14 @@ supernode_t fill_missing_root_chain_cluster_info( [](const id_t &index, std::pair &value, const id_t &num_points_added, - const distance_t &sum_lambda_edges_added) { + const distance_t + &sum_lambda_edges_added_including_min_edge) { value.second.num_points_added = num_points_added; value.second.sum_lambda_edges_added = - sum_lambda_edges_added; + sum_lambda_edges_added_including_min_edge - + value.second.lambda_min_edge; }, - num_points_added, sum_lambda_edges_added); + num_points_added, sum_lambda_edges_added_including_min_edge); }; root_chain_cluster_edges_map.async_visit( From b9faecd9578bc6e97174d749cdabef15854ca21b Mon Sep 17 00:00:00 2001 From: Grace Li Date: Fri, 25 Sep 2026 17:46:21 -0700 Subject: [PATCH 2/4] Finalize changing edge contraction and endpoint YGM maps to YGM arrays, fix bug of overcounting lambda_min_edge contribution to root chain cluster stabilities --- ...distributed_clustering_build_hierarchy.hpp | 20 +-- .../distributed_clustering_hdbscan.hpp | 70 ++++---- ...distributed_clustering_mst_contraction.hpp | 151 +++++++++--------- .../distributed_clustering_types.hpp | 12 +- .../run_distributed_hdbscan_clustering.cpp | 130 ++++++++++----- 5 files changed, 216 insertions(+), 167 deletions(-) diff --git a/src/clustering/distributed_clustering_build_hierarchy.hpp b/src/clustering/distributed_clustering_build_hierarchy.hpp index 6fa5503..b6cb5a4 100644 --- a/src/clustering/distributed_clustering_build_hierarchy.hpp +++ b/src/clustering/distributed_clustering_build_hierarchy.hpp @@ -25,13 +25,13 @@ namespace clams::clustering { * contraction map so we can clear it and save space * * @param alpha_edge_map An empty YGM map of alpha edge id -> alpha edge info. - * @param edge_contraction_map YGM map of edge id -> edge contraction info. + * @param edge_contraction_array YGM array of edge id -> edge contraction info. * @param root_chain_supernode Supernode name for the root chain. */ void fill_alpha_edge_map( - ygm::container::map &alpha_edge_map, - ygm::container::map &edge_contraction_map, - const supernode_t &root_chain_supernode) { + ygm::container::map &alpha_edge_map, + ygm::container::array &edge_contraction_array, + const supernode_t &root_chain_supernode) { ygm::comm &comm = alpha_edge_map.comm(); auto set_alpha_edge_info = [&alpha_edge_map, &root_chain_supernode]( const id_t &edge_id, @@ -51,7 +51,7 @@ void fill_alpha_edge_map( chain_supernode); } }; - edge_contraction_map.for_all(set_alpha_edge_info); + edge_contraction_array.for_all(set_alpha_edge_info); comm.barrier(); } @@ -59,7 +59,7 @@ void fill_alpha_edge_map( * @brief Goes through the edge contraction map and assigns all edges to chains, * represented by their supernode. * - * @param edge_contraction_map YGM map of edge id -> edge contraction info. + * @param edge_contraction_array YGM array of edge id -> edge contraction info. * @param chain_map An empty YGM map of chain name (supernode) -> pair of * (cluster map, chain info). The cluster map goes from edge id -> full cluster * info, and the chain info stores the required chain info that doesn't go in @@ -75,15 +75,15 @@ void fill_alpha_edge_map( * tell which edges belong to the root chain. */ void assign_edges_to_chains( - ygm::container::map &edge_contraction_map, + ygm::container::array &edge_contraction_array, ygm::container::map, - full_chain_info>> &chain_map, + full_chain_info>> &chain_map, std::vector &local_root_chain_alpha_edges, ygm::container::set &root_chain_non_alpha_edge_set, ygm::container::map &leaf_cluster_map, const uint32_t final_round) { - ygm::comm &comm = edge_contraction_map.comm(); + ygm::comm &comm = edge_contraction_array.comm(); static auto add_alpha_edge_to_chain_lambda = []([[maybe_unused]] const supernode_t &chain_name, @@ -162,7 +162,7 @@ void assign_edges_to_chains( } } }; - edge_contraction_map.for_all(assign_edge_to_chain_lambda); + edge_contraction_array.for_all(assign_edge_to_chain_lambda); comm.barrier(); } diff --git a/src/clustering/distributed_clustering_hdbscan.hpp b/src/clustering/distributed_clustering_hdbscan.hpp index 25792e0..7e8899c 100644 --- a/src/clustering/distributed_clustering_hdbscan.hpp +++ b/src/clustering/distributed_clustering_hdbscan.hpp @@ -1341,11 +1341,13 @@ calculate_root_chain_size_stability( for (auto it = root_chain_array.local_begin(); it != root_chain_array.local_end(); ++it) { root_chain_cluster_info cluster_info = it->value.second; - float stability = cluster_info.lambda_min_edge * - (cluster_info.size - cluster_info.num_points_added) + - cluster_info.sum_lambda_edges_added - - cluster_info.lambda_birth * cluster_info.size; + float stability = + (cluster_info.lambda_min_edge - cluster_info.lambda_birth) * + (cluster_info.size - cluster_info.num_points_added) + + cluster_info.sum_lambda_edges_added - + (cluster_info.lambda_birth * cluster_info.num_points_added); it->value.second.stability = stability; + if (it->value.second.stability >= it->value.second.stability_traversing_up) { local_possible_clusters_for_selection.push_back(it->value); @@ -1717,7 +1719,7 @@ std::vector traverse_down_hierarchy_and_select_clusters( * @param root_chain_cluster_edges_map YGM map of root chain cluster edge id -> * vector of edges added to this cluster. * @param point_to_cluster_id_map An empty YGM map of point id -> cluster label. - * @param edge_endpoints_map YGM map of edge id -> pair of point ids of its + * @param edge_endpoints_array YGM array of edge id -> pair of point ids of its * endpoints. * @param local_selected_clusters Vector of selected clusters to process on this * rank. @@ -1735,23 +1737,23 @@ void assign_points_cluster_ids( ygm::container::map &leaf_cluster_map, ygm::container::map &root_chain_cluster_map, ygm::container::map> - &root_chain_cluster_edges_map, - ygm::container::map &point_to_cluster_id_map, - ygm::container::map> &edge_endpoints_map, - const std::vector &local_selected_clusters, + &root_chain_cluster_edges_map, + ygm::container::map &point_to_cluster_id_map, + ygm::container::array> &edge_endpoints_array, + const std::vector &local_selected_clusters, const cluster_id_t start_cluster_id, const supernode_t root_chain_supernode, const supernode_t root_chain_second_child_name) { - ygm::comm &comm = chain_map.comm(); - auto edge_endpoints_map_ptr = edge_endpoints_map.get_ygm_ptr(); - auto point_to_cluster_id_map_ptr = point_to_cluster_id_map.get_ygm_ptr(); - auto leaf_cluster_map_ptr = leaf_cluster_map.get_ygm_ptr(); - auto chain_map_ptr = chain_map.get_ygm_ptr(); + ygm::comm &comm = chain_map.comm(); + auto edge_endpoints_array_ptr = edge_endpoints_array.get_ygm_ptr(); + auto point_to_cluster_id_map_ptr = point_to_cluster_id_map.get_ygm_ptr(); + auto leaf_cluster_map_ptr = leaf_cluster_map.get_ygm_ptr(); + auto chain_map_ptr = chain_map.get_ygm_ptr(); auto root_chain_cluster_edges_map_ptr = root_chain_cluster_edges_map.get_ygm_ptr(); // All points in the data set start out as noise points and when we // pick clusters, this will be over written with the cluster id - edge_endpoints_map.for_all( + edge_endpoints_array.for_all( [&point_to_cluster_id_map]([[maybe_unused]] const id_t &edge_id, const std::pair &edge_endpoints) { point_to_cluster_id_map.async_insert(edge_endpoints.first, @@ -1783,10 +1785,10 @@ void assign_points_cluster_ids( static auto label_leaf_cluster_edge_endpoints_lambda = [](const supernode_t &cluster_name, const full_leaf_cluster_info &cluster_info, - const cluster_id_t &cluster_id, auto edge_endpoints_map_ptr, + const cluster_id_t &cluster_id, auto edge_endpoints_array_ptr, auto point_to_cluster_id_map_ptr) { for (const edge_id_with_dist_t &edge : cluster_info.edges) { - edge_endpoints_map_ptr->async_visit( + edge_endpoints_array_ptr->async_visit( edge.first, label_edge_endpoints_lambda, cluster_id, point_to_cluster_id_map_ptr); } @@ -1796,10 +1798,10 @@ void assign_points_cluster_ids( // points with the provided cluster id static auto label_root_chain_cluster_edge_endpoints_lambda = [](const id_t &cluster_edge_id, std::vector &edges, - const cluster_id_t &cluster_id, auto edge_endpoints_map_ptr, + const cluster_id_t &cluster_id, auto edge_endpoints_array_ptr, auto point_to_cluster_id_map_ptr) { for (const edge_id_with_dist_t &edge : edges) { - edge_endpoints_map_ptr->async_visit( + edge_endpoints_array_ptr->async_visit( edge.first, label_edge_endpoints_lambda, cluster_id, point_to_cluster_id_map_ptr); } @@ -1820,8 +1822,8 @@ void assign_points_cluster_ids( const id_t &starting_cluster_edge_id, const cluster_id_t &cluster_id, ygm::ygm_ptr> leaf_cluster_map_ptr, - ygm::ygm_ptr>> - edge_endpoints_map_ptr, + ygm::ygm_ptr>> + edge_endpoints_array_ptr, ygm::ygm_ptr> point_to_cluster_id_map_ptr) { // For all clusters in the chain below or equal to the provided starting @@ -1838,7 +1840,7 @@ void assign_points_cluster_ids( // Label each edge in the cluster with the provided label for (const edge_id_with_dist_t &edge : cluster_info.edges) { - edge_endpoints_map_ptr->async_visit( + edge_endpoints_array_ptr->async_visit( edge.first, label_edge_endpoints_lambda, cluster_id, point_to_cluster_id_map_ptr); } @@ -1848,13 +1850,13 @@ void assign_points_cluster_ids( if (cluster_info.child.second == 1) { leaf_cluster_map_ptr->async_visit( cluster_info.child, label_leaf_cluster_edge_endpoints_lambda, - cluster_id, edge_endpoints_map_ptr, + cluster_id, edge_endpoints_array_ptr, point_to_cluster_id_map_ptr); } else { chain_map_ptr->async_visit( cluster_info.child, label_points_functor(), std::numeric_limits::max(), cluster_id, - leaf_cluster_map_ptr, edge_endpoints_map_ptr, + leaf_cluster_map_ptr, edge_endpoints_array_ptr, point_to_cluster_id_map_ptr); } } @@ -1864,12 +1866,12 @@ void assign_points_cluster_ids( if (chain.second.children[1].second == 1) { leaf_cluster_map_ptr->async_visit( chain.second.children[1], label_leaf_cluster_edge_endpoints_lambda, - cluster_id, edge_endpoints_map_ptr, point_to_cluster_id_map_ptr); + cluster_id, edge_endpoints_array_ptr, point_to_cluster_id_map_ptr); } else { chain_map_ptr->async_visit( chain.second.children[1], label_points_functor(), std::numeric_limits::max(), cluster_id, leaf_cluster_map_ptr, - edge_endpoints_map_ptr, point_to_cluster_id_map_ptr); + edge_endpoints_array_ptr, point_to_cluster_id_map_ptr); } } }; @@ -1887,7 +1889,7 @@ void assign_points_cluster_ids( if (cluster_name.first.second == 1) { leaf_cluster_map.async_visit( cluster_name.first, label_leaf_cluster_edge_endpoints_lambda, - cluster_id, edge_endpoints_map_ptr, point_to_cluster_id_map_ptr); + cluster_id, edge_endpoints_array_ptr, point_to_cluster_id_map_ptr); } // If the selected cluster is in the root chain (there will be at most 1), // then store its edge id and cluster id for processing separately @@ -1900,7 +1902,7 @@ void assign_points_cluster_ids( else { chain_map.async_visit(cluster_name.first, label_points_functor(), cluster_name.second, cluster_id, - leaf_cluster_map_ptr, edge_endpoints_map_ptr, + leaf_cluster_map_ptr, edge_endpoints_array_ptr, point_to_cluster_id_map_ptr); } } @@ -1916,7 +1918,7 @@ void assign_points_cluster_ids( auto label_root_chain_cluster_lambda = [selected_root_chain_cluster_edge_id, selected_root_chain_cluster_id, chain_map_ptr, leaf_cluster_map_ptr, root_chain_cluster_edges_map_ptr, - edge_endpoints_map_ptr, + edge_endpoints_array_ptr, point_to_cluster_id_map_ptr](const id_t &cluster_edge_id, root_chain_cluster_info &cluster_info) { // If this root chain cluster is the selected cluster or below it, @@ -1928,7 +1930,7 @@ void assign_points_cluster_ids( // Label this cluster's edges added root_chain_cluster_edges_map_ptr->async_visit( cluster_edge_id, label_root_chain_cluster_edge_endpoints_lambda, - selected_root_chain_cluster_id, edge_endpoints_map_ptr, + selected_root_chain_cluster_id, edge_endpoints_array_ptr, point_to_cluster_id_map_ptr); // Visit this cluster's non-root-chain child and label it @@ -1936,7 +1938,7 @@ void assign_points_cluster_ids( if (cluster_info.child.second == 1) { leaf_cluster_map_ptr->async_visit( cluster_info.child, label_leaf_cluster_edge_endpoints_lambda, - selected_root_chain_cluster_id, edge_endpoints_map_ptr, + selected_root_chain_cluster_id, edge_endpoints_array_ptr, point_to_cluster_id_map_ptr); } // If its a chain, go visit it with the label points functor @@ -1945,7 +1947,7 @@ void assign_points_cluster_ids( cluster_info.child, label_points_functor(), std::numeric_limits::max(), selected_root_chain_cluster_id, leaf_cluster_map_ptr, - edge_endpoints_map_ptr, point_to_cluster_id_map_ptr); + edge_endpoints_array_ptr, point_to_cluster_id_map_ptr); } } }; @@ -1956,13 +1958,13 @@ void assign_points_cluster_ids( leaf_cluster_map.async_visit(root_chain_second_child_name, label_leaf_cluster_edge_endpoints_lambda, selected_root_chain_cluster_id, - edge_endpoints_map_ptr, + edge_endpoints_array_ptr, point_to_cluster_id_map_ptr); } else { chain_map.async_visit( root_chain_second_child_name, label_points_functor(), std::numeric_limits::max(), selected_root_chain_cluster_id, - leaf_cluster_map_ptr, edge_endpoints_map_ptr, + leaf_cluster_map_ptr, edge_endpoints_array_ptr, point_to_cluster_id_map_ptr); } } diff --git a/src/clustering/distributed_clustering_mst_contraction.hpp b/src/clustering/distributed_clustering_mst_contraction.hpp index a56fc44..e20da98 100644 --- a/src/clustering/distributed_clustering_mst_contraction.hpp +++ b/src/clustering/distributed_clustering_mst_contraction.hpp @@ -26,16 +26,18 @@ namespace clams::clustering { * @param mst_edge_vector Vector of MST edges (distance, (endpoint1, * endpoint2)). Each rank has its own vector of edges and edges should not be * duplicated between ranks. - * @param edge_endpoints_map An empty YGM map of edge id -> (edge_endpoint1, - * edge_endpoint2) - * @param edge_contraction_map An empty YGM map of edge id -> - * edge_contraction_info + * @param edge_endpoints_array A YGM array of edge id -> (edge_endpoint1, + * edge_endpoint2). This array is not filled with values yet, but needs to have + * size equal to the number of edges when intialized. + * @param edge_contraction_array An YGM array of edge id -> + * edge_contraction_info. This array is not filled with values yet, but needs to + * have size equal to the number of edges when intialized. */ void sort_and_process_mst_edges_into_maps( std::vector>> &mst_edge_vector, - ygm::container::map> &edge_endpoints_map, - ygm::container::map &edge_contraction_map) { - ygm::comm &comm = edge_endpoints_map.comm(); + ygm::container::array> &edge_endpoints_array, + ygm::container::array &edge_contraction_array) { + ygm::comm &comm = edge_endpoints_array.comm(); // Create a YGM array with the MST edges ygm::container::array>> @@ -47,35 +49,35 @@ void sort_and_process_mst_edges_into_maps( comm.barrier(); // Fill in the edge endpoints and contraction map - auto visit_array_lambda = - [&edge_endpoints_map, &edge_contraction_map]( + auto visit_sorted_array_lambda = + [&edge_endpoints_array, &edge_contraction_array]( const std::size_t &index, const std::pair> &value) { // Add this edge to the edge map with its sorted position as its id distance_t distance = value.first; std::pair edge_endpoints = value.second; - edge_endpoints_map.async_insert(index, edge_endpoints); - edge_contraction_map.async_insert( + edge_endpoints_array.local_insert(index, edge_endpoints); + edge_contraction_array.local_insert( index, edge_contraction_info{.endpoint_supernode_reps = edge_endpoints, .distance = distance}); }; - edge_array.for_all(visit_array_lambda); + edge_array.for_all(visit_sorted_array_lambda); comm.barrier(); } /** - * @brief Fill the initial min_incident_edge_map from the edge_endpoints_map + * @brief Fill the initial min_incident_edge_map from the edge_endpoints_array * - * @param edge_endpoints_map YGM map of edge id -> (edge_endpoint1, + * @param edge_endpoints_array YGM array of edge id -> (edge_endpoint1, * edge_endpoint2) * @param min_incident_edge_map An empty YGM map of supernode representative -> * min incident edge id. Initially, each supernode representative is an * original node in the graph. */ void fill_init_min_incident_edge_map( - ygm::container::map &min_incident_edge_map, - ygm::container::map> &edge_endpoints_map) { + ygm::container::map &min_incident_edge_map, + ygm::container::array> &edge_endpoints_array) { ygm::comm &comm = min_incident_edge_map.comm(); static auto check_incident_edge_lambda = @@ -95,7 +97,7 @@ void fill_init_min_incident_edge_map( min_incident_edge_map.async_insert(edge_endpoints.second, std::numeric_limits::max()); }; - edge_endpoints_map.for_all(fill_init_min_incident_edge_map_supernode_reps); + edge_endpoints_array.for_all(fill_init_min_incident_edge_map_supernode_reps); comm.barrier(); // Fill the initial incidence map from the original MST edges @@ -107,7 +109,7 @@ void fill_init_min_incident_edge_map( min_incident_edge_map.async_visit(edge_endpoints.second, check_incident_edge_lambda, edge_id); }; - edge_endpoints_map.for_all(fill_init_min_incident_edge_map); + edge_endpoints_array.for_all(fill_init_min_incident_edge_map); comm.barrier(); } @@ -117,8 +119,7 @@ void fill_init_min_incident_edge_map( * @param _round The round of MST contraction we're on. * @param min_incident_edge_map YGM map of supernode representative -> * min incident edge id. - * @param edge_contraction_map YGM map of edge id -> - * edge_contraction_info + * @param edge_contraction_array YGM array of edge id -> edge_contraction_info * @param alpha_edge_map_ptr Pointer to YGM map of alpha edge id -> * alpha_edge_info. Alpha edges are ones that are not contracted in the first * round. As we contract alpha edges, we update their child supernodes in the @@ -128,10 +129,10 @@ void fill_init_min_incident_edge_map( */ void contract_edges( uint32_t _round, ygm::container::map &min_incident_edge_map, - ygm::container::map &edge_contraction_map, + ygm::container::array &edge_contraction_array, ygm::ygm_ptr> alpha_edge_map_ptr, ygm::container::disjoint_set &tree_components_djset) { - ygm::comm &comm = edge_contraction_map.comm(); + ygm::comm &comm = edge_contraction_array.comm(); auto tree_components_djset_ptr = tree_components_djset.get_ygm_ptr(); static uint32_t round; @@ -147,7 +148,7 @@ void contract_edges( } }; - auto contract_edges_lambda = [&edge_contraction_map, alpha_edge_map_ptr, + auto contract_edges_lambda = [&edge_contraction_array, alpha_edge_map_ptr, tree_components_djset_ptr]( [[maybe_unused]] const id_t &supernode_rep, const id_t min_incident_edge_id) { @@ -164,9 +165,9 @@ void contract_edges( edge_info.endpoint_supernode_reps.first, edge_info.endpoint_supernode_reps.second); }; - edge_contraction_map.async_visit(min_incident_edge_id, - add_edge_to_djset_lambda, - tree_components_djset_ptr); + edge_contraction_array.async_visit(min_incident_edge_id, + add_edge_to_djset_lambda, + tree_components_djset_ptr); // Set this supernode as the dendrogram child of the min incident edge if (round >= 2) { @@ -188,20 +189,19 @@ void contract_edges( * directly visits the YGM min_incident_edge_map. * * @param round The round of MST contraction we're on. - * @param edge_contraction_map YGM map of edge id -> - * edge_contraction_info + * @param edge_contraction_array YGM array of edge id -> edge_contraction_info * @param min_incident_edge_map YGM map of supernode representative -> * min incident edge id. */ void update_edge_chain_parent_edge_id( - uint32_t round, - ygm::container::map &edge_contraction_map, - ygm::container::map &min_incident_edge_map) { - ygm::comm &comm = edge_contraction_map.comm(); - auto edge_contraction_map_ptr = edge_contraction_map.get_ygm_ptr(); + uint32_t round, + ygm::container::array &edge_contraction_array, + ygm::container::map &min_incident_edge_map) { + ygm::comm &comm = edge_contraction_array.comm(); + auto edge_contraction_array_ptr = edge_contraction_array.get_ygm_ptr(); auto update_chain_parents_lambda = - [&min_incident_edge_map, round, edge_contraction_map_ptr]( + [&min_incident_edge_map, round, edge_contraction_array_ptr]( const id_t &edge_id, edge_contraction_info &edge_info) { // Only see if we can update an edge's chain parent edge id if not // assigned yet (if its 0) and if the edge is already contracted @@ -212,11 +212,11 @@ void update_edge_chain_parent_edge_id( edge_info.chain_supernode.first, []([[maybe_unused]] const id_t &supernode_rep, const id_t &min_incident_edge_id, const id_t &edge_id, - auto edge_contraction_map_ptr) { + auto edge_contraction_array_ptr) { // If the supernode's min incident edge id is larger than edge // id, we found the chain's parent edge id in the dendrogram if (min_incident_edge_id > edge_id) { - edge_contraction_map_ptr->async_visit( + edge_contraction_array_ptr->async_visit( edge_id, []([[maybe_unused]] const id_t &edge_id, edge_contraction_info &edge_info, @@ -226,10 +226,10 @@ void update_edge_chain_parent_edge_id( min_incident_edge_id); } }, - edge_id, edge_contraction_map_ptr); + edge_id, edge_contraction_array_ptr); } }; - edge_contraction_map.for_all(update_chain_parents_lambda); + edge_contraction_array.for_all(update_chain_parents_lambda); comm.barrier(); return; @@ -244,16 +244,15 @@ void update_edge_chain_parent_edge_id( * update_edge_chain_parent_edge_id when the min_incident_edge_map is small. * * @param round The round of MST contraction we're on. - * @param edge_contraction_map YGM map of edge id -> - * edge_contraction_info + * @param edge_contraction_array YGM array of edge id -> edge_contraction_info * @param min_incident_edge_map YGM map of supernode representative -> * min incident edge id. */ void update_edge_chain_parent_edge_id_from_local_map( - uint32_t round, - ygm::container::map &edge_contraction_map, - ygm::container::map &min_incident_edge_map) { - ygm::comm &comm = edge_contraction_map.comm(); + uint32_t round, + ygm::container::array &edge_contraction_array, + ygm::container::map &min_incident_edge_map) { + ygm::comm &comm = edge_contraction_array.comm(); // Give all ranks a full copy of min_incident_edge_map std::map local_min_incident_edge_map; @@ -290,36 +289,35 @@ void update_edge_chain_parent_edge_id_from_local_map( } } }; - edge_contraction_map.for_all(update_chain_parents_lambda); + edge_contraction_array.for_all(update_chain_parents_lambda); comm.barrier(); return; } /** - * @brief Updates edge_contraction_map by udpating the supernode endpoints of + * @brief Updates edge_contraction_array by updating the supernode endpoints of * uncontracted edges and updates the current chain supernode of contracted * edges if the chain parent edge id is not yet found. This version directly * visits the tree_conmponents_djset. * * @param _round The round of MST contraction we're on. - * @param edge_contraction_map YGM map of edge id -> - * edge_contraction_info + * @param edge_contraction_array YGM array of edge id -> edge_contraction_info * @param tree_components_djset YGM disjoint_set of connected components of the * contracted tree. We use this to get the new supernode reps. */ void update_edge_endpoints_and_chain_supernode( - uint32_t _round, - ygm::container::map &edge_contraction_map, - ygm::container::disjoint_set &tree_components_djset) { + uint32_t _round, + ygm::container::array &edge_contraction_array, + ygm::container::disjoint_set &tree_components_djset) { static uint32_t round; round = _round; - ygm::comm &comm = edge_contraction_map.comm(); - auto edge_contraction_map_ptr = edge_contraction_map.get_ygm_ptr(); + ygm::comm &comm = edge_contraction_array.comm(); + auto edge_contraction_array_ptr = edge_contraction_array.get_ygm_ptr(); auto update_edge_endpoints_and_chain_supernode_lambda = - [&tree_components_djset, edge_contraction_map_ptr]( + [&tree_components_djset, edge_contraction_array_ptr]( const id_t &edge_id, edge_contraction_info &edge_info) { // If the edge is not contracted yet, update its supernode endpoints if (edge_info.contraction_round == 0) { @@ -327,8 +325,8 @@ void update_edge_endpoints_and_chain_supernode( tree_components_djset.async_visit( edge_info.endpoint_supernode_reps.first, []([[maybe_unused]] const auto &djset_item_data_pair, - const id_t &edge_id, auto edge_contraction_map_ptr) { - edge_contraction_map_ptr->async_visit( + const id_t &edge_id, auto edge_contraction_array_ptr) { + edge_contraction_array_ptr->async_visit( edge_id, []([[maybe_unused]] const id_t &edge_id, edge_contraction_info &edge_info, @@ -338,14 +336,14 @@ void update_edge_endpoints_and_chain_supernode( }, djset_item_data_pair.second.get_parent()); }, - edge_id, edge_contraction_map_ptr); + edge_id, edge_contraction_array_ptr); // Update the second supernode endpoint of this edge tree_components_djset.async_visit( edge_info.endpoint_supernode_reps.second, []([[maybe_unused]] const auto &djset_item_data_pair, - const id_t &edge_id, auto edge_contraction_map_ptr) { - edge_contraction_map_ptr->async_visit( + const id_t &edge_id, auto edge_contraction_array_ptr) { + edge_contraction_array_ptr->async_visit( edge_id, []([[maybe_unused]] const id_t &edge_id, edge_contraction_info &edge_info, @@ -355,7 +353,7 @@ void update_edge_endpoints_and_chain_supernode( }, djset_item_data_pair.second.get_parent()); }, - edge_id, edge_contraction_map_ptr); + edge_id, edge_contraction_array_ptr); } // If an edge is already contracted, see if it already has a known chain @@ -378,8 +376,8 @@ void update_edge_endpoints_and_chain_supernode( tree_components_djset.async_visit( supernode_rep_to_visit, []([[maybe_unused]] const auto &djset_item_data_pair, - const id_t edge_id, auto edge_contraction_map_ptr) { - edge_contraction_map_ptr->async_visit( + const id_t edge_id, auto edge_contraction_array_ptr) { + edge_contraction_array_ptr->async_visit( edge_id, []([[maybe_unused]] const id_t &edge_id, edge_contraction_info &edge_info, @@ -389,11 +387,11 @@ void update_edge_endpoints_and_chain_supernode( std::make_pair(djset_item_data_pair.second.get_parent(), round)); }, - edge_id, edge_contraction_map_ptr); + edge_id, edge_contraction_array_ptr); } } }; - edge_contraction_map.for_all( + edge_contraction_array.for_all( update_edge_endpoints_and_chain_supernode_lambda); comm.barrier(); @@ -401,7 +399,7 @@ void update_edge_endpoints_and_chain_supernode( } /** - * @brief Updates edge_contraction_map by udpating the supernode endpoints of + * @brief Updates edge_contraction_array by updating the supernode endpoints of * uncontracted edges and updates the current chain supernode of contracted * edges if the chain parent edge id is not yet found. Makes a * local copy of the tree_components_djset representatives on each rank to avoid @@ -410,16 +408,15 @@ void update_edge_endpoints_and_chain_supernode( * is small. * * @param _round The round of MST contraction we're on. - * @param edge_contraction_map YGM map of edge id -> - * edge_contraction_info + * @param edge_contraction_array YGM array of edge id -> edge_contraction_info * @param tree_components_djset YGM disjoint_set of connected components of the * contracted tree. We use this to get the new supernode reps. */ void update_edge_endpoints_and_chain_supernode_from_local_map( - uint32_t round, - ygm::container::map &edge_contraction_map, - ygm::container::disjoint_set &tree_components_djset) { - ygm::comm &comm = edge_contraction_map.comm(); + uint32_t round, + ygm::container::array &edge_contraction_array, + ygm::container::disjoint_set &tree_components_djset) { + ygm::comm &comm = edge_contraction_array.comm(); static std::map old_to_new_supernode_rep_map; static std::map local_old_to_new_supernode_rep_map; @@ -478,7 +475,7 @@ void update_edge_endpoints_and_chain_supernode_from_local_map( } } }; - edge_contraction_map.for_all( + edge_contraction_array.for_all( update_edge_endpoints_and_chain_supernode_lambda); comm.barrier(); @@ -519,16 +516,16 @@ void reinit_min_incident_edge_map( * min_incident_edge_map already has the correct supernode reps as keys, e.g., * by running reinit_min_incident_edge_map first. * - * @param edge_contraction_map YGM map of edge id -> edge contraction info. + * @param edge_contraction_array YGM map of edge id -> edge contraction info. * Assumes that edges already have their supernode endpoints updated. * @param min_incident_edge_map YGM map of supernode representative -> * min incident edge id. Assumes that the map already has the correct supernode * reps as keys. */ void update_min_incident_edge_map( - ygm::container::map &edge_contraction_map, - ygm::container::map &min_incident_edge_map) { - ygm::comm &comm = edge_contraction_map.comm(); + ygm::container::array &edge_contraction_array, + ygm::container::map &min_incident_edge_map) { + ygm::comm &comm = edge_contraction_array.comm(); static auto check_incident_edge_lambda = []([[maybe_unused]] const id_t &supernode_rep, id_t &min_incident_edge_id, @@ -551,7 +548,7 @@ void update_min_incident_edge_map( check_incident_edge_lambda, edge_id); } }; - edge_contraction_map.for_all(update_min_incident_edge_map_lambda); + edge_contraction_array.for_all(update_min_incident_edge_map_lambda); comm.barrier(); } diff --git a/src/clustering/distributed_clustering_types.hpp b/src/clustering/distributed_clustering_types.hpp index 7673f43..f904774 100644 --- a/src/clustering/distributed_clustering_types.hpp +++ b/src/clustering/distributed_clustering_types.hpp @@ -453,12 +453,16 @@ struct root_chain_cluster_info { // overload operator<< for root-chain cluster info std::ostream &operator<<(std::ostream &os, const root_chain_cluster_info &cluster) { - os << "stability = " << cluster.stability - << ", stability traversing up = " << cluster.stability_traversing_up + os << "stability = " << std::format("{:.10g}", cluster.stability) + << ", stability traversing up = " + << std::format("{:.9g}", cluster.stability_traversing_up) << ", size = " << cluster.size << ", selected = " << cluster.selected << ", num_points_added = " << cluster.num_points_added - << ", sum_lambda_edges_added = " << cluster.sum_lambda_edges_added - << ", lambda birth = " << cluster.lambda_birth + << ", sum_lambda_edges_added = " + << std::format("{:.10g}", cluster.sum_lambda_edges_added) + << ", lambda min edge = " + << std::format("{:.10g}", cluster.lambda_min_edge) + << ", lambda birth = " << std::format("{:.10g}", cluster.lambda_birth) << ", parent edge id = " << cluster.parent_edge_id << ", non-chain child: " << cluster.child << " with size " << cluster.child_size; diff --git a/src/clustering/run_distributed_hdbscan_clustering.cpp b/src/clustering/run_distributed_hdbscan_clustering.cpp index a5e6f4b..eb4b95a 100644 --- a/src/clustering/run_distributed_hdbscan_clustering.cpp +++ b/src/clustering/run_distributed_hdbscan_clustering.cpp @@ -110,6 +110,11 @@ std::pair> parse_option(int argc, char *argv[], } } + // Metall MST currently doesn't support already assigned edge ids + if (opt.metall_mst) { + opt.read_mst_with_edge_ids = false; + } + return std::make_pair(show_help, unknown_opts); } @@ -171,14 +176,17 @@ int main(int argc, char *argv[]) { float select_clusters_time = 0.0; float label_points_time = 0.0; - // Initialize edge map of basic endpoints - // and edge contraction map with edge_contraction_info - ygm::container::map> edge_endpoints_map(world); - ygm::container::map edge_contraction_map(world); - - // Read MST edges, sort and assign edge ids + // Read MST edges sw_round.reset(); sw_step.reset(); + + // Vector to store mst edges read by this rank + // Edges stored in the form (distance, endpoint pair (node 1, node2)) + static std::vector>> + mst_edge_vector; + static id_t local_num_edges_read; + local_num_edges_read = 0; + if (opt.metall_mst) { if (world.rank() == 0) { spdlog::info("Attaching MST in Metall datastore"); @@ -203,8 +211,6 @@ int main(int argc, char *argv[]) { uint64_t end_idx = std::min(chunk_size * (world.rank() + 1) - 1, num_edges - 1); - static std::vector>> - mst_edge_vector; mst_edge_vector.clear(); for (uint64_t i = start_idx; i <= end_idx; ++i) { std::pair> array_item = @@ -213,19 +219,11 @@ int main(int argc, char *argv[]) { (*input_mst_edges)[i].ids[1])); mst_edge_vector.push_back(array_item); } + local_num_edges_read = mst_edge_vector.size(); if ((world.rank() == 0) & (opt.verbose)) { spdlog::info(" Time to read edges (s): {:.3f}", sw_step); } - sw_step.reset(); - - sort_and_process_mst_edges_into_maps(mst_edge_vector, edge_endpoints_map, - edge_contraction_map); - - if ((world.rank() == 0) & (opt.verbose)) { - spdlog::info(" Time to sort edges and fill YGM maps (s): {:.3f}", - sw_step); - } } else { if (opt.read_mst_with_edge_ids) { @@ -238,8 +236,7 @@ int main(int argc, char *argv[]) { std::vector input_paths{opt.mst_edges_path.c_str()}; ygm::io::line_parser mst_line_parser(world, input_paths); - auto line_parser_lambda = [&edge_endpoints_map, &edge_contraction_map]( - const std::string &line) { + auto line_parser_lambda = [](const std::string &line) { if (std::isdigit(line[0])) { try { std::stringstream ss(line); @@ -247,18 +244,15 @@ int main(int argc, char *argv[]) { id_t node1, node2; distance_t dist; ss >> edge_id >> node1 >> node2 >> dist; - std::pair edge_endpoints = std::make_pair(node1, node2); - edge_endpoints_map.async_insert(edge_id, edge_endpoints); - edge_contraction_map.async_insert( - edge_id, - edge_contraction_info{.endpoint_supernode_reps = edge_endpoints, - .distance = dist}); + ++local_num_edges_read; } catch (...) { std::cout << "Error reading mst line: " << line << std::endl; } } else { - std::cout << "Read comment line in mst file: " << line << std::endl; + if (line[0] != '#') { + std::cout << "Read comment line in mst file: " << line << std::endl; + } } }; mst_line_parser.for_all(line_parser_lambda); @@ -273,10 +267,6 @@ int main(int argc, char *argv[]) { std::vector input_paths{opt.mst_edges_path.c_str()}; ygm::io::line_parser mst_line_parser(world, input_paths); - // Vector to store mst edges read by this rank - // Edges stored in the form (distance, endpoint pair (node 1, node2)) - static std::vector>> - mst_edge_vector; mst_edge_vector.clear(); auto line_parser_lambda = [](const std::string &line) { @@ -299,28 +289,83 @@ int main(int argc, char *argv[]) { mst_line_parser.for_all(line_parser_lambda); world.barrier(); + local_num_edges_read = mst_edge_vector.size(); + if ((world.rank() == 0) & (opt.verbose)) { spdlog::info(" Time to read edges (s): {:.3f}", sw_step); } - sw_step.reset(); + } + } - sort_and_process_mst_edges_into_maps(mst_edge_vector, edge_endpoints_map, - edge_contraction_map); + // Sort and ingest MST edges + sw_step.reset(); - if ((world.rank() == 0) & (opt.verbose)) { - spdlog::info(" Time to sort edges and fill YGM maps (s): {:.3f}", - sw_step); - } + id_t total_num_edges_read = ygm::sum(local_num_edges_read, world); + + // Initialize edge map of basic endpoints + // and edge contraction map with edge_contraction_info + ygm::container::array> edge_endpoints_array( + world, total_num_edges_read); + ygm::container::array edge_contraction_map( + world, total_num_edges_read); + + { + // Re-read the input file if edge ids are already assigned and add edges to + // the YGM arrays + if (opt.read_mst_with_edge_ids) { + std::vector input_paths{opt.mst_edges_path.c_str()}; + ygm::io::line_parser mst_line_parser(world, input_paths); + + auto line_parser_lambda = [&edge_endpoints_array, &edge_contraction_map]( + const std::string &line) { + if (std::isdigit(line[0])) { + try { + std::stringstream ss(line); + id_t edge_id; + id_t node1, node2; + distance_t dist; + ss >> edge_id >> node1 >> node2 >> dist; + ++local_num_edges_read; + std::pair edge_endpoints = std::make_pair(node1, node2); + edge_endpoints_array.async_insert(edge_id, edge_endpoints); + edge_contraction_map.async_insert( + edge_id, + edge_contraction_info{.endpoint_supernode_reps = edge_endpoints, + .distance = dist}); + + } catch (...) { + std::cout << "Error reading mst line: " << line << std::endl; + } + } else { + if (line[0] != '#') { + std::cout << "Read comment line in mst file: " << line << std::endl; + } + } + }; + mst_line_parser.for_all(line_parser_lambda); + world.barrier(); + + } + // Sort the MST edges and add them to our edge arrays + else { + sort_and_process_mst_edges_into_maps( + mst_edge_vector, edge_endpoints_array, edge_contraction_map); + } + + if ((world.rank() == 0) & (opt.verbose)) { + spdlog::info(" Time to sort edges and fill YGM maps (s): {:.3f}", + sw_step); } } + std::stringstream ss; - ss << " Number of MST edges: " << edge_endpoints_map.size(); + ss << " Number of MST edges: " << edge_endpoints_array.size(); if (world.rank() == 0) { spdlog::info(ss.str()); spdlog::info(" Time to ingest MST edges (s): {:.3f}", sw_round); } - if (edge_endpoints_map.size() == 0) { + if (edge_endpoints_array.size() == 0) { if (world.rank() == 0) { spdlog::info("Error: 0 MST edges read"); show_help(); @@ -357,13 +402,14 @@ int main(int argc, char *argv[]) { // Initial set up sw_step.reset(); - fill_init_min_incident_edge_map(min_incident_edge_map, edge_endpoints_map); + fill_init_min_incident_edge_map(min_incident_edge_map, + edge_endpoints_array); incidence_map_time += sw_step.elapsed().count(); // Max possible number of contraction rounds is log2(number of edges in // MST) uint32_t max_possible_rounds = - static_cast(std::ceil(std::log2(edge_endpoints_map.size()))); + static_cast(std::ceil(std::log2(edge_endpoints_array.size()))); // Rounds of successive MST contraction static uint32_t round; @@ -1140,7 +1186,7 @@ int main(int argc, char *argv[]) { assign_points_cluster_ids( chain_map, leaf_cluster_map, root_chain_cluster_map, root_chain_cluster_edges_map, point_to_cluster_id_map, - edge_endpoints_map, local_selected_clusters, start_cluster_id, + edge_endpoints_array, local_selected_clusters, start_cluster_id, root_chain_supernode, root_chain_second_child.name); label_points_time += sw_step.elapsed().count(); From c314c017dc6abe5e8501336c35767b79d190eb67 Mon Sep 17 00:00:00 2001 From: Grace Li Date: Fri, 25 Sep 2026 18:36:55 -0700 Subject: [PATCH 3/4] Fix name for edge_contraction_array --- .../run_distributed_hdbscan_clustering.cpp | 30 ++++++++++--------- 1 file changed, 16 insertions(+), 14 deletions(-) diff --git a/src/clustering/run_distributed_hdbscan_clustering.cpp b/src/clustering/run_distributed_hdbscan_clustering.cpp index eb4b95a..e407c6f 100644 --- a/src/clustering/run_distributed_hdbscan_clustering.cpp +++ b/src/clustering/run_distributed_hdbscan_clustering.cpp @@ -306,7 +306,7 @@ int main(int argc, char *argv[]) { // and edge contraction map with edge_contraction_info ygm::container::array> edge_endpoints_array( world, total_num_edges_read); - ygm::container::array edge_contraction_map( + ygm::container::array edge_contraction_array( world, total_num_edges_read); { @@ -316,7 +316,8 @@ int main(int argc, char *argv[]) { std::vector input_paths{opt.mst_edges_path.c_str()}; ygm::io::line_parser mst_line_parser(world, input_paths); - auto line_parser_lambda = [&edge_endpoints_array, &edge_contraction_map]( + auto line_parser_lambda = [&edge_endpoints_array, + &edge_contraction_array]( const std::string &line) { if (std::isdigit(line[0])) { try { @@ -328,7 +329,7 @@ int main(int argc, char *argv[]) { ++local_num_edges_read; std::pair edge_endpoints = std::make_pair(node1, node2); edge_endpoints_array.async_insert(edge_id, edge_endpoints); - edge_contraction_map.async_insert( + edge_contraction_array.async_insert( edge_id, edge_contraction_info{.endpoint_supernode_reps = edge_endpoints, .distance = dist}); @@ -349,7 +350,7 @@ int main(int argc, char *argv[]) { // Sort the MST edges and add them to our edge arrays else { sort_and_process_mst_edges_into_maps( - mst_edge_vector, edge_endpoints_array, edge_contraction_map); + mst_edge_vector, edge_endpoints_array, edge_contraction_array); } if ((world.rank() == 0) & (opt.verbose)) { @@ -422,12 +423,12 @@ int main(int argc, char *argv[]) { // Find all edges to contract and add to disjoint set sw_step.reset(); - contract_edges(round, min_incident_edge_map, edge_contraction_map, + contract_edges(round, min_incident_edge_map, edge_contraction_array, alpha_edge_map_ptr, tree_components_djset); // Count how many edges were contracted this round id_t num_edges_contracted = 0; - edge_contraction_map.for_all( + edge_contraction_array.for_all( [&num_edges_contracted]([[maybe_unused]] const id_t &edge_id, const edge_contraction_info edge_info) { if (edge_info.contraction_round == round) { @@ -445,9 +446,9 @@ int main(int argc, char *argv[]) { if (min_incident_edge_map.size() <= LOCAL_COPY_NUM_SUPERNODES_THRESHOLD) { update_edge_chain_parent_edge_id_from_local_map( - round, edge_contraction_map, min_incident_edge_map); + round, edge_contraction_array, min_incident_edge_map); } else { - update_edge_chain_parent_edge_id(round, edge_contraction_map, + update_edge_chain_parent_edge_id(round, edge_contraction_array, min_incident_edge_map); } } @@ -507,12 +508,12 @@ int main(int argc, char *argv[]) { if (tree_components_djset.num_sets() <= LOCAL_COPY_NUM_SUPERNODES_THRESHOLD) { update_edge_endpoints_and_chain_supernode_from_local_map( - round, edge_contraction_map, tree_components_djset); + round, edge_contraction_array, tree_components_djset); } // Otherwise, avoid creating local copies and look up new edge info in // the supernode_map else { - update_edge_endpoints_and_chain_supernode(round, edge_contraction_map, + update_edge_endpoints_and_chain_supernode(round, edge_contraction_array, tree_components_djset); } reinit_min_incident_edge_map(min_incident_edge_map, @@ -524,7 +525,8 @@ int main(int argc, char *argv[]) { sw_step.reset(); // Update the incidence map - update_min_incident_edge_map(edge_contraction_map, min_incident_edge_map); + update_min_incident_edge_map(edge_contraction_array, + min_incident_edge_map); incidence_map_time += sw_step.elapsed().count(); ++round; @@ -604,7 +606,7 @@ int main(int argc, char *argv[]) { sw_step.reset(); // Fill in the missing alpha edge info - fill_alpha_edge_map(alpha_edge_map, edge_contraction_map, + fill_alpha_edge_map(alpha_edge_map, edge_contraction_array, root_chain_supernode); // Assign edges to chains @@ -614,12 +616,12 @@ int main(int argc, char *argv[]) { } assign_edges_to_chains( - edge_contraction_map, chain_map, local_root_chain_alpha_edges, + edge_contraction_array, chain_map, local_root_chain_alpha_edges, root_chain_non_alpha_edge_set, leaf_cluster_map, final_round); assign_edges_to_chains_time += sw_step.elapsed().count(); - edge_contraction_map.clear(); + edge_contraction_array.clear(); world.barrier(); if (opt.verbose) { From 6ed9207a75d84b693b9bcd4e09ca40594b872edc Mon Sep 17 00:00:00 2001 From: Grace Li Date: Fri, 25 Sep 2026 19:15:31 -0700 Subject: [PATCH 4/4] Remove prescision printouts --- src/clustering/distributed_clustering_types.hpp | 13 +++++-------- 1 file changed, 5 insertions(+), 8 deletions(-) diff --git a/src/clustering/distributed_clustering_types.hpp b/src/clustering/distributed_clustering_types.hpp index f904774..85cb1e1 100644 --- a/src/clustering/distributed_clustering_types.hpp +++ b/src/clustering/distributed_clustering_types.hpp @@ -453,16 +453,13 @@ struct root_chain_cluster_info { // overload operator<< for root-chain cluster info std::ostream &operator<<(std::ostream &os, const root_chain_cluster_info &cluster) { - os << "stability = " << std::format("{:.10g}", cluster.stability) - << ", stability traversing up = " - << std::format("{:.9g}", cluster.stability_traversing_up) + os << "stability = " << cluster.stability + << ", stability traversing up = " << cluster.stability_traversing_up << ", size = " << cluster.size << ", selected = " << cluster.selected << ", num_points_added = " << cluster.num_points_added - << ", sum_lambda_edges_added = " - << std::format("{:.10g}", cluster.sum_lambda_edges_added) - << ", lambda min edge = " - << std::format("{:.10g}", cluster.lambda_min_edge) - << ", lambda birth = " << std::format("{:.10g}", cluster.lambda_birth) + << ", sum_lambda_edges_added = " << cluster.sum_lambda_edges_added + << ", lambda min edge = " << cluster.lambda_min_edge + << ", lambda birth = " << cluster.lambda_birth << ", parent edge id = " << cluster.parent_edge_id << ", non-chain child: " << cluster.child << " with size " << cluster.child_size;