<commit_before>package com.tresata.spark.sorted

import scala.reflect.ClassTag

import org.apache.spark.{ Logging, Partitioner, HashPartitioner }
import org.apache.spark.Partitioner.defaultPartitioner
import org.apache.spark.rdd.{ RDD, ShuffledRDD }

object PairRDDFunctions {
  implicit def rddToSparkSortedPairRDDFunctions[K: ClassTag, V: ClassTag](rdd: RDD[(K, V)]) = new PairRDDFunctions(rdd)
}

class PairRDDFunctions[K: ClassTag, V: ClassTag](self: RDD[(K, V)]) extends Logging with Serializable {
  def groupSort(partitioner: Partitioner, valueOrdering: Option[Ordering[V]])(implicit keyOrdering: Ordering[K]): GroupSorted[K, V] = 
    (self, valueOrdering) match {
      case (gs: GroupSorted[K, V], vo) if gs.valueOrdering == vo && (gs.partitioner match {
        case Some(KeyPartitioner(p)) if p == partitioner => true
        case _ => false
      }) =>
        log.info("re-using existing GroupSorted")
        gs
      case (rdd, None) =>
        log.info("creating GroupSorted without value ordering")
        val shuffled = new ShuffledRDD[K, V, V](self, partitioner)
          .setKeyOrdering(keyOrdering)<commit_msg>i think this should always have been a HashOrdering because its potentially faster and has better distribution of keys<commit_after>package com.tresata.spark.sorted

import scala.reflect.ClassTag

import org.apache.spark.{ Logging, Partitioner, HashPartitioner }
import org.apache.spark.Partitioner.defaultPartitioner
import org.apache.spark.rdd.{ RDD, ShuffledRDD }

object PairRDDFunctions {
  implicit def rddToSparkSortedPairRDDFunctions[K: ClassTag, V: ClassTag](rdd: RDD[(K, V)]) = new PairRDDFunctions(rdd)
}

class PairRDDFunctions[K: ClassTag, V: ClassTag](self: RDD[(K, V)]) extends Logging with Serializable {
  def groupSort(partitioner: Partitioner, valueOrdering: Option[Ordering[V]])(implicit keyOrdering: Ordering[K]): GroupSorted[K, V] = 
    (self, valueOrdering) match {
      case (gs: GroupSorted[K, V], vo) if gs.valueOrdering == vo && (gs.partitioner match {
        case Some(KeyPartitioner(p)) if p == partitioner => true
        case _ => false
      }) =>
        log.info("re-using existing GroupSorted")
        gs
      case (rdd, None) =>
        log.info("creating GroupSorted without value ordering")
        val shuffled = new ShuffledRDD[K, V, V](self, partitioner)
          .setKeyOrdering(new HashOrdering(keyOrdering))
{'fraction_non_alphanumeric': 0.11543287327478043, 'fraction_numerical': 0.0, 'mean_word_length': 4.2687224669603525, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1692629', 'n_tokens_mistral': 784, 'n_tokens_neox': 766, 'n_words': 194}