feat: MIS

README.md

@@ -11,4 +11,5 @@ An experimental single node out-of core graph algorithms.
 1. Low-level `Pregel API`: *Malewicz, Grzegorz, et al. "Pregel: a system for large-scale graph processing." Proceedings of the 2010 ACM SIGMOD International Conference on Management of data. 2010.*
    - Pregel-based PageRank: *Zadeh, R., et al. "Cme 323: Distributed algorithms and optimization, spring 2015." University Lecture (2015).*
    - Pregel-based Multi Source Shortest Paths: *Malewicz, Grzegorz, et al. "Pregel: a system for large-scale graph processing." Proceedings of the 2010 ACM SIGMOD International Conference on Management of data. 2010.*
-3. Weakly Connected Components: *Bögeholz, Harald, Michael Brand, and Radu-Alexandru Todor. "In-database connected component analysis." 2020 IEEE 36th International Conference on Data Engineering (ICDE). IEEE, 2020.*
+2. Weakly Connected Components: *Bögeholz, Harald, Michael Brand, and Radu-Alexandru Todor. "In-database connected component analysis." 2020 IEEE 36th International Conference on Data Engineering (ICDE). IEEE, 2020.*
+3. Maximal Independent Set: *Ghaffari, Mohsen. "An improved distributed algorithm for maximal independent set." Proceedings of the twenty-seventh annual ACM-SIAM symposium on Discrete algorithms. Society for Industrial and Applied Mathematics, 2016.*

src/algorithm.rs

@@ -1,3 +1,4 @@
 mod centrality;
 mod connectivity;
 mod pregel;
+mod subgraph;

src/algorithm/connectivity/connected_components.rs

@@ -14,7 +14,7 @@
 
 use crate::expressions::{axpb, finite_axpb};
 use crate::memory::{CheckpointConfig, ParquetCheckpointer};
-use crate::utils::{GraphFramesConfig, scoped_ctx};
+use crate::utils::{GraphFramesConfig, scoped_ctx, symmetrize};
 use crate::{EDGE_DST, EDGE_SRC, GraphFrame, VERTEX_ID};
 use datafusion::dataframe::DataFrameWriteOptions;
 use datafusion::error::Result;
@@ -39,18 +39,6 @@ const CC_FR_REP: &str = "__cc_fr_rep";
 const CC_COMP_KEY: &str = "__cc_comp_key";
 const CC_NEW_COMPONENT: &str = "__cc_new_component";
 
-/// Prepares the edge set for contraction: drops self-loops, symmetrizes
-/// (adds the reverse of every edge), and deduplicates. The result is the
-/// undirected simple graph the algorithm operates on.
-fn prepare_edges(edges: &DataFrame) -> Result<DataFrame> {
-    let no_loops = edges.clone().filter(col(EDGE_SRC).not_eq(col(EDGE_DST)))?;
-    let reversed = no_loops.clone().select(vec![
-        col(EDGE_DST).alias(EDGE_SRC),
-        col(EDGE_SRC).alias(EDGE_DST),
-    ])?;
-    no_loops.union(reversed)?.distinct()
-}
-
 /// Computes the per-iteration component representatives for every source
 /// vertex under the affine hash `(rA, rB)`:
 ///   `rep(v) = least(axpb(rA, v, rB), min over neighbours of axpb(rA, u, rB))`.
@@ -226,7 +214,14 @@ impl<'a> ConnectedComponentsBuilder<'a> {
         // ---- prepare: drop self-loops, symmetrize, deduplicate ----
         let run_id = Uuid::new_v4().to_string();
         log::info!("start WCC with run-id {run_id}");
-        let prepared_edges = prepare_edges(&original_edges)?;
+        // it may look conter-intuitive
+        // but IRL distinct is more expensive here.
+        //
+        // With distinct: full-scan over huge edges, slightly less on the first iter;
+        // Without distinct: fast flow here, slightly slower on the first iter.
+        //
+        // TODO: benchmark it and see which one is better IRL, not in theory
+        let prepared_edges = symmetrize(&original_edges, false)?;
 
         let ckpt_base = self.checkpoint_config.dir.clone().join(run_id.clone());
         let store_url = self.checkpoint_config.store_url.clone();

src/algorithm/subgraph.rs

@@ -0,0 +1 @@
+mod maximal_independent_set;