Major refactoring to make certain Cypher is more lazy

systay · Nov 3, 2012 · 5c592e9 · 5c592e9
1 parent b23f934
commit 5c592e9
Show file tree

Hide file tree

Showing 36 changed files with 524 additions and 161 deletions.
diff --git a/cypher/CHANGES.txt b/cypher/CHANGES.txt
@@ -2,6 +2,7 @@
 -------------------
 o The traversal pattern matcher now supports variable length relationship patterns
 o Fixes #946 - HAS(...) fails with ThisShouldNotHappenException for some patterns
+o Major refactoring to make certain Cypher is more lazy
 
 1.9.M01 (2012-10-23)
 --------------------

diff --git a/cypher/src/main/scala/org/neo4j/cypher/EagerPipeExecutionResult.scala b/cypher/src/main/scala/org/neo4j/cypher/EagerPipeExecutionResult.scala
@@ -23,25 +23,26 @@ import internal.pipes.QueryState
 import org.neo4j.graphdb.GraphDatabaseService
 import collection.Map
 
-class EagerPipeExecutionResult(r: => Traversable[Map[String, Any]],
+class EagerPipeExecutionResult(result: Iterator[Map[String, Any]],
                                columns: List[String],
                                state: QueryState,
                                db: GraphDatabaseService)
-  extends PipeExecutionResult(r, columns) {
+  extends PipeExecutionResult(result, columns) {
 
-  override lazy val queryStatistics = QueryStatistics(
+  val (eagerResult,timeTaken) = super.createTimedResults
-    nodesCreated = state.createdNodes.count,
+  lazy val inner = eagerResult.iterator
-    relationshipsCreated = state.createdRelationships.count,
-    propertiesSet = state.propertySet.count,
-    deletedNodes = state.deletedNodes.count,
-    deletedRelationships = state.deletedRelationships.count)
 
-  override val createTimedResults = {
+  override def next() = inner.next().toMap
-    val start = System.currentTimeMillis()
+  override def hasNext = inner.hasNext
-    val eagerResult = immutableResult.toList
 
-    val ms = System.currentTimeMillis() - start
+  override def queryStatistics = {
-
+    QueryStatistics(
-    (eagerResult, ms.toString)
+      nodesCreated = state.createdNodes.count,
+      relationshipsCreated = state.createdRelationships.count,
+      propertiesSet = state.propertySet.count,
+      deletedNodes = state.deletedNodes.count,
+      deletedRelationships = state.deletedRelationships.count)
   }
+
+  override def createTimedResults = (eagerResult,timeTaken)
 }
diff --git a/cypher/src/main/scala/org/neo4j/cypher/PipeExecutionResult.scala b/cypher/src/main/scala/org/neo4j/cypher/PipeExecutionResult.scala
@@ -27,23 +27,21 @@ import java.io.{StringWriter, PrintWriter}
 import collection.Map
 import collection.immutable.{Map => ImmutableMap}
 
-class PipeExecutionResult(r: => Traversable[Map[String, Any]], val columns: List[String])
+class PipeExecutionResult(result: Iterator[Map[String, Any]], val columns: List[String])
   extends ExecutionResult
   with StringExtras
   with CollectionSupport
   with StringHelper {
 
-  lazy val immutableResult = r.map(m => m.toMap)
-
   def javaColumns: java.util.List[String] = columns.asJava
 
   def javaColumnAs[T](column: String): java.util.Iterator[T] = columnAs[T](column).map(x => makeValueJavaCompatible(x).asInstanceOf[T]).asJava
 
-  def columnAs[T](column: String): Iterator[T] = {
+  def columnAs[T](column: String): Iterator[T] = map {
-    this.map(m => {
+    case m => {
       val item: Any = m.getOrElse(column, throw new EntityNotFoundException("No column named '" + column + "' was found. Found: " + m.keys.mkString("(\"", "\", \"", "\")")))
       item.asInstanceOf[T]
-    }).toIterator
+    }
   }
 
   private def makeValueJavaCompatible(value: Any): Any = value match {
@@ -69,9 +67,9 @@ class PipeExecutionResult(r: => Traversable[Map[String, Any]], val columns: List
     columnSizes.toMap
   }
 
-  protected def createTimedResults = {
+  protected def createTimedResults: (List[Map[String, Any]], String) = {
     val start = System.currentTimeMillis()
-    val eagerResult = immutableResult.toList
+    val eagerResult = result.toList
     val ms = System.currentTimeMillis() - start
 
     (eagerResult, ms.toString)
@@ -137,12 +135,10 @@ class PipeExecutionResult(r: => Traversable[Map[String, Any]], val columns: List
     }).mkString("| ", " | ", " |")
   }
 
-  lazy val iterator = immutableResult.toIterator
+  def hasNext: Boolean = result.hasNext
-
-  def hasNext: Boolean = iterator.hasNext
 
-  def next(): ImmutableMap[String, Any] = iterator.next()
+  def next(): ImmutableMap[String, Any] = result.next().toMap
 
-  lazy val queryStatistics = QueryStatistics.empty
+  def queryStatistics = QueryStatistics.empty
 }
 
diff --git a/cypher/src/main/scala/org/neo4j/cypher/internal/executionplan/ExecutionPlanImpl.scala b/cypher/src/main/scala/org/neo4j/cypher/internal/executionplan/ExecutionPlanImpl.scala
@@ -124,7 +124,8 @@ class ExecutionPlanImpl(inputQuery: Query, graph: GraphDatabaseService) extends
   private def getLazyReadonlyQuery(pipe: Pipe, columns: List[String]): Map[String, Any] => ExecutionResult = {
     val func = (params: Map[String, Any]) => {
       val state = new QueryState(graph, params)
-      new PipeExecutionResult(pipe.createResults(state), columns)
+      val results = pipe.createResults(state)
+      new PipeExecutionResult(results, columns)
     }
 
     func
@@ -133,7 +134,8 @@ class ExecutionPlanImpl(inputQuery: Query, graph: GraphDatabaseService) extends
   private def getEagerReadWriteQuery(pipe: Pipe, columns: List[String]): Map[String, Any] => ExecutionResult = {
     val func = (params: Map[String, Any]) => {
       val state = new QueryState(graph, params)
-      new EagerPipeExecutionResult(pipe.createResults(state), columns, state, graph)
+      val results = pipe.createResults(state)
+      new EagerPipeExecutionResult(results, columns, state, graph)
     }
 
     func

diff --git a/cypher/src/main/scala/org/neo4j/cypher/internal/pipes/CommitPipe.scala b/cypher/src/main/scala/org/neo4j/cypher/internal/pipes/CommitPipe.scala
@@ -34,11 +34,11 @@ class CommitPipe(source: Pipe, graph: GraphDatabaseService) extends PipeWithSour
     }
     try {
       try {
-        val result = source.createResults(state).toList
+        val result = source.createResults(state).toList.iterator
         tx.success()
         result
       } catch {
-        case e => {
+        case e: Throwable => {
           tx.failure()
           throw e
         }

diff --git a/cypher/src/main/scala/org/neo4j/cypher/internal/pipes/EagerAggregationPipe.scala b/cypher/src/main/scala/org/neo4j/cypher/internal/pipes/EagerAggregationPipe.scala
@@ -44,7 +44,7 @@ class EagerAggregationPipe(source: Pipe, val keyExpressions: Map[String, Express
     new SymbolTable(keyIdentifiers ++ aggrIdentifiers)
   }
 
-  def createResults(state: QueryState): Traversable[ExecutionContext] = {
+  def createResults(state: QueryState) = {
     // This is the temporary storage used while the aggregation is going on
     val result = MutableMap[NiceHasher, (ExecutionContext, Seq[AggregationFunction])]()
     val keyNames: Seq[String] = keyExpressions.map(_._1).toSeq
@@ -62,13 +62,13 @@ class EagerAggregationPipe(source: Pipe, val keyExpressions: Map[String, Express
       ctx.newFrom(newMap)
     }
 
-    def createEmptyResult(params:Map[String,Any]): Traversable[ExecutionContext] = {
+    def createEmptyResult(params:Map[String,Any]): Iterator[ExecutionContext] = {
       val newMap = MutableMaps.empty
       val aggregationNamesAndFunctions = aggregationNames zip aggregations.map(_._2.createAggregationFunction.result)
 
       aggregationNamesAndFunctions.toMap
         .foreach { case (name, zeroValue) => newMap += name -> zeroValue  }
-      Traversable(ExecutionContext(newMap, params = params))
+      Iterator(ExecutionContext(newMap, params = params))
     }
 
 
@@ -79,15 +79,13 @@ class EagerAggregationPipe(source: Pipe, val keyExpressions: Map[String, Express
       functions.foreach(func => func(ctx))
     })
 
-    val a = if (result.isEmpty && keyNames.isEmpty) {
+    if (result.isEmpty && keyNames.isEmpty) {
       createEmptyResult(state.params)
     } else {
       result.map {
         case (key, (ctx, aggregator)) => createResults(key, aggregator, ctx)
-      }
+      }.toIterator
     }
-
-    a
   }
 
   override def executionPlan(): String = source.executionPlan() + "\r\n" + "EagerAggregation( keys: [" + oldKeyExpressions.mkString(", ") + "], aggregates: [" + aggregations.mkString(", ") + "])"

diff --git a/cypher/src/main/scala/org/neo4j/cypher/internal/pipes/EmptyResultPipe.scala b/cypher/src/main/scala/org/neo4j/cypher/internal/pipes/EmptyResultPipe.scala
@@ -25,13 +25,14 @@ class EmptyResultPipe(source: Pipe)
   extends PipeWithSource(source) {
 
   def createResults(state: QueryState) = {
-    source.createResults(state)
+    val iter = source.createResults(state)
+    while(iter.hasNext) {
+      iter.next()
+    }
 
-    Seq()
+    Iterator()
   }
 
-//  def symbols = new SymbolTable()
-
   override def executionPlan(): String = source.executionPlan() + "\nEmptyResult()"
 
   def dependencies = Seq()

diff --git a/cypher/src/main/scala/org/neo4j/cypher/internal/pipes/ParameterPipe.scala b/cypher/src/main/scala/org/neo4j/cypher/internal/pipes/ParameterPipe.scala
@@ -23,7 +23,7 @@ import java.lang.String
 import org.neo4j.cypher.internal.symbols.SymbolTable
 
 class ParameterPipe() extends Pipe {
-  def createResults(state: QueryState) = Seq(ExecutionContext(params = state.params))
+  def createResults(state: QueryState) = Iterator(ExecutionContext(params = state.params))
 
   val symbols = new SymbolTable()
 

diff --git a/cypher/src/main/scala/org/neo4j/cypher/internal/pipes/Pipe.scala b/cypher/src/main/scala/org/neo4j/cypher/internal/pipes/Pipe.scala
@@ -38,15 +38,15 @@ import java.util.concurrent.atomic.AtomicInteger
  * the execute the query.
  */
 trait Pipe {
-  def createResults(state: QueryState): Traversable[ExecutionContext]
+  def createResults(state: QueryState): Iterator[ExecutionContext]
 
   def symbols: SymbolTable
 
   def executionPlan(): String
 }
 
 class NullPipe extends Pipe {
-  def createResults(state: QueryState) = Seq(ExecutionContext.empty)
+  def createResults(state: QueryState) = Seq(ExecutionContext.empty).toIterator
 
   def symbols: SymbolTable = new SymbolTable()
 

diff --git a/cypher/src/main/scala/org/neo4j/cypher/internal/pipes/SlicePipe.scala b/cypher/src/main/scala/org/neo4j/cypher/internal/pipes/SlicePipe.scala
@@ -20,31 +20,30 @@
 package org.neo4j.cypher.internal.pipes
 
 import org.neo4j.cypher.internal.commands.expressions.Expression
-import java.lang.String
 import org.neo4j.helpers.ThisShouldNotHappenError
-import collection.mutable.Map
 
 class SlicePipe(source:Pipe, skip:Option[Expression], limit:Option[Expression]) extends Pipe {
-//  val symbols = source.symbols
+
   val symbols = source.symbols
 
-  //TODO: Make this nicer. I'm sure it's expensive and silly.
+  def createResults(state: QueryState) : Iterator[ExecutionContext] = {
-  def createResults(state: QueryState): Traversable[ExecutionContext] = {
     val sourceTraversable = source.createResults(state)
 
     if(sourceTraversable.isEmpty)
-      return Seq()
+      return Iterator()
+
+    val first: ExecutionContext = sourceTraversable.next()
 
-    val first: ExecutionContext = sourceTraversable.head
+    val sourceIter = new HeadAndTail[ExecutionContext](first, sourceTraversable)
 
     def asInt(v:Expression)=v(first).asInstanceOf[Int]
 
     (skip, limit) match {
-      case (Some(x), None) => sourceTraversable.drop(asInt(x))
+      case (Some(x), None) => sourceIter.drop(asInt(x))
-      case (None, Some(x)) => sourceTraversable.take(asInt(x))
+      case (None, Some(x)) => sourceIter.take(asInt(x))
       case (Some(startAt), Some(count)) => {
         val start = asInt(startAt)
-        sourceTraversable.slice(start, start + asInt(count))
+        sourceIter.slice(start, start + asInt(count))
       }
       case (None, None)=>throw new ThisShouldNotHappenError("Andres Taylor", "A slice pipe that doesn't slice should never exist.")
     }
@@ -60,4 +59,18 @@ class SlicePipe(source:Pipe, skip:Option[Expression], limit:Option[Expression])
     }
     source.executionPlan() + "\r\n" + "Slice(" + info + ")"
   }
+}
+
+class HeadAndTail[T](head:T, tail:Iterator[T]) extends Iterator[T] {
+  var usedHead = false
+  def headUnused = !usedHead
+
+  def hasNext = headUnused || tail.hasNext
+
+  def next() = if (headUnused) {
+    usedHead = true
+    head
+  } else {
+    tail.next()
+  }
 }
diff --git a/cypher/src/main/scala/org/neo4j/cypher/internal/pipes/SortPipe.scala b/cypher/src/main/scala/org/neo4j/cypher/internal/pipes/SortPipe.scala
@@ -35,7 +35,9 @@ class SortPipe(source: Pipe, sortDescription: List[SortItem]) extends PipeWithSo
     }
   }
 
-  def createResults(state:QueryState) = source.createResults(state).toList.sortWith((a, b) => compareBy(a, b, sortDescription))
+  def createResults(state:QueryState) =
+    source.createResults(state).toList.
+    sortWith((a, b) => compareBy(a, b, sortDescription)).iterator
 
   def compareBy(a: Map[String, Any], b: Map[String, Any], order: Seq[SortItem]): Boolean = order match {
     case Nil => false

diff --git a/cypher/src/main/scala/org/neo4j/cypher/internal/pipes/StartPipe.scala b/cypher/src/main/scala/org/neo4j/cypher/internal/pipes/StartPipe.scala
@@ -30,14 +30,13 @@ abstract class StartPipe[T <: PropertyContainer](inner: Pipe, name: String, crea
 
   val symbols = inner.symbols.add(name, identifierType)
 
-  def createResults(state: QueryState): Traversable[ExecutionContext] = {
+  def createResults(state: QueryState) = {
-    val map = inner.createResults(state).flatMap(ctx => {
+    inner.createResults(state).flatMap(ctx => {
       val source: Iterable[T] = createSource(ctx)
       source.map(x => {
         ctx.newWith(name -> x)
       })
     })
-    map
   }
 
   def visibleName: String