This is an automated email from the ASF dual-hosted git repository.
andygrove pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
The following commit(s) were added to refs/heads/main by this push:
new 5627ab8c01 perf: reduce allocations when collecting cache statistics
(#5734)
5627ab8c01 is described below
commit 5627ab8c01473b186db2fbe73c91c599b1387f9b
Author: Peter Lee <[email protected]>
AuthorDate: Wed Sep 9 20:10:36 2026 +0800
perf: reduce allocations when collecting cache statistics (#5734)
* perf: specialize native cache statistics collection
* refactor: address cache statistics review feedback
---
.../arrow/ArrowCachedBatchSerializer.scala | 258 ++++++++++++++++-----
.../comet/exec/CometInMemoryCacheSuite.scala | 151 ++++++++++++
.../benchmark/CometInMemoryCacheBenchmark.scala | 43 ++++
3 files changed, 394 insertions(+), 58 deletions(-)
diff --git
a/spark/src/main/scala/org/apache/spark/sql/comet/execution/arrow/ArrowCachedBatchSerializer.scala
b/spark/src/main/scala/org/apache/spark/sql/comet/execution/arrow/ArrowCachedBatchSerializer.scala
index 46a51ad877..821f84e0c2 100644
---
a/spark/src/main/scala/org/apache/spark/sql/comet/execution/arrow/ArrowCachedBatchSerializer.scala
+++
b/spark/src/main/scala/org/apache/spark/sql/comet/execution/arrow/ArrowCachedBatchSerializer.scala
@@ -19,6 +19,8 @@
package org.apache.spark.sql.comet.execution.arrow
+import java.lang.{Boolean => JBoolean, Byte => JByte, Double => JDouble, Float
=> JFloat, Integer => JInteger, Long => JLong, Short => JShort}
+
import scala.collection.JavaConverters._
import scala.util.control.NonFatal
@@ -26,6 +28,7 @@ import org.apache.spark.TaskContext
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.expressions.{Attribute, Expression,
GenericInternalRow, IsNotNull, IsNull, UnsafeProjection}
+import org.apache.spark.sql.catalyst.util.TypeUtils
import org.apache.spark.sql.columnar.{CachedBatch, SimpleMetricsCachedBatch,
SimpleMetricsCachedBatchSerializer}
import org.apache.spark.sql.comet.util.Utils
import org.apache.spark.sql.execution.columnar.{DefaultCachedBatch,
DefaultCachedBatchSerializer}
@@ -33,7 +36,7 @@ import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types._
import org.apache.spark.sql.vectorized.{ColumnarBatch, ColumnVector}
import org.apache.spark.storage.StorageLevel
-import org.apache.spark.unsafe.types.{ByteArray, UTF8String}
+import org.apache.spark.unsafe.types.UTF8String
import org.apache.spark.util.io.ChunkedByteBuffer
import org.apache.comet.CometArrowAllocator
@@ -78,40 +81,220 @@ class ArrowCachedBatchSerializer extends
SimpleMetricsCachedBatchSerializer {
// Bounds and null counts per column, gathered before the batch is
serialized: serializing
// clears the batch's vectors, and the per-column byte sizes that complete
the statistics row
// are only known afterwards. See statsRow.
- private def gatherColumnStats(
+ private[sql] def gatherColumnStats(
batch: ColumnarBatch,
attrs: Seq[Attribute]): (Array[Any], Array[Any], Array[Int]) = {
val numCols = attrs.length
val lower = new Array[Any](numCols)
val upper = new Array[Any](numCols)
- val nulls = Array.fill[Int](numCols)(0)
+ val nulls = new Array[Int](numCols)
val numRows = batch.numRows()
var c = 0
while (c < numCols) {
- val dt = attrs(c).dataType
val col = batch.column(c)
- var r = 0
- while (r < numRows) {
- if (col.isNullAt(r)) {
- nulls(c) += 1
- } else if (tracksBounds(dt)) {
- val value = readValue(col, dt, r)
- if (lower(c) == null || compare(dt, value, lower(c)) < 0) {
- lower(c) = value
- }
- if (upper(c) == null || compare(dt, value, upper(c)) > 0) {
- upper(c) = value
+ val (min, max, nullCount) = attrs(c).dataType match {
+ case BooleanType => gatherBooleanStats(col, numRows)
+ case ByteType => gatherByteStats(col, numRows)
+ case ShortType => gatherShortStats(col, numRows)
+ case IntegerType | DateType => gatherIntStats(col, numRows)
+ case LongType | TimestampType | TimestampNTZType =>
gatherLongStats(col, numRows)
+ case FloatType => gatherFloatStats(col, numRows)
+ case DoubleType => gatherDoubleStats(col, numRows)
+ case d: DecimalType => gatherDecimalStats(col, numRows, d)
+ case StringType => gatherStringStats(col, numRows)
+ case other =>
+ assert(!tracksBounds(other), s"Missing cache bounds implementation
for $other")
+ var nullCount = 0
+ var r = 0
+ while (r < numRows) {
+ if (col.isNullAt(r)) nullCount += 1
+ r += 1
}
- }
- r += 1
+ (null, null, nullCount)
}
+ if (nullCount < numRows) {
+ lower(c) = min
+ upper(c) = max
+ }
+ nulls(c) = nullCount
c += 1
}
(lower, upper, nulls)
}
+ // Keep each loop specialized and box only its final bounds. r == nullCount
identifies
+ // the first non-null value, including when a column starts with nulls.
+ private def gatherBooleanStats(col: ColumnVector, numRows: Int): (Boolean,
Boolean, Int) = {
+ var min = false
+ var max = false
+ var nullCount = 0
+ var r = 0
+ while (r < numRows) {
+ if (col.isNullAt(r)) {
+ nullCount += 1
+ } else {
+ val value = col.getBoolean(r)
+ if (r == nullCount || JBoolean.compare(value, min) < 0) min = value
+ if (r == nullCount || JBoolean.compare(value, max) > 0) max = value
+ }
+ r += 1
+ }
+ (min, max, nullCount)
+ }
+
+ private def gatherByteStats(col: ColumnVector, numRows: Int): (Byte, Byte,
Int) = {
+ var min = 0.toByte
+ var max = 0.toByte
+ var nullCount = 0
+ var r = 0
+ while (r < numRows) {
+ if (col.isNullAt(r)) {
+ nullCount += 1
+ } else {
+ val value = col.getByte(r)
+ if (r == nullCount || JByte.compare(value, min) < 0) min = value
+ if (r == nullCount || JByte.compare(value, max) > 0) max = value
+ }
+ r += 1
+ }
+ (min, max, nullCount)
+ }
+
+ private def gatherShortStats(col: ColumnVector, numRows: Int): (Short,
Short, Int) = {
+ var min = 0.toShort
+ var max = 0.toShort
+ var nullCount = 0
+ var r = 0
+ while (r < numRows) {
+ if (col.isNullAt(r)) {
+ nullCount += 1
+ } else {
+ val value = col.getShort(r)
+ if (r == nullCount || JShort.compare(value, min) < 0) min = value
+ if (r == nullCount || JShort.compare(value, max) > 0) max = value
+ }
+ r += 1
+ }
+ (min, max, nullCount)
+ }
+
+ private def gatherIntStats(col: ColumnVector, numRows: Int): (Int, Int, Int)
= {
+ var min = 0
+ var max = 0
+ var nullCount = 0
+ var r = 0
+ while (r < numRows) {
+ if (col.isNullAt(r)) {
+ nullCount += 1
+ } else {
+ val value = col.getInt(r)
+ if (r == nullCount || JInteger.compare(value, min) < 0) min = value
+ if (r == nullCount || JInteger.compare(value, max) > 0) max = value
+ }
+ r += 1
+ }
+ (min, max, nullCount)
+ }
+
+ private def gatherLongStats(col: ColumnVector, numRows: Int): (Long, Long,
Int) = {
+ var min = 0L
+ var max = 0L
+ var nullCount = 0
+ var r = 0
+ while (r < numRows) {
+ if (col.isNullAt(r)) {
+ nullCount += 1
+ } else {
+ val value = col.getLong(r)
+ if (r == nullCount || JLong.compare(value, min) < 0) min = value
+ if (r == nullCount || JLong.compare(value, max) > 0) max = value
+ }
+ r += 1
+ }
+ (min, max, nullCount)
+ }
+
+ private def gatherFloatStats(col: ColumnVector, numRows: Int): (Float,
Float, Int) = {
+ var min = 0.0f
+ var max = 0.0f
+ var nullCount = 0
+ var r = 0
+ while (r < numRows) {
+ if (col.isNullAt(r)) {
+ nullCount += 1
+ } else {
+ val value = col.getFloat(r)
+ if (r == nullCount || JFloat.compare(value, min) < 0) min = value
+ if (r == nullCount || JFloat.compare(value, max) > 0) max = value
+ }
+ r += 1
+ }
+ (min, max, nullCount)
+ }
+
+ private def gatherDoubleStats(col: ColumnVector, numRows: Int): (Double,
Double, Int) = {
+ var min = 0.0d
+ var max = 0.0d
+ var nullCount = 0
+ var r = 0
+ while (r < numRows) {
+ if (col.isNullAt(r)) {
+ nullCount += 1
+ } else {
+ val value = col.getDouble(r)
+ if (r == nullCount || JDouble.compare(value, min) < 0) min = value
+ if (r == nullCount || JDouble.compare(value, max) > 0) max = value
+ }
+ r += 1
+ }
+ (min, max, nullCount)
+ }
+
+ private def gatherDecimalStats(
+ col: ColumnVector,
+ numRows: Int,
+ dt: DecimalType): (Decimal, Decimal, Int) = {
+ var min: Decimal = null
+ var max: Decimal = null
+ var nullCount = 0
+ var r = 0
+ while (r < numRows) {
+ if (col.isNullAt(r)) {
+ nullCount += 1
+ } else {
+ val value = col.getDecimal(r, dt.precision, dt.scale)
+ if (r == nullCount || value.compare(min) < 0) min = value
+ if (r == nullCount || value.compare(max) > 0) max = value
+ }
+ r += 1
+ }
+ (min, max, nullCount)
+ }
+
+ private def gatherStringStats(
+ col: ColumnVector,
+ numRows: Int): (UTF8String, UTF8String, Int) = {
+ val ordering = TypeUtils.getInterpretedOrdering(StringType)
+ var min: UTF8String = null
+ var max: UTF8String = null
+ var nullCount = 0
+ var r = 0
+ while (r < numRows) {
+ if (col.isNullAt(r)) {
+ nullCount += 1
+ } else {
+ val value = col.getUTF8String(r)
+ // Borrow the value for comparison, but retain owned copies of the
bounds.
+ if (r == nullCount || ordering.compare(value, min) < 0) min =
value.copy()
+ if (r == nullCount || ordering.compare(value, max) > 0) max =
value.copy()
+ }
+ r += 1
+ }
+ (min, max, nullCount)
+ }
+
// Build the statistics row expected by SimpleMetricsCachedBatchSerializer.
// For each cached column Spark expects five values in this order:
// lower bound, upper bound, null count, row count, and size in bytes.
@@ -149,47 +332,6 @@ class ArrowCachedBatchSerializer extends
SimpleMetricsCachedBatchSerializer {
case _ => false
}
- // Read a non-null value from a ColumnVector using Spark's internal value
type
- // for the corresponding DataType.
- private def readValue(col: ColumnVector, dt: DataType, rowId: Int): Any = dt
match {
- case BooleanType => col.getBoolean(rowId)
- case ByteType => col.getByte(rowId)
- case ShortType => col.getShort(rowId)
- case IntegerType | DateType => col.getInt(rowId)
- case LongType | TimestampType | TimestampNTZType => col.getLong(rowId)
- case FloatType => col.getFloat(rowId)
- case DoubleType => col.getDouble(rowId)
- case d: DecimalType => col.getDecimal(rowId, d.precision, d.scale)
- case StringType => col.getUTF8String(rowId).copy()
- case _ => null
- }
-
- // Compare values using the same physical representation used in the stats
row.
- private def compare(dt: DataType, left: Any, right: Any): Int = dt match {
- case BooleanType =>
- java.lang.Boolean.compare(left.asInstanceOf[Boolean],
right.asInstanceOf[Boolean])
- case ByteType =>
- java.lang.Byte.compare(left.asInstanceOf[Byte], right.asInstanceOf[Byte])
- case ShortType =>
- java.lang.Short.compare(left.asInstanceOf[Short],
right.asInstanceOf[Short])
- case IntegerType | DateType =>
- java.lang.Integer.compare(left.asInstanceOf[Int],
right.asInstanceOf[Int])
- case LongType | TimestampType | TimestampNTZType =>
- java.lang.Long.compare(left.asInstanceOf[Long], right.asInstanceOf[Long])
- case FloatType =>
- java.lang.Float.compare(left.asInstanceOf[Float],
right.asInstanceOf[Float])
- case DoubleType =>
- java.lang.Double.compare(left.asInstanceOf[Double],
right.asInstanceOf[Double])
- case _: DecimalType =>
- left.asInstanceOf[Decimal].compare(right.asInstanceOf[Decimal])
- case StringType =>
- ByteArray.compareBinary(
- left.asInstanceOf[UTF8String].getBytes,
- right.asInstanceOf[UTF8String].getBytes)
- case other =>
- throw new IllegalStateException(s"compare called for unsupported type
$other")
- }
-
// Compute Spark-compatible cache stats before serializing each batch to
Arrow.
// The stats are stored beside the Arrow bytes so Spark's cache filter can
prune
// CometCachedBatch without decoding the batch first.
diff --git
a/spark/src/test/scala/org/apache/comet/exec/CometInMemoryCacheSuite.scala
b/spark/src/test/scala/org/apache/comet/exec/CometInMemoryCacheSuite.scala
index 065299f101..e72f4d10f7 100644
--- a/spark/src/test/scala/org/apache/comet/exec/CometInMemoryCacheSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/exec/CometInMemoryCacheSuite.scala
@@ -36,6 +36,7 @@ import org.apache.spark.sql.execution.exchange.{Exchange,
ReusedExchangeExec, Sh
import org.apache.spark.sql.execution.joins.{BroadcastHashJoinExec,
SortMergeJoinExec}
import org.apache.spark.sql.functions.max
import org.apache.spark.sql.internal.{SQLConf, StaticSQLConf}
+import org.apache.spark.sql.types._
import org.apache.spark.storage.StorageLevel
import org.apache.comet.{CometArrowAllocator, CometConf}
@@ -483,6 +484,156 @@ class CometInMemoryCacheSuite extends CometTestBase {
}
}
+ test("Comet in-memory cache statistics preserve typed bounds and null
counts") {
+ withSQLConf(
+ CometConf.COMET_ENABLED.key -> "false",
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
+ val types = Seq(
+ "boolean",
+ "tinyint",
+ "smallint",
+ "int",
+ "bigint",
+ "float",
+ "double",
+ "decimal(10,2)",
+ "decimal(38,2)",
+ "string",
+ "date",
+ "timestamp",
+ "timestamp_ntz",
+ "binary")
+ val expressions = types.zipWithIndex.map { case (dt, i) =>
+ val value = dt match {
+ case "date" => "date_add(DATE '2000-01-01', cast(v AS INT))"
+ case "timestamp" | "timestamp_ntz" =>
+ s"cast(date_add(DATE '2000-01-01', cast(v AS INT)) AS $dt)"
+ case "string" | "binary" => s"cast(concat('字', cast(v AS STRING)) AS
$dt)"
+ case _ => s"cast(v AS $dt)"
+ }
+ s"$value AS c$i"
+ }
+ // Leading nulls, updates in both directions, duplicate values, all-null
and single-value
+ // columns exercise initialization as well as the primitive and
reference bounds loops.
+ Seq("(NULL), (2), (-3), (0), (1), (2), (NULL)", "(NULL), (NULL)",
"(NULL), (1)").foreach {
+ values =>
+ val df = spark
+ .sql(s"SELECT ${expressions.mkString(", ")} FROM VALUES $values AS
t(v)")
+ .coalesce(1)
+ // Compute the reference through Spark before caching, using its
internal value types.
+ df.createOrReplaceTempView("typed_stats_input")
+ val expected = spark
+ .sql(
+ s"SELECT ${types.indices.flatMap(i => Seq(s"min(c$i)",
s"max(c$i)")).mkString(", ")} " +
+ "FROM typed_stats_input")
+ .queryExecution
+ .toRdd
+ .map(_.copy())
+ .collect()
+ .head
+ val expectedNulls = df.filter("c0 IS NULL").count().toInt
+ val expectedRows = df.count().toInt
+ def checkStats(view: String): Unit = {
+ val relation =
spark.sharedState.cacheManager.lookupCachedData(spark.table(view)).get
+ val batches =
relation.cachedRepresentation.cacheBuilder.cachedColumnBuffers.collect()
+ assert(batches.length == 1)
+ val stats =
batches.head.asInstanceOf[SimpleMetricsCachedBatch].stats
+ df.schema.fields.zipWithIndex.foreach { case (field, i) =>
+ if (field.dataType == BinaryType) {
+ assert(stats.isNullAt(i * 5) && stats.isNullAt(i * 5 + 1))
+ } else {
+ assert(stats.get(i * 5, field.dataType) == expected.get(i * 2,
field.dataType))
+ assert(
+ stats.get(i * 5 + 1, field.dataType) ==
+ expected.get(i * 2 + 1, field.dataType))
+ }
+ assert(stats.getInt(i * 5 + 2) == expectedNulls)
+ assert(stats.getInt(i * 5 + 3) == expectedRows)
+ }
+ }
+
+ df.cache()
+ try {
+ df.count()
+ checkStats("typed_stats_input")
+ } finally {
+ df.unpersist(blocking = true)
+ spark.catalog.dropTempView("typed_stats_input")
+ }
+
+ withSparkColumnarCache("typed_stats_columnar")(path =>
df.write.parquet(path)) {
+ val relation = spark.sharedState.cacheManager
+ .lookupCachedData(spark.table("typed_stats_columnar"))
+ .get
+ .cachedRepresentation
+ assert(relation.cacheBuilder.cachedPlan.supportsColumnar)
+ checkStats("typed_stats_columnar")
+ }
+ }
+ }
+ }
+
+ test("Comet in-memory cache statistics preserve numeric extremes and
floating-point ordering") {
+ withSQLConf(
+ CometConf.COMET_ENABLED.key -> "false",
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
+ val df = spark
+ .sql("""
+ SELECT
+ CAST(if(id = 0, -128, 127) AS TINYINT) AS b,
+ CAST(if(id = 0, -32768, 32767) AS SMALLINT) AS s,
+ CAST(if(id = 0, -2147483648, 2147483647) AS INT) AS i,
+ if(id = 0, -9223372036854775808L, 9223372036854775807L) AS l,
+ CAST(v AS FLOAT) AS f,
+ CAST(v AS DOUBLE) AS d,
+ CAST(if(id = 0, '0.0', '-0.0') AS FLOAT) AS fz,
+ CAST(if(id = 0, '0.0', '-0.0') AS DOUBLE) AS dz
+ FROM VALUES (0, '0.0'), (1, '-0.0'), (2, '-Infinity'), (3,
'Infinity'), (4, 'NaN')
+ AS t(id, v)
+ """)
+ .coalesce(1)
+
+ def checkStats(view: String): Unit = {
+ val relation =
spark.sharedState.cacheManager.lookupCachedData(spark.table(view)).get
+ val batches =
relation.cachedRepresentation.cacheBuilder.cachedColumnBuffers.collect()
+ assert(batches.length == 1)
+ val stats = batches.head.asInstanceOf[SimpleMetricsCachedBatch].stats
+ assert(stats.getByte(0) == Byte.MinValue && stats.getByte(1) ==
Byte.MaxValue)
+ assert(stats.getShort(5) == Short.MinValue && stats.getShort(6) ==
Short.MaxValue)
+ assert(stats.getInt(10) == Int.MinValue && stats.getInt(11) ==
Int.MaxValue)
+ assert(stats.getLong(15) == Long.MinValue && stats.getLong(16) ==
Long.MaxValue)
+ assert(stats.getFloat(20) == Float.NegativeInfinity &&
stats.getFloat(21).isNaN)
+ assert(stats.getDouble(25) == Double.NegativeInfinity &&
stats.getDouble(26).isNaN)
+ // Spark aggregates consider signed zeros equal; the cache stores
Java's total ordering.
+ assert(
+ java.lang.Float.floatToRawIntBits(stats.getFloat(30)) ==
+ java.lang.Float.floatToRawIntBits(-0.0f))
+ assert(java.lang.Float.floatToRawIntBits(stats.getFloat(31)) == 0)
+ assert(
+ java.lang.Double.doubleToRawLongBits(stats.getDouble(35)) ==
+ java.lang.Double.doubleToRawLongBits(-0.0d))
+ assert(java.lang.Double.doubleToRawLongBits(stats.getDouble(36)) == 0L)
+ (0 until 8).foreach { c =>
+ assert(stats.getInt(c * 5 + 2) == 0)
+ assert(stats.getInt(c * 5 + 3) == 5)
+ }
+ }
+
+ df.createOrReplaceTempView("extreme_stats_input")
+ df.cache()
+ try {
+ df.count()
+ checkStats("extreme_stats_input")
+ } finally {
+ df.unpersist(blocking = true)
+ spark.catalog.dropTempView("extreme_stats_input")
+ }
+ withSparkColumnarCache("extreme_stats_columnar")(path =>
df.write.parquet(path)) {
+ checkStats("extreme_stats_columnar")
+ }
+ }
+ }
+
test("Comet in-memory cache supports stats-based batch pruning") {
withSQLConf(
SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
diff --git
a/spark/src/test/scala/org/apache/spark/sql/benchmark/CometInMemoryCacheBenchmark.scala
b/spark/src/test/scala/org/apache/spark/sql/benchmark/CometInMemoryCacheBenchmark.scala
index 959b5590b3..7d171ac26e 100644
---
a/spark/src/test/scala/org/apache/spark/sql/benchmark/CometInMemoryCacheBenchmark.scala
+++
b/spark/src/test/scala/org/apache/spark/sql/benchmark/CometInMemoryCacheBenchmark.scala
@@ -19,10 +19,17 @@
package org.apache.spark.sql.benchmark
+import java.nio.charset.StandardCharsets
+
import org.apache.spark.SparkConf
import org.apache.spark.benchmark.Benchmark
import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.expressions.AttributeReference
+import org.apache.spark.sql.comet.execution.arrow.ArrowCachedBatchSerializer
+import org.apache.spark.sql.execution.vectorized.OnHeapColumnVector
import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.{DataType, LongType, StringType}
+import org.apache.spark.sql.vectorized.{ColumnarBatch, ColumnVector}
import org.apache.comet.{CometConf, CometSparkSessionExtensions}
@@ -30,6 +37,7 @@ object CometInMemoryCacheBenchmark extends CometBenchmarkBase
{
private val numRows = 5 * 1000 * 1000
private val cacheTable = "comet_cache_bench"
private val sourceTable = "comet_cache_bench_src"
+ @volatile private var statsResult: (Array[Any], Array[Any], Array[Int]) = _
override def getSparkSession: SparkSession = {
val conf = new SparkConf()
@@ -61,6 +69,10 @@ object CometInMemoryCacheBenchmark extends
CometBenchmarkBase {
}
override def runCometBenchmark(args: Array[String]): Unit = {
+ runStatsBenchmark()
+ // Run just the JVM statistics loop without constructing or scanning a
cached relation.
+ if (args.contains("--stats-only")) return
+
withTempTable(sourceTable, cacheTable) {
spark
.range(0, numRows, 1, 16)
@@ -102,6 +114,37 @@ object CometInMemoryCacheBenchmark extends
CometBenchmarkBase {
}
}
+ private def runStatsBenchmark(): Unit = {
+ val batchSize = 10000
+ val types: Seq[DataType] = Seq.fill(3)(LongType) ++ Seq.fill(3)(StringType)
+ val attrs = types.zipWithIndex.map { case (dt, i) =>
AttributeReference(s"c$i", dt)() }
+ val columns = types.map(dt => new OnHeapColumnVector(batchSize, dt))
+ val batch = new ColumnarBatch(columns.map(c => c: ColumnVector).toArray,
batchSize)
+ try {
+ var r = 0
+ while (r < batchSize) {
+ columns(0).putLong(r, r.toLong)
+ columns(1).putLong(r, r % 1000)
+ columns(2).putLong(r, r + 1)
+ columns(3).putByteArray(r, s"str_a_${r %
100000}".getBytes(StandardCharsets.UTF_8))
+ columns(4).putByteArray(r, s"str_b_${r %
7919}".getBytes(StandardCharsets.UTF_8))
+ columns(5).putByteArray(r,
s"str_c_$r".getBytes(StandardCharsets.UTF_8))
+ r += 1
+ }
+ val serializer = new ArrowCachedBatchSerializer
+ val benchmark = new Benchmark("in-memory cache statistics", numRows,
output = output)
+ // One case measures this collector across commits; Spark's default
cache has its own collector.
+ benchmark.addCase("Comet statistics collector") { _ =>
+ var i = 0
+ while (i < numRows / batchSize) {
+ statsResult = serializer.gatherColumnStats(batch, attrs)
+ i += 1
+ }
+ }
+ benchmark.run()
+ } finally batch.close()
+ }
+
private def runCacheBenchmark(name: String, query: String): Unit = {
withCachedTable {
withSQLConf(cacheConf(nativeCacheEnabled = false): _*) {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
