Skip to content

SKIPME Cancelable range partitioner #47

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

Closed
markhamstra wants to merge 1 commit into from
Closed

SKIPME Cancelable range partitioner #47

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
10 changes: 5 additions & 5 deletions core/src/main/scala/org/apache/spark/Partitioner.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -110,7 +110,7 @@ class RangePartitioner[K : Ordering : ClassTag, V](
// We allow partitions = 0, which happens when sorting an empty RDD under the default settings.
require(partitions >= 0, s"Number of partitions cannot be negative but found $partitions.")

private var ordering = implicitly[Ordering[K]]
protected var ordering = implicitly[Ordering[K]]

// An array of upper bounds for the first (partitions - 1) partitions
private var rangeBounds: Array[K] = {
Expand Down Expand Up @@ -156,7 +156,7 @@ class RangePartitioner[K : Ordering : ClassTag, V](

def numPartitions = rangeBounds.length + 1

private var binarySearch: ((Array[K], K) => Int) = CollectionsUtils.makeBinarySearch[K]
protected var binarySearch: ((Array[K], K) => Int) = CollectionsUtils.makeBinarySearch[K]

def getPartition(key: Any): Int = {
val k = key.asInstanceOf[K]
Expand Down Expand Up @@ -204,7 +204,7 @@ class RangePartitioner[K : Ordering : ClassTag, V](
}

@throws(classOf[IOException])
private def writeObject(out: ObjectOutputStream): Unit = Utils.tryOrIOException {
protected def writeObject(out: ObjectOutputStream): Unit = Utils.tryOrIOException {
val sfactory = SparkEnv.get.serializer
sfactory match {
case js: JavaSerializer => out.defaultWriteObject()
Expand All @@ -222,7 +222,7 @@ class RangePartitioner[K : Ordering : ClassTag, V](
}

@throws(classOf[IOException])
private def readObject(in: ObjectInputStream): Unit = Utils.tryOrIOException {
protected def readObject(in: ObjectInputStream): Unit = Utils.tryOrIOException {
val sfactory = SparkEnv.get.serializer
sfactory match {
case js: JavaSerializer => in.defaultReadObject()
Expand All @@ -240,7 +240,7 @@ class RangePartitioner[K : Ordering : ClassTag, V](
}
}

private[spark] object RangePartitioner {
object RangePartitioner {

/**
* Sketches the input RDD via reservoir sampling on each partition.
Expand Down
2 changes: 1 addition & 1 deletion core/src/main/scala/org/apache/spark/util/CollectionsUtils.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,7 @@ import java.util

import scala.reflect.{classTag, ClassTag}

private[spark] object CollectionsUtils {
object CollectionsUtils {
def makeBinarySearch[K : Ordering : ClassTag] : (Array[K], K) => Int = {
// For primitive keys, we can use the natural ordering. Otherwise, use the Ordering comparator.
classTag[K] match {
Expand Down
2 changes: 1 addition & 1 deletion core/src/main/scala/org/apache/spark/util/random/SamplingUtils.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@ package org.apache.spark.util.random
import scala.reflect.ClassTag
import scala.util.Random

private[spark] object SamplingUtils {
object SamplingUtils {

/**
* Reservoir sampling implementation that also returns the input size.
Expand Down