feat: path compaction - #865
Conversation
Sorry, something went wrong.
|
I'm not sure that the 'indegree == outdegree == 1' is quite the right rule depending on the graph. like an undirected graph should have 2 distinct undirected edges. so a <-> b <-> c should be compacted to a<->c but wouldnt b be considered to have an in-degree of 2 here? This is still true in a directed graph: a <-> b <-> c -> d I would say the correct result would be a<->c->d. |
Sorry, something went wrong.
|
Similarly I think dead ends would be incorrectly eliminated: ...a<->b should not eliminate b even though it has one in and one out in the directed graph. |
Sorry, something went wrong.
james-willis
left a comment
There was a problem hiding this comment.
initial quick review with AI. will come back later for a more complete review
Sorry, something went wrong.
| def requiredVertexAttributes(value: Seq[String]): this.type = { | ||
| requiredVertexAttributes = value.filter(_ != GraphFrame.ID).toSeq | ||
| this |
There was a problem hiding this comment.
you need to provide these columns for always remove/always keep filters as well
Sorry, something went wrong.
| /** | ||
| * Register a named aggregation expression to be evaluated over the interior vertices of each | ||
| * compacted path. The expression is applied inside a `groupBy(component).agg(...)`; it may | ||
| * reference any vertex attribute column of the original graph as well as `orig_src` and |
There was a problem hiding this comment.
I think it needs to be a column saved by requiredVertexAttributes
Sorry, something went wrong.
| .join( | ||
| persistedBaseEdges.select(col(GraphFrame.SRC), col(GraphFrame.DST).as("orig_dst")), | ||
| col(GraphFrame.SRC) === col(GraphFrame.ID), | ||
| "inner") |
There was a problem hiding this comment.
should we drop the src and dst from the join here? I think theyre dupes of the id
Sorry, something went wrong.
| * The returned `DataFrame` has columns `src`, `dst`, `count`, followed by one column per | ||
| * additionally registered aggregation expression. |
There was a problem hiding this comment.
I dont think the Map will give us the ordering semantics youre claiming here. need some kind of order map implementation for this
Sorry, something went wrong.
| GraphFrame.DST, | ||
| when(col("is_cycle"), array_min(col("inner_ids"))) | ||
| .otherwise(element_at(col("compacted_dst"), 1))) | ||
|
|
There was a problem hiding this comment.
We need to do count = count - 1 if is_cycle because we keep one of the candidates
Sorry, something went wrong.
|
|
||
| /** | ||
| * If `true` (default), cycles of candidates are not compacted: they are absent from the set of | ||
| * compacted edges, so their original edges survive unchanged in the output. If `false`, each |
There was a problem hiding this comment.
dont we drop all columns other than src,dst?
Sorry, something went wrong.
| compactedInteriorIds.unpersist(blocking = true) | ||
|
|
||
| resultIsPersistent() | ||
| allEdges |
There was a problem hiding this comment.
should we return a graphframe instead?
Sorry, something went wrong.
There was a problem hiding this comment.
Please no 😂
There is one API (old PageRank) that returns GraphFrame and I have no idea what do with it. It is (almost) impossible to support anything that returns GraphFrame in Spark-Connect.
Sorry, something went wrong.
Reviewers
At least 1 approving review is required to merge this pull request.
What changes were proposed in this pull request?
The new paths compaction API; only core -- public API will follow after pre-approve
Why are the changes needed?
Close #827