feat: GQL support - #849
Conversation
|
Even if spark uses https://github.com/ihji/sbt-antlr4 the plugin looks unmaintained (last commit 5 years ago, staled PRs and unanswered issues). And the whole plugin is less than 200 lines (and we need only part of it). So, my decision was to add our own internal plugin. |
|
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## main #849 +/- ##
==========================================
+ Coverage 79.26% 80.33% +1.06%
==========================================
Files 81 90 +9
Lines 4712 5293 +581
Branches 554 646 +92
==========================================
+ Hits 3735 4252 +517
- Misses 977 1041 +64 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
optimizations will follow up
FunctionsRegistry, parsing, tests.
There was a problem hiding this comment.
Pull request overview
Adds bounded GQL MATCH support to PropertyGraphFrame, compiling graph patterns into Spark DataFrame joins.
Changes:
- Adds ANTLR parsing, AST, resolution, planning, execution, functions, and explain output.
- Adds public query, schema, options, and case-insensitive group APIs.
- Adds extensive unit and integration coverage plus build configuration.
Reviewed changes
Copilot reviewed 31 out of 33 changed files in this pull request and generated 20 comments.
Show a summary per file
| File | Description |
|---|---|
python/pyproject.toml |
Updates tutorial archive dependency. |
project/plugins.sbt |
Adds versioned ANTLR tooling. |
project/GraphFramesAntlr4Plugin.scala |
Generates parser sources. |
build.sbt |
Configures ANTLR and core generation. |
core/src/main/antlr4/org/graphframes/propertygraph/internal/GqlLexer.g4 |
Defines GQL tokens. |
core/src/main/antlr4/org/graphframes/propertygraph/internal/GqlParser.g4 |
Defines supported GQL grammar. |
core/src/main/scala/org/graphframes/propertygraph/QueryOptions.scala |
Adds query configuration and explain modes. |
core/src/main/scala/org/graphframes/propertygraph/PropertyGraphFrame.scala |
Exposes query, explain, and schema APIs. |
core/src/main/scala/org/graphframes/propertygraph/property/PropertyGroup.scala |
Supports requested property columns. |
core/src/main/scala/org/graphframes/propertygraph/property/VertexPropertyGroup.scala |
Extends vertex scans with properties. |
core/src/main/scala/org/graphframes/propertygraph/property/EdgePropertyGroup.scala |
Extends edge scans with properties. |
core/src/main/scala/org/graphframes/propertygraph/internal/GqlAst.scala |
Defines the query AST. |
core/src/main/scala/org/graphframes/propertygraph/internal/AstBuilder.scala |
Converts parse trees into ASTs. |
core/src/main/scala/org/graphframes/propertygraph/internal/SchemaGraphSnapshot.scala |
Models and renders graph schemas. |
core/src/main/scala/org/graphframes/propertygraph/internal/Resolver.scala |
Resolves patterns and predicates. |
core/src/main/scala/org/graphframes/propertygraph/internal/QueryIr.scala |
Defines resolved query IR. |
core/src/main/scala/org/graphframes/propertygraph/internal/JoinPlan.scala |
Defines physical join plans. |
core/src/main/scala/org/graphframes/propertygraph/internal/JoinOptimizer.scala |
Produces join ordering. |
core/src/main/scala/org/graphframes/propertygraph/internal/GraphStatistics.scala |
Adds statistics abstractions. |
core/src/main/scala/org/graphframes/propertygraph/internal/QueryExecutor.scala |
Executes joins and projections. |
core/src/main/scala/org/graphframes/propertygraph/internal/FunctionRegistry.scala |
Lowers whitelisted Spark functions. |
core/src/main/scala/org/graphframes/propertygraph/internal/GqlExplain.scala |
Renders logical and physical plans. |
core/src/test/scala/org/graphframes/propertygraph/PropertyGraphFrameTest.scala |
Tests schema rendering. |
core/src/test/scala/org/graphframes/propertygraph/PropertyGraphFrameQuerySuite.scala |
Tests public query behavior. |
core/src/test/scala/org/graphframes/propertygraph/PropertyGraphFrameCaseInsensitiveSuite.scala |
Tests case-insensitive APIs. |
core/src/test/scala/org/graphframes/propertygraph/internal/SchemaGraphSnapshotSuite.scala |
Tests schema snapshots. |
core/src/test/scala/org/graphframes/propertygraph/internal/ResolverCaseInsensitiveSuite.scala |
Tests label resolution casing. |
core/src/test/scala/org/graphframes/propertygraph/internal/QueryExecutorSuite.scala |
Tests execution behavior. |
core/src/test/scala/org/graphframes/propertygraph/internal/JoinOptimizerSuite.scala |
Tests physical planning. |
core/src/test/scala/org/graphframes/propertygraph/internal/GqlExplainSuite.scala |
Tests explain rendering. |
core/src/test/scala/org/graphframes/propertygraph/internal/AstBuilderSuite.scala |
Tests parsing and lowering. |
Suppressed comments (2)
core/src/main/scala/org/graphframes/propertygraph/PropertyGraphFrame.scala:50
- These
toMapcalls silently collapse groups whose names differ only by case. The schema snapshot still retains both original groups, so resolution can choose one canonical name while execution reads the other map entry. Reject case-insensitive duplicate names when constructing the graph, or otherwise preserve an unambiguous canonical mapping.
lazy private[propertygraph] val vertexGroups: Map[String, VertexPropertyGroup] =
vertexPropertyGroups.map(pg => pg.name.toLowerCase -> pg).toMap
lazy private[propertygraph] val edgeGroups: Map[String, EdgePropertyGroup] =
edgesPropertyGroups.map(pg => pg.name.toLowerCase -> pg).toMap
core/src/main/scala/org/graphframes/propertygraph/internal/QueryExecutor.scala:106
- The debug execution path has the same duplicate-column failure as the public executor:
unionByNamecannot combine multi-plan projections such as two unaliasednamefields. Keep it consistent with positional union.
perPlan.reduce(_ unionByName _)
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
| lazy private[propertygraph] val vertexGroups: Map[String, VertexPropertyGroup] = | ||
| vertexPropertyGroups.map(pg => pg.name.toLowerCase -> pg).toMap | ||
| lazy private[propertygraph] val edgeGroups: Map[String, EdgePropertyGroup] = | ||
| edgesPropertyGroups.map(pg => pg.name.toLowerCase -> pg).toMap |
| if (resolved.paths.isEmpty) { | ||
| return QueryExecutor.execute(this, Seq.empty) |
| options.maxSchemaPathLength > 0, | ||
| s"maxSchemaPathLength must be positive, got ${options.maxSchemaPathLength}") | ||
| val ast = AstBuilder.parse(gql) | ||
| val resolved = Resolver.resolve(ast, schemaGraphSnapshot, options) |
|
|
||
| private def renderProjection(projection: Projection): String = projection match { | ||
| case Projection.Default => "(default: matched IDs of first/last named nodes)" | ||
| case Projection.Star => "* (all matched variables)" |
Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>
GQL MATCH on PropertyGraphFrame — motivation, philosophy, and the plan model
1. Motivation & philosophy
The guiding principle is don't rebuild what the platform already does well. A property-graph query engine on top of Spark sits between two mature systems, and the design
draws a hard line on either side:
WHERE a.age > 30becomescol("age") > 30; aRETURN year(x.born)becomesfunctions.year(...). Catalyst keeps doing the cost-based physical planning, the codegen, the shuffles. We emit logical DataFrame ops and let the existing engine optimize them.What's left in the middle — and the only thing this engine actually owns — is the schema-aware translation from a graph pattern to a relational join plan with pushing down all the predicates from the GQL. That is the entire value-add: taking
(a:Person)-[:KNOWS]->(b)and figuring out, against the declared LPG schema, which concrete vertex/edge groups it can bind to and how they join. This is the part neither Spark SQL nor the algorithm collection provides.2. Decisions and hard limitations that follow
The philosophy dictates the boundaries directly. These are not gaps to be filled later; they are the shape of the thing:
MATCH <linear pattern> [WHERE] [RETURN]. NoWITH, no aggregation, noORDER BY/LIMIT, no multi-MATCH, noOPTIONAL. Anything that is "just SQL on the result" the user can do on the returnedDataFrame— so we don't absorb it.spark.sql.functions, fail-fast on unknown name/arity. No UDFs. The whitelist is the scope boundary: if Spark SQL has it, we expose it by name; if it doesn't, we don't invent it. We do not expose all, just a subset that I think is useful forWHERE. Anything else does not make any sense (see motivation): if user wants to apply a function on top of results the result isDataFramealready.start_*,end_*,edge_property_group,patharray). We surface the matched topology; arbitrary projection reshaping is the user's to do downstream in SQL.3. The model: query → plan → optimization
Three narrowing IRs, each a typed boundary with one owner. The progression mirrors a compiler, but each stage exists only to get closer to "emit
DataFrameops."Query (syntactic):
GqlAst. A hand-written sealed ADT, completely firewalled from ANTLR — no generated *Context type escapesAstBuilder. Pure syntax; carries no schema knowledge and no precedence in its nodes (precedence lives in the grammar tiers). This is what makes the rest of the engine testable against plain case classes and the grammar replaceable.Resolved (logical): ResolvedQuery. This is the only schema-aware stage — the heart of the engine. Resolution does two things:
SchemaPathsby a bounded DFS over theSchemaGraphSnapshot. An untyped/ambiguous element fans out into one path per compatible vertex/edge group. Direction (traversedForward) is recorded per step so<-[e]-and undirected edges join correctly. Disconnected → zero paths.Physical:
JoinPlan. One self-contained plan per path (path + element-level join order + predicates + projection + the stats that drove it). Self-contained so explain(Physical) renders without re-resolving.JoinOptimizeris the single place order is chosen; it's structured as a Planner + PlanRefiner SPI threadingOption[GraphStatistics], but the default planner uses written order and consumes no stats. The seam for CBO exists; the policy today is "trust Catalyst."Execution.
QueryExecutorwalks each plan's order, scanning each element once and joining onto the growing frame. Two refinements keep the emitted SQL lean, both consistent with "let Spark do the work but don't hand it garbage":Plans are UNION ALL-ed into one
DataFramewith the fixed schema.4. Notes on the code
GqlAst→ QueryIr/SchemaGraphSnapshot →Resolver.classifyWhere→QueryExecutor.executePlan/classifyElementProps. Grammar +AstBuilderare mechanical.