ODIn/spark-instrumented-optimizer
1
0
Fork 0
Code Issues Pull requests Projects Releases Wiki Activity

Code review

Browse source
This commit is contained in:
harshars 2013-07-26 15:23:15 -07:00
parent 72cf7ec0e5
commit 392d7474fd
2 changed files with 3 additions and 3 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
core/src/main/scala/spark/rdd/PartitionPruningRDD.scala
View file

@ -17,8 +17,8 @@ class PruneDependency[T](rdd: RDD[T], @transient partitionFilterFunc: Int => Boo
@transient
val partitions: Array[Partition] = rdd.partitions.zipWithIndex
.filter(s => partitionFilterFunc(s._2))
.map { case(split, idx) => new PartitionPruningRDDPartition(idx, split) : Partition }
.filter(s => partitionFilterFunc(s._2))
.map { case(split, idx) => new PartitionPruningRDDPartition(idx, split) : Partition }
override def getParents(partitionId: Int) = List(partitions(partitionId).index)
}

2
core/src/test/scala/spark/PartitionPruningRDDSuite.scala
View file

@ -21,8 +21,8 @@ class PartitionPruningRDDSuite extends FunSuite with SharedSparkContext {
def compute(split: Partition, context: TaskContext) = {Iterator()}
}
val prunedRDD = PartitionPruningRDD.create(rdd, {x => if (x==2) true else false})
println(prunedRDD.partitions.length)
val p = prunedRDD.partitions(0)
assert(p.index == 2)
assert(prunedRDD.partitions.length == 1)
}
}