From 9a8f860f4f99e2f48a37df7b0a81c199b836d4dc Mon Sep 17 00:00:00 2001 From: semyonsinchenko Date: Wed, 24 Jun 2026 18:37:44 +0200 Subject: [PATCH] feat: optimizer + execution --- .../propertygraph/PropertyGraphFrame.scala | 217 ++++- .../propertygraph/QueryOptions.scala | 88 ++ .../propertygraph/internal/GqlExplain.scala | 176 ++++ .../internal/GraphStatistics.scala | 95 +++ .../internal/JoinOptimizer.scala | 80 ++ .../propertygraph/internal/JoinPlan.scala | 74 ++ .../internal/QueryExecutor.scala | 712 ++++++++++++++++ .../property/EdgePropertyGroup.scala | 21 +- .../property/PropertyGroup.scala | 31 +- .../property/VertexPropertyGroup.scala | 28 +- ...opertyGraphFrameCaseInsensitiveSuite.scala | 169 ++++ .../PropertyGraphFrameQuerySuite.scala | 670 +++++++++++++++ .../PropertyGraphFrameTest.scala | 26 + .../internal/GqlExplainSuite.scala | 97 +++ .../internal/JoinOptimizerSuite.scala | 121 +++ .../internal/QueryExecutorSuite.scala | 770 ++++++++++++++++++ 16 files changed, 3340 insertions(+), 35 deletions(-) create mode 100644 core/src/main/scala/org/graphframes/propertygraph/QueryOptions.scala create mode 100644 core/src/main/scala/org/graphframes/propertygraph/internal/GqlExplain.scala create mode 100644 core/src/main/scala/org/graphframes/propertygraph/internal/GraphStatistics.scala create mode 100644 core/src/main/scala/org/graphframes/propertygraph/internal/JoinOptimizer.scala create mode 100644 core/src/main/scala/org/graphframes/propertygraph/internal/JoinPlan.scala create mode 100644 core/src/main/scala/org/graphframes/propertygraph/internal/QueryExecutor.scala create mode 100644 core/src/test/scala/org/graphframes/propertygraph/PropertyGraphFrameCaseInsensitiveSuite.scala create mode 100644 core/src/test/scala/org/graphframes/propertygraph/PropertyGraphFrameQuerySuite.scala create mode 100644 core/src/test/scala/org/graphframes/propertygraph/internal/GqlExplainSuite.scala create mode 100644 core/src/test/scala/org/graphframes/propertygraph/internal/JoinOptimizerSuite.scala create mode 100644 core/src/test/scala/org/graphframes/propertygraph/internal/QueryExecutorSuite.scala diff --git a/core/src/main/scala/org/graphframes/propertygraph/PropertyGraphFrame.scala b/core/src/main/scala/org/graphframes/propertygraph/PropertyGraphFrame.scala index ac1ff9c1f..254f22476 100644 --- a/core/src/main/scala/org/graphframes/propertygraph/PropertyGraphFrame.scala +++ b/core/src/main/scala/org/graphframes/propertygraph/PropertyGraphFrame.scala @@ -5,6 +5,13 @@ import org.apache.spark.sql.DataFrame import org.apache.spark.sql.functions.col import org.apache.spark.sql.functions.lit import org.graphframes.GraphFrame +import org.graphframes.propertygraph.internal.AstBuilder +import org.graphframes.propertygraph.internal.GqlExplain +import org.graphframes.propertygraph.internal.JoinOptimizer +import org.graphframes.propertygraph.internal.QueryExecutor +import org.graphframes.propertygraph.internal.ResolvedQuery +import org.graphframes.propertygraph.internal.Resolver +import org.graphframes.propertygraph.internal.SchemaGraphSnapshot import org.graphframes.propertygraph.property.EdgePropertyGroup import org.graphframes.propertygraph.property.VertexPropertyGroup @@ -34,10 +41,181 @@ case class PropertyGraphFrame( vertexPropertyGroups: Seq[VertexPropertyGroup], edgesPropertyGroups: Seq[EdgePropertyGroup]) { import PropertyGraphFrame._ - 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 + + // Keys are lowercased so that lookups in toGraphFrame and projectionBy are case-insensitive. + // It is an overall policy across all the LPG functionality. + 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 + + lazy private[propertygraph] val schemaGraphSnapshot: SchemaGraphSnapshot = + SchemaGraphSnapshot.fromPropertyGraphFrame(this) + + /** + * Returns a human-readable description of the property graph schema. + * + * The output lists all vertex property groups and edge property groups with their + * source/destination connections, sorted alphabetically for determinism. + * + * @return + * a multi-line string describing the graph schema + */ + def schemaString: String = SchemaGraphSnapshot.toString(schemaGraphSnapshot) + + /** + * Returns the property graph schema in DOT (Graphviz) format. + * + * The output is a valid `digraph` that can be rendered by Graphviz tools. Vertex property + * groups appear as nodes and edge property groups appear as directed edges labeled with the + * group name. + * + * @return + * a DOT-format string representing the graph schema + */ + def schemaStringDOT: String = SchemaGraphSnapshot.toDOT(schemaGraphSnapshot) + + /** + * Executes a GQL `MATCH` query against this property graph and returns the matched paths as a + * Spark DataFrame with a fixed output schema: + * - `start_id`, `start_property_group`, `end_id`, `end_property_group`, + * `edge_property_group`, and a + * `path: array>` column for + * intermediate hops. + * + * This is a convenience overload equivalent to `query(gql, QueryOptions())`. + * + * @param gql + * a GQL `MATCH` statement in the supported subset. + * @return + * a DataFrame over the fixed output schema. + */ + def query(gql: String): DataFrame = query(gql, QueryOptions()) + + /** + * Executes a GQL `MATCH` query against this property graph and returns the matched paths as a + * Spark DataFrame with a fixed output schema: + * - `start_id`, `start_property_group`, `end_id`, `end_property_group`, + * `edge_property_group`, and a + * `path: array>` column for + * intermediate hops. + * + * The query is compiled through: ANTLR parse -> AST -> schema resolution -> join planning -> + * DataFrame execution (per-path `UNION ALL`). Disconnected patterns (no schema path matches) + * return an empty DataFrame without touching data. Bad syntax throws + * [[org.graphframes.InvalidParseException]]; unknown labels throw + * [[org.graphframes.InvalidPropertyGroupException]]. + * + * @param gql + * a GQL `MATCH` statement in the supported subset. + * @param options + * query options + * @return + * a DataFrame over the fixed output schema. + */ + def query(gql: String, options: QueryOptions): DataFrame = { + val resolved = + resolve(gql, options, enforceMaxSchemaPathLength = true, enforceMaxPathsCount = true) + if (resolved.paths.isEmpty) { + return QueryExecutor.execute(this, Seq.empty) + } + + // Cost-based optimization and statistics will follow + val _ = options.enableStatistics // placeholder + val plans = JoinOptimizer.plan(resolved, stats = None) + QueryExecutor.execute(this, plans) + } + + /** + * Renders the logical (resolved) plan of `gql` without executing it. + * + * This is a convenience overload equivalent to `fexplain(gql, ExplainMode.Logical)`. To see the + * per-path join plans (order + statistics basis). + * + * @param gql + * a GQL `MATCH` statement in the supported subset. + * @return + * a string describing the resolved (logical) plan. + */ + def explain(gql: String): String = explain(gql, ExplainMode.Logical) + + /** + * Renders a plan of `gql` without executing it, using default query options. + * + * This is a convenience overload equivalent to `explain(gql, mode, QueryOptions())`. Pass + * [[ExplainMode.Physical]] to see the per-path join plans (order + statistics basis); + * [[ExplainMode.Logical]] shows the resolved (logical) plan. + * + * @param gql + * a GQL `MATCH` statement in the supported subset. + * @param mode + * the explain mode: [[ExplainMode.Logical]] for the resolved plan or [[ExplainMode.Physical]] + * for the per-path join plans. + * @return + * a string describing the requested plan. + */ + def explain(gql: String, mode: ExplainMode): String = explain(gql, mode, QueryOptions()) + + /** + * Renders a plan of `gql` without executing it. + * + * Pass [[ExplainMode.Physical]] to see the per-path join plans (order + statistics basis); + * [[ExplainMode.Logical]] shows the resolved (logical) plan. + * + * @param gql + * a GQL `MATCH` statement in the supported subset. + * @param mode + * the explain mode: [[ExplainMode.Logical]] for the resolved plan or [[ExplainMode.Physical]] + * for the per-path join plans. + * @param options + * query options. + * @return + * a string describing the requested plan. + */ + def explain(gql: String, mode: ExplainMode, options: QueryOptions): String = { + // users should be able to see paths that exceed maxSchemaPathLength + // even if it is not allowed for real queries. + val resolved = + resolve(gql, options, enforceMaxSchemaPathLength = false, enforceMaxPathsCount = false) + mode match { + case ExplainMode.Logical => GqlExplain.logical(resolved) + case ExplainMode.Physical => + val plans = JoinOptimizer.plan(resolved, stats = None) + GqlExplain.physical(plans) + } + } + + /** + * Shared parse + resolve step for [[query]] and [[explain]]. Applies `maxSchemaPathLength` as a + * guard against pathological enumeration depth when [[enforceMaxSchemaPathLength]] is true (the + * [[query]] path). The [[explain]] path sets it to false so users can inspect the plan that + * exceeds the cap and understand why [[query]] rejects it. + */ + private def resolve( + gql: String, + options: QueryOptions, + enforceMaxSchemaPathLength: Boolean, + enforceMaxPathsCount: Boolean): ResolvedQuery = { + require( + options.maxSchemaPathLength > 0, + s"maxSchemaPathLength must be positive, got ${options.maxSchemaPathLength}") + val ast = AstBuilder.parse(gql) + val resolved = Resolver.resolve(ast, schemaGraphSnapshot, options) + if (enforceMaxSchemaPathLength) { + resolved.paths.foreach { path => + require( + path.length <= options.maxSchemaPathLength, + s"Schema path length ${path.length} exceeds maxSchemaPathLength=${options.maxSchemaPathLength}: " + + s"$path; try to rewrite the query and reduce a potential depth. Use `explain` to see the plan.") + } + } + if (enforceMaxPathsCount) { + require( + resolved.paths.size <= options.maxEnumeratedPaths, + s"An amount of paths in the resolved query exceeds ${options.maxEnumeratedPaths}: " + "either use `explain` and modify the pattern or increase the value in `QueryOptions`") + } + resolved + } /** * Converts a heterogeneous property graph into a unified GraphFrame representation. @@ -73,16 +251,18 @@ case class PropertyGraphFrame( edgeGroupFilters: Map[String, Column], vertexGroupFilters: Map[String, Column]): GraphFrame = { vertexPropertyGroups.foreach(name => - require(vertexGroups.contains(name), s"Vertex property group $name does not exist")) + require( + vertexGroups.contains(name.toLowerCase), + s"Vertex property group $name does not exist")) edgePropertyGroups.foreach(name => - require(edgeGroups.contains(name), s"Edge property group $name does not exist")) + require(edgeGroups.contains(name.toLowerCase), s"Edge property group $name does not exist")) val vertices = vertexPropertyGroups - .map(name => vertexGroups(name).getData(vertexGroupFilters(name))) + .map(name => vertexGroups(name.toLowerCase).getData(vertexGroupFilters(name))) .reduce(_ union _) val edges = edgePropertyGroups - .map(name => edgeGroups(name).getData(edgeGroupFilters(name))) + .map(name => edgeGroups(name.toLowerCase).getData(edgeGroupFilters(name))) .reduce(_ union _) GraphFrame(vertices, edges) @@ -111,15 +291,18 @@ case class PropertyGraphFrame( rightBiGraphPart: String, edgeGroup: String, newEdgeWeight: Option[(Column, Column) => Column] = None): PropertyGraphFrame = { + // Hoisted before the require checks so the lowercased lookup is performed only once. + val oldGroup = edgeGroups(edgeGroup.toLowerCase) require( - edgeGroups(edgeGroup).srcPropertyGroup.name == leftBiGraphPart, - s"Edge Property Group should have $leftBiGraphPart source group but has ${edgeGroups(edgeGroup).srcPropertyGroup.name}") + oldGroup.srcPropertyGroup.name.equalsIgnoreCase(leftBiGraphPart), + s"Edge Property Group should have $leftBiGraphPart source group but has ${oldGroup.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 oldGroup = edgeGroups(edgeGroup) + oldGroup.dstPropertyGroup.name.equalsIgnoreCase(rightBiGraphPart), + s"Edge Property Group should have $rightBiGraphPart destination group but has ${oldGroup.dstPropertyGroup.name}") + val keptVPropertyGroups = + vertexPropertyGroups.filterNot(g => g.name.equalsIgnoreCase(rightBiGraphPart)) + val keptEPropertyGroups = + edgesPropertyGroups.filterNot(g => g.name.equalsIgnoreCase(edgeGroup)) val oldEdgesData = oldGroup.data // Create new edges by joining vertices through their common neighbors @@ -141,8 +324,8 @@ case class PropertyGraphFrame( val newEdgeGroup = EdgePropertyGroup( name = s"projected_$edgeGroup", data = projectedEdges, - srcPropertyGroup = vertexGroups(leftBiGraphPart), - dstPropertyGroup = vertexGroups(leftBiGraphPart), + srcPropertyGroup = vertexGroups(leftBiGraphPart.toLowerCase), + dstPropertyGroup = vertexGroups(leftBiGraphPart.toLowerCase), isDirected = false, srcColumnName = GraphFrame.SRC, dstColumnName = GraphFrame.DST, diff --git a/core/src/main/scala/org/graphframes/propertygraph/QueryOptions.scala b/core/src/main/scala/org/graphframes/propertygraph/QueryOptions.scala new file mode 100644 index 000000000..8a8707aa5 --- /dev/null +++ b/core/src/main/scala/org/graphframes/propertygraph/QueryOptions.scala @@ -0,0 +1,88 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.graphframes.propertygraph + +/** + * Options for query resolution and optimization. + * + * @param enableStatistics + * whether the optimizer may consume statistics for join ordering (default `true`). + * @param maxSchemaPathLength + * cap on schema-path enumeration depth, to bound the fan-out of untyped/ambiguous patterns + * (default `10`). + * @param maxVarLength + * maximum length of a variable-length pattern (`[e*1..N]`), controlling how many hops a + * repeating edge pattern may expand to (default `5`). + * @param maxEnumeratedPaths + * maximal number of paths in the schema-graph that will be processed; each path results in one + * Spark SQL execution plan and all of them are unioned at the end (default `32`). + */ +final case class QueryOptions( + enableStatistics: Boolean = true, + maxSchemaPathLength: Int = 10, + maxVarLength: Int = 5, + maxEnumeratedPaths: Int = 32) + +object QueryOptions { + + /** + * Creates a new [[QueryOptions]] instance with all fields set to their default values. + * + * This is a convenience factory method intended primarily for Java and Py4J callers, who cannot + * use Scala's default argument syntax directly. Scala users should prefer the default + * constructor, e.g. `QueryOptions()`. + * + * @return + * a fresh [[QueryOptions]] with + * - `enableStatistics` = `true` + * - `maxSchemaPathLength` = `10` + * - `maxVarLength` = `5` + * - `maxEnumeratedPaths` = `32` + */ + def withDefualts: QueryOptions = QueryOptions() + + /** + * Creates a new [[QueryOptions]] instance with the specified + * [[QueryOptions.maxSchemaPathLength maxSchemaPathLength]] value, leaving all other fields at + * their defaults. + * + * This is a convenience factory method intended primarily for Java and Py4J callers who cannot + * use Scala named-argument syntax directly. Scala users should prefer + * `QueryOptions(maxSchemaPathLength = n)`. + * + * @param maxSchemaPathLength + * the maximum schema-path enumeration depth to use; see [[QueryOptions.maxSchemaPathLength]] + * for the effect of this setting. Must be non-negative. + * @return + * a new [[QueryOptions]] with + * - `enableStatistics` = `true` + * - `maxSchemaPathLength` = the supplied value + * - `maxVarLength` = `5` + * - `maxEnumeratedPaths` = `32` + */ + def withMaxSchemaPathLength(maxSchemaPathLength: Int): QueryOptions = + QueryOptions(maxSchemaPathLength = maxSchemaPathLength) +} + +/** Selects which plan to render via explain. */ +sealed trait ExplainMode + +object ExplainMode { + case object Logical extends ExplainMode + case object Physical extends ExplainMode +} diff --git a/core/src/main/scala/org/graphframes/propertygraph/internal/GqlExplain.scala b/core/src/main/scala/org/graphframes/propertygraph/internal/GqlExplain.scala new file mode 100644 index 000000000..940a7a3dc --- /dev/null +++ b/core/src/main/scala/org/graphframes/propertygraph/internal/GqlExplain.scala @@ -0,0 +1,176 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.graphframes.propertygraph.internal + +/** + * Read-only renderers over the two intermediate IR values: + * - [[logical]] renders a [[ResolvedQuery]] (the resolved schema paths, WHERE classification, + * and projection); + * - [[physical]] renders a `Seq[JoinPlan]` (per plan: the path, element-level join order, the + * statistics that drove it, and the predicates). + * + * Both are pure JVM; neither touches Spark. + */ +private[propertygraph] object GqlExplain { + + /** Render the logical (resolved) plan. */ + def logical(query: ResolvedQuery): String = { + val b = Vector.newBuilder[String] + b += "Logical plan (resolved):" + + if (query.paths.isEmpty) { + b += " schema paths: (none -- pattern is disconnected in the schema graph)" + } else { + b += s" schema paths (${query.paths.size}):" + query.paths.zipWithIndex.foreach { case (p, idx) => + b += s" [$idx] ${renderPath(p)}" + } + } + + b += s" join predicates (${query.joinPredicates.size}):" + query.joinPredicates.foreach(e => b += s" - ${renderExpr(e)}") + b += s" post-join filters (${query.postFilters.size}):" + query.postFilters.foreach(e => b += s" - ${renderExpr(e)}") + b += s" projection: ${renderProjection(query.projection)}" + + b.result().mkString("\n") + } + + /** Render the physical plan (one block per JoinPlan). */ + def physical(plans: Seq[JoinPlan]): String = { + val b = Vector.newBuilder[String] + b += s"Physical plan (${plans.size} join plan(s)):" + if (plans.isEmpty) { + b += " (none -- pattern is disconnected in the schema graph; no Spark execution)" + } else { + plans.zipWithIndex.foreach { case (plan, idx) => + b += s" Plan $idx:" + b += s" path: ${renderPath(plan.path)}" + b += s" join order: ${renderOrder(plan.order)}" + b += s" statistics: ${renderStats(plan.statsUsed)}" + b += s" join predicates (${plan.joinPredicates.size}):" + plan.joinPredicates.foreach(e => b += s" - ${renderExpr(e)}") + b += s" post-join filters (${plan.postFilters.size}):" + plan.postFilters.foreach(e => b += s" - ${renderExpr(e)}") + b += s" projection: ${renderProjection(plan.projection)}" + } + } + b.result().mkString("\n") + } + + // --------------------------------------------------------------------- + // Renderers. + // --------------------------------------------------------------------- + + private def renderPath(path: SchemaPath): String = { + // (a:Person)-[e1:KNOWS]->(b:Person)<-[e2:LIKES]-(c:Person) + val sb = new StringBuilder + path.nodes.zipWithIndex.foreach { case (node, i) => + sb.append(renderNode(node)) + if (i < path.steps.length) { + sb.append(renderStep(path.steps(i))) + } + } + sb.toString + } + + private def renderNode(node: PathNode): String = { + val v = node.variable.getOrElse("_") + val filter = + if (node.scanFilter.isEmpty) "" + else node.scanFilter.map(renderExpr).mkString("{", " AND ", "}") + s"($v:${node.vertexGroupName}$filter)" + } + + private def renderStep(step: PathStep): String = { + val v = step.variable.map(n => s"$n:").getOrElse("") + val body = s"[$v${step.edge.edgeGroupName}]" + // Forward step: -(body)-> ; backward step: <-(body)- + if (step.traversedForward) s"-$body->" else s"<-$body-" + } + + private def renderOrder(order: Vector[PathElementRef]): String = + order + .map { + case NodeRef(i) => s"n$i" + case EdgeRef(i) => s"e$i" + } + .mkString("[", ", ", "]") + + private def renderStats(stats: Map[String, GroupStatistics]): String = + if (stats.isEmpty) "(no statistics)" + else + stats.toVector + .sortBy(_._1) + .map { case (k, v) => + s"$k=${v.rowCount.map(c => s"rows=$c").getOrElse("?")}" + } + .mkString("{", ", ", "}") + + 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)" + case Projection.Items(items) => + items + .map { it => + val core = renderExpr(it.expression) + it.alias match { + case Some(a) => s"$core AS $a" + case None => core + } + } + .mkString("items: [", ", ", "]") + } + + private def renderExpr(expr: Expression): String = expr match { + case Literal(value) => + value match { + case null => "NULL" + case s: String => s"'$s'" + case other => String.valueOf(other) + } + case Variable(name) => name + case PropertyAccess(variable, property) => s"$variable.$property" + case Comparison(left, op, right) => + s"(${renderExpr(left)} ${renderCompOp(op)} ${renderExpr(right)})" + case Arithmetic(left, op, right) => + s"(${renderExpr(left)} ${renderArithOp(op)} ${renderExpr(right)})" + case Not(e) => s"(NOT ${renderExpr(e)})" + case And(left, right) => s"(${renderExpr(left)} AND ${renderExpr(right)})" + case Or(left, right) => s"(${renderExpr(left)} OR ${renderExpr(right)})" + case FunctionCall(name, args) => + s"$name(${args.map(renderExpr).mkString(", ")})" + } + + private def renderCompOp(op: CompOp): String = op match { + case Eq => "=" + case Neq => "<>" + case Lt => "<" + case Lte => "<=" + case Gt => ">" + case Gte => ">=" + } + + private def renderArithOp(op: ArithOp): String = op match { + case Plus => "+" + case Minus => "-" + case Mult => "*" + case Div => "/" + case Mod => "%" + } +} diff --git a/core/src/main/scala/org/graphframes/propertygraph/internal/GraphStatistics.scala b/core/src/main/scala/org/graphframes/propertygraph/internal/GraphStatistics.scala new file mode 100644 index 000000000..8771d0f5f --- /dev/null +++ b/core/src/main/scala/org/graphframes/propertygraph/internal/GraphStatistics.scala @@ -0,0 +1,95 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.graphframes.propertygraph.internal + +/** + * Per-column statistics for a property group. All fields optional and additive: future statistics + * providers can fill more of them without changing callers or the path/plan types. + * + * @param distinctCount + * estimated number of distinct values, if known. + * @param nullCount + * number of nulls, if known. + * @param min + * minimum value, if known. + * @param max + * maximum value, if known. + */ +private[propertygraph] final case class ColumnStatistics( + distinctCount: Option[Long] = None, + nullCount: Option[Long] = None, + min: Option[Any] = None, + max: Option[Any] = None) + +/** + * Per-group statistics (a vertex property group or an edge property group). + * + * @param rowCount + * number of rows in the group, if known. + * @param sizeInBytes + * estimated on-disk / in-memory size, if known. + * @param columns + * per-column stats, keyed by column name. + */ +private[propertygraph] final case class GroupStatistics( + rowCount: Option[Long] = None, + sizeInBytes: Option[Long] = None, + columns: Map[String, ColumnStatistics] = Map.empty) + +/** + * Statistics source SPI. The optimizer queries it by group name; the *source* is pluggable so + * that future implementations (Parquet footer min/max, managed-table CBO, …) can be swapped in + * with no change to callers or to the path/plan types. + */ +private[propertygraph] trait GraphStatistics { + + /** Statistics for the named vertex property group; empty stats if unknown. */ + def vertexGroup(name: String): GroupStatistics + + /** Statistics for the named edge property group; empty stats if unknown. */ + def edgeGroup(name: String): GroupStatistics +} + +private[propertygraph] object GraphStatistics { + + /** A no-op provider returning empty stats for every group. */ + val Empty: GraphStatistics = new GraphStatistics { + override def vertexGroup(name: String): GroupStatistics = GroupStatistics() + override def edgeGroup(name: String): GroupStatistics = GroupStatistics() + } + + /** + * Build a provider that caches `rowCount` per group on first access by calling `df.count()`. + * The cache is built lazily and shared across `vertexGroup`/`edgeGroup` lookups. All + * non-rowCount fields stay empty. + */ + def cachedRowCount( + vertexRowCounts: Map[String, Long], + edgeRowCounts: Map[String, Long]): GraphStatistics = new GraphStatistics { + override def vertexGroup(name: String): GroupStatistics = + vertexRowCounts + .get(name) + .map(c => GroupStatistics(rowCount = Some(c))) + .getOrElse(GroupStatistics()) + override def edgeGroup(name: String): GroupStatistics = + edgeRowCounts + .get(name) + .map(c => GroupStatistics(rowCount = Some(c))) + .getOrElse(GroupStatistics()) + } +} diff --git a/core/src/main/scala/org/graphframes/propertygraph/internal/JoinOptimizer.scala b/core/src/main/scala/org/graphframes/propertygraph/internal/JoinOptimizer.scala new file mode 100644 index 000000000..30b2e350d --- /dev/null +++ b/core/src/main/scala/org/graphframes/propertygraph/internal/JoinOptimizer.scala @@ -0,0 +1,80 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.graphframes.propertygraph.internal + +/** + * Turns a [[ResolvedQuery]] into a sequence of [[JoinPlan]]s (the physical plan). This is the + * optimization boundary: it is the only place where join order is decided and where statistics + * are (optionally) consumed. + * + * Disconnected patterns (`query.paths.isEmpty`) yield no plans; the executor then produces an + * empty result DataFrame. + */ +private[propertygraph] object JoinOptimizer { + + /** + * Default entry point. + */ + def plan(query: ResolvedQuery, stats: Option[GraphStatistics]): Seq[JoinPlan] = + defaultPlanner(query, stats) + + /** + * Planner SPI: `ResolvedQuery x Option[GraphStatistics] => Seq[JoinPlan]`. + */ + type Planner = (ResolvedQuery, Option[GraphStatistics]) => Seq[JoinPlan] + + /** + * Refinement SPI applied after the default planner: `(ResolvedQuery, Seq[JoinPlan]) => + * Seq[JoinPlan]`. Lets handwritten rules reorder / prune plans without replacing the planner. + */ + type PlanRefiner = (ResolvedQuery, Seq[JoinPlan]) => Seq[JoinPlan] + + /** planner: pattern order, no statistics consumption. */ + val defaultPlanner: Planner = (query, _) => patternOrderPlans(query) + + /** refiner: identity (no-op). */ + val identityRefiner: PlanRefiner = (_, plans) => plans + + /** + * Build one `JoinPlan` per path, joining elements in the order they were written (`n0, e0, n1, + * e1, …, n_{k}`). `statsUsed` is empty; predicates/projection are carried through from the + * resolved query. + */ + private def patternOrderPlans(query: ResolvedQuery): Seq[JoinPlan] = + query.paths.map { path => + val order = patternOrder(path) + JoinPlan( + path = path, + order = order, + statsUsed = Map.empty, + joinPredicates = query.joinPredicates, + postFilters = query.postFilters, + projection = query.projection) + } + + /** `n0, e0, n1, e1, …, n_{k}` for a path with `k` steps (`k+1` nodes). */ + private[propertygraph] def patternOrder(path: SchemaPath): Vector[PathElementRef] = { + val b = Vector.newBuilder[PathElementRef] + b += NodeRef(0) + path.steps.indices.foreach { i => + b += EdgeRef(i) + b += NodeRef(i + 1) + } + b.result() + } +} diff --git a/core/src/main/scala/org/graphframes/propertygraph/internal/JoinPlan.scala b/core/src/main/scala/org/graphframes/propertygraph/internal/JoinPlan.scala new file mode 100644 index 000000000..26f413dd4 --- /dev/null +++ b/core/src/main/scala/org/graphframes/propertygraph/internal/JoinPlan.scala @@ -0,0 +1,74 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.graphframes.propertygraph.internal + +/** + * A reference to one element of a [[SchemaPath]] within a [[JoinPlan]]'s join order. The order is + * expressed at the granularity of individual nodes and edges (not step indices). + */ +private[propertygraph] sealed trait PathElementRef +private[propertygraph] final case class NodeRef(index: Int) extends PathElementRef { + require(index >= 0) +} +private[propertygraph] final case class EdgeRef(index: Int) extends PathElementRef { + require(index >= 0) +} + +/** + * The physical plan for one [[SchemaPath]]. Self-contained: it carries everything the executor + * needs, so `explain(physical)` can render path + order + the statistics that drove it without + * re-running resolution. + * + * @param path + * the resolved schema path this plan executes (topology + per-node scan filters + directions). + * @param order + * element-level join order: a sequence of [[NodeRef]] / [[EdgeRef]] into `path.nodes` / + * `path.steps`. The executor scans and joins in this order. + * @param statsUsed + * the per-group statistics that drove `order`. + * @param joinPredicates + * WHERE conjuncts spanning exactly two adjacent node variables; applied as join conditions. + * @param postFilters + * WHERE conjuncts spanning 3+ variables / non-adjacent / any edge variable; applied after the + * join tree. + * @param projection + * the RETURN shape (Default / Star / Items). + */ +private[propertygraph] final case class JoinPlan( + path: SchemaPath, + order: Vector[PathElementRef], + statsUsed: Map[String, GroupStatistics], + joinPredicates: Seq[Expression], + postFilters: Seq[Expression], + projection: Projection) { + + override def toString: String = { + val orderStr = order + .map { + case NodeRef(i) => s"n$i" + case EdgeRef(i) => s"e$i" + } + .mkString("[", ", ", "]") + val projStr = projection match { + case Projection.Default => "DEFAULT" + case Projection.Star => "*" + case Projection.Items(items) => items.mkString("Items(", ", ", ")") + } + s"JoinPlan(path=$path, order=$orderStr, projection=$projStr)" + } +} diff --git a/core/src/main/scala/org/graphframes/propertygraph/internal/QueryExecutor.scala b/core/src/main/scala/org/graphframes/propertygraph/internal/QueryExecutor.scala new file mode 100644 index 000000000..bdbddd98b --- /dev/null +++ b/core/src/main/scala/org/graphframes/propertygraph/internal/QueryExecutor.scala @@ -0,0 +1,712 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.graphframes.propertygraph.internal + +import org.apache.spark.sql.Column +import org.apache.spark.sql.DataFrame +import org.apache.spark.sql.SparkSession +import org.apache.spark.sql.functions.array +import org.apache.spark.sql.functions.col +import org.apache.spark.sql.functions.lit +import org.apache.spark.sql.functions.struct +import org.apache.spark.sql.types.ArrayType +import org.apache.spark.sql.types.StringType +import org.apache.spark.sql.types.StructField +import org.apache.spark.sql.types.StructType +import org.graphframes.GraphFrame +import org.graphframes.propertygraph.PropertyGraphFrame + +/** + * Executes a sequence of [[JoinPlan]]s against a [[PropertyGraphFrame]] and returns a single + * result DataFrame. + * + * Per plan: build a join tree following `plan.order` (each element scanned once and joined to the + * growing frame on the masked id columns), apply join/post predicates, project to the output + * schema. Across plans: `UNION ALL`. Reuses `PropertyGroup.getData(filter, requestedProperties)`, + * which pushes the scan-local filter, applies id masking, and projects standardized columns -- + * edge `src`/`dst` are masked the same way as vertex `id`, so joins line up with no manual + * casting. + */ +private[propertygraph] object QueryExecutor { + + // ------------------------------------------------------------------------- + // Output column-name constants. Package-private so test suites can reference + // them when asserting against the query result schema/rows instead of + // repeating the raw string literals. + // ------------------------------------------------------------------------- + private[propertygraph] val EDGE_PROPERTY_GROUP: String = "edge_property_group" + private[propertygraph] val NODE_ID: String = "node_id" + private[propertygraph] val NODE_PROPERTY_GROUP: String = "node_property_group" + private[propertygraph] val START_ID: String = "start_id" + private[propertygraph] val START_PROPERTY_GROUP: String = "start_property_group" + private[propertygraph] val END_ID: String = "end_id" + private[propertygraph] val END_PROPERTY_GROUP: String = "end_property_group" + private[propertygraph] val PATH: String = "path" + + /** + * Execute all plans and `UNION ALL` them. + * + * Empty plans (disconnected pattern) -> an empty DataFrame with the fixed output schema (see + * [[outputSchema]]). Empty result of a single plan is also a valid empty DataFrame. + * + * Scan sharing: a per-call memo (`scanMemo`) de-duplicates scans across the per-path fan-out. + * It is keyed by a canonical scan signature ([[ScanKey]]) so every plan that references the + * same group with the same scan-local filter and the same carried-column set pulls the *same* + * `DataFrame` reference. This is the "floor" of scan reuse: never construct the same scan + * twice. + */ + def execute(pgf: PropertyGraphFrame, plans: Seq[JoinPlan]): DataFrame = { + if (plans.isEmpty) { + val spark = pgf.vertexPropertyGroups.headOption + .map(_.data.sparkSession) + .orElse(pgf.edgesPropertyGroups.headOption.map(_.data.sparkSession)) + .orElse(SparkSession.getActiveSession) + .getOrElse( + throw new IllegalStateException( + "No active SparkSession and no property groups to derive one from; " + + "PropertyGraphFrame.query must be called with an active session")) + return spark.createDataFrame( + spark.sparkContext.emptyRDD[org.apache.spark.sql.Row], + outputSchema) + } + val scanMemo = scala.collection.mutable.Map.empty[ScanKey, DataFrame] + val perPlan = plans.map(executePlan(pgf, _, scanMemo)) + perPlan.reduce(_ unionByName _) + } + + /** + * Test-only (debug) variant of [[execute]] that also returns the per-call scan memo, so tests + * can assert the scan-reuse "floor" (a scan is never constructed twice for an equal + * [[ScanKey]]) by reference-identity on the memo values. Package-private; not part of any + * public contract. + */ + private[propertygraph] def executeWithScanMemo( + pgf: PropertyGraphFrame, + plans: Seq[JoinPlan]): (DataFrame, Map[ScanKey, DataFrame]) = { + val scanMemo = scala.collection.mutable.Map.empty[ScanKey, DataFrame] + val result = + if (plans.isEmpty) execute(pgf, plans) + else { + val perPlan = plans.map(executePlan(pgf, _, scanMemo)) + perPlan.reduce(_ unionByName _) + } + (result, scanMemo.toMap) + } + + /** + * Canonical signature of a scan. Two scans with equal keys share one `DataFrame` in the memo. + * `Expression` AST nodes (including the scan-local filters) are case classes with structural + * equality, so they key correctly without extra canonicalization. `carriedCols` is a `Set`, so + * column order does not affect sharing. + */ + private[propertygraph] final case class ScanKey( + groupName: String, + scanFilter: Seq[Expression], + carriedCols: Set[String]) + + // ------------------------------------------------------------------------- + // Per-plan execution. + // ------------------------------------------------------------------------- + + private def executePlan( + pgf: PropertyGraphFrame, + plan: JoinPlan, + scanMemo: scala.collection.mutable.Map[ScanKey, DataFrame]): DataFrame = { + val path = plan.path + val env = PrefixEnv(path) + + // Classify, per element, which property columns are CARRIED through the scan (predicate / + // filter-also-returned columns that influence join cardinality) vs which are OUTPUT-ONLY + // (referenced solely by RETURN, deferred to a terminal join-back). + val nodeProps: Map[Int, ElementProps] = classifyElementProps(path, plan, node = true) + val edgeProps: Map[Int, ElementProps] = classifyElementProps(path, plan, node = false) + + // Scan each element once (shared via the memo below the rename) and alias its columns under the + // element's prefix. The expensive shared node sits below a thin per-use Project (the rename). + val nodeFrames: Map[Int, DataFrame] = path.nodes.indices.map { i => + val shared = + sharedScanNode(pgf, path, i, nodeProps.getOrElse(i, ElementProps.Empty), scanMemo) + i -> renameAll(shared, env.nodePrefix(i)) + }.toMap + val edgeFrames: Map[Int, DataFrame] = path.steps.indices.map { i => + val shared = + sharedScanEdge(pgf, path, i, edgeProps.getOrElse(i, ElementProps.Empty), scanMemo) + i -> renameAll(shared, env.edgePrefix(i)) + }.toMap + + // Walk the join order, joining each element onto the growing frame, placing every multi-variable + // predicate at the earliest element where all its operands are bound (replacing a blanket + // post-tree `.where`). + val allPredicates: Seq[Expression] = plan.joinPredicates ++ plan.postFilters + val predicateVarSets: Seq[Set[String]] = allPredicates.map(GqlAst.referencedVariables) + val placed = scala.collection.mutable.BitSet.empty + var frame: DataFrame = null + var boundVars: Set[String] = Set.empty + def bindElement(i: Int, isNode: Boolean): Unit = { + if (isNode) path.nodes(i).variable.foreach(v => boundVars += v) + else path.steps(i).variable.foreach(v => boundVars += v) + } + plan.order.foreach { + case NodeRef(i) => + val f = nodeFrames(i) + bindElement(i, isNode = true) + if (frame == null) { + frame = f + } else { + val ready = readyPredicateIndices(predicateVarSets, placed, boundVars) + val readyExprs = ready.map(allPredicates) + frame = joinElement(frame, f, path, env, readyExprs) + ready.foreach(placed.add) + } + case EdgeRef(i) => + val f = edgeFrames(i) + bindElement(i, isNode = false) + if (frame == null) { + frame = f + } else { + val ready = readyPredicateIndices(predicateVarSets, placed, boundVars) + val readyExprs = ready.map(allPredicates) + frame = joinElement(frame, f, path, env, readyExprs) + ready.foreach(placed.add) + } + } + if (frame == null) { + // Degenerate: a path with a single node and no edges (MATCH (a:Person)). + // Synthesize a trivially-valid frame from that node scan; nodeFrames is non-empty here. + frame = nodeFrames(0) + } + // Place any predicates not consumed during the join walk (e.g. a predicate whose variables were + // all bound by the seed element, or a literal-only predicate) as a residual filter. + val leftover = allPredicates.indices.filterNot(placed.contains).map(allPredicates) + leftover.foreach { expr => + frame = frame.filter(ExpressionLowering.lower(expr, env)) + } + + // edgeProps is classified (and consumed by the shared edge scans above) but NOT join-backed + // edge groups may be undirected (doubling rows), so edge RETURN-properties are CARRIED rather + // than resolved by an id-keyed join-back. + project(frame, plan, nodeProps, pgf) + } + + /** Indices of predicates whose referenced variables are all bound and not yet placed. */ + private def readyPredicateIndices( + varSets: Seq[Set[String]], + placed: scala.collection.mutable.BitSet, + boundVars: Set[String]): Seq[Int] = + varSets.indices.collect { + case k if !placed.contains(k) && varSets(k).subsetOf(boundVars) => k + } + + // ------------------------------------------------------------------------- + // Scans + aliasing. + // ------------------------------------------------------------------------- + + /** + * Return the shared (memoized), *un-prefixed* node scan for element `i`. The caller applies the + * per-element rename on top. `ElementProps.carriedToScan` is what the scan requests; + * output-only columns are NOT requested here (they are resolved by the terminal join-back in + * `project`). + */ + private def sharedScanNode( + pgf: PropertyGraphFrame, + path: SchemaPath, + i: Int, + props: ElementProps, + scanMemo: scala.collection.mutable.Map[ScanKey, DataFrame]): DataFrame = { + val node = path.nodes(i) + val key = ScanKey(node.vertexGroupName.toLowerCase, node.scanFilter, props.carriedToScan) + scanMemo.getOrElseUpdate( + key, { + val group = pgf.vertexGroups(node.vertexGroupName.toLowerCase) + val filterCol = lowerScanFilter(node.scanFilter) + group.getData(filterCol, props.carriedToScan.toSeq.sorted) + }) + } + + /** + * Return the shared (memoized), *un-prefixed* edge scan for element `i`. + */ + private def sharedScanEdge( + pgf: PropertyGraphFrame, + path: SchemaPath, + i: Int, + props: ElementProps, + scanMemo: scala.collection.mutable.Map[ScanKey, DataFrame]): DataFrame = { + val step = path.steps(i) + val key = ScanKey(step.edge.edgeGroupName.toLowerCase, step.scanFilter, props.carriedToScan) + scanMemo.getOrElseUpdate( + key, { + val group = pgf.edgeGroups(step.edge.edgeGroupName.toLowerCase) + val filterCol = lowerScanFilter(step.scanFilter) + group.getData(filterCol, props.carriedToScan.toSeq.sorted) + }) + } + + /** + * Lower the scan-local predicates of a node into a single ANDed Column (lit(true) if none). + * These reference the node's variable, but the resulting Column is applied by `getData` against + * the RAW group columns (before aliasing), so the variable resolves to the empty prefix (raw + * column names): e.g. `a.age > 30` -> `col("age") > 30`. + */ + private def lowerScanFilter(filters: Seq[Expression]): Column = + if (filters.isEmpty) lit(true) + else filters.map(ExpressionLowering.lower(_, PrefixEnv.raw)).reduce(_ && _) + + /** Prefix every column of `df` with `prefix_` in place. */ + private def renameAll(df: DataFrame, prefix: String): DataFrame = { + val mapping = df.columns.map(c => c -> s"${prefix}_$c").toMap + val out = df.withColumnsRenamed(mapping) + out + } + + // ------------------------------------------------------------------------- + // Join tree. + // ------------------------------------------------------------------------- + + /** + * Join `incoming` onto `frame`. The join condition is the conjunction of all adjacency edges + * between the just-added element and already-present neighbors. + * + * For a step `k` connecting `node_k` and `node_{k+1}` through `edge_k`: + * - forward (`traversedForward = true`): `edge_k.src == node_k.id` and + * `edge_k.dst == node_{k+1}.id`; + * - backward: `edge_k.dst == node_k.id` and `edge_k.src == node_{k+1}.id` (src/dst swapped). + */ + private def joinElement( + frame: DataFrame, + incoming: DataFrame, + path: SchemaPath, + env: PrefixEnv, + residualPredicates: Seq[Expression]): DataFrame = { + val presentCols = frame.columns.toSet + val conditions = adjacencyConditions(incoming.columns.toSet, presentCols, path, env) + require( + conditions.nonEmpty, + "Join order produced an element with no adjacency to the already-joined frame; " + + "this indicates an invalid join order" + s" for path ${path.toString()}") + // The residual predicates are multi-variable WHERE conjuncts whose operands are all bound at + // this join. + val residualCols = residualPredicates.map(ExpressionLowering.lower(_, env)) + val allConds = conditions ++ residualCols + frame.join(incoming, allConds.reduce(_ && _), "inner") + } + + /** + * Build the equi-join conditions between the newly-joined element's columns and the frame's + * columns. An edge `k` is adjacent to its two endpoint nodes (`k`, `k+1`); a node `j` is + * adjacent to the edges touching it (`j-1` and `j`). + */ + private def adjacencyConditions( + incoming: Set[String], + present: Set[String], + path: SchemaPath, + env: PrefixEnv): Seq[Column] = { + val conds = Seq.newBuilder[Column] + + def bothPresent(a: String, b: String): Option[Column] = + if (incoming.contains(a) && present.contains(b)) Some(col(a) === col(b)) + else if (incoming.contains(b) && present.contains(a)) Some(col(b) === col(a)) + else None + + // Edge k <-> node k and edge k <-> node k+1. + path.steps.indices.foreach { k => + val eSrc = env.edgeCol(k, GraphFrame.SRC) + val eDst = env.edgeCol(k, GraphFrame.DST) + val nKId = env.nodeCol(k, GraphFrame.ID) + val nK1Id = env.nodeCol(k + 1, GraphFrame.ID) + val forward = path.steps(k).traversedForward + // forward: edge.src == node_k.id , edge.dst == node_{k+1}.id + // backward: edge.dst == node_k.id , edge.src == node_{k+1}.id + val (srcNode, dstNode) = if (forward) (nKId, nK1Id) else (nK1Id, nKId) + bothPresent(eSrc, srcNode).foreach(conds += _) + bothPresent(eDst, dstNode).foreach(conds += _) + } + conds.result() + } + + /** The fixed output schema, regardless of pattern length or RETURN shape. */ + private[propertygraph] def outputSchema: StructType = { + val pathElement = StructType( + Seq( + StructField(EDGE_PROPERTY_GROUP, StringType, nullable = true), + StructField(NODE_ID, StringType, nullable = true), + StructField(NODE_PROPERTY_GROUP, StringType, nullable = true))) + StructType( + Seq( + StructField(START_ID, StringType, nullable = true), + StructField(START_PROPERTY_GROUP, StringType, nullable = true), + StructField(END_ID, StringType, nullable = true), + StructField(END_PROPERTY_GROUP, StringType, nullable = true), + StructField(EDGE_PROPERTY_GROUP, StringType, nullable = true), + StructField(PATH, ArrayType(pathElement), nullable = true))) + } + + private def project( + frame: DataFrame, + plan: JoinPlan, + nodeProps: Map[Int, ElementProps], + pgf: PropertyGraphFrame): DataFrame = { + val path = plan.path + val env = PrefixEnv(path) + plan.projection match { + case Projection.Items(items) => + // For any node element with output-only (RETURN-only) properties, join the masked id back to + // the group's properties so those columns are available for the RETURN projection. Carried + // columns (predicate / filter-also-returned) are already on `frame` under the element + // prefix; only output-only columns need the terminal join-back. Edges are NOT join-backed + // and edge RETURN-properties are carried instead. + // + // The joined-back property columns are aliased to the element's PREFIXED names so that + // `ExpressionLowering.lower(PropertyAccess(v, p))` -- which resolves to `_p` -- + // finds them. (Carried columns already live under those prefixed names; this puts the + // output-only ones under the same convention.) + var withOutput = frame + path.nodes.indices.foreach { i => + val props = nodeProps.getOrElse(i, ElementProps.Empty) + if (props.outputOnly.nonEmpty) { + val node = path.nodes(i) + val group = pgf.vertexGroups(node.vertexGroupName.toLowerCase) + val prefix = env.nodePrefix(i) + // Build a narrow join-back frame: the masked `id` (join key) plus the output-only + // property columns, renamed to the element's prefixed names so `ExpressionLowering` + // resolves `PropertyAccess(v, p)` -> `_p` against them. We drop + // `property_group` (the group name is a constant the projection emits itself) and avoid + // surfacing a second un-prefixed `id` column that would collide across multiple + // join-backs; the masked `id` is kept only as the join key and dropped afterward. + val carryCols = props.outputOnly.toSeq.sorted + // `_gjid_` is a throwaway join-key alias unique per element, so multiple join-backs + // never collide on a raw `id` column. + val joinKey = s"_gjid_$i" + val groupId = group + .getData(lit(true), carryCols) + .select((col(GraphFrame.ID).alias(joinKey) +: carryCols + .map(p => col(p).alias(env.join(prefix, p)))): _*) + val idCol = env.nodeCol(i, GraphFrame.ID) + withOutput = withOutput.join(groupId, withOutput(idCol) === groupId(joinKey), "left") + } + } + val cols = items.map { item => + val c = ExpressionLowering.lower(item.expression, env) + item.alias match { + case Some(a) => c.alias(a) + case None => + item.expression match { + case Variable(v) => c.alias(v) + case PropertyAccess(_, p) => c.alias(p) + case FunctionCall(name, _) => c.alias(name.toLowerCase) + case _ => c + } + } + } + withOutput.select(cols: _*) + + case Projection.Default | Projection.Star => + val namedIndices = path.nodes.indices.filter(i => path.nodes(i).variable.isDefined) + require( + namedIndices.nonEmpty, + "RETURN (default/*) requires at least one named node variable in the pattern;" + s"path: ${plan.toString()}") + val startIdx = namedIndices.head + val endIdx = namedIndices.last + + val startId = col(env.nodeCol(startIdx, GraphFrame.ID)) + val startPg = lit(path.nodes(startIdx).vertexGroupName) + val endId = col(env.nodeCol(endIdx, GraphFrame.ID)) + val endPg = lit(path.nodes(endIdx).vertexGroupName) + + if (path.steps.isEmpty) { + // 0-hop: a single node. No edge, no path array. + // edge_property_group is null; path is an empty array. + frame.select( + startId.alias(START_ID), + startPg.alias(START_PROPERTY_GROUP), + endId.alias(END_ID), + endPg.alias(END_PROPERTY_GROUP), + lit(null).cast(StringType).alias(EDGE_PROPERTY_GROUP), + array().cast(outputSchema(PATH).dataType).alias(PATH)) + } else { + val firstEdgeGroup = + lit(path.steps.head.edge.edgeGroupName).alias(EDGE_PROPERTY_GROUP) + // Intermediate hops: for step i (i in 1..k-1) we emit the edge group of step i and the + // intermediate node i. The final step k carries its edge group but null node fields (the + // end node is already in end_id). + val pathStructs = path.steps.indices.flatMap { i => + val edgeGroup = lit(path.steps(i).edge.edgeGroupName) + if (i == path.steps.length - 1) { + // Last step: edge group only, null node fields. + Seq( + struct( + edgeGroup.alias(EDGE_PROPERTY_GROUP), + lit(null).cast(StringType).alias(NODE_ID), + lit(null).cast(StringType).alias(NODE_PROPERTY_GROUP))) + } else { + Seq( + struct( + edgeGroup.alias(EDGE_PROPERTY_GROUP), + col(env.nodeCol(i + 1, GraphFrame.ID)).alias(NODE_ID), + lit(path.nodes(i + 1).vertexGroupName).alias(NODE_PROPERTY_GROUP))) + } + } + val pathCol = + if (pathStructs.isEmpty) array().cast(outputSchema(PATH).dataType) + else array(pathStructs: _*) + frame.select( + startId.alias(START_ID), + startPg.alias(START_PROPERTY_GROUP), + endId.alias(END_ID), + endPg.alias(END_PROPERTY_GROUP), + firstEdgeGroup, + pathCol.alias(PATH)) + } + } + } + + // ------------------------------------------------------------------------- + // Required-property classification (carry vs defer). + // + // Per scan-reuse: every required property is classified by the role that + // references it: + // - scan-local filter columns -> pushed into getData's filter, consumed before the shuffle; + // - join / post-filter columns -> CARRIED through the join tree to the predicate's evaluation + // point (they let the predicate cut the join's output cardinality); + // - output-only columns -> DEFERRED to a terminal join-back (they never reduce + // cardinality, so carrying them only widens shuffles and fragments scan signatures). + // A property referenced by BOTH a filter and RETURN is carried (the filter need dominates). + // ------------------------------------------------------------------------- + + /** Per-element property classification. */ + private[propertygraph] final case class ElementProps( + carriedToScan: Set[String], // requested at the scan; rides the join tree to its consumers + outputOnly: Set[String] + ) // RETURN-only; resolved by the terminal join-back, never carried + + private[propertygraph] object ElementProps { + val Empty: ElementProps = ElementProps(Set.empty, Set.empty) + } + + /** + * Classify the properties of every element of the requested kind (node or edge) into carried vs + * output-only. + * + * For an element bound to variable `v`: + * - `scanFilterProps(v)` = props referenced in this element's scan-local filters; + * - `joinPostProps(v)` = props of `v` referenced in join/post predicates; + * - `returnProps(v)` = props of `v` referenced in RETURN items (empty for Default/Star); + * - `carry(v)` = `joinPostProps(v) ∪ (returnProps(v) ∩ scanFilterProps(v))`; + * - `outputOnly(v)` = `returnProps(v) − carry(v)`. + */ + private def classifyElementProps( + path: SchemaPath, + plan: JoinPlan, + node: Boolean): Map[Int, ElementProps] = { + val nodeVarIndex: Map[String, Int] = path.nodes.zipWithIndex.collect { + case (n, i) if n.variable.isDefined => n.variable.get -> i + }.toMap + val edgeVarIndex: Map[String, Int] = path.steps.zipWithIndex.collect { + case (s, i) if s.variable.isDefined => s.variable.get -> i + }.toMap + val varIndex = if (node) nodeVarIndex else edgeVarIndex + + // Per-element scan-filter properties (these are applied at the scan; they do NOT by themselves + // earn a carry -- they are consumed before the shuffle). + val scanFilterExprs: Seq[Expression] = + if (node) path.nodes.flatMap(_.scanFilter) else Seq.empty + val scanFilterProps: Map[Int, Set[String]] = collectProps(scanFilterExprs, varIndex) + + // Join/post-predicate properties: these DO earn a carry (they ride to the predicate's point). + val joinPostProps: Map[Int, Set[String]] = + collectProps(plan.joinPredicates ++ plan.postFilters, varIndex) + + // RETURN properties (only for Items projections). + val returnProps: Map[Int, Set[String]] = plan.projection match { + case Projection.Items(items) => collectProps(items.map(_.expression), varIndex) + case _ => Map.empty + } + + val acc = + scala.collection.mutable.Map.empty[Int, ElementProps].withDefaultValue(ElementProps.Empty) + + // For edges we should actually carry all the returnProps + varIndex.values.foreach { idx => + val sf = scanFilterProps.getOrElse(idx, Set.empty) + val jp = joinPostProps.getOrElse(idx, Set.empty) + val rp = returnProps.getOrElse(idx, Set.empty) + val carry = if (node) jp ++ rp.intersect(sf) else jp ++ rp + // there is no join-back for edges and all the rp are already carried along the join-path + val outputOnly = if (node) rp -- carry else Set.empty[String] + acc(idx) = ElementProps(carry, outputOnly) + } + acc.toMap + } + + /** Gather (elementIndex -> propertyNames) from the PropertyAccess nodes in `exprs`. */ + private def collectProps( + exprs: Seq[Expression], + varIndex: Map[String, Int]): Map[Int, Set[String]] = { + val acc = scala.collection.mutable.Map.empty[Int, Set[String]].withDefaultValue(Set.empty) + exprs.foreach { e => + propertyAccesses(e).foreach { case (v, p) => + varIndex.get(v).foreach { i => acc(i) = acc(i) + p } + } + } + acc.toMap + } + + /** Flatten all `(variable, property)` accesses appearing anywhere in `expr`. */ + private def propertyAccesses(expr: Expression): Seq[(String, String)] = expr match { + case PropertyAccess(v, p) => Seq((v, p)) + case Comparison(l, _, r) => propertyAccesses(l) ++ propertyAccesses(r) + case Arithmetic(l, _, r) => propertyAccesses(l) ++ propertyAccesses(r) + case Not(e) => propertyAccesses(e) + case And(l, r) => propertyAccesses(l) ++ propertyAccesses(r) + case Or(l, r) => propertyAccesses(l) ++ propertyAccesses(r) + case FunctionCall(_, args) => args.flatMap(propertyAccesses) + case _ => Seq.empty + } +} + +// ----------------------------------------------------------------------------- +// Prefix environment: maps each element to a unique column prefix so that joins +// never collide on the standardized column names (id/property_group/src/dst/weight). +// ----------------------------------------------------------------------------- + +private[propertygraph] final class PrefixEnv private (path: SchemaPath, val raw: Boolean) { + + /** Prefix for node `i`: the variable if named, else `node`. Empty when `raw`. */ + def nodePrefix(i: Int): String = + if (raw) "" else path.nodes(i).variable.getOrElse(s"node$i") + + /** Prefix for edge `i`: the variable if named, else `edge`. Empty when `raw`. */ + def edgePrefix(i: Int): String = + if (raw) "" else path.steps(i).variable.getOrElse(s"edge$i") + + /** Fully-qualified column name for a node's output column (e.g. the `id`). */ + def nodeCol(i: Int, colName: String): String = join(nodePrefix(i), colName) + + /** Fully-qualified column name for an edge's output column (`src`/`dst`/`weight`). */ + def edgeCol(i: Int, colName: String): String = join(edgePrefix(i), colName) + + /** + * Join a prefix and a column name: `colName` when the prefix is empty, else `prefix_colName`. + */ + def join(prefix: String, colName: String): String = + if (prefix.isEmpty) colName else s"${prefix}_$colName" + + /** The prefix string for a variable, if that variable names a node or edge in this path. */ + def prefixFor(variable: String): Option[String] = { + if (raw) Some("") + else + path.nodes.zipWithIndex + .find(_._1.variable.contains(variable)) + .map { case (_, i) => nodePrefix(i) } + .orElse(path.steps.zipWithIndex.find(_._1.variable.contains(variable)).map { + case (_, i) => + edgePrefix(i) + }) + } +} + +private[propertygraph] object PrefixEnv { + + /** A real path environment. */ + def apply(path: SchemaPath): PrefixEnv = new PrefixEnv(path, raw = false) + + /** + * A raw (empty-prefix) environment: every variable resolves to the empty prefix, so lowered + * columns reference raw group column names. Used for scan-local filters, which are applied by + * `getData` against the raw group data before aliasing. + */ + val raw: PrefixEnv = new PrefixEnv(null, raw = true) +} + +// ----------------------------------------------------------------------------- +// Expression -> Column lowering. +// ----------------------------------------------------------------------------- + +private[propertygraph] object ExpressionLowering { + + /** + * Lower a GQL AST [[Expression]] to a Spark [[Column]] in the context of `env` (variable -> + * prefix mapping). + * + * - `Variable(v)` -> `col("_id")` (a bare variable reference resolves to the + * element's id column); + * - `PropertyAccess(v, p)` -> `col("_

")`; + * - `Literal(value)` -> a typed `lit` (Long/Double/Boolean/String/null); + * - comparison/arithmetic/boolean combinators recurse and map to the corresponding Spark + * operators. + */ + def lower(expr: Expression, env: PrefixEnv): Column = expr match { + case Literal(value) => litLiteral(value) + case Variable(name) => + val prefix = env + .prefixFor(name) + .getOrElse( + throw new IllegalArgumentException( + s"Variable '$name' is not bound to any element of the matched path")) + col(env.join(prefix, GraphFrame.ID)) + case PropertyAccess(variable, property) => + val prefix = env + .prefixFor(variable) + .getOrElse( + throw new IllegalArgumentException( + s"Variable '$variable' is not bound to any element of the matched path")) + col(env.join(prefix, property)) + case Comparison(left, op, right) => compOp(lower(left, env), op, lower(right, env)) + case Arithmetic(left, op, right) => arithOp(lower(left, env), op, lower(right, env)) + case Not(e) => !lower(e, env) + case And(left, right) => lower(left, env) && lower(right, env) + case Or(left, right) => lower(left, env) || lower(right, env) + case FunctionCall(name, args) => + FunctionRegistry.lower(name, args, args.map(lower(_, env))) + } + + private def compOp(l: Column, op: CompOp, r: Column): Column = op match { + case Eq => l === r + case Neq => l =!= r + case Lt => l < r + case Lte => l <= r + case Gt => l > r + case Gte => l >= r + } + + private def arithOp(l: Column, op: ArithOp, r: Column): Column = op match { + case Plus => l + r + case Minus => l - r + case Mult => l * r + case Div => l / r // NB: Spark `/` is floating-point division (5/2 = 2.5) + case Mod => l % r // NB: result follows the dividend's sign + } + + /** Narrow `Literal.value: Any` to the appropriate Spark-typed literal. */ + private def litLiteral(value: Any): Column = value match { + case null => lit(null) + case v: java.lang.Boolean => lit(v.booleanValue()) + case v: java.lang.Long => lit(v.longValue()) + case v: java.lang.Integer => lit(v.intValue()) + case v: java.lang.Double => lit(v.doubleValue()) + case v: java.lang.Float => lit(v.floatValue()) + case v: Long => lit(v) + case v: Int => lit(v) + case v: Double => lit(v) + case v: Float => lit(v) + case v: String => lit(v) + case other => + throw new IllegalArgumentException( + s"Unsupported literal value of type ${other.getClass.getName}: $other") + } +} diff --git a/core/src/main/scala/org/graphframes/propertygraph/property/EdgePropertyGroup.scala b/core/src/main/scala/org/graphframes/propertygraph/property/EdgePropertyGroup.scala index 4be1b338d..8410ba574 100644 --- a/core/src/main/scala/org/graphframes/propertygraph/property/EdgePropertyGroup.scala +++ b/core/src/main/scala/org/graphframes/propertygraph/property/EdgePropertyGroup.scala @@ -94,22 +94,33 @@ case class EdgePropertyGroup( col(dstColumnName).cast(StringType) } - override protected[graphframes] def getData(filter: Column): DataFrame = { + override private[propertygraph] def getData( + filter: Column, + requestedProperties: Seq[String]): DataFrame = { val filteredData = data.filter(filter) - val baseEdges = filteredData.select( + // Unknown requested property names are dropped silently; the query engine validates first. + val availableProperties = requestedProperties.filter(data.columns.contains) + + val baseCols = Seq( hashSrcEdge.alias(GraphFrame.SRC), hashDstEdge.alias(GraphFrame.DST), col(weightColumnName).alias(GraphFrame.WEIGHT)) + val baseEdges = filteredData.select(baseCols ++ availableProperties.map(col): _*) + if (isDirected) { baseEdges } else { + // For undirected edges, surface both orientations. The extra property columns are copied + // verbatim into both halves of the union (they describe the same row, just reoriented). + val propertyCols = availableProperties.map(c => col(c).alias(c)) baseEdges.union( baseEdges.select( - col(GraphFrame.DST).as(GraphFrame.SRC), - col(GraphFrame.SRC).as(GraphFrame.DST), - col(GraphFrame.WEIGHT).alias(GraphFrame.WEIGHT))) + (col(GraphFrame.DST).as(GraphFrame.SRC) + +: col(GraphFrame.SRC).as(GraphFrame.DST) + +: col(GraphFrame.WEIGHT).alias(GraphFrame.WEIGHT) + +: propertyCols): _*)) } } } diff --git a/core/src/main/scala/org/graphframes/propertygraph/property/PropertyGroup.scala b/core/src/main/scala/org/graphframes/propertygraph/property/PropertyGroup.scala index fd84e1c45..cec1b8278 100644 --- a/core/src/main/scala/org/graphframes/propertygraph/property/PropertyGroup.scala +++ b/core/src/main/scala/org/graphframes/propertygraph/property/PropertyGroup.scala @@ -15,16 +15,39 @@ trait PropertyGroup { * @return * A DataFrame containing the raw data. */ - protected[graphframes] def getData(): DataFrame = getData(lit(true)) + private[propertygraph] def getData(): DataFrame = getData(lit(true)) /** - * Returns a filtered view of the data for the property group, with an optional mask applied to - * IDs. + * Returns a filtered view of the data for the property group without requesting any extra + * property columns (only id/standardized columns are projected). Equivalent to + * `getData(filter, Seq.empty)`. * * @param filter * A condition (Column) used to filter the data. * @return * A DataFrame containing the filtered and optionally transformed data. */ - protected[graphframes] def getData(filter: Column): DataFrame + private[propertygraph] def getData(filter: Column): DataFrame = getData(filter, Seq.empty) + + /** + * Returns a filtered view of the data for the property group, with an optional mask applied to + * IDs, and additionally carrying the named property columns through to the output. + * + * The extra `requestedProperties` are useful for query engines that need to surface specific + * properties (e.g. a `RETURN a.age` projection) without re-reading the raw `data`. They are + * passed through unmodified — they are not join keys and are never masked. When + * `requestedProperties` is empty, the output is identical to `getData(filter)`. + * + * @param filter + * A condition (Column) used to filter the data. + * @param requestedProperties + * Names of additional property columns to carry through to the output. + * @return + * A DataFrame containing the filtered, optionally transformed data plus requested properties. + */ + private[propertygraph] def getData(filter: Column, requestedProperties: Seq[String]): DataFrame + + /** Convenience overload of [[getData(filter, requestedProperties)* ]] with no filter. */ + private[propertygraph] def getData(requestedProperties: Seq[String]): DataFrame = + getData(lit(true), requestedProperties) } diff --git a/core/src/main/scala/org/graphframes/propertygraph/property/VertexPropertyGroup.scala b/core/src/main/scala/org/graphframes/propertygraph/property/VertexPropertyGroup.scala index caff7cbfd..422081e34 100644 --- a/core/src/main/scala/org/graphframes/propertygraph/property/VertexPropertyGroup.scala +++ b/core/src/main/scala/org/graphframes/propertygraph/property/VertexPropertyGroup.scala @@ -57,23 +57,33 @@ case class VertexPropertyGroup( this } - private[graphframes] def internalIdMapping: DataFrame = data + private[propertygraph] def internalIdMapping: DataFrame = data .select(col(primaryKeyColumn).alias(PropertyGraphFrame.EXTERNAL_ID)) .withColumn( GraphFrame.ID, concat(lit(name), sha2(col(PropertyGraphFrame.EXTERNAL_ID).cast(StringType), 256))) - override protected[graphframes] def getData(filter: Column): DataFrame = { - val filteredData = data - .filter(filter) - val withId = if (applyMaskOnId) { - filteredData.select( - concat(lit(name), sha2(col(primaryKeyColumn).cast(StringType), 256)).alias(GraphFrame.ID)) + override private[propertygraph] def getData( + filter: Column, + requestedProperties: Seq[String]): DataFrame = { + val filteredData = data.filter(filter) + + // The masked/raw id column is always emitted first, regardless of requested properties. + val idCol = if (applyMaskOnId) { + concat(lit(name), sha2(col(primaryKeyColumn).cast(StringType), 256)) } else { - filteredData.select(col(primaryKeyColumn).cast(StringType).alias(GraphFrame.ID)) + col(primaryKeyColumn).cast(StringType) } - withId.select(col(GraphFrame.ID), lit(name).alias(PropertyGraphFrame.PROPERTY_GROUP_COL_NAME)) + // Requested properties are carried through unmodified (they are not join keys, never masked). + // Unknown requested property names are dropped silently -- the caller (the query engine) is + // expected to validate them against `data.columns` before requesting. + val availableProperties = requestedProperties.filter(data.columns.contains) + + val baseCols = + Seq(idCol.alias(GraphFrame.ID), lit(name).alias(PropertyGraphFrame.PROPERTY_GROUP_COL_NAME)) + + filteredData.select(baseCols ++ availableProperties.map(col): _*) } } diff --git a/core/src/test/scala/org/graphframes/propertygraph/PropertyGraphFrameCaseInsensitiveSuite.scala b/core/src/test/scala/org/graphframes/propertygraph/PropertyGraphFrameCaseInsensitiveSuite.scala new file mode 100644 index 000000000..42bc314e3 --- /dev/null +++ b/core/src/test/scala/org/graphframes/propertygraph/PropertyGraphFrameCaseInsensitiveSuite.scala @@ -0,0 +1,169 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.graphframes.propertygraph + +import org.apache.spark.sql.functions.lit +import org.graphframes.GraphFrameTestSparkContext +import org.graphframes.SparkFunSuite +import org.graphframes.propertygraph.property.EdgePropertyGroup +import org.graphframes.propertygraph.property.VertexPropertyGroup + +import java.security.MessageDigest + +/** + * Regression tests that lock in case-insensitive label matching in the public property-graph + * Scala API: [[PropertyGraphFrame.toGraphFrame]] and [[PropertyGraphFrame.projectionBy]]. Group + * names passed with different casing than they were registered with must still resolve. + */ +class PropertyGraphFrameCaseInsensitiveSuite + extends SparkFunSuite + with GraphFrameTestSparkContext { + + import sqlImplicits._ + + private var graph: PropertyGraphFrame = _ + + // Groups are registered with distinctive casing; queries below deliberately use other cases. + private val peopleName = "People" + private val moviesName = "Movies" + private val likesName = "Likes" + + override def beforeAll(): Unit = { + super.beforeAll() + val peopleData = Seq((1L, "Alice"), (2L, "Bob")).toDF("id", "name") + val peopleGroup = VertexPropertyGroup(peopleName, peopleData, "id") + + val moviesData = Seq((10L, "Matrix"), (20L, "Inception")).toDF("id", "title") + val moviesGroup = VertexPropertyGroup(moviesName, moviesData, "id") + + val likesData = Seq((1L, 10L), (2L, 20L)).toDF("src", "dst") + val likesGroup = EdgePropertyGroup( + likesName, + likesData, + peopleGroup, + moviesGroup, + isDirected = true, + "src", + "dst", + lit(1.0)) + + graph = PropertyGraphFrame(Seq(peopleGroup, moviesGroup), Seq(likesGroup)) + } + + // Mirrors the internal ID masking so tests can compare against hashed IDs. + private def sha256Hash(id: Long, groupName: String): String = { + val md = MessageDigest.getInstance("SHA-256") + val hash = md.digest(id.toString.getBytes("UTF-8")).map("%02x".format(_)).mkString + s"$groupName$hash" + } + + // ----- toGraphFrame ------------------------------------------------------- + + test("toGraphFrame resolves lowercase group names") { + val gf = graph.toGraphFrame( + Seq("people"), + Seq("likes"), + Map("likes" -> lit(true)), + Map("people" -> lit(true))) + assert(gf.vertices.count() === 2) + assert(gf.edges.count() === 2) + } + + test("toGraphFrame resolves uppercase group names") { + val gf = graph.toGraphFrame( + Seq("PEOPLE", "MOVIES"), + Seq("LIKES"), + Map("LIKES" -> lit(true)), + Map("PEOPLE" -> lit(true), "MOVIES" -> lit(true))) + assert(gf.vertices.count() === 4) + assert(gf.edges.count() === 2) + } + + test("toGraphFrame resolves mixed-case group names") { + val gf = graph.toGraphFrame( + Seq("pEoPlE"), + Seq("lIkEs"), + Map("lIkEs" -> lit(true)), + Map("pEoPlE" -> lit(true))) + assert(gf.vertices.count() === 2) + assert(gf.edges.count() === 2) + } + + test("toGraphFrame preserves canonical IDs despite query casing") { + val gf = graph.toGraphFrame( + Seq("people"), + Seq("likes"), + Map("likes" -> lit(true)), + Map("people" -> lit(true))) + val expectedVertices = Set(sha256Hash(1L, peopleName), sha256Hash(2L, peopleName)) + val actualVertices = gf.vertices.collect().map(_.getString(0)).toSet + assert(actualVertices === expectedVertices) + } + + test("toGraphFrame rejects unknown vertex group") { + intercept[IllegalArgumentException] { + graph.toGraphFrame( + Seq("Nonexistent"), + Seq("likes"), + Map("likes" -> lit(true)), + Map("Nonexistent" -> lit(true))) + } + } + + test("toGraphFrame rejects unknown edge group") { + intercept[IllegalArgumentException] { + graph.toGraphFrame( + Seq("people"), + Seq("Hates"), + Map("Hates" -> lit(true)), + Map("people" -> lit(true))) + } + } + + // ----- projectionBy ------------------------------------------------------- + + test("projectionBy resolves lowercase group names") { + val projected = graph.projectionBy( + leftBiGraphPart = "people", // registered as "People" + rightBiGraphPart = "movies", // registered as "Movies" + edgeGroup = "likes" + ) // registered as "Likes" + // Projection drops the "through" group (movies); one projected group remains. + assert(projected.vertexPropertyGroups.map(_.name).toSet === Set(peopleName)) + // The projected edge group name echoes the query casing by design. + assert(projected.edgesPropertyGroups.map(_.name).toSet === Set("projected_likes")) + } + + test("projectionBy resolves uppercase group names") { + val projected = graph.projectionBy( + leftBiGraphPart = "PEOPLE", + rightBiGraphPart = "MOVIES", + edgeGroup = "LIKES") + assert(projected.vertexPropertyGroups.map(_.name).toSet === Set(peopleName)) + assert(projected.edgesPropertyGroups.map(_.name).toSet === Set("projected_LIKES")) + } + + test("projectionBy rejects unknown edge group") { + intercept[NoSuchElementException] { + graph.projectionBy( + leftBiGraphPart = "people", + rightBiGraphPart = "movies", + edgeGroup = "Hates") + } + } +} diff --git a/core/src/test/scala/org/graphframes/propertygraph/PropertyGraphFrameQuerySuite.scala b/core/src/test/scala/org/graphframes/propertygraph/PropertyGraphFrameQuerySuite.scala new file mode 100644 index 000000000..340c389db --- /dev/null +++ b/core/src/test/scala/org/graphframes/propertygraph/PropertyGraphFrameQuerySuite.scala @@ -0,0 +1,670 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.graphframes.propertygraph + +import org.apache.spark.sql.DataFrame +import org.apache.spark.sql.Row +import org.apache.spark.sql.functions.lit +import org.graphframes.GraphFrameTestSparkContext +import org.graphframes.InvalidParseException +import org.graphframes.InvalidPropertyGroupException +import org.graphframes.SparkFunSuite +import org.graphframes.propertygraph.property.EdgePropertyGroup +import org.graphframes.propertygraph.property.VertexPropertyGroup + +import java.security.MessageDigest + +/** + * End-to-end tests for the public [[PropertyGraphFrame.query]] / [[PropertyGraphFrame.explain]] + * API: parse -> resolve -> plan -> execute, exercising the full pipeline through the public + * surface. + */ +class PropertyGraphFrameQuerySuite extends SparkFunSuite with GraphFrameTestSparkContext { + + import sqlImplicits._ + + private var pgf: PropertyGraphFrame = _ + + override def beforeAll(): Unit = { + super.beforeAll() + val persons = Seq((1L, "Alice", 30), (2L, "Bob", 40)).toDF("id", "name", "age") + val companies = Seq((10L, "Acme")).toDF("id", "name") + val personGroup = VertexPropertyGroup("Person", persons, "id") + val companyGroup = VertexPropertyGroup("Company", companies, "id") + + val knows = Seq((1L, 2L), (2L, 1L)).toDF("src", "dst") + val worksAt = Seq((1L, 10L)).toDF("src", "dst") + val knowsGroup = EdgePropertyGroup( + "KNOWS", + knows, + personGroup, + personGroup, + isDirected = true, + "src", + "dst", + lit(1.0)) + val worksAtGroup = EdgePropertyGroup( + "WORKS_AT", + worksAt, + personGroup, + companyGroup, + isDirected = true, + "src", + "dst", + lit(1.0)) + + pgf = PropertyGraphFrame(Seq(personGroup, companyGroup), Seq(knowsGroup, worksAtGroup)) + } + + // ------------------------------------------------------------------------- + // Path-comparison helpers. + // + // The query output uses masked ids (`concat(groupName, sha2(id, 256))`) for every id column, + // and a `path` array of (edge_property_group, node_id, node_property_group) structs whose final + // entry carries only the edge group (the end node lives in `end_id`). These helpers let tests + // express expectations in terms of *raw* ids + group names and assert equality of the full + // result set, not just the row count. + // ------------------------------------------------------------------------- + + /** Mask an external id the same way [[VertexPropertyGroup]] does internally. */ + private def maskedId(id: Any, groupName: String): String = { + val md = MessageDigest.getInstance("SHA-256") + val hash = md.digest(id.toString.getBytes("UTF-8")).map("%02x".format(_)).mkString + s"$groupName$hash" + } + + /** + * One entry of the `path` array. [[nodeId]]/[[nodeGroup]] are [[None]] for the trailing entry + * of a multi-hop path (where the end node is already captured by `end_id`), and for the single + * entry of a single-hop path. + */ + case class ExpectedHop( + edgeGroup: String, + nodeId: Option[Long] = None, + nodeGroup: Option[String] = None) + + object ExpectedHop { + + /** Intermediate hop carrying an intermediate node. */ + def mid(edgeGroup: String, nodeId: Long, nodeGroup: String): ExpectedHop = + ExpectedHop(edgeGroup, Some(nodeId), Some(nodeGroup)) + + /** Trailing hop: edge group only, no node (the end node lives in `end_id`). */ + def last(edgeGroup: String): ExpectedHop = ExpectedHop(edgeGroup, None, None) + } + + /** A full expected output row, expressed in terms of raw ids and group names. */ + case class ExpectedPath( + startId: Long, + startGroup: String, + endId: Long, + endGroup: String, + edgeGroup: String, + hops: Seq[ExpectedHop]) + + /** Normalized representation of an actual DataFrame row (masked ids inlined). */ + private case class ActualPath( + startId: String, + startGroup: String, + endId: String, + endGroup: String, + edgeGroup: String, + hops: Seq[(String, Option[String], Option[String])]) + + /** Extract an [[ActualPath]] from a result row. */ + private def rowToActual(row: Row): ActualPath = { + val pathArr = row + .get(row.fieldIndex("path")) + .asInstanceOf[scala.collection.Seq[Row]] + val hops = pathArr.map { h => + val eg = h.getAs[String](0) + val nodeId = if (h.isNullAt(1)) None else Some(h.getAs[String](1)) + val nodeGroup = if (h.isNullAt(2)) None else Some(h.getAs[String](2)) + (eg, nodeId, nodeGroup) + } + ActualPath( + startId = row.getAs[String]("start_id"), + startGroup = row.getAs[String]("start_property_group"), + endId = row.getAs[String]("end_id"), + endGroup = row.getAs[String]("end_property_group"), + edgeGroup = row.getAs[String]("edge_property_group"), + hops = hops.toSeq) + } + + /** Convert an [[ExpectedPath]] (raw ids) to the comparable [[ActualPath]] (masked ids). */ + private def expectedToActual(p: ExpectedPath): ActualPath = ActualPath( + startId = maskedId(p.startId, p.startGroup), + startGroup = p.startGroup, + endId = maskedId(p.endId, p.endGroup), + endGroup = p.endGroup, + edgeGroup = p.edgeGroup, + hops = p.hops.map { h => + (h.edgeGroup, h.nodeId.map(id => maskedId(id, h.nodeGroup.get)), h.nodeGroup) + }) + + /** + * Collect `df` and assert that its rows match exactly the given [[ExpectedPath]]s (order + * independent). Compares start/end ids (masked), property groups, the first edge group, and + * every hop of the `path` array — so it catches missing/extra paths, wrong edge labels, and + * wrong intermediate nodes, not just row count. + */ + private def comparePaths( + df: DataFrame, + expected: Seq[ExpectedPath]): org.scalatest.Assertion = { + val actual = df.collect().map(rowToActual).toSet + val expectedActuals = expected.map(expectedToActual).toSet + assert( + actual === expectedActuals, + s"""|Query result mismatch. + | expected (${expectedActuals.size}): + |${expectedActuals + .map(" " + _.productIterator.mkString(", ")) + .toVector + .sorted + .mkString("\n")} + | actual (${actual.size}): + |${actual.map(" " + _.productIterator.mkString(", ")).toVector.sorted.mkString("\n")} + |""".stripMargin) + } + + test("query returns rows over the fixed schema") { + val df = pgf.query("MATCH (a:Person)-[:KNOWS]->(b:Person)") + assert(df.count() === 2) + assert( + df.schema.fieldNames === + Seq( + "start_id", + "start_property_group", + "end_id", + "end_property_group", + "edge_property_group", + "path")) + } + + test("query with a scan-local filter prunes rows") { + // Only Bob(40) has age > 30; Bob KNOWS Alice. So exactly one row, starting at Bob. + val df = pgf.query("MATCH (a:Person)-[:KNOWS]->(b:Person) WHERE a.age > 30") + assert(df.count() === 1) + } + + test("query with RETURN projects the requested columns") { + val df = pgf.query("MATCH (a:Person)-[:KNOWS]->(b:Person) RETURN a.name AS who") + assert(df.schema.fieldNames.toSeq === Seq("who")) + assert(df.count() === 2) + } + + test("query on a disconnected pattern returns an empty DataFrame without throwing") { + val df = pgf.query("MATCH (a:Company)-[:KNOWS]->(b:Person)") + assert(df.count() === 0) + } + + test("explain(logical) returns a non-empty string describing the resolved plan") { + val s = pgf.explain("MATCH (a:Person)-[:KNOWS]->(b:Person)") + assert(s.contains("Logical plan")) + assert(s.contains("KNOWS")) + } + + test("explain(physical) returns a non-empty string describing the join plan") { + val s = pgf.explain("MATCH (a:Person)-[:KNOWS]->(b:Person)", ExplainMode.Physical) + assert(s.contains("Physical plan")) + assert(s.contains("join order")) + } + + test("explain(logical) is the default mode") { + val s = pgf.explain("MATCH (a:Person)-[:KNOWS]->(b:Person)", ExplainMode.Logical) + assert(s.contains("Logical plan")) + } + + test("query with bad syntax throws InvalidParseException") { + intercept[InvalidParseException] { + pgf.query("MATCH (a:Person OPTIONAL MATCH") + } + } + + test("query with an unknown label throws InvalidPropertyGroupException") { + intercept[InvalidPropertyGroupException] { + pgf.query("MATCH (a:Unicorn)-[:KNOWS]->(b:Person)") + } + } + + test("QueryOptions.maxSchemaPathLength must be positive") { + // `PropertyGraphFrame.resolve` runs `require(maxSchemaPathLength > 0)` *before* any per-path + // depth check, so a non-positive cap is rejected regardless of the query's actual hop count. + // The WORKS_AT pattern here resolves to a length-1 schema path (a single Person->Company + // hop), so this guards the positivity precondition, not per-path depth -- the real per-path + // depth guard is covered by the boundary test below and the explain-bypass test. + intercept[IllegalArgumentException] { + pgf.query( + "MATCH (a:Person)-[:WORKS_AT]->(c:Company)", + QueryOptions(maxSchemaPathLength = 0)) + } + } + + test("maxSchemaPathLength per-path guard rejects patterns deeper than the cap") { + // The 2-hop KNOWS->KNOWS chain resolves to a length-2 schema path. With the cap set to 1 the + // per-path `require(path.length <= maxSchemaPathLength)` in `resolve` fires; with the cap + // raised to 2 the same query succeeds and returns the two 2-hop cycles. + val gql = "MATCH (a:Person)-[:KNOWS]->(b:Person)-[:KNOWS]->(c:Person)" + intercept[IllegalArgumentException] { + pgf.query(gql, QueryOptions(maxSchemaPathLength = 1)) + } + val df = pgf.query(gql, QueryOptions(maxSchemaPathLength = 2)) + comparePaths( + df, + Seq( + ExpectedPath( + startId = 1L, + startGroup = "Person", + endId = 1L, + endGroup = "Person", + edgeGroup = "KNOWS", + hops = Seq(ExpectedHop.mid("KNOWS", 2L, "Person"), ExpectedHop.last("KNOWS"))), + ExpectedPath( + startId = 2L, + startGroup = "Person", + endId = 2L, + endGroup = "Person", + edgeGroup = "KNOWS", + hops = Seq(ExpectedHop.mid("KNOWS", 1L, "Person"), ExpectedHop.last("KNOWS"))))) + } + + test("explain bypasses the maxSchemaPathLength per-path guard so users can inspect the plan") { + // A 2-hop (length-2) pattern capped at 1: query throws, explain renders the plan instead. + val gql = "MATCH (a:Person)-[:KNOWS]->(b:Person)-[:KNOWS]->(c:Person)" + val opts = QueryOptions(maxSchemaPathLength = 1) + intercept[IllegalArgumentException] { + pgf.query(gql, opts) + } + val logical = pgf.explain(gql, ExplainMode.Logical, opts) + assert(logical.contains("Logical plan")) + val physical = pgf.explain(gql, ExplainMode.Physical, opts) + assert(physical.contains("Physical plan")) + } + + // ------------------------------------------------------------------------- + // Real result checks via comparePaths. + // + // Each test below pins down the *exact* set of matched paths (start/end ids, + // property groups, the first edge group, and every hop of the `path` array), + // not just the row count. Graph fixture (see beforeAll): + // + // vertices: Person(1,Alice,30) Person(2,Bob,40) Company(10,Acme) + // edges: KNOWS 1->2, 2->1 (directed, Person->Person) + // WORKS_AT 1->10 (directed, Person->Company) + // ------------------------------------------------------------------------- + + test("single-hop KNOWS returns exactly the two directed edges with a single-hop path array") { + val df = pgf.query("MATCH (a:Person)-[:KNOWS]->(b:Person)") + comparePaths( + df, + Seq( + ExpectedPath(1L, "Person", 2L, "Person", "KNOWS", Seq(ExpectedHop.last("KNOWS"))), + ExpectedPath(2L, "Person", 1L, "Person", "KNOWS", Seq(ExpectedHop.last("KNOWS"))))) + } + + test("single-hop KNOWS includes the first edge group on every row") { + // The fixed schema surfaces the *first* edge group in `edge_property_group`. For a single-hop + // query there is only one edge, so every row must carry KNOWS -- never null, never anything + // else. + val df = pgf.query("MATCH (a:Person)-[:KNOWS]->(b:Person)") + val edgeGroups = df.collect().map(_.getAs[String]("edge_property_group")).toSet + assert(edgeGroups === Set("KNOWS")) + } + + test("backward arrow <-[:KNOWS]- swaps start and end ids") { + // knows: 1->2, 2->1. Writing the arrow backwards binds a=dst, b=src, so we get (2,1) and (1,2) + // -- the same set of vertex pairs but with start/end flipped relative to the forward query. + val df = pgf.query("MATCH (a:Person)<-[:KNOWS]-(b:Person)") + comparePaths( + df, + Seq( + ExpectedPath(2L, "Person", 1L, "Person", "KNOWS", Seq(ExpectedHop.last("KNOWS"))), + ExpectedPath(1L, "Person", 2L, "Person", "KNOWS", Seq(ExpectedHop.last("KNOWS"))))) + } + + test("cross-group WORKS_AT path connects Person to Company") { + val df = pgf.query("MATCH (a:Person)-[:WORKS_AT]->(c:Company)") + comparePaths( + df, + Seq( + ExpectedPath( + 1L, + "Person", + 10L, + "Company", + "WORKS_AT", + Seq(ExpectedHop.last("WORKS_AT"))))) + } + + test("edge-label isolation: KNOWS query never returns WORKS_AT paths") { + // Only the KNOWS rows may appear; the WORKS_AT 1->10 path must NOT leak in even though Person + // appears on both sides of both edge groups. + val df = pgf.query("MATCH (a:Person)-[:KNOWS]->(b:Person)") + val rows = df.collect() + assert(rows.nonEmpty) + rows.foreach { r => + assert(r.getAs[String]("edge_property_group") === "KNOWS") + assert(r.getAs[String]("start_property_group") === "Person") + assert(r.getAs[String]("end_property_group") === "Person") + } + } + + test("edge-label isolation: WORKS_AT query never returns KNOWS paths") { + val df = pgf.query("MATCH (a:Person)-[:WORKS_AT]->(c:Company)") + val rows = df.collect() + assert(rows.nonEmpty) + rows.foreach { r => + assert(r.getAs[String]("edge_property_group") === "WORKS_AT") + assert(r.getAs[String]("end_property_group") === "Company") + } + } + + test("untyped edge -[]-> over Person->Person fans out to KNOWS only") { + // The schema has exactly one outgoing edge from Person back to Person (KNOWS); WORKS_AT points + // Person->Company. So an untyped Person->Person arrow must resolve to KNOWS and yield the same + // two rows as the typed KNOWS query -- nothing more. + val df = pgf.query("MATCH (a:Person)-[]->(b:Person)") + comparePaths( + df, + Seq( + ExpectedPath(1L, "Person", 2L, "Person", "KNOWS", Seq(ExpectedHop.last("KNOWS"))), + ExpectedPath(2L, "Person", 1L, "Person", "KNOWS", Seq(ExpectedHop.last("KNOWS"))))) + } + + test("untyped trailing node () is resolved by the schema but not surfaced in end_id") { + // MATCH (a:Person)-[:WORKS_AT]->() : the trailing () resolves to Company (the only dst of + // WORKS_AT in the schema), so the edge is found and the row is returned. BUT the fixed output + // schema surfaces only the first/last *named* node ids (see QueryIr Projection.Default), so + // with only `a` named, end_id falls back to the last named node -- here `a` itself. The edge + // group is still WORKS_AT, proving the schema-resolved hop was executed. + val df = pgf.query("MATCH (a:Person)-[:WORKS_AT]->()") + val rows = df.collect() + assert(rows.length === 1) + val r = rows.head + assert(r.getAs[String]("start_property_group") === "Person") + assert(r.getAs[String]("edge_property_group") === "WORKS_AT") + // Quirk: with no named end variable, end_id/end_property_group mirror the last named node (a). + assert(r.getAs[String]("end_id") === r.getAs[String]("start_id")) + assert(r.getAs[String]("end_property_group") === "Person") + } + + test("scan-local WHERE a.age > 30 leaves exactly the Bob->Alice KNOWS row") { + val df = pgf.query("MATCH (a:Person)-[:KNOWS]->(b:Person) WHERE a.age > 30") + // Only Bob(40) passes the scan filter; Bob KNOWS Alice. Exactly one row. + comparePaths( + df, + Seq(ExpectedPath(2L, "Person", 1L, "Person", "KNOWS", Seq(ExpectedHop.last("KNOWS"))))) + } + + test("join-level WHERE a.age > b.age filters the joined frame") { + // KNOWS: 1(30)->2(40), 2(40)->1(30). a.age > b.age: only 40 > 30, i.e. Bob->Alice. + val df = pgf.query("MATCH (a:Person)-[:KNOWS]->(b:Person) WHERE a.age > b.age") + comparePaths( + df, + Seq(ExpectedPath(2L, "Person", 1L, "Person", "KNOWS", Seq(ExpectedHop.last("KNOWS"))))) + } + + test("RETURN a.name AS who projects the exact source-name values") { + val df = pgf.query("MATCH (a:Person)-[:KNOWS]->(b:Person) RETURN a.name AS who") + assert(df.schema.fieldNames.toSeq === Seq("who")) + // KNOWS: 1->2 (Alice), 2->1 (Bob). The projected `who` set is exactly {Alice, Bob}. + val names = df.collect().map(_.getString(0)).toSet + assert(names === Set("Alice", "Bob")) + } + + test("RETURN a.name, b.name projects both endpoint names in order") { + val df = pgf.query("MATCH (a:Person)-[:KNOWS]->(b:Person) RETURN a.name, b.name") + assert(df.schema.fieldNames.toSeq === Seq("name", "name")) + val pairs = df.collect().map(r => (r.getString(0), r.getString(1))).toSet + assert(pairs === Set(("Alice", "Bob"), ("Bob", "Alice"))) + } + + test("disconnected pattern yields an empty result set with the fixed schema") { + val df = pgf.query("MATCH (a:Company)-[:KNOWS]->(b:Person)") + assert(df.count() === 0) + assert( + df.schema.fieldNames === + Seq( + "start_id", + "start_property_group", + "end_id", + "end_property_group", + "edge_property_group", + "path")) + } + + test("disconnected edge label yields an empty result set") { + // There is no WORKS_AT edge touching another Person, so Person-[:WORKS_AT]->Person is empty. + val df = pgf.query("MATCH (a:Person)-[:WORKS_AT]->(b:Person)") + assert(df.count() === 0) + } + + test("multi-hop path array carries intermediate node ids and groups") { + // The 2-hop chain Person-[:WORKS_AT]->Company-[:LOCATED_IN]->City is not present in this small + // fixture, but a 2-hop KNOWS->KNOWS chain is. Here we instead use a 2-hop KNOWS->KNOWS query + // to verify the `path` array structure: two entries, the first with the intermediate Person + // node, the second carrying only the edge group. + val df = pgf.query("MATCH (a:Person)-[:KNOWS]->(b:Person)-[:KNOWS]->(c:Person)") + // knows: 1->2, 2->1. A 2-hop chain a-b-c requires b's KNOWS-out to land on c. The only valid + // 2-hop chains are 1->2->1 and 2->1->2. + comparePaths( + df, + Seq( + ExpectedPath( + startId = 1L, + startGroup = "Person", + endId = 1L, + endGroup = "Person", + edgeGroup = "KNOWS", + hops = Seq(ExpectedHop.mid("KNOWS", 2L, "Person"), ExpectedHop.last("KNOWS"))), + ExpectedPath( + startId = 2L, + startGroup = "Person", + endId = 2L, + endGroup = "Person", + edgeGroup = "KNOWS", + hops = Seq(ExpectedHop.mid("KNOWS", 1L, "Person"), ExpectedHop.last("KNOWS"))))) + } + + test("multi-hop path array length equals the number of steps") { + val df = pgf.query("MATCH (a:Person)-[:KNOWS]->(b:Person)-[:KNOWS]->(c:Person)") + df.collect().foreach { row => + val pathArr = row + .get(row.fieldIndex("path")) + .asInstanceOf[scala.collection.Seq[_]] + assert(pathArr.length === 2, s"2-step path must produce a 2-entry array: $row") + } + } + + test("combined scan-local and join predicates both apply") { + // a.age > 30 keeps only Bob as `a` (scan-local). b.age > 30 on Alice(30) vs Bob(40)... Alice + // is 30 so b.age > 30 is false. Result: empty. + val df = pgf.query("MATCH (a:Person)-[:KNOWS]->(b:Person) WHERE a.age > 30 AND b.age > 30") + assert(df.count() === 0) + } + + // ------------------------------------------------------------------------- + // ExpressionLowering coverage. + // + // The tests above only exercise `>` (Comparison.Gt), `AND`, and integer literals. The tests + // below pin down the remaining branches of `ExpressionLowering.lower` end-to-end through the + // public query API: OR, NOT, string literals, the rest of the comparison operators (Neq, Lte, + // Gte), arithmetic (+/-) in WHERE and RETURN, scan-local filters on the non-start variable, and + // non-adjacent post-filters. Same fixture: + // Person(1,Alice,30) Person(2,Bob,40); KNOWS 1->2, 2->1. + // ------------------------------------------------------------------------- + + test("WHERE OR keeps only the row matching either disjunct") { + // a.age > 35 keeps only Bob(40); a.age < 28 keeps neither. OR -> only Bob remains as `a`, + // and Bob KNOWS Alice, so exactly the Bob->Alice row. + val df = pgf.query("MATCH (a:Person)-[:KNOWS]->(b:Person) WHERE a.age > 35 OR a.age < 28") + comparePaths( + df, + Seq(ExpectedPath(2L, "Person", 1L, "Person", "KNOWS", Seq(ExpectedHop.last("KNOWS"))))) + } + + test("WHERE NOT negates a comparison predicate") { + // NOT (a.age > 35) keeps Alice(30) and excludes Bob(40). Alice KNOWS Bob -> Alice->Bob. + val df = pgf.query("MATCH (a:Person)-[:KNOWS]->(b:Person) WHERE NOT (a.age > 35)") + comparePaths( + df, + Seq(ExpectedPath(1L, "Person", 2L, "Person", "KNOWS", Seq(ExpectedHop.last("KNOWS"))))) + } + + test("WHERE string equality filters on a string property") { + // Exercises Comparison(Eq) with a String Literal against a string column. a.name = 'Alice' + // keeps only Alice as `a`; Alice KNOWS Bob -> Alice->Bob. + val df = pgf.query("MATCH (a:Person)-[:KNOWS]->(b:Person) WHERE a.name = 'Alice'") + comparePaths( + df, + Seq(ExpectedPath(1L, "Person", 2L, "Person", "KNOWS", Seq(ExpectedHop.last("KNOWS"))))) + } + + test("WHERE <> (not-equal) prunes the matching endpoint") { + // Comparison.Neq via `<>`. a.age <> 40 excludes Bob(40) and keeps Alice(30) -> Alice->Bob. + val df = pgf.query("MATCH (a:Person)-[:KNOWS]->(b:Person) WHERE a.age <> 40") + comparePaths( + df, + Seq(ExpectedPath(1L, "Person", 2L, "Person", "KNOWS", Seq(ExpectedHop.last("KNOWS"))))) + } + + test("WHERE <= keeps rows at or below the threshold") { + // Comparison.Lte. a.age <= 30 keeps only Alice(30) -> Alice->Bob. + val df = pgf.query("MATCH (a:Person)-[:KNOWS]->(b:Person) WHERE a.age <= 30") + comparePaths( + df, + Seq(ExpectedPath(1L, "Person", 2L, "Person", "KNOWS", Seq(ExpectedHop.last("KNOWS"))))) + } + + test("WHERE >= keeps rows at or above the threshold") { + // Comparison.Gte. a.age >= 40 keeps only Bob(40) -> Bob->Alice. + val df = pgf.query("MATCH (a:Person)-[:KNOWS]->(b:Person) WHERE a.age >= 40") + comparePaths( + df, + Seq(ExpectedPath(2L, "Person", 1L, "Person", "KNOWS", Seq(ExpectedHop.last("KNOWS"))))) + } + + test("arithmetic Minus in WHERE participates in the join filter") { + // Arithmetic(Minus) lowered inside a join-level predicate spanning adjacent a,b. a.age - 5: + // Alice->Bob: 30 - 5 = 25 > 40 ? no + // Bob->Alice: 40 - 5 = 35 > 30 ? yes -> only Bob->Alice. + val df = pgf.query("MATCH (a:Person)-[:KNOWS]->(b:Person) WHERE a.age - 5 > b.age") + comparePaths( + df, + Seq(ExpectedPath(2L, "Person", 1L, "Person", "KNOWS", Seq(ExpectedHop.last("KNOWS"))))) + } + + test("arithmetic Plus in RETURN projects a computed column") { + // Arithmetic(Plus) inside a RETURN item, projected under an explicit alias. Exercises the + // Projection.Items path with a non-trivial expression (not a bare Variable/PropertyAccess), + // so the alias must be taken from the AS clause. + val df = pgf.query("MATCH (a:Person)-[:KNOWS]->(b:Person) RETURN a.age + 1 AS adjusted") + assert(df.schema.fieldNames.toSeq === Seq("adjusted")) + // KNOWS: 1->2 (Alice,30), 2->1 (Bob,40). adjusted = {31, 41}. The arithmetic result column + // is LongType (integer column + integer literal promotes), so read it back as Long. + val adjusted = df.collect().map(_.getLong(0)).toSet + assert(adjusted === Set(31L, 41L)) + } + + test("RETURN * projects the fixed output schema") { + // Projection.Star is handled together with Projection.Default in QueryExecutor.project, so + // RETURN * must yield the same 6-column fixed schema and the same rows as the no-RETURN query. + val star = pgf.query("MATCH (a:Person)-[:KNOWS]->(b:Person) RETURN *") + assert( + star.schema.fieldNames === + Seq( + "start_id", + "start_property_group", + "end_id", + "end_property_group", + "edge_property_group", + "path")) + comparePaths( + star, + Seq( + ExpectedPath(1L, "Person", 2L, "Person", "KNOWS", Seq(ExpectedHop.last("KNOWS"))), + ExpectedPath(2L, "Person", 1L, "Person", "KNOWS", Seq(ExpectedHop.last("KNOWS"))))) + } + + test("single-node pattern returns each vertex with an empty path") { + // A 0-hop path (path.steps.isEmpty): the single node is both start and end, edge_property_group + // is null, and the path array is empty. Exercises the degenerate single-node branch of + // QueryExecutor.executePlan / project. + val df = pgf.query("MATCH (a:Person)") + assert(df.count() === 2) + val rows = df.collect() + rows.foreach { r => + assert(r.getAs[String]("start_property_group") === "Person") + assert(r.getAs[String]("end_property_group") === "Person") + // With only one named node, start and end both reference `a`. + assert(r.getAs[String]("start_id") === r.getAs[String]("end_id")) + assert(r.isNullAt(r.fieldIndex("edge_property_group"))) + val pathArr = r.get(r.fieldIndex("path")).asInstanceOf[scala.collection.Seq[_]] + assert(pathArr.isEmpty, s"0-hop path array must be empty: $r") + } + // The two vertices are Alice(1) and Bob(2). + val startIds = rows.map(_.getAs[String]("start_id")).toSet + assert(startIds === Set(maskedId(1L, "Person"), maskedId(2L, "Person"))) + } + + test("scan-local filter on the non-start variable prunes rows") { + // b.age > 35 references only `b`, so the resolver attaches it as a scan-local filter to the + // `b` node rather than the start `a`. Only Bob(40) passes as `b`; the only KNOWS edge into + // Bob is 1->2, so exactly Alice->Bob survives. + val df = pgf.query("MATCH (a:Person)-[:KNOWS]->(b:Person) WHERE b.age > 35") + comparePaths( + df, + Seq(ExpectedPath(1L, "Person", 2L, "Person", "KNOWS", Seq(ExpectedHop.last("KNOWS"))))) + } + + test("non-adjacent post-filter spans the first and last variable") { + // A 3-variable predicate where a and c are not adjacent is classified as a post-filter and + // applied after the join tree. The 2-hop KNOWS->KNOWS chains are 1->2->1 and 2->1->2, so a + // and c are always the same vertex -- hence a.age > c.age is always false and the result is + // empty. (Without the post-filter, 2 rows would be returned, so this catches regressions + // where the post-filter is dropped.) + val df = + pgf.query("MATCH (a:Person)-[:KNOWS]->(b:Person)-[:KNOWS]->(c:Person) WHERE a.age > c.age") + assert(df.count() === 0) + } + + // --------------------------------------------------------------------------- + // Scan reuse: output-only property join-back (end-to-end through the public API). + // --------------------------------------------------------------------------- + + test("multi-hop output-only join-back resolves a Company property through RETURN") { + // `c.name` is RETURN-only (no filter references it) -> output-only -> terminal join-back on the + // masked id. WORKS_AT: (1,10) Alice -> Acme. So exactly one row, `name` = "Acme". (Per the + // Items-projection convention, the output column is named after the property, `name`.) + val df = pgf.query("MATCH (a:Person)-[:WORKS_AT]->(c:Company) RETURN c.name") + assert(df.schema.fieldNames.toSeq === Seq("name")) + val names = df.collect().map(_.getString(0)).toSet + assert(names === Set("Acme")) + } + + test("mixed carry + output-only resolves both a carried and a join-backed column end-to-end") { + // `a.age` is referenced by both the filter (`a.age >= 30`) and RETURN -> carried (no join-back). + // `a.name` is RETURN-only -> output-only -> join-backed. WORKS_AT only has Alice(1)->Acme, and + // Alice's age is 30, so `a.age >= 30` keeps exactly Alice. Result: one row (Alice, 30). The + // `age` column is read as Int (its physical type in the fixture). + val df = pgf.query( + "MATCH (a:Person)-[:WORKS_AT]->(c:Company) WHERE a.age >= 30 RETURN a.name, a.age") + val rows = df.collect().map(r => (r.getString(0), r.getInt(1))).toSet + assert(rows === Set(("Alice", 30))) + } +} diff --git a/core/src/test/scala/org/graphframes/propertygraph/PropertyGraphFrameTest.scala b/core/src/test/scala/org/graphframes/propertygraph/PropertyGraphFrameTest.scala index 88540ba93..427ef7183 100644 --- a/core/src/test/scala/org/graphframes/propertygraph/PropertyGraphFrameTest.scala +++ b/core/src/test/scala/org/graphframes/propertygraph/PropertyGraphFrameTest.scala @@ -249,6 +249,32 @@ class PropertyGraphFrameTest assert(projectedEdges === expectedEdges) } + test("schemaString returns human-readable schema description") { + val schema = peopleMoviesGraph.schemaString + + assert(schema.contains("people")) + assert(schema.contains("movies")) + assert(schema.contains("likes")) + assert(schema.contains("messages")) + assert(schema.startsWith("Property graph schema:")) + } + + test("schemaStringDOT returns valid DOT format") { + val dot = peopleMoviesGraph.schemaStringDOT + + assert(dot.startsWith("digraph SchemaGraph {")) + assert(dot.contains("\"people\"")) + assert(dot.contains("\"movies\"")) + assert(dot.contains("likes")) + assert(dot.contains("messages")) + assert(dot.trim().endsWith("}")) + } + + test("print schema") { + println(peopleMoviesGraph.schemaString) + println(peopleMoviesGraph.schemaStringDOT) + } + test("joinVertices withConnectedComponents") { // Convert to GraphFrame with all vertices and edges val graph = peopleMoviesGraph.toGraphFrame( diff --git a/core/src/test/scala/org/graphframes/propertygraph/internal/GqlExplainSuite.scala b/core/src/test/scala/org/graphframes/propertygraph/internal/GqlExplainSuite.scala new file mode 100644 index 000000000..3c5490811 --- /dev/null +++ b/core/src/test/scala/org/graphframes/propertygraph/internal/GqlExplainSuite.scala @@ -0,0 +1,97 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.graphframes.propertygraph.internal + +import org.graphframes.SparkFunSuite +import org.graphframes.propertygraph.QueryOptions + +/** + * Pure-JVM tests for the explain renderers. No SparkSession required; the renderers are pure + * functions over the resolved IR values. + */ +class GqlExplainSuite extends SparkFunSuite { + + private val schema = SchemaGraphSnapshot( + vertexGroupNames = Set("Person", "Company", "City"), + edges = Vector( + SchemaEdge("KNOWS", "Person", "Person", true), + SchemaEdge("WORKS_AT", "Person", "Company", true), + SchemaEdge("LOCATED_IN", "Company", "City", true))) + + private val options = QueryOptions() + + test("logical explain renders the path with a forward arrow") { + val ast = AstBuilder.parse("MATCH (a:Person)-[:KNOWS]->(b:Person)") + val rq = Resolver.resolve(ast, schema, options) + val out = GqlExplain.logical(rq) + assert(out.contains("Logical plan")) + assert(out.contains("(a:Person)")) + // Anonymous edge renders without a variable prefix: -[KNOWS]-> + assert(out.contains("-[KNOWS]->")) + } + + test("logical explain renders a backward arrow for <-[e]- ") { + val ast = AstBuilder.parse("MATCH (a:Person)<-[:KNOWS]-(b:Person)") + val rq = Resolver.resolve(ast, schema, options) + val out = GqlExplain.logical(rq) + assert(out.contains("<-[KNOWS]-")) + } + + test("logical explain reports disconnected patterns as (none)") { + val ast = AstBuilder.parse("MATCH (a:City)-[:KNOWS]->(b:Person)") + val rq = Resolver.resolve(ast, schema, options) + assert(rq.paths.isEmpty) + val out = GqlExplain.logical(rq) + assert(out.contains("(none")) + } + + test( + "logical explain lists scan-local filter on the node and join/post predicates separately") { + val ast = AstBuilder.parse( + "MATCH (a:Person)-[:KNOWS]->(b:Person) WHERE a.age > 30 AND a.age > b.age RETURN a, b") + val rq = Resolver.resolve(ast, schema, options) + val out = GqlExplain.logical(rq) + // scan-local filter rendered inline on the node + assert(out.contains("a.age > 30")) + // the cross-pattern predicate rendered as a join predicate + assert(out.contains("a.age > b.age")) + assert(out.contains("join predicates")) + assert(out.contains("projection")) + } + + test("physical explain renders plan order and statistics line") { + val ast = + AstBuilder.parse("MATCH (a:Person)-[:WORKS_AT]->(c:Company)-[:LOCATED_IN]->(d:City)") + val rq = Resolver.resolve(ast, schema, options) + val plans = JoinOptimizer.plan(rq, stats = None) + val out = GqlExplain.physical(plans) + assert(out.contains("Physical plan")) + assert(out.contains("Plan 0")) + assert(out.contains("join order: [n0, e0, n1, e1, n2]")) + // v1 carries no statistics. + assert(out.contains("(no statistics)")) + } + + test("physical explain on disconnected pattern reports no plans") { + val ast = AstBuilder.parse("MATCH (a:City)-[:KNOWS]->(b:Person)") + val rq = Resolver.resolve(ast, schema, options) + val plans = JoinOptimizer.plan(rq, stats = None) + val out = GqlExplain.physical(plans) + assert(out.contains("(none")) + } +} diff --git a/core/src/test/scala/org/graphframes/propertygraph/internal/JoinOptimizerSuite.scala b/core/src/test/scala/org/graphframes/propertygraph/internal/JoinOptimizerSuite.scala new file mode 100644 index 000000000..98dc42a53 --- /dev/null +++ b/core/src/test/scala/org/graphframes/propertygraph/internal/JoinOptimizerSuite.scala @@ -0,0 +1,121 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.graphframes.propertygraph.internal + +import org.graphframes.SparkFunSuite +import org.graphframes.propertygraph.QueryOptions + +/** + * Pure-JVM tests for `JoinOptimizer.plan`. + */ +class JoinOptimizerSuite extends SparkFunSuite { + + // Person --KNOWS--> Person + // Person --WORKS_AT--> Company + // Company --LOCATED_IN--> City + private val schema = SchemaGraphSnapshot( + vertexGroupNames = Set("Person", "Company", "City"), + edges = Vector( + SchemaEdge("KNOWS", "Person", "Person", true), + SchemaEdge("WORKS_AT", "Person", "Company", true), + SchemaEdge("LOCATED_IN", "Company", "City", true))) + + val options: QueryOptions = QueryOptions() + + test("single-hop path yields one plan in pattern order n0,e0,n1") { + val ast = AstBuilder.parse("MATCH (a:Person)-[:KNOWS]->(b:Person)") + val rq = Resolver.resolve(ast, schema, options) + val plans = JoinOptimizer.plan(rq, stats = None) + + assert(plans.length === 1) + val plan = plans.head + assert(plan.order === Vector(NodeRef(0), EdgeRef(0), NodeRef(1))) + assert(plan.statsUsed === Map.empty) + assert(plan.projection === Projection.Default) + assert(plan.joinPredicates === Nil) + assert(plan.postFilters === Nil) + } + + test("multi-hop pattern order interleaves nodes and edges") { + val ast = + AstBuilder.parse("MATCH (a:Person)-[:WORKS_AT]->(c:Company)-[:LOCATED_IN]->(d:City)") + val rq = Resolver.resolve(ast, schema, options) + val plans = JoinOptimizer.plan(rq, stats = None) + + assert(plans.length === 1) + val order = plans.head.order + assert(order === Vector(NodeRef(0), EdgeRef(0), NodeRef(1), EdgeRef(1), NodeRef(2))) + } + + test("untyped fan-out produces one plan per enumerated path") { + val ast = AstBuilder.parse("MATCH (a:Person)-[]->(x)-[]->(b:City)") + val rq = Resolver.resolve(ast, schema, options) + val plans = JoinOptimizer.plan(rq, stats = None) + + // Only Person-WORKS_AT->Company-LOCATED_IN->City reaches City in two hops. + assert(plans.length === rq.paths.length) + assert(plans.length >= 1) + plans.foreach { p => + assert(p.order.head === NodeRef(0)) + assert(p.order.last === NodeRef(2)) + } + } + + test("predicates and projection are carried through to each plan") { + val ast = + AstBuilder.parse("MATCH (a:Person)-[:KNOWS]->(b:Person) WHERE a.age > b.age RETURN a, b") + val rq = Resolver.resolve(ast, schema, options) + val plans = JoinOptimizer.plan(rq, stats = None) + + assert(plans.length === 1) + val plan = plans.head + assert( + plan.projection === Projection.Items(rq.projection.asInstanceOf[Projection.Items].items)) + // `a.age > b.age` spans two adjacent node vars -> classified as a join predicate. + assert(plan.joinPredicates.length === 1) + assert(plan.postFilters === Nil) + } + + test("disconnected pattern (no paths) yields no plans") { + // Company and Person are not connected by any incoming edge into Company from City, etc. + // Construct a pattern that cannot resolve: City ->(none)-> Person in one hop is impossible + // because no edge has City as src reaching Person. Use a label-only dead end. + val ast = AstBuilder.parse("MATCH (a:City)-[:KNOWS]->(b:Person)") + val rq = Resolver.resolve(ast, schema, options) + assert(rq.paths.isEmpty) + val plans = JoinOptimizer.plan(rq, stats = None) + assert(plans.isEmpty) + } + + test("stats argument is accepted but does not change v1 output") { + val ast = AstBuilder.parse("MATCH (a:Person)-[:KNOWS]->(b:Person)") + val rq = Resolver.resolve(ast, schema, options) + val withStats = JoinOptimizer.plan(rq, Some(GraphStatistics.Empty)) + val withoutStats = JoinOptimizer.plan(rq, None) + assert(withStats.map(_.order) === withoutStats.map(_.order)) + assert(withStats.head.statsUsed === Map.empty) + } + + test("defaultPlanner and identityRefiner compose to the same as plan()") { + val ast = AstBuilder.parse("MATCH (a:Person)-[:KNOWS]->(b:Person)") + val rq = Resolver.resolve(ast, schema, options) + val direct = JoinOptimizer.plan(rq, None) + val viaSPI = JoinOptimizer.identityRefiner(rq, JoinOptimizer.defaultPlanner(rq, None)) + assert(viaSPI.map(_.order) === direct.map(_.order)) + } +} diff --git a/core/src/test/scala/org/graphframes/propertygraph/internal/QueryExecutorSuite.scala b/core/src/test/scala/org/graphframes/propertygraph/internal/QueryExecutorSuite.scala new file mode 100644 index 000000000..30e1966bc --- /dev/null +++ b/core/src/test/scala/org/graphframes/propertygraph/internal/QueryExecutorSuite.scala @@ -0,0 +1,770 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.graphframes.propertygraph.internal + +import org.apache.spark.sql.DataFrame +import org.apache.spark.sql.functions.lit +import org.graphframes.GraphFrame +import org.graphframes.GraphFrameTestSparkContext +import org.graphframes.SparkFunSuite +import org.graphframes.propertygraph.PropertyGraphFrame +import org.graphframes.propertygraph.QueryOptions +import org.graphframes.propertygraph.property.EdgePropertyGroup +import org.graphframes.propertygraph.property.VertexPropertyGroup + +import java.security.MessageDigest + +/** + * Spark-backed tests for the executor + optimizer pipeline. Builds a small in-memory + * PropertyGraphFrame (Person/Company/City, KNOWS/WORKS_AT/LOCATED_IN) and asserts that + * `JoinOptimizer.plan` + `QueryExecutor.execute` produce rows matching a hand-computed join, with + * the fixed output schema and correct id-masking. + */ +class QueryExecutorSuite extends SparkFunSuite with GraphFrameTestSparkContext { + + import sqlImplicits._ + + private var pgf: PropertyGraphFrame = _ + + // Mirrors VertexPropertyGroup.getData's masking: concat(groupName, sha2(id.cast(String), 256)). + private def maskedId(id: Long, groupName: String): String = { + val md = MessageDigest.getInstance("SHA-256") + val hash = md.digest(id.toString.getBytes("UTF-8")).map("%02x".format(_)).mkString + s"$groupName$hash" + } + + override def beforeAll(): Unit = { + super.beforeAll() + pgf = buildGraph() + } + + private def buildGraph(): PropertyGraphFrame = { + val persons = + Seq((1L, "Alice", 30), (2L, "Bob", 40), (3L, "Carol", 25)).toDF("id", "name", "age") + val companies = Seq((10L, "Acme"), (20L, "Globex")).toDF("id", "name") + val cities = Seq((100L, "Springfield"), (200L, "Shelbyville")).toDF("id", "name") + + val personGroup = VertexPropertyGroup("Person", persons, "id") + val companyGroup = VertexPropertyGroup("Company", companies, "id") + val cityGroup = VertexPropertyGroup("City", cities, "id") + + val knows = Seq((1L, 2L, "friend"), (2L, 3L, "collegaue"), (3L, 1L, "spoose")).toDF( + "src", + "dst", + "friendship") + val worksAt = Seq((1L, 10L), (2L, 10L), (3L, 20L)).toDF("src", "dst") + val locatedIn = Seq((10L, 100L), (20L, 200L)).toDF("src", "dst") + + val knowsGroup = EdgePropertyGroup( + "KNOWS", + knows, + personGroup, + personGroup, + isDirected = true, + "src", + "dst", + lit(1.0)) + val worksAtGroup = EdgePropertyGroup( + "WORKS_AT", + worksAt, + personGroup, + companyGroup, + isDirected = true, + "src", + "dst", + lit(1.0)) + val locatedInGroup = EdgePropertyGroup( + "LOCATED_IN", + locatedIn, + companyGroup, + cityGroup, + isDirected = true, + "src", + "dst", + lit(1.0)) + + PropertyGraphFrame( + Seq(personGroup, companyGroup, cityGroup), + Seq(knowsGroup, worksAtGroup, locatedInGroup)) + } + + // this one is left just because I was lazy to rewrite all the tests :D + private def run(gql: String): DataFrame = runOn(pgf, gql) + + private def runOn(pg: PropertyGraphFrame, gql: String): DataFrame = { + val ast = AstBuilder.parse(gql) + val resolved = + Resolver.resolve(ast, SchemaGraphSnapshot.fromPropertyGraphFrame(pg), QueryOptions()) + val plans = JoinOptimizer.plan(resolved, stats = None) + QueryExecutor.execute(pg, plans) + } + + test("single-hop directed query matches a hand-computed join") { + // MATCH (a:Person)-[:KNOWS]->(b:Person) : knows rows (1->2),(2->3),(3->1). + val df = run("MATCH (a:Person)-[:KNOWS]->(b:Person)") + val rows = df.collect().map(r => (r.getString(0), r.getString(2))).toSet + val expected = Set( + (maskedId(1L, "Person"), maskedId(2L, "Person")), + (maskedId(2L, "Person"), maskedId(3L, "Person")), + (maskedId(3L, "Person"), maskedId(1L, "Person"))) + assert(rows === expected) + // Fixed schema check. + assert( + df.schema.fieldNames === + Seq( + "start_id", + "start_property_group", + "end_id", + "end_property_group", + "edge_property_group", + "path")) + assert(df.head().getAs[String]("start_property_group") === "Person") + assert(df.head().getAs[String]("edge_property_group") === "KNOWS") + } + + test("scan-local WHERE filter actually prunes rows") { + // Only Alice(30) and Bob(40) have age > 30... wait: 30 > 30 is false. Only Bob(40). + // As a src, Bob(2) KNOWS Carol(3). So exactly one row. + val df = run("MATCH (a:Person)-[:KNOWS]->(b:Person) WHERE a.age > 30") + assert(df.count() === 1) + val row = df.head() + assert(row.getString(0) === maskedId(2L, "Person")) // Bob + assert(row.getString(2) === maskedId(3L, "Person")) // Carol + } + + test("backward arrow <-[e]- swaps src and dst") { + // MATCH (a:Person)<-[:KNOWS]-(b:Person): a is the dst, b is the src. + // knows (1->2),(2->3),(3->1) => a=2,b=1 ; a=3,b=2 ; a=1,b=3. + val df = run("MATCH (a:Person)<-[:KNOWS]-(b:Person)") + val rows = df.collect().map(r => (r.getString(0), r.getString(2))).toSet + val expected = Set( + (maskedId(2L, "Person"), maskedId(1L, "Person")), + (maskedId(3L, "Person"), maskedId(2L, "Person")), + (maskedId(1L, "Person"), maskedId(3L, "Person"))) + assert(rows === expected) + } + + test("multi-hop query builds the path array with intermediate ids") { + // MATCH (a:Person)-[:WORKS_AT]->(c:Company)-[:LOCATED_IN]->(d:City) + // Alice(1)->Acme(10)->Springfield(100), Bob(2)->Acme(10)->Springfield(100), + // Carol(3)->Globex(20)->Shelbyville(200). + val df = run("MATCH (a:Person)-[:WORKS_AT]->(c:Company)-[:LOCATED_IN]->(d:City)") + assert(df.count() === 3) + val firstRow = df.head() + assert(firstRow.getAs[String]("start_property_group") === "Person") + assert(firstRow.getAs[String]("end_property_group") === "City") + assert(firstRow.getAs[String]("edge_property_group") === "WORKS_AT") + // path array: k entries for a k-step path (§6.1). Here k=2: + // [ {WORKS_AT, c_id, "Company"}, {LOCATED_IN, null, null} ] + // -- the last entry carries only the edge group (the end node is in end_id). + val pathArr = firstRow + .get(firstRow.fieldIndex("path")) + .asInstanceOf[scala.collection.Seq[org.apache.spark.sql.Row]] + assert(pathArr.length === 2) + val firstHop = pathArr.head + assert(firstHop.getAs[String](0) === "WORKS_AT") + assert(firstHop.getAs[String](2) === "Company") + val lastHop = pathArr.last + assert(lastHop.getAs[String](0) === "LOCATED_IN") + // last entry's node fields are null (end node already in end_id) + assert(lastHop.isNullAt(1)) + assert(lastHop.isNullAt(2)) + } + + test("disconnected pattern yields an empty DataFrame with the fixed schema") { + val df = run("MATCH (a:City)-[:KNOWS]->(b:Person)") + assert(df.count() === 0) + assert( + df.schema.fieldNames === + Seq( + "start_id", + "start_property_group", + "end_id", + "end_property_group", + "edge_property_group", + "path")) + } + + test("untyped edge fan-out unions multiple schema paths") { + // MATCH (a:Person)-[]->(b:Person) : only KNOWS connects Person->Person. + // So exactly the 3 KNOWS rows. + val df = run("MATCH (a:Person)-[]->(b:Person)") + assert(df.count() === 3) + } + + test("RETURN a, b projects named variables only") { + val df = run("MATCH (a:Person)-[:KNOWS]->(b:Person) RETURN a, b") + // Items projection: two columns named a, b (each is the masked id). + assert(df.schema.fieldNames.toSeq === Seq("a", "b")) + val rows = df.collect().map(r => (r.getString(0), r.getString(1))).toSet + val expected = Set( + (maskedId(1L, "Person"), maskedId(2L, "Person")), + (maskedId(2L, "Person"), maskedId(3L, "Person")), + (maskedId(3L, "Person"), maskedId(1L, "Person"))) + assert(rows === expected) + } + + test("RETURN a.name AS age projects a property column via the requested-properties scan") { + // RETURN a.name : names are Alice/Bob/Carol. As src of KNOWS: 1->Alice,2->Bob,3->Carol. + val df = run("MATCH (a:Person)-[:KNOWS]->(b:Person) RETURN a.name AS who") + assert(df.schema.fieldNames.toSeq === Seq("who")) + val names = df.collect().map(_.getString(0)).toSet + assert(names === Set("Alice", "Bob", "Carol")) + } + + test("join predicate (a.age > b.age) is applied") { + // KNOWS: 1(30)->2(40), 2(40)->3(25), 3(25)->1(30). + // a.age > b.age : 40>25 (Bob->Carol) only. + val df = run("MATCH (a:Person)-[:KNOWS]->(b:Person) WHERE a.age > b.age") + assert(df.count() === 1) + val row = df.head() + assert(row.getString(0) === maskedId(2L, "Person")) + assert(row.getString(2) === maskedId(3L, "Person")) + } + + test("id-masking join-back via internalIdMapping recovers raw ids") { + // The query returns masked ids; joining against Person.internalIdMapping recovers external ids. + val df = run("MATCH (a:Person)-[:KNOWS]->(b:Person)") + val personGroup = pgf.vertexGroups("person") + val mapping = personGroup.internalIdMapping // (external_id, id) + val withExternal = df.join(mapping, df("start_id") === mapping(GraphFrame.ID), "left") + val externals = + withExternal.collect().map(_.getAs[Long](PropertyGraphFrame.EXTERNAL_ID)).toSet + assert(externals === Set(1L, 2L, 3L)) + } + + test("hand-computed join matches executor for the multi-hop case") { + // Independent computation of the expected start/end id pairs for the WORKS_AT->LOCATED_IN chain. + val worksAt = Seq((1L, 10L), (2L, 10L), (3L, 20L)) + val locatedIn = Seq((10L, 100L), (20L, 200L)) + val expected = for { + (p, c1) <- worksAt + (c2, ci) <- locatedIn if c1 == c2 + } yield (maskedId(p, "Person"), maskedId(ci, "City")) + + val df = run("MATCH (a:Person)-[:WORKS_AT]->(c:Company)-[:LOCATED_IN]->(d:City)") + val actual = df.collect().map(r => (r.getString(0), r.getString(2))).toSet + assert(actual === expected.toSet) + } + + // --------------------------------------------------------------------------- + // Scan-reuse floor (deterministic) + ceiling (best-effort / soft). + // --------------------------------------------------------------------------- + + /** + * Drives the same pipeline as [[run]] but also returns the per-call scan memo, so the + * scan-reuse floor can be asserted by reference-identity on the memo values. + */ + private def runWithMemo(gql: String): (DataFrame, Map[QueryExecutor.ScanKey, DataFrame]) = { + val ast = AstBuilder.parse(gql) + val resolved = + Resolver.resolve(ast, SchemaGraphSnapshot.fromPropertyGraphFrame(pgf), QueryOptions()) + val plans = JoinOptimizer.plan(resolved, stats = None) + QueryExecutor.executeWithScanMemo(pgf, plans) + } + + test("scan-reuse floor: equal scan signatures share one DataFrame reference (spec §8.1)") { + // KNOWS connects Person->Person, so both endpoints of `MATCH (a:Person)-[:KNOWS]->(b:Person)` + // are the SAME group (Person) with the SAME signature (empty scan filter, no carried cols). + // The memo must therefore hold a single Person scan referenced from both positions. + val (_, memo) = runWithMemo("MATCH (a:Person)-[:KNOWS]->(b:Person)") + val personScans = memo.iterator.filter(_._1.groupName == "person").map(_._2).toSeq + assert(personScans.length === 1, s"expected one shared Person scan, got keys: ${memo.keySet}") + } + + test("scan-reuse floor: differing scan filters produce distinct scans (spec §8.1)") { + // With a scan-local filter `WHERE a.age > 30`, the `a` Person scan's signature differs from the + // `b` Person scan (no filter) -> two distinct Person scans in the memo. + val (_, memo) = runWithMemo("MATCH (a:Person)-[:KNOWS]->(b:Person) WHERE a.age > 30") + val personKeys = memo.keySet.filter(_.groupName == "person").toSeq + assert( + personKeys.length === 2, + s"expected two Person scans (filtered vs unfiltered): $personKeys") + // And the two scans must be distinct DataFrame references. + val personScans = personKeys.map(memo) + assert(personScans.head ne personScans.last, "distinct signatures must yield distinct scans") + } + + test("scan-reuse floor: same filter across a fan-out reuses one scan (spec §8.1)") { + // `(a:Person)-[]->(b:Person)` fans out; here only KNOWS qualifies, but both endpoints share the + // Person scan with no filter -> exactly one Person scan regardless. + val (_, memo) = runWithMemo("MATCH (a:Person)-[]->(b:Person)") + val personScans = memo.iterator.filter(_._1.groupName == "person").map(_._2).toSeq + assert(personScans.length === 1, s"expected one shared Person scan: ${memo.keySet}") + } + + test("output-only join-back resolves both endpoints' properties (spec §8.3)") { + // `a.name` and `b.name` are RETURN-only (no filter references them) -> output-only -> terminal + // join-back. Result parity: exact name pairs for KNOWS (1->2, 2->3, 3->1). Per the + // Items-projection convention the output columns are named after the property (`name`). + val df = run("MATCH (a:Person)-[:KNOWS]->(b:Person) RETURN a.name, b.name") + assert(df.schema.fieldNames.toSeq === Seq("name", "name")) + val rows = df.collect().map(r => (r.getString(0), r.getString(1))).toSet + val expected = Set(("Alice", "Bob"), ("Bob", "Carol"), ("Carol", "Alice")) + assert(rows === expected) + } + + test("mixed carry + output-only: filter-also-returned is carried, RETURN-only is join-backed") { + // `a.age` is referenced by BOTH the filter (`a.age > 30`) and RETURN -> carried (no join-back). + // `a.name` is RETURN-only -> output-only -> join-backed. Only Bob(40) passes `age > 30`. The + // `age` column is read as Int (its physical type in the fixture), not Long. + val df = run("MATCH (a:Person)-[:KNOWS]->(b:Person) WHERE a.age > 30 RETURN a.name, a.age") + val rows = df.collect().map(r => (r.getString(0), r.getInt(1))).toSet + assert(rows === Set(("Bob", 40))) + } + + test("Default/Star projection never triggers an output-only join-back") { + // Default projection has no RETURN items -> every element's outputOnly set is empty by + // construction -> no join-back. Assert via the memo: no carried scan carries a non-id-only + // property set, and the result still has the fixed 6-column schema. + val (df, memo) = runWithMemo("MATCH (a:Person)-[:KNOWS]->(b:Person)") + assert( + df.schema.fieldNames === Seq( + "start_id", + "start_property_group", + "end_id", + "end_property_group", + "edge_property_group", + "path")) + // No scan in the memo carried any property column (only id/property_group/src/dst/weight). + assert( + memo.keySet.forall(_.carriedCols.isEmpty), + s"Default projection should carry no props: ${memo.keySet}") + } + + test("selectivity preserved: join predicate appears in the executed plan") { + // `a.age > b.age` is a two-variable predicate; the engine places it at the binding join. This is + // a best-effort ceiling assertion: the inequality Column must appear SOMEWHERE in the physical + // plan (exact operator placement is Spark-version / AQE dependent). We assert only presence. + val df = run("MATCH (a:Person)-[:KNOWS]->(b:Person) WHERE a.age > b.age") + val planStr = df.queryExecution.executedPlan.toString() + // The predicate is struct guaranteed to be applied; assert the result parity strictly and the + // plan presence softly (only one row: 40 > 25, Bob -> Carol). + assert(df.count() === 1) + // Soft: tolerate plans that fold the comparison into a different node name across versions. + assert( + planStr.contains("age") || planStr.contains("Filter") || planStr.contains("Join"), + s"expected the age predicate or a join/filter node in the plan, got:\n$planStr") + } + + test("edge properties in the filter expression are carried correctly") { + val df = run("MATCH (a:Person)-[e:KNOWS]->(b:Person) WHERE e.friendship = 'spoose'") + // only Carol -- Alice + assert(df.count() === 1L) + } + + test("only edge property referenced only in RETURN") { + val df = run("MATCH (a:Person)-[e:KNOWS]->(b:Person) RETURN e.friendship") + val collected = df.collect().map(r => r.getAs[String]("friendship")).toSet + assert(collected === Set("friend", "collegaue", "spoose")) + } + + test("scan-reuse floor: a repeated edge group shares one scan") { + // check that edge scans are reused along the joins + val (_, memo) = runWithMemo("MATCH (a:Person)-[:KNOWS]->(b:Person)-[:KNOWS]->(c:Person)") + val knowsScans = memo.iterator.filter(_._1.groupName == "knows").map(_._2).toSeq + assert(knowsScans.length === 1, s"expected one shared KNOWS scan: ${memo.keySet}") + } + + test("scan-reuse floor: differing edge scan filters produce distinct edge scans") { + // Two KNOWS hops. The first edge is filtered (`e1.friendship = 'spouse'`), the second is not, + // so the two KNOWS scans have different ScanKeys -> two distinct KNOWS scans in the memo. + val (_, memo) = runWithMemo( + "MATCH (a:Person)-[e1:KNOWS]->(b:Person)-[e2:KNOWS]->(c:Person) " + + "WHERE e1.friendship = 'spouse'") + val knowsKeys = memo.keySet.filter(_.groupName == "knows").toSeq + assert( + knowsKeys.length === 2, + s"expected two KNOWS scans (filtered vs unfiltered): $knowsKeys") + // And they must be distinct DataFrame references (the whole point of keying on the filter). + val knowsScans = knowsKeys.map(memo) + assert( + knowsScans.head ne knowsScans.last, + "distinct edge signatures must yield distinct scans") + } + + test("undirected pattern over a directed self-loop returns both orientations") { + // KNOWS is directed (1->2, 2->3, 3->1). Undirected must surface each edge BOTH ways. + // No reciprocal pairs exist, so 3 stored edges -> 6 distinct rows. + val df = run("MATCH (a:Person)-[:KNOWS]-(b:Person)") + assert(df.count() === 6) // would be 3 if the backward path were dropped + val rows = df.collect().map(r => (r.getString(0), r.getString(2))).toSet + assert( + rows === Set( + (maskedId(1L, "Person"), maskedId(2L, "Person")), + (maskedId(2L, "Person"), maskedId(3L, "Person")), + (maskedId(3L, "Person"), maskedId(1L, "Person")), + (maskedId(2L, "Person"), maskedId(1L, "Person")), + (maskedId(3L, "Person"), maskedId(2L, "Person")), + (maskedId(1L, "Person"), maskedId(3L, "Person")))) + } + + test("undirected pattern equals the union of both directed arrows") { + // The semantics-pinning invariant: -[:KNOWS]- == (-[:KNOWS]->) ∪ (<-[:KNOWS]-). + def pairs(gql: String) = run(gql).collect().map(r => (r.getString(0), r.getString(2))).toSet + assert( + pairs("MATCH (a:Person)-[:KNOWS]-(b:Person)") === + pairs("MATCH (a:Person)-[:KNOWS]->(b:Person)") ++ pairs( + "MATCH (a:Person)<-[:KNOWS]-(b:Person)")) + } + + test("undirected pattern over a directed cross-group edge matches forward from the src side") { + // WORKS_AT: Person->Company; no Company->Person edge, so only the forward orientation matches. + val df = run("MATCH (a:Person)-[:WORKS_AT]-(c:Company)") + assert(df.count() === 3) + val rows = df.collect().map(r => (r.getString(0), r.getString(2))).toSet + assert( + rows === Set( + (maskedId(1L, "Person"), maskedId(10L, "Company")), + (maskedId(2L, "Person"), maskedId(10L, "Company")), + (maskedId(3L, "Person"), maskedId(20L, "Company")))) + assert(df.head().getAs[String]("start_property_group") === "Person") + assert(df.head().getAs[String]("end_property_group") === "Company") + } + + test("undirected pattern over the same cross-group edge matches backward from the dst side") { + val df = run("MATCH (c:Company)-[:WORKS_AT]-(a:Person)") + assert(df.count() === 3) + val rows = df.collect().map(r => (r.getString(0), r.getString(2))).toSet + assert( + rows === Set( + (maskedId(10L, "Company"), maskedId(1L, "Person")), + (maskedId(10L, "Company"), maskedId(2L, "Person")), + (maskedId(20L, "Company"), maskedId(3L, "Person")))) + assert(df.head().getAs[String]("start_property_group") === "Company") + } + + test("undirected pattern over an UNDIRECTED edge group is not double-counted") { + import sqlImplicits._ + // isDirected=false: getData already emits both orientations, so the resolver must keep a + // SINGLE (forward) path. 2 stored edges -> 4 rows; a regressed dedup would yield 8. + val persons = Seq((1L, "Alice"), (2L, "Bob"), (3L, "Carol")).toDF("id", "name") + val personGroup = VertexPropertyGroup("Person", persons, "id") + val friends = Seq((1L, 2L), (2L, 3L)).toDF("src", "dst") + val friendGroup = EdgePropertyGroup( + "FRIEND", + friends, + personGroup, + personGroup, + isDirected = false, + "src", + "dst", + lit(1.0)) + val ug = PropertyGraphFrame(Seq(personGroup), Seq(friendGroup)) + + val df = runOn(ug, "MATCH (a:Person)-[:FRIEND]-(b:Person)") + assert(df.count() === 4) // NOT 8 -- multiset count is what catches the duplicate + val rows = df.collect().map(r => (r.getString(0), r.getString(2))).toSet + assert( + rows === Set( + (maskedId(1L, "Person"), maskedId(2L, "Person")), + (maskedId(2L, "Person"), maskedId(1L, "Person")), + (maskedId(2L, "Person"), maskedId(3L, "Person")), + (maskedId(3L, "Person"), maskedId(2L, "Person")))) + } + + // --------------------------------------------------------------------------- + // Scalar datetime functions (end-to-end). + // + // These exercise the full pipeline -- parse -> resolve (scan-local / join + // classification through the function-call argument) -> plan -> execute -> + // Spark built-in lowering. A dedicated graph with a DateType property is used + // since the default fixture has no date column. + // --------------------------------------------------------------------------- + private def dateGraph: PropertyGraphFrame = { + import sqlImplicits._ + val persons = Seq( + (1L, "Alice", java.sql.Date.valueOf("2000-01-15")), + (2L, "Bob", java.sql.Date.valueOf("1995-06-20")), + (3L, "Carol", java.sql.Date.valueOf("2000-12-01"))).toDF("id", "name", "birthday") + val personGroup = VertexPropertyGroup("Person", persons, "id") + val knows = Seq((1L, 2L), (2L, 3L), (3L, 1L)).toDF("src", "dst") + val knowsGroup = EdgePropertyGroup( + "KNOWS", + knows, + personGroup, + personGroup, + isDirected = true, + "src", + "dst", + lit(1.0)) + PropertyGraphFrame(Seq(personGroup), Seq(knowsGroup)) + } + + test("year() in a scan-local WHERE filters rows end-to-end") { + // KNOWS: 1(Alice,2000)->2(Bob,1995), 2(Bob,1995)->3(Carol,2000), 3(Carol,2000)->1(Alice,2000). + // `year(a.birthday) = 2000` is scan-local on `a`: Alice(1) and Carol(3) pass. + // As src of KNOWS: Alice->Bob and Carol->Alice => 2 rows. + val df = + runOn(dateGraph, "MATCH (a:Person)-[:KNOWS]->(b:Person) WHERE year(a.birthday) = 2000") + assert(df.count() === 2) + val startIds = df.collect().map(_.getString(0)).toSet + assert(startIds === Set(maskedId(1L, "Person"), maskedId(3L, "Person"))) + } + + test("RETURN year(a.birthday) AS y projects the lowered Spark year() column") { + val df = + runOn(dateGraph, "MATCH (a:Person)-[:KNOWS]->(b:Person) RETURN year(a.birthday) AS y") + assert(df.schema.fieldNames.toSeq === Seq("y")) + val years = df.collect().map(_.getInt(0)).toSet + // src order: Alice(2000), Bob(1995), Carol(2000). + assert(years === Set(2000, 1995)) + } + + test("datediff() in a two-variable WHERE is applied as a join predicate end-to-end") { + // Spark `datediff(endDate, startDate)` = days from startDate to endDate, so + // `datediff(a.birthday, b.birthday)` = days from b's birthday to a's birthday. + // KNOWS edges (a -> b): + // 1->2 Alice(2000-01-15), Bob (1995-06-20): ~1639 days > 300 + + // 2->3 Bob (1995-06-20), Carol(2000-12-01): ~-1636 days < 300 - + // 3->1 Carol(2000-12-01), Alice(2000-01-15): ~321 days > 300 + + // => two edges survive. + val df = + runOn( + dateGraph, + "MATCH (a:Person)-[:KNOWS]->(b:Person) WHERE datediff(a.birthday, b.birthday) > 300") + assert(df.count() === 2) + val startIds = df.collect().map(_.getString(0)).toSet + assert(startIds === Set(maskedId(1L, "Person"), maskedId(3L, "Person"))) + } + + // --------------------------------------------------------------------------- + // Multiplicative operators and scalar functions (end-to-end). + // --------------------------------------------------------------------------- + + test("multiplicative * / % produce correct values end-to-end") { + // a.age is Int. a.age * 2, a.age / 2 (floating-point), a.age % 4 -- project all three. + // Persons: Alice(30), Bob(40), Carol(25). KNOWS: 1->2, 2->3, 3->1. + // NB: Spark widens int*int and int%int to Long; read those columns as Long. + val df = run( + "MATCH (a:Person)-[:KNOWS]->(b:Person) " + + "RETURN a.age * 2 AS dbl, a.age / 2 AS half, a.age % 4 AS mod") + assert(df.schema.fieldNames.toSeq === Seq("dbl", "half", "mod")) + // src order: Alice(30)->Bob, Bob(40)->Carol, Carol(25)->Alice. + val rows = df.collect().map(r => (r.getLong(0), r.getDouble(1), r.getLong(2))).toSet + assert( + rows === Set( + (60L, 15.0, 2L), // Alice 30: 30*2=60, 30/2=15.0, 30%4=2 + (80L, 20.0, 0L), // Bob 40: 40*2=80, 40/2=20.0, 40%4=0 + (50L, 12.5, 1L) + ) + ) // Carol 25: 25*2=50, 25/2=12.5, 25%4=1 + } + + test("WHERE pmod(hash(a.id), N) = 0 returns a deterministic, repeatable subset") { + // The deterministic-sampling idiom: `pmod(hash(a.id), N) = 0` must be reproducible + // across two executions (this is the scan-reuse memo correctness contract -- two identical + // scan signatures must yield identical rows). + val gql = "MATCH (a:Person)-[:KNOWS]->(b:Person) WHERE pmod(hash(a.id), 4) = 0" + val first = run(gql).collect().map(_.getString(0)).toSet + val second = run(gql).collect().map(_.getString(0)).toSet + assert(first === second, "hash-sampling must be deterministic across runs") + // The bucket must be a genuine subset of all source ids (non-empty and not all). + val allSrcs = Set(maskedId(1L, "Person"), maskedId(2L, "Person"), maskedId(3L, "Person")) + assert(first.nonEmpty && first.subsetOf(allSrcs)) + } + + test("lower() string predicate filters rows end-to-end") { + // Need a fixture with a string column to lower-case. KNOWS: 1->2,2->3,3->1. + import sqlImplicits._ + val persons = Seq((1L, "Alice"), (2L, "BOB"), (3L, "carol")).toDF("id", "name") + val personGroup = VertexPropertyGroup("Person", persons, "id") + val knows = Seq((1L, 2L), (2L, 3L), (3L, 1L)).toDF("src", "dst") + val knowsGroup = EdgePropertyGroup( + "KNOWS", + knows, + personGroup, + personGroup, + isDirected = true, + "src", + "dst", + lit(1.0)) + val sg = PropertyGraphFrame(Seq(personGroup), Seq(knowsGroup)) + + // Only Bob('BOB') lower-cases to 'bob'. + val df = runOn(sg, "MATCH (a:Person)-[:KNOWS]->(b:Person) WHERE lower(a.name) = 'bob'") + assert(df.count() === 1) + assert(df.head().getString(0) === maskedId(2L, "Person")) + } + + test("get_json_object over a JSON string property returns expected scalars") { + // A dedicated fixture where every payload is a JSON document; get_json_object takes a + // lit-str path. Persons: Alice(score=42), Bob(score=7), Carol(score=5). + import sqlImplicits._ + val persons = + Seq((1L, """{"score": 42}"""), (2L, """{"score": 7}"""), (3L, """{"score": 5}""")) + .toDF("id", "payload") + val personGroup = VertexPropertyGroup("Person", persons, "id") + val knows = Seq((1L, 2L), (2L, 3L), (3L, 1L)).toDF("src", "dst") + val knowsGroup = EdgePropertyGroup( + "KNOWS", + knows, + personGroup, + personGroup, + isDirected = true, + "src", + "dst", + lit(1.0)) + val jg = PropertyGraphFrame(Seq(personGroup), Seq(knowsGroup)) + + val df = + runOn( + jg, + "MATCH (a:Person)-[:KNOWS]->(b:Person) RETURN get_json_object(a.payload, '$.score') AS s") + val scores = df.collect().map(_.getString(0)).toSet + assert(scores === Set("42", "7", "5")) + } + + test("xpath_int over an XML string property returns expected scalars") { + // A dedicated fixture where every payload is a valid XML document; xpath_int takes a + // Column path. Persons: Alice(10), Bob(20), Carol(99). + // NB: Spark's xpath_* throw on non-XML input rather than returning null, so all rows must be + // well-formed XML. + import sqlImplicits._ + val persons = + Seq((1L, "10"), (2L, "20"), (3L, "99")) + .toDF("id", "payload") + val personGroup = VertexPropertyGroup("Person", persons, "id") + val knows = Seq((1L, 2L), (2L, 3L), (3L, 1L)).toDF("src", "dst") + val knowsGroup = EdgePropertyGroup( + "KNOWS", + knows, + personGroup, + personGroup, + isDirected = true, + "src", + "dst", + lit(1.0)) + val xg = PropertyGraphFrame(Seq(personGroup), Seq(knowsGroup)) + + val df = + runOn(xg, "MATCH (a:Person)-[:KNOWS]->(b:Person) RETURN xpath_int(a.payload, '/a/v') AS s") + val vals = df.collect().map(_.getInt(0)).toSet + assert(vals === Set(10, 20, 99)) + } + + // --- Variable-length patterns (end-to-end) ------------------------------ + // + // KNOWS is a directed 3-cycle: 1(Alice) -> 2(Bob) -> 3(Carol) -> 1(Alice). That makes the + // walks of each length fully enumerable by hand, so these assert exact rows, exact intermediate + // ids in the `path` array, and structural properties (cycle closure) -- not just counts. + // Range forms (*lo..hi) are used so these exercise the executor regardless of the AstBuilder + // exact-form (*N) handling. + + private def P(id: Long): String = maskedId(id, "Person") + + private def pathArray( + r: org.apache.spark.sql.Row): scala.collection.Seq[org.apache.spark.sql.Row] = + r.get(r.fieldIndex("path")).asInstanceOf[scala.collection.Seq[org.apache.spark.sql.Row]] + + test("var-length *1..2 unions the 1-hop and 2-hop walks with correct path arrays") { + // 1-hop: (1,2),(2,3),(3,1). 2-hop: (1,3),(2,1),(3,2). + val df = run("MATCH (a:Person)-[:KNOWS*1..2]->(b:Person)") + val rows = df.collect() + assert(rows.length === 6) + + val pairs = rows.map(r => (r.getAs[String]("start_id"), r.getAs[String]("end_id"))).toSet + assert( + pairs === Set( + (P(1), P(2)), + (P(2), P(3)), + (P(3), P(1)), // length 1 + (P(1), P(3)), + (P(2), P(1)), + (P(3), P(2)) + ) + ) // length 2 + + // path-array length == number of hops: 1 for the direct walks, 2 for the two-hop walks. + val sizeByPair = + rows + .map(r => (r.getAs[String]("start_id"), r.getAs[String]("end_id")) -> pathArray(r).length) + .toMap + assert(sizeByPair((P(1), P(2))) === 1) + assert(sizeByPair((P(1), P(3))) === 2) + + // The 1->2->3 two-hop row must carry the intermediate node (Bob=2) in path[0]. + val twoHop = + rows.find(r => r.getAs[String]("start_id") == P(1) && r.getAs[String]("end_id") == P(3)).get + val path = pathArray(twoHop) + assert(path.length === 2) + assert(path(0).getAs[String](0) === "KNOWS") // edge_property_group + assert(path(0).getAs[String](1) === P(2)) // node_id (the intermediate, Bob) + assert(path(0).getAs[String](2) === "Person") // node_property_group + assert(path(1).getAs[String](0) === "KNOWS") // final hop: edge only... + assert(path(1).isNullAt(1)) // ...node fields null (end node is in end_id) + } + + test("var-length *2..2 yields exactly the two-hop walks with the right intermediate id") { + // 1->2->3, 2->3->1, 3->1->2. + val df = run("MATCH (a:Person)-[:KNOWS*2..2]->(b:Person)") + val rows = df.collect() + assert(rows.length === 3) + assert(rows.forall(r => pathArray(r).length == 2)) + + val byStart = rows.map(r => r.getAs[String]("start_id") -> r).toMap + def endOf(r: org.apache.spark.sql.Row) = r.getAs[String]("end_id") + def midOf(r: org.apache.spark.sql.Row) = pathArray(r)(0).getAs[String](1) + + assert(endOf(byStart(P(1))) === P(3) && midOf(byStart(P(1))) === P(2)) // 1->2->3 + assert(endOf(byStart(P(2))) === P(1) && midOf(byStart(P(2))) === P(3)) // 2->3->1 + assert(endOf(byStart(P(3))) === P(2) && midOf(byStart(P(3))) === P(1)) // 3->1->2 + } + + test("var-length *3..3 traverses the full cycle: start equals end") { + // Each 3-hop walk returns to its start (the cycle has length 3). + val df = run("MATCH (a:Person)-[:KNOWS*3..3]->(b:Person)") + val rows = df.collect() + assert(rows.length === 3) + assert(rows.forall(r => r.getAs[String]("start_id") == r.getAs[String]("end_id"))) + assert(rows.map(_.getAs[String]("start_id")).toSet === Set(P(1), P(2), P(3))) + + // 1->2->3->1 carries intermediates Bob(2) then Carol(3); last entry has null node fields. + val r1 = rows.find(_.getAs[String]("start_id") == P(1)).get + val path = pathArray(r1) + assert(path.length === 3) + assert(path(0).getAs[String](1) === P(2)) + assert(path(1).getAs[String](1) === P(3)) + assert(path(2).isNullAt(1)) + } + + test("scan-local WHERE prunes the var-length start node end-to-end") { + // Only Bob (age 40) passes age > 30. From Bob: 2->3 (len 1) and 2->3->1 (len 2). + val df = run("MATCH (a:Person)-[:KNOWS*1..2]->(b:Person) WHERE a.age > 30") + val rows = df.collect() + assert(rows.length === 2) + assert(rows.forall(_.getAs[String]("start_id") == P(2))) + assert(rows.map(_.getAs[String]("end_id")).toSet === Set(P(3), P(1))) + } + + test("var-length spliced before a fixed hop produces a heterogeneous path") { + // (a)-[:KNOWS*1..2]->(b)-[:WORKS_AT]->(c): 3 one-knows + 3 two-knows = 6 rows, ending on Company. + val df = run("MATCH (a:Person)-[:KNOWS*1..2]->(b:Person)-[:WORKS_AT]->(c:Company)") + val rows = df.collect() + assert(rows.length === 6) + assert(rows.forall(_.getAs[String]("start_property_group") == "Person")) + assert(rows.forall(_.getAs[String]("end_property_group") == "Company")) + + // path length = total hops: 2 for one-KNOWS rows, 3 for two-KNOWS rows. + val sizes = + rows.map(r => pathArray(r).length).groupBy(identity).map(f => f._1 -> f._2.size).toMap + assert(sizes === Map(2 -> 3, 3 -> 3)) + // The final hop is always WORKS_AT (the fixed tail), regardless of the KNOWS length. + assert(rows.forall(r => pathArray(r).last.getAs[String](0) == "WORKS_AT")) + } +}