SemyonSinchenko · Aug 11, 2025 · Jun 29, 2025 · Jun 29, 2025 · Jun 29, 2025 · Jul 2, 2025
diff --git a/.gitignore b/.gitignore
@@ -56,3 +56,4 @@ tmp/*
 
 # db-connect targets
 graphframes-connect-databricks/*
+/workspace/
diff --git a/src/main/scala/org/graphframes/GraphFrame.scala b/src/main/scala/org/graphframes/GraphFrame.scala
@@ -749,6 +749,15 @@ object GraphFrame extends Serializable with Logging {
    */
   val EDGE: String = "edge"
 
+  /**
+   * Column name representing the weight attribute of edges in a graph.
+   *
+   * This field is used to identify and represent the weight associated with edges in a
+   * GraphFrame. The weight generally encodes the strength or importance of the connection between
+   * two nodes in a graph.
+   */
+  val WEIGHT: String = "weight"
+
   // ============================ Constructors and converters =================================
 
   /**

diff --git a/src/main/scala/org/graphframes/exceptions.scala b/src/main/scala/org/graphframes/exceptions.scala
@@ -23,3 +23,5 @@ class InvalidPatternException() extends Exception()
  */
 class GraphFramesUnreachableException()
     extends Exception("This exception should not be reachable")
+
+class InvalidPropertyGroupException(message: String) extends Exception(message)
diff --git a/src/main/scala/org/graphframes/propertygraph/PropertyGraphFrame.scala b/src/main/scala/org/graphframes/propertygraph/PropertyGraphFrame.scala
@@ -0,0 +1,138 @@
+package org.graphframes.propertygraph
+
+import org.apache.spark.sql.Column
+import org.apache.spark.sql.functions.col
+import org.apache.spark.sql.functions.lit
+import org.graphframes.GraphFrame
+import org.graphframes.propertygraph.property.EdgePropertyGroup
+import org.graphframes.propertygraph.property.VertexPropertyGroup
+
+/**
+ * A high-level abstraction for working with property graphs that simplifies interaction with the
+ * GraphFrames library.
+ *
+ * PropertyGraphFrame serves as a logical structure that manages collections of vertex and edge
+ * property groups, providing a user-friendly API for graph operations. It handles various
+ * internal complexities such as:
+ *   - ID conversion and collision prevention
+ *   - Management of directed/undirected graph representations
+ *   - Handling of weighted/unweighted edges
+ *   - Data consistency across different property groups
+ *
+ * The class maintains separate collections for vertex and edge properties, allowing for flexible
+ * graph construction while ensuring data integrity. Each property (vertex or edge) handles its
+ * data internally, while this class provides a simplified interface for working with the
+ * underlying GraphFrame structure.
+ *
+ * @param vertexPropertyGroups
+ *   Sequence of vertex property groups that define the graph's vertices
+ * @param edgesPropertyGroups
+ *   Sequence of edge property groups that define the graph's edges
+ */
+case class PropertyGraphFrame(
+    vertexPropertyGroups: Seq[VertexPropertyGroup],
+    edgesPropertyGroups: Seq[EdgePropertyGroup]) {
+  lazy private val vertexGroups: Map[String, VertexPropertyGroup] =
+    vertexPropertyGroups.map(pg => pg.name -> pg).toMap
+  lazy private val edgeGroups: Map[String, EdgePropertyGroup] =
+    edgesPropertyGroups.map(pg => pg.name -> pg).toMap
+
+  /**
+   * Converts a heterogeneous property graph into a unified GraphFrame representation.
+   *
+   * This method transforms a property graph that may contain multiple vertex types and both
+   * directed and undirected edges into a single GraphFrame object where all vertices and edges
+   * share the same schema. The conversion process handles:
+   *
+   *   - Internal ID generation and collision prevention by hashing vertex/edge IDs with their
+   *     group names
+   *   - Merging of different vertex types into a unified vertex DataFrame
+   *   - Conversion of directed/undirected edge relationships into a consistent edge DataFrame
+   *   - Filtering of vertices and edges based on provided predicates
+   *
+   * The method allows selecting a subset of property groups and applying filters to control which
+   * data is included in the final GraphFrame.
+   *
+   * @param vertexPropertyGroups
+   *   Sequence of vertex property group names to include in the GraphFrame
+   * @param edgePropertyGroups
+   *   Sequence of edge property group names to include in the GraphFrame
+   * @param edgeGroupFilters
+   *   Map of edge property group names to filter predicates (Column expressions)
+   * @param vertexGroupFilters
+   *   Map of vertex property group names to filter predicates (Column expressions)
+   * @return
+   *   A GraphFrame containing the unified representation of the selected and filtered property
+   *   groups
+   * @throws IllegalArgumentException
+   *   if any specified property group name doesn't exist
+   */
+  def toGraphFrame(
+      vertexPropertyGroups: Seq[String],
+      edgePropertyGroups: Seq[String],
+      edgeGroupFilters: Map[String, Column],
+      vertexGroupFilters: Map[String, Column]): GraphFrame = {
+    vertexPropertyGroups.foreach(name =>
+      require(vertexGroups.contains(name), s"Vertex property group $name does not exist"))
+    edgePropertyGroups.foreach(name =>
+      require(edgeGroups.contains(name), s"Edge property group $name does not exist"))
+
+    val vertices = vertexPropertyGroups
+      .map(name => vertexGroups(name).getData(vertexGroupFilters(name)))
+      .reduce(_ union _)
+
+    val edges = edgePropertyGroups
+      .map(name => edgeGroups(name).getData(edgeGroupFilters(name)))
+      .reduce(_ union _)
+
+    GraphFrame(vertices, edges)
+  }
+
+  /**
+   * Projects a bipartite graph onto one of its parts, creating edges between vertices that share
+   * neighbors in the other part. Drops the property group used for projection through and returns
+   * a new property graph.
+   *
+   * @param leftBiGraphPart
+   *   Name of the vertex property group to project onto
+   * @param rightBiGraphPart
+   *   Name of the vertex property group to project through
+   * @param edgeGroup
+   *   Name of the edge property group connecting the two parts
+   * @return
+   *   A new PropertyGraphFrame containing the projected graph
+   */
+  def projectionBy(
+      leftBiGraphPart: String,
+      rightBiGraphPart: String,
+      edgeGroup: String): PropertyGraphFrame = {
+    require(
+      edgeGroups(edgeGroup).srcPropertyGroup.name == leftBiGraphPart,
+      s"Edge Property Group should have $leftBiGraphPart source group but has ${edgeGroups(edgeGroup).srcPropertyGroup.name}")
+    require(
+      edgeGroups(edgeGroup).dstPropertyGroup.name == rightBiGraphPart,
+      s"Edge Property Group should have $rightBiGraphPart destination group but has ${edgeGroups(edgeGroup).dstPropertyGroup.name}")
+    val keptVPropertyGroups = vertexPropertyGroups.filterNot(g => g.name == rightBiGraphPart)
+    val keptEPropertyGroups = edgesPropertyGroups.filterNot(g => g.name == edgeGroup)
+    val oldEdgesData = edgeGroups(edgeGroup).data
+
+    // Create new edges by joining vertices through their common neighbors
+    val projectedEdges = oldEdgesData
+      .as("e1")
+      .join(oldEdgesData.as("e2"), col("e1.dst") === col("e2.dst"))
+      .where("e1.src < e2.src")
+      .select(col("e1.src").alias(GraphFrame.SRC), col("e2.src").alias(GraphFrame.DST))
+
+    val newEdgeGroup = EdgePropertyGroup(
+      name = s"projected_$edgeGroup",
+      data = projectedEdges,
+      srcPropertyGroup = vertexGroups(leftBiGraphPart),
+      dstPropertyGroup = vertexGroups(leftBiGraphPart),
+      isDirected = false,
+      srcColumnName = GraphFrame.SRC,
+      dstColumnName = GraphFrame.DST,
+      weightColumn = lit(1.0))
+
+    PropertyGraphFrame(keptVPropertyGroups, keptEPropertyGroups :+ newEdgeGroup)
+  }
+}
diff --git a/src/main/scala/org/graphframes/propertygraph/property/EdgePropertyGroup.scala b/src/main/scala/org/graphframes/propertygraph/property/EdgePropertyGroup.scala
@@ -0,0 +1,151 @@
+package org.graphframes.propertygraph.property
+
+import org.apache.spark.sql.Column
+import org.apache.spark.sql.DataFrame
+import org.apache.spark.sql.functions.col
+import org.apache.spark.sql.functions.concat
+import org.apache.spark.sql.functions.lit
+import org.apache.spark.sql.functions.sha2
+import org.apache.spark.sql.types._
+import org.graphframes.GraphFrame
+import org.graphframes.InvalidPropertyGroupException
+
+/**
+ * Represents a logical group of edges in a property graph with associated metadata and data.
+ *
+ * EdgePropertyGroup encapsulates edge data stored in a DataFrame along with metadata describing
+ * how to interpret the data as graph edges. Each edge group has:
+ *
+ *   - A unique name identifier
+ *   - DataFrame containing the actual edge data
+ *   - Source and destination vertex property groups
+ *   - Direction flag indicating if edges are directed or undirected
+ *   - Column names specifying source vertex, destination vertex and edge weight columns
+ *
+ * The class validates that required columns exist in the provided DataFrame on creation. Required
+ * columns are:
+ *   - Source vertex column
+ *   - Destination vertex column
+ *   - Weight column
+ *
+ * @param name
+ *   Unique identifier for this edge property group
+ * @param data
+ *   DataFrame containing the edge data with required columns
+ * @param srcPropertyGroup
+ *   Source vertex property group
+ * @param dstPropertyGroup
+ *   Destination vertex property group
+ * @param isDirected
+ *   Whether edges should be treated as directed (true) or undirected (false)
+ * @param srcColumnName
+ *   Name of the source vertex column in the data
+ * @param dstColumnName
+ *   Name of the destination vertex column in the data
+ * @param weightColumnName
+ *   Name of the edge weight column in the data
+ * @note
+ *   When edges from different groups are combined into a GraphFrame, their SRCs and DSTs are
+ *   hashed with the group name to prevent collisions in the same way as ID of the corresponded
+ *   vertex group is hashed.
+ */
+case class EdgePropertyGroup private (
+    name: String,
+    data: DataFrame,
+    srcPropertyGroup: VertexPropertyGroup,
+    dstPropertyGroup: VertexPropertyGroup,
+    isDirected: Boolean,
+    srcColumnName: String,
+    dstColumnName: String,
+    weightColumnName: String)
+    extends PropertyGroup {
+
+  override protected def validate(): this.type = {
+    if (!data.columns.contains(srcColumnName)) {
+      throw new InvalidPropertyGroupException(
+        s"source column $srcColumnName does not exist, existed columns [${data.columns.mkString(", ")}]")
+    }
+    if (!data.columns.contains(dstColumnName)) {
+      throw new InvalidPropertyGroupException(
+        s"dest column $dstColumnName does not exist, existed columns [${data.columns.mkString(", ")}]")
+    }
+    if (!data.columns.contains(weightColumnName)) {
+      throw new InvalidPropertyGroupException(
+        s"weight column $weightColumnName does not exist, existed columns [${data.columns.mkString(", ")}]")
+    }
+    val weightColumnType = data.schema(weightColumnName).dataType
+    if (!weightColumnType.isInstanceOf[NumericType]) {
+      throw new InvalidPropertyGroupException(
+        s"weight column $weightColumnName must be numeric type, but was $weightColumnType")
+    }
+
+    this
+  }
+
+  private def hashSrcEdge: Column =
+    concat(lit(srcPropertyGroup.name), sha2(col(srcColumnName).cast("string"), 256))
+  private def hashDstEdge: Column =
+    concat(lit(dstPropertyGroup.name), sha2(col(dstColumnName).cast("string"), 256))
+
+  override protected[graphframes] def getData(filter: Column): DataFrame = {
+    val filteredData = data.filter(filter)
+
+    val baseEdges = filteredData.select(
+      hashSrcEdge.alias(GraphFrame.SRC),
+      hashDstEdge.alias(GraphFrame.DST),
+      col(weightColumnName).alias(GraphFrame.WEIGHT))
+
+    if (isDirected) {
+      baseEdges
+    } else {
+      baseEdges.union(
+        baseEdges.select(
+          col(GraphFrame.DST).as(GraphFrame.SRC),
+          col(GraphFrame.SRC).as(GraphFrame.DST),
+          col(GraphFrame.WEIGHT).alias(GraphFrame.WEIGHT)))
+    }
+  }
+}
+
+object EdgePropertyGroup {
+  def apply(
+      name: String,
+      data: DataFrame,
+      srcPropertyGroup: VertexPropertyGroup,
+      dstPropertyGroup: VertexPropertyGroup,
+      isDirected: Boolean,
+      srcColumnName: String,
+      dstColumnName: String,
+      weightColumnName: String): EdgePropertyGroup = {
+    new EdgePropertyGroup(
+      name,
+      data,
+      srcPropertyGroup,
+      dstPropertyGroup,
+      isDirected,
+      srcColumnName,
+      dstColumnName,
+      weightColumnName).validate()
+  }
+
+  def apply(
+      name: String,
+      data: DataFrame,
+      srcPropertyGroup: VertexPropertyGroup,
+      dstPropertyGroup: VertexPropertyGroup,
+      isDirected: Boolean,
+      srcColumnName: String,
+      dstColumnName: String,
+      weightColumn: Column): EdgePropertyGroup = {
+    val dataWithWeight = data.withColumn(GraphFrame.WEIGHT, weightColumn)
+    apply(
+      name,
+      dataWithWeight,
+      srcPropertyGroup,
+      dstPropertyGroup,
+      isDirected,
+      srcColumnName,
+      dstColumnName,
+      GraphFrame.WEIGHT)
+  }
+}
diff --git a/src/main/scala/org/graphframes/propertygraph/property/PropertyGroup.scala b/src/main/scala/org/graphframes/propertygraph/property/PropertyGroup.scala
@@ -0,0 +1,14 @@
+package org.graphframes.propertygraph.property
+
+import org.apache.spark.sql.Column
+import org.apache.spark.sql.DataFrame
+import org.apache.spark.sql.functions.lit
+
+trait PropertyGroup {
+  val name: String
+  val data: DataFrame
+  protected def validate(): this.type
+
+  protected[graphframes] def getData: DataFrame = getData(lit(true))
+  protected[graphframes] def getData(filter: Column): DataFrame
+}
Original file line number	Diff line number	Diff line change
Expand Up		@@ -56,3 +56,4 @@ tmp/*

		# db-connect targets
		graphframes-connect-databricks/*
		/workspace/