Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Fix bug in orderedRDD #3

Merged
merged 3 commits into from
Jul 20, 2017
Merged

Fix bug in orderedRDD #3

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
cd58d8f
Fix bug in orderedRDD
tpoterba Jul 20, 2017
9b729bf
Added single test
tpoterba Jul 20, 2017
95dc06a
Added single test
tpoterba Jul 20, 2017
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
59 changes: 39 additions & 20 deletions src/main/scala/is/hail/check/Prop.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@ package is.hail.check
import org.apache.commons.math3.random.RandomDataGenerator

import scala.collection.mutable.ArrayBuffer
import scala.util.Random
import scala.util.{Failure, Random, Success, Try}

abstract class Prop {
def apply(p: Parameters, name: Option[String] = None): Unit
Expand All @@ -25,11 +25,17 @@ class GenProp1[T1](g1: Gen[T1], f: (T1) => Boolean) extends Prop {
val prefix = name.map(_ + ": ").getOrElse("")
for (i <- 0 until p.count) {
val v1 = g1(p)
val r = f(v1)
if (!r) {
println(s"""! ${ prefix }Falsified after $i passed tests.""")
println(s"> ARG_0: $v1")
assert(r)
val r = Try(f(v1))
r match {
case Success(true) =>
case Success(false) =>
println(s"""! ${ prefix }Falsified after $i passed tests.""")
println(s"> ARG_0: $v1")
throw new AssertionError(null)
case Failure(e) =>
println(s"""! ${ prefix }Error after $i passed tests.""")
println(s"> ARG_0: $v1")
throw new AssertionError(e)
}
}
println(s" + ${ prefix }OK, passed ${ p.count } tests.")
Expand All @@ -42,12 +48,17 @@ class GenProp2[T1, T2](g1: Gen[T1], g2: Gen[T2], f: (T1, T2) => Boolean) extends
for (i <- 0 until p.count) {
val v1 = g1(p)
val v2 = g2(p)
val r = f(v1, v2)
if (!r) {
println(s"! ${ prefix }Falsified after $i passed tests.")
println(s"> ARG_0: $v1")
println(s"> ARG_1: $v2")
assert(r)
val r = Try(f(v1, v2))
r match {
case Success(true) =>
case Success(false) =>
println(s"""! ${ prefix }Falsified after $i passed tests.""")
println(s"> ARG_0: $v1")
throw new AssertionError(null)
case Failure(e) =>
println(s"""! ${ prefix }Error after $i passed tests.""")
println(s"> ARG_0: $v1")
throw new AssertionError(e)
}
}
println(s" + ${ prefix }OK, passed ${ p.count } tests.")
Expand All @@ -61,13 +72,17 @@ class GenProp3[T1, T2, T3](g1: Gen[T1], g2: Gen[T2], g3: Gen[T3], f: (T1, T2, T3
val v1 = g1(p)
val v2 = g2(p)
val v3 = g3(p)
val r = f(v1, v2, v3)
if (!r) {
println(s"! ${ prefix }Falsified after $i passed tests.")
println(s"> ARG_0: $v1")
println(s"> ARG_1: $v2")
println(s"> ARG_2: $v3")
assert(r)
val r = Try(f(v1, v2, v3))
r match {
case Success(true) =>
case Success(false) =>
println(s"""! ${ prefix }Falsified after $i passed tests.""")
println(s"> ARG_0: $v1")
throw new AssertionError(null)
case Failure(e) =>
println(s"""! ${ prefix }Error after $i passed tests.""")
println(s"> ARG_0: $v1")
throw new AssertionError(e)
}
}
println(s" + ${ prefix }OK, passed ${ p.count } tests.")
Expand All @@ -87,7 +102,11 @@ class Properties(val name: String) extends Prop {

override def apply(p: Parameters, prefix: Option[String]) {
for ((propName, prop) <- properties)
prop(p, prefix.map(_ + "." + propName).orElse(Some(propName)))
prop.apply(p, prefix.map(_ + "." + propName).orElse(Some(propName)))
}

def checkSingle(propName: String) {

Copy link
Owner

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@tpoterba ???

Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

oops, meant to actually finish this. I want a way to run one specific method from a Spec without running them all.

}
}

Expand Down
27 changes: 17 additions & 10 deletions src/main/scala/is/hail/sparkextras/OrderedRDD.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -88,14 +88,13 @@ object OrderedRDD {
}

if (partitionsSorted) {
val (adjustedPartitions, rangeBounds) = rangesAndAdjustments[PK, K, V](sortedKeyInfo)
val partitioner = OrderedPartitioner[PK, K](rangeBounds, adjustedPartitions.length)
val sortedness = sortedKeyInfo.map(_.sortedness).min
val (adjustedPartitions, rangeBounds, adjSortedness) = rangesAndAdjustments[PK, K, V](sortedKeyInfo, sortedness)
val partitioner = OrderedPartitioner[PK, K](rangeBounds, adjustedPartitions.length)
val reorderedPartitionsRDD = rdd.reorderPartitions(sortedKeyInfo.map(_.partIndex))
val adjustedRDD = new AdjustedPartitionsRDD(reorderedPartitionsRDD, adjustedPartitions)
(sortedness: @unchecked) match {
(adjSortedness: @unchecked) match {
case PartitionKeyInfo.KSORTED =>
assert(sortedness == PartitionKeyInfo.KSORTED)
info("Coerced sorted dataset")
(AS_IS, OrderedRDD(adjustedRDD, partitioner))

Expand Down Expand Up @@ -123,15 +122,17 @@ object OrderedRDD {
}
}

def rangesAndAdjustments[PK, K, V](sortedKeyInfo: Array[PartitionKeyInfo[PK]])
(implicit kOk: OrderedKey[PK, K], vct: ClassTag[V]): (IndexedSeq[Array[Adjustment[(K, V)]]], Array[PK]) = {
def rangesAndAdjustments[PK, K, V](sortedKeyInfo: Array[PartitionKeyInfo[PK]], sortedness: Int)
(implicit kOk: OrderedKey[PK, K], vct: ClassTag[V]): (IndexedSeq[Array[Adjustment[(K, V)]]], Array[PK], Int) = {
import kOk._
import scala.Ordering.Implicits._

val rangeBounds = mutable.ArrayBuilder.make[PK]
val adjustmentsBuffer = new mutable.ArrayBuffer[Array[Adjustment[(K, V)]]]
val indicesBuilder = mutable.ArrayBuilder.make[Int]

var anyOverlaps = false

val it = sortedKeyInfo.indices.iterator.buffered

while (it.nonEmpty) {
Expand All @@ -148,11 +149,15 @@ object OrderedRDD {
info.min == info.max && info.min == max
}

while (it.hasNext && overlaps(sortedKeyInfo(it.head)))
while (it.hasNext && overlaps(sortedKeyInfo(it.head))) {
anyOverlaps = true
indicesBuilder += it.next()
}

if (it.hasNext && sortedKeyInfo(it.head).min == max)
if (it.hasNext && sortedKeyInfo(it.head).min == max) {
anyOverlaps = true
indicesBuilder += it.head
}

val adjustments = indicesBuilder.result().zipWithIndex.map { case (partitionIndex, index) =>
val f: (Iterator[(K, V)]) => Iterator[(K, V)] =
Expand All @@ -177,7 +182,9 @@ object OrderedRDD {
rangeBounds += max
}

(adjustmentsBuffer, rangeBounds.result())
val adjSortedness = if (anyOverlaps) sortedness.min(PartitionKeyInfo.TSORTED) else sortedness

(adjustmentsBuffer, rangeBounds.result(), adjSortedness)
}

def apply[PK, K, V](rdd: RDD[(K, V)],
Expand Down Expand Up @@ -210,7 +217,7 @@ object OrderedRDD {
if (first)
first = false
else
assert(prevK <= r._1)
assert(prevK <= r._1, s"$prevK is greater than ${r._1}")

prevK = r._1
r
Expand Down
15 changes: 8 additions & 7 deletions src/test/scala/is/hail/utils/OrderedRDDSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -87,12 +87,12 @@ class OrderedRDDSuite extends SparkSuite {
val check1 = leftDistinct.toSet == leftJoin.map { case (k, (v1, _)) => (k, v1) }.toSet
val check2 = leftJoin.forall { case (k, (_, v2)) =>
val eq = v2.toSet == m2.getOrElse(k, Set.empty[String])
if (!eq)
println(
s"""key=$k
| v1 = ${v2.toSet}
| v2 = ${m2.getOrElse(k, Set.empty[String])}""".stripMargin)
eq
if (!eq)
println(
s"""key=$k
| v1 = ${ v2.toSet }
| v2 = ${ m2.getOrElse(k, Set.empty[String]) }""".stripMargin)
eq
}

val p = check1 && check2
Expand Down Expand Up @@ -160,7 +160,8 @@ class OrderedRDDSuite extends SparkSuite {

property("fullySorted") = Prop.forAll(sorted) { rdd =>
val (status, ordered) = OrderedRDD.coerce(rdd)
check(ordered, rdd) && status == OrderedRDD.AS_IS
// could be local sort if variant is split across partitions
check(ordered, rdd) && status <= OrderedRDD.LOCAL_SORT
}

property("join1") = Prop.forAll(g, g) { case ((nPar1, is1), (nPar2, is2)) =>
Expand Down