feat: path compaction - #865
Conversation
|
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 This is still true in a directed graph: |
|
Similarly I think dead ends would be incorrectly eliminated: |
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
| 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
| /** | ||
| * 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
| .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
| * 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
| 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
|
|
||
| /** | ||
| * 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?
| compactedInteriorIds.unpersist(blocking = true) | ||
|
|
||
| resultIsPersistent() | ||
| allEdges |
There was a problem hiding this comment.
should we return a graphframe instead?
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.
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