Merge pull request twitter-archive#86 from twitter/fix_difference_query

Make DifferenceQuery obey page size
rgaidot · Apr 2, 2012 · 48fcca1 · 48fcca1
2 parents be69af8 + e1fb576
commit 48fcca1
Show file tree

Hide file tree

Showing 3 changed files with 12 additions and 8 deletions.
diff --git a/src/main/scala/com/twitter/flockdb/ResultWindow.scala b/src/main/scala/com/twitter/flockdb/ResultWindow.scala
@@ -58,27 +58,32 @@ class ResultWindow[T](val data: ResultWindowRows[T], val inNextCursor: Cursor, v
   val nextCursor = if (nextChanged && !page.isEmpty) page(page.size - 1).cursor else inNextCursor
   val prevCursor = if (prevChanged && !page.isEmpty) page(0).cursor.reverse else inPrevCursor
 
-  def ++(other: ResultWindow[T]) = {
+  def ++(other: ResultWindow[T]) = concat(other)
+
+  def concat(other: ResultWindow[T], newCount: Int = count) = {
     if (cursor < Cursor.Start) {
-      new ResultWindow(new ResultWindowRows(other.page ++ page), nextCursor, other.prevCursor, count, cursor)
+      new ResultWindow(new ResultWindowRows(other.page ++ page), nextCursor, other.prevCursor, newCount, cursor)
     } else {
-      new ResultWindow(new ResultWindowRows(page ++ other.page), other.nextCursor, prevCursor, count, cursor)
+      new ResultWindow(new ResultWindowRows(page ++ other.page), other.nextCursor, prevCursor, newCount, cursor)
     }
   }
 
+
   def merge(other: ResultWindow[T]) = {
     val newPage = Sorting.stableSort((Set((page ++ other.page): _*)).toSeq)
     val newNextCursor = if (nextCursor == Cursor.End && other.nextCursor == Cursor.End) Cursor.End else newPage(newPage.size - 1).cursor
     val newPrevCursor = if (prevCursor == Cursor.End && other.prevCursor == Cursor.End) Cursor.End else newPage(0).cursor.reverse
     new ResultWindow(new ResultWindowRows(newPage), newNextCursor, newPrevCursor, count, cursor)
   }
 
-  def --(values: Seq[T]) = {
+  def --(values: Seq[T]) = diff(values)
+
+  def diff(values: Seq[T], newCount: Int = count) = {
     val rejects = Set(values: _*)
     val newPage = page.filter { row => !rejects.contains(row.id) }
     val newNextCursor = if (nextCursor == Cursor.End || newPage.size == 0) Cursor.End else newPage(newPage.size - 1).cursor
     val newPrevCursor = if (prevCursor == Cursor.End || newPage.size == 0) Cursor.End else newPage(0).cursor.reverse
-    new ResultWindow(new ResultWindowRows(newPage), newNextCursor, newPrevCursor, count, cursor)
+    new ResultWindow(new ResultWindowRows(newPage), newNextCursor, newPrevCursor, newCount, cursor)
   }
 
   def length = page.length

diff --git a/src/main/scala/com/twitter/flockdb/queries/DifferenceQuery.scala b/src/main/scala/com/twitter/flockdb/queries/DifferenceQuery.scala
@@ -61,7 +61,7 @@ class DifferenceQuery(query1: QueryTree, query2: QueryTree, averageIntersectionP
     for {
       results <- query1.selectPageByDestinationId(internalPageSize, cursor)
       rejects <- query2.selectWhereIn(results.view)
-    } yield results -- rejects
+    } yield results.diff(rejects, count)
   }
 
   override def toString =

diff --git a/src/test/scala/com/twitter/flockdb/unit/IntersectionQuerySpec.scala b/src/test/scala/com/twitter/flockdb/unit/IntersectionQuerySpec.scala
@@ -24,11 +24,10 @@ object IntersectionQuerySpec extends ConfiguredSpecification with JMocker {
     val query1 = new queries.SeqQuery(List(1,2,3,4,5,6,7,8,9,10))
     val query2 = new queries.SeqQuery(List(1,2,3,4,11))
     val queryConfig = config.intersectionQuery
-    queryConfig.averageIntersectionProportion = 1.0
 
     "sizeEstimate" in {
       val intersectionQuery = queryConfig.intersect(query1, query2)
-      intersectionQuery.sizeEstimate()() mustEqual 5
+      intersectionQuery.sizeEstimate()() mustEqual (5 * queryConfig.averageIntersectionProportion).toInt
     }
 
     "selectWhereIn" in {