10746

Build # 10746

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/6f7fb4af802748c45877ace2d87e93049eb9297f">6f7fb4af8<a href="https://github.com/apache/carbondata/commit/6f7fb4af802748c45877ace2d87e93049eb9297f">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/eaa42f7238a9202a03b7d900571a0baa24dc72a7">eaa42f723</a>">eaa42f723</a><a href="https://github.com/apache/carbondata/commit/6f7fb4af802748c45877ace2d87e93049eb9297f"> into c55f1d7f6">c55f1d7f6</a>

Run Details

11 of 11 new or added lines in 1 file covered. (100.0%)

62420 of 78806 relevant lines covered (79.21%)

1.05 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

72.0

/integration/spark-common/src/main/scala/org/apache/spark/DataSkewRangePartitioner.scala

/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.spark

import java.io.{IOException, ObjectInputStream, ObjectOutputStream}

import scala.collection.mutable
import scala.collection.mutable.ArrayBuffer
import scala.reflect.ClassTag
import scala.util.hashing.byteswap32

import org.apache.spark.rdd.{PartitionPruningRDD, RDD}
import org.apache.spark.serializer.JavaSerializer
import org.apache.spark.util.{CollectionsUtils, Utils}

/**
 * support data skew scenario
 * copy from spark: RangePartitioner
 *
 * RangePartitioner:
 * the rangeBounds are the distinct values, each rangeBound has a partition.
 * so for the data skew scenario, some partitions will include more data.
 *
 * DataSkewRangePartitioner:
 * the rangeBounds are also the distinct values, but it calculates the skew weight.
 * So some rangeBounds maybe have more than one partitions.
 *
 * for example, split following CSV file to 5 partitions:
 * ---------------
 * col1,col2
 * 1,
 * 2,
 * 3,
 * 4,
 * 5,e
 * 6,f
 * 7,g
 * 8,h
 * 9,i
 * 10,j
 * ---------------
 * RangePartitioner will give the following rangeBounds.
 * -------------------------------------------------
 * | range bound| partition range| number of values|
 * -------------------------------------------------
 * | null       | <= null        | 4               |
 * | e          | (null, e]      | 1               |
 * | f          | (e, f]         | 1               |
 * | h          | (f, h]         | 2               |
 * |            | > h            | 2               |
 * -------------------------------------------------
 *
 * DataSkewRangePartitioner will give the following rangeBounds.
 * --------------------------------------------------------------
 * | range bound| skew weight| partition range| number of values|
 * --------------------------------------------------------------
 * | null       | 2          | <= null        | 2(4/2)          |
 * | f          |            | (null, f]      | 2               |
 * | h          |            | (f, h]         | 2               |
 * |            |            | > h            | 2               |
 * |            |            | <= null        | 2(4/2)          |
 * --------------------------------------------------------------
 * The skew weight of range bound "null" is 2.
 * So it will start two tasks for range bound "null" to create two partitions.
 */
class DataSkewRangePartitioner[K: Ordering : ClassTag, V](
    partitions: Int,
    rdd: RDD[_ <: Product2[K, V]])
  extends Partitioner {

  // 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]]

  // An array of upper bounds for the first (partitions - 1) partitions
  // dataSkewCount: how many bounds happened data skew
  // dataSkewIndex: the index of data skew bounds
  // dataSkewNum: how many partition of each data skew bound
  private var (rangeBounds: Array[K], skewCount: Int, skewIndexes: Array[Int],
  skewWeights: Array[Int]) = {
    if (partitions <= 1) {
      (Array.empty[K], 0, Array.empty[Int], Array.empty[Int])
    } else {
      // This is the sample size we need to have roughly balanced output partitions, capped at 1M.
      val sampleSize = math.min(20.0 * partitions, 1e6)
      // Assume the input partitions are roughly balanced and over-sample a little bit.
      val sampleSizePerPartition = math.ceil(3.0 * sampleSize / rdd.partitions.length).toInt
      val (numItems, sketched) = RangePartitioner.sketch(rdd.map(_._1), sampleSizePerPartition)
      if (numItems == 0L) {
        (Array.empty[K], 0, Array.empty[Int], Array.empty[Int])
      } else {
        // If a partition contains much more than the average number of items, we re-sample from it
        // to ensure that enough items are collected from that partition.
        val fraction = math.min(sampleSize / math.max(numItems, 1L), 1.0)
        val candidates = ArrayBuffer.empty[(K, Float)]
        val imbalancedPartitions = mutable.Set.empty[Int]
        sketched.foreach { case (idx, n, sample) =>
          if (fraction * n > sampleSizePerPartition) {
            imbalancedPartitions += idx
          } else {
            // The weight is 1 over the sampling probability.
            val weight = (n.toDouble / sample.length).toFloat
            for (key <- sample) {
              candidates += ((key, weight))
            }
          }
        }
        if (imbalancedPartitions.nonEmpty) {
          // Re-sample imbalanced partitions with the desired sampling probability.
          val imbalanced = new PartitionPruningRDD(rdd.map(_._1), imbalancedPartitions.contains)
          val seed = byteswap32(-rdd.id - 1)
          val reSampled = imbalanced.sample(withReplacement = false, fraction, seed).collect()
          val weight = (1.0 / fraction).toFloat
          candidates ++= reSampled.map(x => (x, weight))
        }
        determineBounds(candidates, partitions)
      }
    }
  }

  def determineBounds(
      candidates: ArrayBuffer[(K, Float)],
      partitions: Int): (Array[K], Int, Array[Int], Array[Int]) = {
    val ordered = candidates.sortBy(_._1)
    val numCandidates = ordered.size
    val sumWeights = ordered.map(_._2.toDouble).sum
    val step = sumWeights / partitions
    var cumWeight = 0.0
    var target = step
    val bounds = ArrayBuffer.empty[K]
    var i = 0
    var j = 0
    var previousBound = Option.empty[K]
    while ((i < numCandidates) && (j < partitions - 1)) {
      val (key, weight) = ordered(i)
      cumWeight += weight
      if (cumWeight >= target) {
        // Skip duplicate values.
        if (previousBound.isEmpty || ordering.gteq(key, previousBound.get)) {
          bounds += key
          target += step
          j += 1
          previousBound = Some(key)
        }
      }
      i += 1
    }

    if (bounds.size >= 2) {
      combineDataSkew(bounds)
    } else {
      (bounds.toArray, 0, Array.empty[Int], Array.empty[Int])
    }
  }

  def combineDataSkew(bounds: ArrayBuffer[K]): (Array[K], Int, Array[Int], Array[Int]) = {
    val finalBounds = ArrayBuffer.empty[K]
    var preBound = bounds(0)
    finalBounds += preBound
    val dataSkewIndexTmp = ArrayBuffer.empty[Int]
    val dataSkewNumTmp = ArrayBuffer.empty[Int]
    var dataSkewCountTmp = 1
    (1 until bounds.size).foreach { index =>
      val bound = bounds(index)
      if (ordering.equiv(bound, preBound)) {
        if (dataSkewCountTmp == 1) {
          dataSkewIndexTmp += (finalBounds.size - 1)
        }
        dataSkewCountTmp += 1
      } else {
        finalBounds += bound
        preBound = bound
        if (dataSkewCountTmp > 1) {
          dataSkewNumTmp += dataSkewCountTmp
          dataSkewCountTmp = 1
        }
      }
    }
    if (dataSkewCountTmp > 1) {
      dataSkewNumTmp += dataSkewCountTmp
    }
    if (dataSkewIndexTmp.size > 0) {
      (finalBounds.toArray, dataSkewIndexTmp.size, dataSkewIndexTmp.toArray, dataSkewNumTmp.toArray)
    } else {
      (finalBounds.toArray, 0, Array.empty[Int], Array.empty[Int])
    }
  }

  private var skewPartitions: Int = if (skewCount == 0) {
    0
  } else {
    skewWeights.map(_ - 1).sum
  }

  def numPartitions: Int = rangeBounds.length + 1 + skewPartitions

  private var binarySearch: ((Array[K], K) => Int) = CollectionsUtils.makeBinarySearch[K]
  private var skewIndexesLen = 0
  private var dsps: Array[DataSkewPartitioner] = null

  def initialize(): Unit = {
    if (skewCount > 0) {
      skewIndexesLen = skewIndexes.length
      dsps = new Array[DataSkewPartitioner](skewIndexesLen)
      var previousPart = rangeBounds.length
      for (i <- 0 until skewIndexesLen) {
        dsps(i) = new DataSkewPartitioner(skewIndexes(i), previousPart, skewWeights(i))
        previousPart = previousPart + skewWeights(i) - 1
      }
    }
  }

  private var needInit = true

  def getPartition(key: Any): Int = {
    if (needInit) {
      needInit = false
      initialize()
    }
    val k = key.asInstanceOf[K]
    var partition = 0
    if (rangeBounds.length <= 128) {
      // If we have less than 128 partitions naive search
      while (partition < rangeBounds.length && ordering.gt(k, rangeBounds(partition))) {
        partition += 1
      }
    } else {
      // Determine which binary search method to use only once.
      partition = binarySearch(rangeBounds, k)
      // binarySearch either returns the match location or -[insertion point]-1
      if (partition < 0) {
        partition = -partition - 1
      }
      if (partition > rangeBounds.length) {
        partition = rangeBounds.length
      }
    }
    if (skewCount == 0) {
      partition
    } else {
      getDataSkewPartition(partition)
    }
  }

  def getDataSkewPartition(partition: Int): Int = {
    var index = -1
    if (partition <= skewIndexes(skewIndexesLen - 1) && partition >= skewIndexes(0)) {
      for (i <- 0 until skewIndexesLen) {
        if (skewIndexes(i) == partition) {
          index = i
        }
      }
    }
    if (index == -1) {
      partition
    } else {
      nextPartition(index)
    }
  }

  def nextPartition(index: Int): Int = {
    dsps(index).nextPartition()
  }

  override def equals(other: Any): Boolean = {
    other match {
      case r: DataSkewRangePartitioner[_, _] =>
        r.rangeBounds.sameElements(rangeBounds)
      case _ =>
        false
    }
  }

  override def hashCode(): Int = {
    val prime = 31
    var result = 1
    var i = 0
    while (i < rangeBounds.length) {
      result = prime * result + rangeBounds(i).hashCode
      i += 1
    }
    result = prime * result
    result
  }

  @throws(classOf[IOException])
  private def writeObject(out: ObjectOutputStream): Unit = {
    Utils.tryOrIOException {
      val sfactory = SparkEnv.get.serializer
      sfactory match {
        case js: JavaSerializer => out.defaultWriteObject()
        case _ =>
          out.writeInt(skewCount)
          out.writeInt(skewPartitions)
          if (skewCount > 0) {
            out.writeObject(skewIndexes)
            out.writeObject(skewWeights)
          }
          out.writeObject(ordering)
          out.writeObject(binarySearch)

          val ser = sfactory.newInstance()
          Utils.serializeViaNestedStream(out, ser) { stream =>
            stream.writeObject(scala.reflect.classTag[Array[K]])
            stream.writeObject(rangeBounds)
          }
      }
    }
  }

  @throws(classOf[IOException])
  private def readObject(in: ObjectInputStream): Unit = {
    Utils.tryOrIOException {
      needInit = true
      val sfactory = SparkEnv.get.serializer
      sfactory match {
        case js: JavaSerializer => in.defaultReadObject()
        case _ =>
          skewCount = in.readInt()
          skewPartitions = in.readInt()
          if (skewCount > 0) {
            skewIndexes = in.readObject().asInstanceOf[Array[Int]]
            skewWeights = in.readObject().asInstanceOf[Array[Int]]
          }
          ordering = in.readObject().asInstanceOf[Ordering[K]]
          binarySearch = in.readObject().asInstanceOf[(Array[K], K) => Int]

          val ser = sfactory.newInstance()
          Utils.deserializeViaNestedStream(in, ser) { ds =>
            implicit val classTag = ds.readObject[ClassTag[Array[K]]]()
            rangeBounds = ds.readObject[Array[K]]()
          }
      }
    }
  }
}

class DataSkewPartitioner(originPart: Int, previousPart: Int, skewWeight: Int) {
  var index = 0

  def nextPartition(): Int = {
    if (index == 0) {
      index = index + 1
      originPart
    } else {
      val newPartition = previousPart + index
      index = index + 1
      if (index == skewWeight) {
        index = 0
      }
      newPartition
    }
  }
}

1	/*
2	* Licensed to the Apache Software Foundation (ASF) under one or more
3	* contributor license agreements. See the NOTICE file distributed with
4	* this work for additional information regarding copyright ownership.
5	* The ASF licenses this file to You under the Apache License, Version 2.0
6	* (the "License"); you may not use this file except in compliance with
7	* the License. You may obtain a copy of the License at
8	*
9	* http://www.apache.org/licenses/LICENSE-2.0
10	*
11	* Unless required by applicable law or agreed to in writing, software
12	* distributed under the License is distributed on an "AS IS" BASIS,
13	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14	* See the License for the specific language governing permissions and
15	* limitations under the License.
16	*/
17
18	package org.apache.spark
19
20	import java.io.{IOException, ObjectInputStream, ObjectOutputStream}
21
22	import scala.collection.mutable
23	import scala.collection.mutable.ArrayBuffer
24	import scala.reflect.ClassTag
25	import scala.util.hashing.byteswap32
26
27	import org.apache.spark.rdd.{PartitionPruningRDD, RDD}
28	import org.apache.spark.serializer.JavaSerializer
29	import org.apache.spark.util.{CollectionsUtils, Utils}
30
31	/**
32	* support data skew scenario
33	* copy from spark: RangePartitioner
34	*
35	* RangePartitioner:
36	* the rangeBounds are the distinct values, each rangeBound has a partition.
37	* so for the data skew scenario, some partitions will include more data.
38	*
39	* DataSkewRangePartitioner:
40	* the rangeBounds are also the distinct values, but it calculates the skew weight.
41	* So some rangeBounds maybe have more than one partitions.
42	*
43	* for example, split following CSV file to 5 partitions:
44	* ---------------
45	* col1,col2
46	* 1,
47	* 2,
48	* 3,
49	* 4,
50	* 5,e
51	* 6,f
52	* 7,g
53	* 8,h
54	* 9,i
55	* 10,j
56	* ---------------
57	* RangePartitioner will give the following rangeBounds.
58	* -------------------------------------------------
59	* \| range bound\| partition range\| number of values\|
60	* -------------------------------------------------
61	* \| null \| <= null \| 4 \|
62	* \| e \| (null, e] \| 1 \|
63	* \| f \| (e, f] \| 1 \|
64	* \| h \| (f, h] \| 2 \|
65	* \| \| > h \| 2 \|
66	* -------------------------------------------------
67	*
68	* DataSkewRangePartitioner will give the following rangeBounds.
69	* --------------------------------------------------------------
70	* \| range bound\| skew weight\| partition range\| number of values\|
71	* --------------------------------------------------------------
72	* \| null \| 2 \| <= null \| 2(4/2) \|
73	* \| f \| \| (null, f] \| 2 \|
74	* \| h \| \| (f, h] \| 2 \|
75	* \| \| \| > h \| 2 \|
76	* \| \| \| <= null \| 2(4/2) \|
77	* --------------------------------------------------------------
78	* The skew weight of range bound "null" is 2.
79	* So it will start two tasks for range bound "null" to create two partitions.
80	*/
81	class DataSkewRangePartitioner[K: Ordering : ClassTag, V](	1✔
82	partitions: Int,	1✔
83	rdd: RDD[_ <: Product2[K, V]])
84	extends Partitioner {	1✔
85
86	// We allow partitions = 0, which happens when sorting an empty RDD under the default settings.
87	require(partitions >= 0, s"Number of partitions cannot be negative but found $partitions.")	1✔
88
89	private var ordering = implicitly[Ordering[K]]	1✔
90
91	// An array of upper bounds for the first (partitions - 1) partitions
92	// dataSkewCount: how many bounds happened data skew
93	// dataSkewIndex: the index of data skew bounds
94	// dataSkewNum: how many partition of each data skew bound
95	private var (rangeBounds: Array[K], skewCount: Int, skewIndexes: Array[Int],	1✔
96	skewWeights: Array[Int]) = {	1✔
97	if (partitions <= 1) {	1✔
98	(Array.empty[K], 0, Array.empty[Int], Array.empty[Int])	1✔
99	} else {
100	// This is the sample size we need to have roughly balanced output partitions, capped at 1M.
101	val sampleSize = math.min(20.0 * partitions, 1e6)	1✔
102	// Assume the input partitions are roughly balanced and over-sample a little bit.
103	val sampleSizePerPartition = math.ceil(3.0 * sampleSize / rdd.partitions.length).toInt	1✔
104	val (numItems, sketched) = RangePartitioner.sketch(rdd.map(_._1), sampleSizePerPartition)	1✔
105	if (numItems == 0L) {	1✔
106	(Array.empty[K], 0, Array.empty[Int], Array.empty[Int])	×
107	} else {
108	// If a partition contains much more than the average number of items, we re-sample from it
109	// to ensure that enough items are collected from that partition.
110	val fraction = math.min(sampleSize / math.max(numItems, 1L), 1.0)	1✔
111	val candidates = ArrayBuffer.empty[(K, Float)]	1✔
112	val imbalancedPartitions = mutable.Set.empty[Int]	1✔
113	sketched.foreach { case (idx, n, sample) =>	1✔
114	if (fraction * n > sampleSizePerPartition) {	1✔
115	imbalancedPartitions += idx	×
116	} else {
117	// The weight is 1 over the sampling probability.
118	val weight = (n.toDouble / sample.length).toFloat	1✔
119	for (key <- sample) {	1✔
120	candidates += ((key, weight))	1✔
121	}
122	}
123	}
124	if (imbalancedPartitions.nonEmpty) {	1✔
125	// Re-sample imbalanced partitions with the desired sampling probability.
126	val imbalanced = new PartitionPruningRDD(rdd.map(_._1), imbalancedPartitions.contains)	×
127	val seed = byteswap32(-rdd.id - 1)	×
128	val reSampled = imbalanced.sample(withReplacement = false, fraction, seed).collect()	×
129	val weight = (1.0 / fraction).toFloat	×
130	candidates ++= reSampled.map(x => (x, weight))	×
131	}
132	determineBounds(candidates, partitions)	1✔
133	}
134	}
135	}
136
137	def determineBounds(
138	candidates: ArrayBuffer[(K, Float)],
139	partitions: Int): (Array[K], Int, Array[Int], Array[Int]) = {	1✔
140	val ordered = candidates.sortBy(_._1)	1✔
141	val numCandidates = ordered.size	1✔
142	val sumWeights = ordered.map(_._2.toDouble).sum	1✔
143	val step = sumWeights / partitions	1✔
144	var cumWeight = 0.0	1✔
145	var target = step	1✔
146	val bounds = ArrayBuffer.empty[K]	1✔
147	var i = 0	1✔
148	var j = 0	1✔
149	var previousBound = Option.empty[K]	1✔
150	while ((i < numCandidates) && (j < partitions - 1)) {	1✔
151	val (key, weight) = ordered(i)	1✔
152	cumWeight += weight	1✔
153	if (cumWeight >= target) {	1✔
154	// Skip duplicate values.
155	if (previousBound.isEmpty \|\| ordering.gteq(key, previousBound.get)) {	1✔
156	bounds += key	1✔
157	target += step	1✔
158	j += 1	1✔
159	previousBound = Some(key)	1✔
160	}
161	}
162	i += 1	1✔
163	}
164
165	if (bounds.size >= 2) {	1✔
166	combineDataSkew(bounds)	1✔
167	} else {
168	(bounds.toArray, 0, Array.empty[Int], Array.empty[Int])	×
169	}
170	}
171
172	def combineDataSkew(bounds: ArrayBuffer[K]): (Array[K], Int, Array[Int], Array[Int]) = {	1✔
173	val finalBounds = ArrayBuffer.empty[K]	1✔
174	var preBound = bounds(0)	1✔
175	finalBounds += preBound	1✔
176	val dataSkewIndexTmp = ArrayBuffer.empty[Int]	1✔
177	val dataSkewNumTmp = ArrayBuffer.empty[Int]	1✔
178	var dataSkewCountTmp = 1	1✔
179	(1 until bounds.size).foreach { index =>	1✔
180	val bound = bounds(index)	1✔
181	if (ordering.equiv(bound, preBound)) {	1✔
182	if (dataSkewCountTmp == 1) {	1✔
183	dataSkewIndexTmp += (finalBounds.size - 1)	1✔
184	}
185	dataSkewCountTmp += 1	1✔
186	} else {
187	finalBounds += bound	1✔
188	preBound = bound	1✔
189	if (dataSkewCountTmp > 1) {	1✔
190	dataSkewNumTmp += dataSkewCountTmp	1✔
191	dataSkewCountTmp = 1	1✔
192	}
193	}
194	}
195	if (dataSkewCountTmp > 1) {	1✔
196	dataSkewNumTmp += dataSkewCountTmp	1✔
197	}
198	if (dataSkewIndexTmp.size > 0) {	1✔
199	(finalBounds.toArray, dataSkewIndexTmp.size, dataSkewIndexTmp.toArray, dataSkewNumTmp.toArray)	1✔
200	} else {
201	(finalBounds.toArray, 0, Array.empty[Int], Array.empty[Int])	1✔
202	}
203	}
204
205	private var skewPartitions: Int = if (skewCount == 0) {	1✔
206	0	1✔
207	} else {
208	skewWeights.map(_ - 1).sum	1✔
209	}
210
211	def numPartitions: Int = rangeBounds.length + 1 + skewPartitions	1✔
212
213	private var binarySearch: ((Array[K], K) => Int) = CollectionsUtils.makeBinarySearch[K]	1✔
214	private var skewIndexesLen = 0	1✔
215	private var dsps: Array[DataSkewPartitioner] = null	1✔
216
217	def initialize(): Unit = {
218	if (skewCount > 0) {	1✔
219	skewIndexesLen = skewIndexes.length	1✔
220	dsps = new Array[DataSkewPartitioner](skewIndexesLen)	1✔
221	var previousPart = rangeBounds.length	1✔
222	for (i <- 0 until skewIndexesLen) {	1✔
223	dsps(i) = new DataSkewPartitioner(skewIndexes(i), previousPart, skewWeights(i))	1✔
224	previousPart = previousPart + skewWeights(i) - 1	1✔
225	}
226	}
227	}
228
229	private var needInit = true	1✔
230
231	def getPartition(key: Any): Int = {	1✔
232	if (needInit) {	1✔
233	needInit = false	1✔
234	initialize()	1✔
235	}
236	val k = key.asInstanceOf[K]	1✔
237	var partition = 0	1✔
238	if (rangeBounds.length <= 128) {	1✔
239	// If we have less than 128 partitions naive search
240	while (partition < rangeBounds.length && ordering.gt(k, rangeBounds(partition))) {	1✔
241	partition += 1	1✔
242	}
243	} else {
244	// Determine which binary search method to use only once.
245	partition = binarySearch(rangeBounds, k)	×
246	// binarySearch either returns the match location or -[insertion point]-1
247	if (partition < 0) {	×
248	partition = -partition - 1	×
249	}
250	if (partition > rangeBounds.length) {	×
251	partition = rangeBounds.length	×
252	}
253	}
254	if (skewCount == 0) {	1✔
255	partition	1✔
256	} else {
257	getDataSkewPartition(partition)	1✔
258	}
259	}
260
261	def getDataSkewPartition(partition: Int): Int = {	1✔
262	var index = -1	1✔
263	if (partition <= skewIndexes(skewIndexesLen - 1) && partition >= skewIndexes(0)) {	1✔
264	for (i <- 0 until skewIndexesLen) {	1✔
265	if (skewIndexes(i) == partition) {	1✔
266	index = i	1✔
267	}
268	}
269	}
270	if (index == -1) {	1✔
271	partition	1✔
272	} else {
273	nextPartition(index)	1✔
274	}
275	}
276
277	def nextPartition(index: Int): Int = {
278	dsps(index).nextPartition()	1✔
279	}
280
281	override def equals(other: Any): Boolean = {
282	other match {	×
283	case r: DataSkewRangePartitioner[_, _] =>	×
284	r.rangeBounds.sameElements(rangeBounds)	×
285	case _ =>
286	false	×
287	}
288	}
289
290	override def hashCode(): Int = {
291	val prime = 31	×
292	var result = 1	×
293	var i = 0	×
294	while (i < rangeBounds.length) {	×
295	result = prime * result + rangeBounds(i).hashCode	×
296	i += 1	×
297	}
298	result = prime * result	×
299	result	×
300	}
301
302	@throws(classOf[IOException])
303	private def writeObject(out: ObjectOutputStream): Unit = {
304	Utils.tryOrIOException {	1✔
305	val sfactory = SparkEnv.get.serializer	1✔
306	sfactory match {	1✔
307	case js: JavaSerializer => out.defaultWriteObject()	1✔
308	case _ =>	×
309	out.writeInt(skewCount)	×
310	out.writeInt(skewPartitions)	×
311	if (skewCount > 0) {	×
312	out.writeObject(skewIndexes)	×
313	out.writeObject(skewWeights)	×
314	}
315	out.writeObject(ordering)	×
316	out.writeObject(binarySearch)	×
317
318	val ser = sfactory.newInstance()	×
319	Utils.serializeViaNestedStream(out, ser) { stream =>	×
320	stream.writeObject(scala.reflect.classTag[Array[K]])	×
321	stream.writeObject(rangeBounds)	×
322	}
323	}
324	}
325	}
326
327	@throws(classOf[IOException])
328	private def readObject(in: ObjectInputStream): Unit = {
329	Utils.tryOrIOException {	1✔
330	needInit = true	1✔
331	val sfactory = SparkEnv.get.serializer	1✔
332	sfactory match {	1✔
333	case js: JavaSerializer => in.defaultReadObject()	1✔
334	case _ =>	×
335	skewCount = in.readInt()	×
336	skewPartitions = in.readInt()	×
337	if (skewCount > 0) {	×
338	skewIndexes = in.readObject().asInstanceOf[Array[Int]]	×
339	skewWeights = in.readObject().asInstanceOf[Array[Int]]	×
340	}
341	ordering = in.readObject().asInstanceOf[Ordering[K]]	×
342	binarySearch = in.readObject().asInstanceOf[(Array[K], K) => Int]	×
343
344	val ser = sfactory.newInstance()	×
345	Utils.deserializeViaNestedStream(in, ser) { ds =>	×
346	implicit val classTag = ds.readObject[ClassTag[Array[K]]]()	×
347	rangeBounds = ds.readObject[Array[K]]()	×
348	}
349	}
350	}
351	}
352	}
353
354	class DataSkewPartitioner(originPart: Int, previousPart: Int, skewWeight: Int) {	1✔
355	var index = 0	1✔
356
357	def nextPartition(): Int = {
358	if (index == 0) {	1✔
359	index = index + 1	1✔
360	originPart	1✔
361	} else {
362	val newPartition = previousPart + index	1✔
363	index = index + 1	1✔
364	if (index == skewWeight) {	1✔
365	index = 0	1✔
366	}
367	newPartition	1✔
368	}
369	}
370	}	1✔

apache / carbondata / 10746

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous