feat: implement the graph API and core algorithms - #3
Open
nish2292 wants to merge 2 commits into
Open
Conversation
Implements the Discussion daft-engine#1 two class API and the core algorithm set from issue daft-engine#2, ported from the validated internal prototype and shaped to this repo's conventions (Apache-2.0, hatchling, ruff/pydocstyle, Python 3.10+). API: - abstract Graph plus DirectedGraph and UndirectedGraph; the type carries direction, so algorithms declare the flavor they need - construction from edges with optional vertices, configurable columns, opt in validation; transforms preserve the concrete flavor Algorithms: - connected components (regular + strong), pagerank (+ personalized) and parallel_personalized_pagerank, bfs/bfs_paths/shortest_paths/ all_shortest_paths/all_paths, label_propagation, power_iteration_clustering, find (motif DSL), aggregate_messages/pregel, triangle_count, k_core, cycle detection, maximal_independent_set, random_walks, svd_plus_plus, hyper_anf, plus reindex/restore_ids and edge utilities Packaging: - core depends on daft only; numpy and scipy behind the optional local extra - Self imported under TYPE_CHECKING for the 3.10 target - igraph and networkx are test only oracles Verified: 331 tests pass, ruff and mypy clean.
Every distributed iterative algorithm stalled on Daft's Ray/Flotilla runner: partition counts compounded round over round until each shuffle needed a partition-count-squared number of pieces, exhausting the cluster. Daft resolves a shuffle's output partition count to the repartition spec's count or else the input's (unwrap_or(input_num_partitions)) and never lowers it, and union_all sums its inputs' counts, so a step that symmetrizes/unions doubles the count every round. Invisible on the native runner (no partitions), fatal on Ray. Fix: bound the partition count wherever iterative state is carried, via a cheap plan rewrite (into_partitions on the already-optimized plan, no extra execution), gated to the Ray runner (no-op on native): - iterate.py: bound_partitions() (lazy cap) and collect_bounded() (materialize then present at a bounded count). collect_bounded returns the coalesced frame LAZILY - re-collecting after into_partitions makes num_partitions() report 0, which silently disables every downstream cap. - message_passing.py: cap triplets, the aggregate_messages union, and each pregel step (fixes label_propagation, k_core, shortest_paths, pagerank, ...). - connected_components: cap the star step passes, label propagation, adjacency. - Custom-loop algorithms that bypass the shared machinery: strongly_connected_ components (peeling loop + active_v/active_e + union fold), hyper_anf (per-hop HLL), maximal_independent_set (per-round status), shortest_paths (landmark fold), and the shared BFS frontier in _traversal (visited/levels growth). - Static once-collected inputs (adjacency/edges/degrees) joined every round. Also adds tests/test_bfs_scaling.py, examples/, and benchmarks/ from the driver-memory BFS rewrite.
nish2292
marked this pull request as ready for review
August 23, 2026 05:51
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Implements the Discussion #1 API and the core algorithms requested in #2. The package was an empty template, so this provides the initial implementation.
API (the two-class model from #1)
Graphbase withDirectedGraphandUndirectedGraphsubclasses, so the type carries direction and each algorithm declares the flavor it needs.degrees,triplets,filter_vertices/filter_edges,drop_isolated_vertices,degree_by_type, and directed/undirected conversion (reverse,as_undirected,as_directed).Algorithms (the #2 set)
connected_components(weak),strongly_connected_componentspagerank(+ personalized),parallel_personalized_pagerankbfs,bfs_paths,shortest_paths,all_shortest_paths,all_pathslabel_propagation,power_iteration_clusteringfind(GraphFrames-style DSL)aggregate_messages,pregeltriangle_count,k_core, cycle detection (has_cycle,vertices_on_cycles),maximal_independent_set,random_walks,svd_plus_plus,hyper_anf, id reindexing, and edge utilitiesImplementation
minhash-dedupeexample, then hardens it with types and tests.daftalone.numpyandscipyare an optionallocalextra used only for the single-node solves (connected_componentsstrategy="local",svd_plus_plus).Opening as a draft for maintainer feedback on scope and shape. Closes #2.