graphframes · rjurney · Sep 20, 2025 · Sep 13, 2025 · Sep 13, 2025 · Sep 13, 2025
diff --git a/core/src/main/scala/org/graphframes/GraphFrame.scala b/core/src/main/scala/org/graphframes/GraphFrame.scala
@@ -360,6 +360,26 @@ class GraphFrame private (
    * @group motif
    */
   def find(pattern: String): DataFrame = {
+    val VarLengthPattern = """\((\w+)\)-\[(\w*)\*(\d*)\.\.(\d*)\]->\((\w+)\)""".r
+    pattern match {
+      case VarLengthPattern(src, name, min, max, dst) =>
+        if (min.isEmpty || max.isEmpty) {
+          throw new InvalidParseException(
+            s"Unbounded length patten ${pattern} is not supported! " +
+              "Please a pattern of defined length.")
+        }
+        val strToSeq: Seq[String] = (min.toInt to max.toInt).reverse.map { hop =>
+          s"($src)-[$name*$hop]->($dst)"
+        }
+        strToSeq
+          .map(findAugmentedPatterns)
+          .reduce((a, b) => a.unionByName(b, allowMissingColumns = true))
+      case _ =>
+        findAugmentedPatterns(pattern)
+    }
+  }
+
+  def findAugmentedPatterns(pattern: String): DataFrame = {
     val patterns = Pattern.parse(pattern)
 
     // For each named vertex appearing only in a negated term, we augment the positive terms

diff --git a/core/src/main/scala/org/graphframes/pattern/patterns.scala b/core/src/main/scala/org/graphframes/pattern/patterns.scala
@@ -45,8 +45,32 @@ private[graphframes] object PatternParser extends RegexParsers {
     "!" ~ edge ^^ { case _ ~ e =>
       Negation(e)
     }
+  private val fixedLengthPattern: Parser[List[Edge]] =
+    vertex ~ "-" ~ "[" ~ "[a-zA-Z0-9_]*".r ~ "*" ~ "[0-9]+".r ~ "]" ~ "->" ~ vertex ^^ {
+      case src ~ "-" ~ "[" ~ name ~ "*" ~ num ~ "]" ~ "->" ~ dst => {
+        val hop: Int = num.toInt
+        if (hop == 1) {
+          List(if (name.isEmpty) AnonymousEdge(src, dst) else NamedEdge(name, src, dst))
+        } else if (hop > 1) {
+          val midVertices = (1 until hop).map(i => NamedVertex(s"_v$i"))
+          val vertices = src +: midVertices :+ dst
+          vertices
+            .sliding(2)
+            .zipWithIndex
+            .map {
+              case (Seq(v1, v2), i) =>
+                if (name.isEmpty) AnonymousEdge(v1, v2) else NamedEdge(s"_$name${i + 1}", v1, v2)
+              case _ => throw new GraphFramesUnreachableException()
+            }
+            .toList
+        } else {
+          throw new GraphFramesUnreachableException()
+        }
+      }
+      case _ => throw new GraphFramesUnreachableException()
+    }
   private val pattern: Parser[Pattern] = edge | vertex | negatedEdge
-  val patterns: Parser[List[Pattern]] = repsep(pattern, ";")
+  val patterns: Parser[List[Pattern]] = fixedLengthPattern | repsep(pattern, ";")
 }
 
 private[graphframes] object Pattern {

diff --git a/core/src/test/scala/org/graphframes/PatternMatchSuite.scala b/core/src/test/scala/org/graphframes/PatternMatchSuite.scala
@@ -535,6 +535,12 @@ class PatternMatchSuite extends SparkFunSuite with GraphFrameTestSparkContext {
     }
   }
 
+  test("Unbound variable-length pattern (u)->[*..5]->(v)") {
+    intercept[InvalidParseException] {
+      g.find("(u)-[*..5]->(v)")
+    }
+  }
+
   /* ============================= More complex use case examples ============================== */
 
   test("triangles via post-hoc filter") {
@@ -547,6 +553,111 @@ class PatternMatchSuite extends SparkFunSuite with GraphFrameTestSparkContext {
     compareResultToExpected(res, Set(Row(0L, 1L, 2L), Row(2L, 0L, 1L), Row(1L, 2L, 0L)))
   }
 
+  test("fixed-length 3") {
+    val fixedLengthEdge = g
+      .find("(u)-[*3]->(v)")
+      .where("u.id == 0")
+      .select("u.id", "_v1.id", "_v2.id", "v.id")
+
+    val res = fixedLengthEdge.collect().toSet
+    val expected = Set(Row(0L, 1L, 2L, 0L), Row(0L, 1L, 2L, 3L), Row(0L, 1L, 0L, 1L))
+    compareResultToExpected(res, expected)
+  }
+
+  test("fixed-length 3 with named edge") {
+    val fixedLengthNamedEdge = g
+      .find("(u)-[e*3]->(v)")
+      .where("u.id == 0")
+
+    val expectedCols = Seq("u", "_e1", "_v1", "_e2", "_v2", "_e3", "v")
+
+    assert(fixedLengthNamedEdge.schema.map(_.name) == expectedCols)
+  }
+
+  test("fixed-length 5") {
+    val fixedLengthEdge = g
+      .find("(u)-[*5]->(v)")
+      .where("u.id == 0")
+      .select("u.id", "_v1.id", "_v2.id", "_v3.id", "_v4.id", "v.id")
+
+    val res = fixedLengthEdge.collect().toSet
+    val expected = Set(
+      Row(0L, 1L, 2L, 0L, 1L, 0L),
+      Row(0L, 1L, 0L, 1L, 0L, 1L),
+      Row(0L, 1L, 2L, 0L, 1L, 2L),
+      Row(0L, 1L, 0L, 1L, 2L, 0L),
+      Row(0L, 1L, 0L, 1L, 2L, 3L))
+    compareResultToExpected(res, expected)
+  }
+
+  test("var-length pattern 2..2") {
+    val varEdge = g
+      .find("(u)-[*2..2]->(v)")
+      .where("u.id == 0")
+
+    val fixedEdge = g
+      .find("(u)-[*2]->(v)")
+      .where("u.id == 0")
+
+    assert(varEdge.schema == fixedEdge.schema)
+    assert(varEdge.except(fixedEdge).isEmpty && fixedEdge.except(varEdge).isEmpty)
+  }
+
+  test("var-length pattern 2..3") {
+    val varEdge = g
+      .find("(u)-[*2..3]->(v)")
+      .where("u.id == 0")
+
+    val fixedEdge2 = g
+      .find("(u)-[*2]->(v)")
+      .where("u.id == 0")
+
+    val fixedEdge3 = g
+      .find("(u)-[*3]->(v)")
+      .where("u.id == 0")
+
+    val unionEdge = fixedEdge3
+      .unionByName(fixedEdge2, allowMissingColumns = true)
+
+    assert(varEdge.schema == unionEdge.schema)
+    assert(varEdge.except(unionEdge).isEmpty && unionEdge.except(varEdge).isEmpty)
+  }
+
+  test("var-length pattern 2..3 with named edge") {
+    val varEdge = g
+      .find("(u)-[e*2..3]->(v)")
+      .where("u.id == 0")
+
+    val expectedCols = Seq("u", "_e1", "_v1", "_e2", "_v2", "_e3", "v")
+
+    assert(varEdge.schema.map(_.name) == expectedCols)
+  }
+
+  test("var-length pattern 3..5") {
+    val varEdge = g
+      .find("(u)-[*3..5]->(v)")
+      .where("u.id == 0")
+
+    val fixedEdge3 = g
+      .find("(u)-[*3]->(v)")
+      .where("u.id == 0")
+
+    val fixedEdge4 = g
+      .find("(u)-[*4]->(v)")
+      .where("u.id == 0")
+
+    val fixedEdge5 = g
+      .find("(u)-[*5]->(v)")
+      .where("u.id == 0")
+
+    val unionEdge = fixedEdge5
+      .unionByName(fixedEdge4, allowMissingColumns = true)
+      .unionByName(fixedEdge3, allowMissingColumns = true)
+
+    assert(varEdge.schema == unionEdge.schema)
+    assert(varEdge.except(unionEdge).isEmpty && unionEdge.except(varEdge).isEmpty)
+  }
+
   test("stateful predicates via UDFs") {
     val chain4 = g
       .find("(a)-[ab]->(b); (b)-[bc]->(c); (c)-[cd]->(d)")

diff --git a/core/src/test/scala/org/graphframes/pattern/PatternSuite.scala b/core/src/test/scala/org/graphframes/pattern/PatternSuite.scala
@@ -17,6 +17,7 @@
 
 package org.graphframes.pattern
 
+import org.graphframes.GraphFramesUnreachableException
 import org.graphframes.InvalidParseException
 import org.graphframes.SparkFunSuite
 
@@ -51,6 +52,36 @@ class PatternSuite extends SparkFunSuite {
           AnonymousEdge(NamedVertex("u"), NamedVertex("v")),
           AnonymousEdge(NamedVertex("v"), NamedVertex("w")),
           Negation(AnonymousEdge(NamedVertex("u"), NamedVertex("w")))))
+
+    assert(
+      Pattern.parse("(u)-[*3]->(v)") ===
+        Seq(
+          AnonymousEdge(NamedVertex("u"), NamedVertex("_v1")),
+          AnonymousEdge(NamedVertex("_v1"), NamedVertex("_v2")),
+          AnonymousEdge(NamedVertex("_v2"), NamedVertex("v"))))
+
+    assert(
+      Pattern.parse("(u)-[*5]->(v)") ===
+        Seq(
+          AnonymousEdge(NamedVertex("u"), NamedVertex("_v1")),
+          AnonymousEdge(NamedVertex("_v1"), NamedVertex("_v2")),
+          AnonymousEdge(NamedVertex("_v2"), NamedVertex("_v3")),
+          AnonymousEdge(NamedVertex("_v3"), NamedVertex("_v4")),
+          AnonymousEdge(NamedVertex("_v4"), NamedVertex("v"))))
+
+    assert(
+      Pattern.parse("(u)-[*10]->(v)") ===
+        Seq(
+          AnonymousEdge(NamedVertex("u"), NamedVertex("_v1")),
+          AnonymousEdge(NamedVertex("_v1"), NamedVertex("_v2")),
+          AnonymousEdge(NamedVertex("_v2"), NamedVertex("_v3")),
+          AnonymousEdge(NamedVertex("_v3"), NamedVertex("_v4")),
+          AnonymousEdge(NamedVertex("_v4"), NamedVertex("_v5")),
+          AnonymousEdge(NamedVertex("_v5"), NamedVertex("_v6")),
+          AnonymousEdge(NamedVertex("_v6"), NamedVertex("_v7")),
+          AnonymousEdge(NamedVertex("_v7"), NamedVertex("_v8")),
+          AnonymousEdge(NamedVertex("_v8"), NamedVertex("_v9")),
+          AnonymousEdge(NamedVertex("_v9"), NamedVertex("v"))))
   }
 
   test("bad parses") {
@@ -123,6 +154,26 @@ class PatternSuite extends SparkFunSuite {
     }
   }
 
+  test("unsupported parse on the fixed length patterns") {
+    withClue("Failed to catch parse error with graph frame unreachable") {
+      intercept[GraphFramesUnreachableException] {
+        Pattern.parse("(u)-[*0]->(v)")
+      }
+    }
+
+    withClue("Failed to catch parse error with bad motif string") {
+      intercept[InvalidParseException] {
+        Pattern.parse("(u)-[*]->(v)")
+      }
+    }
+
+    withClue("Failed to catch parse error with chaining quantified length pattern") {
+      intercept[InvalidParseException] {
+        Pattern.parse("(u)-[*2]->(v);(v)-[e]->(w)")
+      }
+    }
+  }
+
   test("empty pattern should be parsable") {
     Pattern.parse("")
   }