diff options
-rw-r--r-- | sql/core/src/test/scala/org/apache/spark/sql/execution/columnar/compression/CompressionSchemeBenchmark.scala | 282 |
1 files changed, 193 insertions, 89 deletions
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/columnar/compression/CompressionSchemeBenchmark.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/columnar/compression/CompressionSchemeBenchmark.scala index 95eb5cf912..0000a5d1ef 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/columnar/compression/CompressionSchemeBenchmark.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/columnar/compression/CompressionSchemeBenchmark.scala @@ -17,19 +17,13 @@ package org.apache.spark.sql.execution.columnar.compression -import java.nio.ByteBuffer -import java.nio.ByteOrder +import java.nio.{ByteBuffer, ByteOrder} import org.apache.commons.lang3.RandomStringUtils import org.apache.commons.math3.distribution.LogNormalDistribution import org.apache.spark.sql.catalyst.expressions.{GenericInternalRow, GenericMutableRow} -import org.apache.spark.sql.execution.columnar.BOOLEAN -import org.apache.spark.sql.execution.columnar.INT -import org.apache.spark.sql.execution.columnar.LONG -import org.apache.spark.sql.execution.columnar.NativeColumnType -import org.apache.spark.sql.execution.columnar.SHORT -import org.apache.spark.sql.execution.columnar.STRING +import org.apache.spark.sql.execution.columnar.{BOOLEAN, INT, LONG, NativeColumnType, SHORT, STRING} import org.apache.spark.sql.types.AtomicType import org.apache.spark.util.Benchmark import org.apache.spark.util.Utils._ @@ -53,35 +47,70 @@ object CompressionSchemeBenchmark extends AllCompressionSchemes { () => rng.sample } - private[this] def runBenchmark[T <: AtomicType]( + private[this] def prepareEncodeInternal[T <: AtomicType]( + count: Int, + tpe: NativeColumnType[T], + supportedScheme: CompressionScheme, + input: ByteBuffer): ((ByteBuffer, ByteBuffer) => ByteBuffer, Double, ByteBuffer) = { + assert(supportedScheme.supports(tpe)) + + def toRow(d: Any) = new GenericInternalRow(Array[Any](d)) + val encoder = supportedScheme.encoder(tpe) + for (i <- 0 until count) { + encoder.gatherCompressibilityStats(toRow(tpe.extract(input)), 0) + } + input.rewind() + + val compressedSize = if (encoder.compressedSize == 0) { + input.remaining() + } else { + encoder.compressedSize + } + + (encoder.compress, encoder.compressionRatio, allocateLocal(4 + compressedSize)) + } + + private[this] def runEncodeBenchmark[T <: AtomicType]( name: String, iters: Int, count: Int, tpe: NativeColumnType[T], input: ByteBuffer): Unit = { - val benchmark = new Benchmark(name, iters * count) schemes.filter(_.supports(tpe)).map { scheme => - def toRow(d: Any) = new GenericInternalRow(Array[Any](d)) - val encoder = scheme.encoder(tpe) - for (i <- 0 until count) { - encoder.gatherCompressibilityStats(toRow(tpe.extract(input)), 0) - } - input.rewind() + val (compressFunc, compressionRatio, buf) = prepareEncodeInternal(count, tpe, scheme, input) + val label = s"${getFormattedClassName(scheme)}(${compressionRatio.formatted("%.3f")})" - val label = s"${getFormattedClassName(scheme)}(${encoder.compressionRatio.formatted("%.3f")})" benchmark.addCase(label)({ i: Int => - val compressedSize = if (encoder.compressedSize == 0) { - input.remaining() - } else { - encoder.compressedSize + for (n <- 0L until iters) { + compressFunc(input, buf) + input.rewind() + buf.rewind() } + }) + } + + benchmark.run() + } - val buf = allocateLocal(4 + compressedSize) + private[this] def runDecodeBenchmark[T <: AtomicType]( + name: String, + iters: Int, + count: Int, + tpe: NativeColumnType[T], + input: ByteBuffer): Unit = { + val benchmark = new Benchmark(name, iters * count) + + schemes.filter(_.supports(tpe)).map { scheme => + val (compressFunc, _, buf) = prepareEncodeInternal(count, tpe, scheme, input) + val compressedBuf = compressFunc(input, buf) + val label = s"${getFormattedClassName(scheme)}" + + input.rewind() + + benchmark.addCase(label)({ i: Int => val rowBuf = new GenericMutableRow(1) - val compressedBuf = encoder.compress(input, buf) - input.rewind() for (n <- 0L until iters) { compressedBuf.rewind.position(4) @@ -96,16 +125,10 @@ object CompressionSchemeBenchmark extends AllCompressionSchemes { benchmark.run() } - def bitDecode(iters: Int): Unit = { + def bitEncodingBenchmark(iters: Int): Unit = { val count = 65536 val testData = allocateLocal(count * BOOLEAN.defaultSize) - // Intel(R) Core(TM) i7-4578U CPU @ 3.00GHz - // BOOLEAN Decode: Avg Time(ms) Avg Rate(M/s) Relative Rate - // ------------------------------------------------------------------------------- - // PassThrough(1.000) 124.98 536.96 1.00 X - // RunLengthEncoding(2.494) 631.37 106.29 0.20 X - // BooleanBitSet(0.125) 1200.36 55.91 0.10 X val g = { val rng = genLowerSkewData() () => (rng().toInt % 2).toByte @@ -113,110 +136,176 @@ object CompressionSchemeBenchmark extends AllCompressionSchemes { for (i <- 0 until count) { testData.put(i * BOOLEAN.defaultSize, g()) } - runBenchmark("BOOLEAN Decode", iters, count, BOOLEAN, testData) + + // Intel(R) Core(TM) i7-4578U CPU @ 3.00GHz + // BOOLEAN Encode: Best/Avg Time(ms) Rate(M/s) Per Row(ns) Relative + // ------------------------------------------------------------------------------------------- + // PassThrough(1.000) 3 / 4 19300.2 0.1 1.0X + // RunLengthEncoding(2.491) 923 / 939 72.7 13.8 0.0X + // BooleanBitSet(0.125) 359 / 363 187.1 5.3 0.0X + runEncodeBenchmark("BOOLEAN Encode", iters, count, BOOLEAN, testData) + + + // Intel(R) Core(TM) i7-4578U CPU @ 3.00GHz + // BOOLEAN Decode: Best/Avg Time(ms) Rate(M/s) Per Row(ns) Relative + // ------------------------------------------------------------------------------------------- + // PassThrough 129 / 136 519.8 1.9 1.0X + // RunLengthEncoding 613 / 623 109.4 9.1 0.2X + // BooleanBitSet 1196 / 1222 56.1 17.8 0.1X + runDecodeBenchmark("BOOLEAN Decode", iters, count, BOOLEAN, testData) } - def shortDecode(iters: Int): Unit = { + def shortEncodingBenchmark(iters: Int): Unit = { val count = 65536 val testData = allocateLocal(count * SHORT.defaultSize) - // Intel(R) Core(TM) i7-4578U CPU @ 3.00GHz - // SHORT Decode (Lower Skew): Avg Time(ms) Avg Rate(M/s) Relative Rate - // ------------------------------------------------------------------------------- - // PassThrough(1.000) 376.87 178.07 1.00 X - // RunLengthEncoding(1.498) 831.59 80.70 0.45 X val g1 = genLowerSkewData() for (i <- 0 until count) { testData.putShort(i * SHORT.defaultSize, g1().toShort) } - runBenchmark("SHORT Decode (Lower Skew)", iters, count, SHORT, testData) // Intel(R) Core(TM) i7-4578U CPU @ 3.00GHz - // SHORT Decode (Higher Skew): Avg Time(ms) Avg Rate(M/s) Relative Rate - // ------------------------------------------------------------------------------- - // PassThrough(1.000) 426.83 157.23 1.00 X - // RunLengthEncoding(1.996) 845.56 79.37 0.50 X + // SHORT Encode (Lower Skew): Best/Avg Time(ms) Rate(M/s) Per Row(ns) Relative + // ------------------------------------------------------------------------------------------- + // PassThrough(1.000) 6 / 7 10971.4 0.1 1.0X + // RunLengthEncoding(1.510) 1526 / 1542 44.0 22.7 0.0X + runEncodeBenchmark("SHORT Encode (Lower Skew)", iters, count, SHORT, testData) + + // Intel(R) Core(TM) i7-4578U CPU @ 3.00GHz + // SHORT Decode (Lower Skew): Best/Avg Time(ms) Rate(M/s) Per Row(ns) Relative + // ------------------------------------------------------------------------------------------- + // PassThrough 811 / 837 82.8 12.1 1.0X + // RunLengthEncoding 1219 / 1266 55.1 18.2 0.7X + runDecodeBenchmark("SHORT Decode (Lower Skew)", iters, count, SHORT, testData) + val g2 = genHigherSkewData() for (i <- 0 until count) { testData.putShort(i * SHORT.defaultSize, g2().toShort) } - runBenchmark("SHORT Decode (Higher Skew)", iters, count, SHORT, testData) + + // Intel(R) Core(TM) i7-4578U CPU @ 3.00GHz + // SHORT Encode (Higher Skew): Best/Avg Time(ms) Rate(M/s) Per Row(ns) Relative + // ------------------------------------------------------------------------------------------- + // PassThrough(1.000) 7 / 7 10112.4 0.1 1.0X + // RunLengthEncoding(2.009) 1623 / 1661 41.4 24.2 0.0X + runEncodeBenchmark("SHORT Encode (Higher Skew)", iters, count, SHORT, testData) + + // Intel(R) Core(TM) i7-4578U CPU @ 3.00GHz + // SHORT Decode (Higher Skew): Best/Avg Time(ms) Rate(M/s) Per Row(ns) Relative + // ------------------------------------------------------------------------------------------- + // PassThrough 818 / 827 82.0 12.2 1.0X + // RunLengthEncoding 1202 / 1237 55.8 17.9 0.7X + runDecodeBenchmark("SHORT Decode (Higher Skew)", iters, count, SHORT, testData) } - def intDecode(iters: Int): Unit = { + def intEncodingBenchmark(iters: Int): Unit = { val count = 65536 val testData = allocateLocal(count * INT.defaultSize) - // Intel(R) Core(TM) i7-4578U CPU @ 3.00GHz - // INT Decode(Lower Skew): Avg Time(ms) Avg Rate(M/s) Relative Rate - // ------------------------------------------------------------------------------- - // PassThrough(1.000) 325.16 206.39 1.00 X - // RunLengthEncoding(0.997) 1219.44 55.03 0.27 X - // DictionaryEncoding(0.500) 955.51 70.23 0.34 X - // IntDelta(0.250) 1146.02 58.56 0.28 X val g1 = genLowerSkewData() for (i <- 0 until count) { testData.putInt(i * INT.defaultSize, g1().toInt) } - runBenchmark("INT Decode(Lower Skew)", iters, count, INT, testData) // Intel(R) Core(TM) i7-4578U CPU @ 3.00GHz - // INT Decode(Higher Skew): Avg Time(ms) Avg Rate(M/s) Relative Rate - // ------------------------------------------------------------------------------- - // PassThrough(1.000) 1133.45 59.21 1.00 X - // RunLengthEncoding(1.334) 1399.00 47.97 0.81 X - // DictionaryEncoding(0.501) 1032.87 64.97 1.10 X - // IntDelta(0.250) 948.02 70.79 1.20 X + // INT Encode (Lower Skew): Best/Avg Time(ms) Rate(M/s) Per Row(ns) Relative + // ------------------------------------------------------------------------------------------- + // PassThrough(1.000) 18 / 19 3716.4 0.3 1.0X + // RunLengthEncoding(1.001) 1992 / 2056 33.7 29.7 0.0X + // DictionaryEncoding(0.500) 723 / 739 92.8 10.8 0.0X + // IntDelta(0.250) 368 / 377 182.2 5.5 0.0X + runEncodeBenchmark("INT Encode (Lower Skew)", iters, count, INT, testData) + + // Intel(R) Core(TM) i7-4578U CPU @ 3.00GHz + // INT Decode (Lower Skew): Best/Avg Time(ms) Rate(M/s) Per Row(ns) Relative + // ------------------------------------------------------------------------------------------- + // PassThrough 821 / 845 81.8 12.2 1.0X + // RunLengthEncoding 1246 / 1256 53.9 18.6 0.7X + // DictionaryEncoding 757 / 766 88.6 11.3 1.1X + // IntDelta 680 / 689 98.7 10.1 1.2X + runDecodeBenchmark("INT Decode (Lower Skew)", iters, count, INT, testData) + val g2 = genHigherSkewData() for (i <- 0 until count) { testData.putInt(i * INT.defaultSize, g2().toInt) } - runBenchmark("INT Decode(Higher Skew)", iters, count, INT, testData) + + // Intel(R) Core(TM) i7-4578U CPU @ 3.00GHz + // INT Encode (Higher Skew): Best/Avg Time(ms) Rate(M/s) Per Row(ns) Relative + // ------------------------------------------------------------------------------------------- + // PassThrough(1.000) 17 / 19 3888.4 0.3 1.0X + // RunLengthEncoding(1.339) 2127 / 2148 31.5 31.7 0.0X + // DictionaryEncoding(0.501) 960 / 972 69.9 14.3 0.0X + // IntDelta(0.250) 362 / 366 185.5 5.4 0.0X + runEncodeBenchmark("INT Encode (Higher Skew)", iters, count, INT, testData) + + // Intel(R) Core(TM) i7-4578U CPU @ 3.00GHz + // INT Decode (Higher Skew): Best/Avg Time(ms) Rate(M/s) Per Row(ns) Relative + // ------------------------------------------------------------------------------------------- + // PassThrough 838 / 884 80.1 12.5 1.0X + // RunLengthEncoding 1287 / 1311 52.1 19.2 0.7X + // DictionaryEncoding 844 / 859 79.5 12.6 1.0X + // IntDelta 764 / 784 87.8 11.4 1.1X + runDecodeBenchmark("INT Decode (Higher Skew)", iters, count, INT, testData) } - def longDecode(iters: Int): Unit = { + def longEncodingBenchmark(iters: Int): Unit = { val count = 65536 val testData = allocateLocal(count * LONG.defaultSize) - // Intel(R) Core(TM) i7-4578U CPU @ 3.00GHz - // LONG Decode(Lower Skew): Avg Time(ms) Avg Rate(M/s) Relative Rate - // ------------------------------------------------------------------------------- - // PassThrough(1.000) 1101.07 60.95 1.00 X - // RunLengthEncoding(0.756) 1372.57 48.89 0.80 X - // DictionaryEncoding(0.250) 947.80 70.81 1.16 X - // LongDelta(0.125) 721.51 93.01 1.53 X val g1 = genLowerSkewData() for (i <- 0 until count) { testData.putLong(i * LONG.defaultSize, g1().toLong) } - runBenchmark("LONG Decode(Lower Skew)", iters, count, LONG, testData) // Intel(R) Core(TM) i7-4578U CPU @ 3.00GHz - // LONG Decode(Higher Skew): Avg Time(ms) Avg Rate(M/s) Relative Rate - // ------------------------------------------------------------------------------- - // PassThrough(1.000) 986.71 68.01 1.00 X - // RunLengthEncoding(1.013) 1348.69 49.76 0.73 X - // DictionaryEncoding(0.251) 865.48 77.54 1.14 X - // LongDelta(0.125) 816.90 82.15 1.21 X + // LONG Encode (Lower Skew): Best/Avg Time(ms) Rate(M/s) Per Row(ns) Relative + // ------------------------------------------------------------------------------------------- + // PassThrough(1.000) 37 / 38 1804.8 0.6 1.0X + // RunLengthEncoding(0.748) 2065 / 2094 32.5 30.8 0.0X + // DictionaryEncoding(0.250) 950 / 962 70.6 14.2 0.0X + // LongDelta(0.125) 475 / 482 141.2 7.1 0.1X + runEncodeBenchmark("LONG Encode (Lower Skew)", iters, count, LONG, testData) + + // Intel(R) Core(TM) i7-4578U CPU @ 3.00GHz + // LONG Decode (Lower Skew): Best/Avg Time(ms) Rate(M/s) Per Row(ns) Relative + // ------------------------------------------------------------------------------------------- + // PassThrough 888 / 894 75.5 13.2 1.0X + // RunLengthEncoding 1301 / 1311 51.6 19.4 0.7X + // DictionaryEncoding 887 / 904 75.7 13.2 1.0X + // LongDelta 693 / 735 96.8 10.3 1.3X + runDecodeBenchmark("LONG Decode (Lower Skew)", iters, count, LONG, testData) + val g2 = genHigherSkewData() for (i <- 0 until count) { testData.putLong(i * LONG.defaultSize, g2().toLong) } - runBenchmark("LONG Decode(Higher Skew)", iters, count, LONG, testData) + + // Intel(R) Core(TM) i7-4578U CPU @ 3.00GHz + // LONG Encode (Higher Skew): Best/Avg Time(ms) Rate(M/s) Per Row(ns) Relative + // ------------------------------------------------------------------------------------------- + // PassThrough(1.000) 34 / 35 1963.9 0.5 1.0X + // RunLengthEncoding(0.999) 2260 / 3021 29.7 33.7 0.0X + // DictionaryEncoding(0.251) 1270 / 1438 52.8 18.9 0.0X + // LongDelta(0.125) 496 / 509 135.3 7.4 0.1X + runEncodeBenchmark("LONG Encode (Higher Skew)", iters, count, LONG, testData) + + // Intel(R) Core(TM) i7-4578U CPU @ 3.00GHz + // LONG Decode (Higher Skew): Best/Avg Time(ms) Rate(M/s) Per Row(ns) Relative + // ------------------------------------------------------------------------------------------- + // PassThrough 965 / 1494 69.5 14.4 1.0X + // RunLengthEncoding 1350 / 1378 49.7 20.1 0.7X + // DictionaryEncoding 892 / 924 75.2 13.3 1.1X + // LongDelta 817 / 847 82.2 12.2 1.2X + runDecodeBenchmark("LONG Decode (Higher Skew)", iters, count, LONG, testData) } - def stringDecode(iters: Int): Unit = { + def stringEncodingBenchmark(iters: Int): Unit = { val count = 65536 val strLen = 8 val tableSize = 16 val testData = allocateLocal(count * (4 + strLen)) - // Intel(R) Core(TM) i7-4578U CPU @ 3.00GHz - // STRING Decode: Avg Time(ms) Avg Rate(M/s) Relative Rate - // ------------------------------------------------------------------------------- - // PassThrough(1.000) 2277.05 29.47 1.00 X - // RunLengthEncoding(0.893) 2624.35 25.57 0.87 X - // DictionaryEncoding(0.167) 2672.28 25.11 0.85 X val g = { val dataTable = (0 until tableSize).map(_ => RandomStringUtils.randomAlphabetic(strLen)) val rng = genHigherSkewData() @@ -227,14 +316,29 @@ object CompressionSchemeBenchmark extends AllCompressionSchemes { testData.put(g().getBytes) } testData.rewind() - runBenchmark("STRING Decode", iters, count, STRING, testData) + + // Intel(R) Core(TM) i7-4578U CPU @ 3.00GHz + // STRING Encode: Best/Avg Time(ms) Rate(M/s) Per Row(ns) Relative + // ------------------------------------------------------------------------------------------- + // PassThrough(1.000) 56 / 57 1197.9 0.8 1.0X + // RunLengthEncoding(0.893) 4892 / 4937 13.7 72.9 0.0X + // DictionaryEncoding(0.167) 2968 / 2992 22.6 44.2 0.0X + runEncodeBenchmark("STRING Encode", iters, count, STRING, testData) + + // Intel(R) Core(TM) i7-4578U CPU @ 3.00GHz + // STRING Decode: Best/Avg Time(ms) Rate(M/s) Per Row(ns) Relative + // ------------------------------------------------------------------------------------------- + // PassThrough 2422 / 2449 27.7 36.1 1.0X + // RunLengthEncoding 2885 / 3018 23.3 43.0 0.8X + // DictionaryEncoding 2716 / 2752 24.7 40.5 0.9X + runDecodeBenchmark("STRING Decode", iters, count, STRING, testData) } def main(args: Array[String]): Unit = { - bitDecode(1024) - shortDecode(1024) - intDecode(1024) - longDecode(1024) - stringDecode(1024) + bitEncodingBenchmark(1024) + shortEncodingBenchmark(1024) + intEncodingBenchmark(1024) + longEncodingBenchmark(1024) + stringEncodingBenchmark(1024) } } |