From 8c6020b79c2643c9b9bc9df8512c5a36cdd78561 Mon Sep 17 00:00:00 2001 From: bowenliang123 Date: Tue, 4 Aug 2026 13:42:41 +0800 Subject: [PATCH 1/2] [COMMON] Reduce intermediate allocations in TRowSet generation - getColumnToList: replace the foreach closure with a while loop so the row index and null bitmap stay in registers instead of a captured IntRef, hoist the convertFunc null-check out of the loop, and pre-size the null BitSet - asFloatTColumn/asByteArrayTColumn: build the target value list in a single pass via convertFunc instead of materializing an intermediate boxed list plus JavaConverters wrappers - toColumnBasedSet: stop pre-sizing the always-empty rows list - toRowBasedSet: hoist the column size out of the row loop and append TRows directly into a pre-sized ArrayList instead of building an intermediate Scala collection via map + asJava --- .../engine/result/TColumnGenerator.scala | 40 ++++++++++--------- .../engine/result/TRowSetGenerator.scala | 21 +++++----- 2 files changed, 32 insertions(+), 29 deletions(-) diff --git a/kyuubi-common/src/main/scala/org/apache/kyuubi/engine/result/TColumnGenerator.scala b/kyuubi-common/src/main/scala/org/apache/kyuubi/engine/result/TColumnGenerator.scala index 5a2acd0bdd7..017e7c72712 100644 --- a/kyuubi-common/src/main/scala/org/apache/kyuubi/engine/result/TColumnGenerator.scala +++ b/kyuubi-common/src/main/scala/org/apache/kyuubi/engine/result/TColumnGenerator.scala @@ -20,8 +20,6 @@ import java.lang.{Boolean => JBoolean, Byte => JByte, Double => JDouble, Float = import java.nio.ByteBuffer import java.util.{ArrayList => JArrayList, BitSet => JBitSet, List => JList} -import scala.collection.JavaConverters._ - import org.apache.kyuubi.shaded.hive.service.rpc.thrift._ trait TColumnGenerator[RowT] extends TRowSetColumnGetter[RowT] { @@ -32,19 +30,19 @@ trait TColumnGenerator[RowT] extends TRowSetColumnGetter[RowT] { convertFunc: (RowT, Int) => T = null): (JList[T], ByteBuffer) = { val rowSize = rows.length val ret = new JArrayList[T](rowSize) - val nulls = new JBitSet() + val nulls = new JBitSet(rowSize) + val valueOf: (RowT, Int) => T = + if (convertFunc == null) (row, ord) => getColumnAs[T](row, ord) else convertFunc + val iter = rows.iterator var idx = 0 - val isConvertFuncNull = convertFunc == null - rows.foreach { row => - val value = if (isColumnNullAt(row, ordinal)) { - nulls.set(idx, true) - defaultVal - } else if (isConvertFuncNull) { - getColumnAs[T](row, ordinal) + while (iter.hasNext) { + val row = iter.next() + if (isColumnNullAt(row, ordinal)) { + nulls.set(idx) + ret.add(defaultVal) } else { - convertFunc(row, ordinal) + ret.add(valueOf(row, ordinal)) } - ret.add(value) idx += 1 } (ret, ByteBuffer.wrap(nulls.toByteArray)) @@ -85,9 +83,12 @@ trait TColumnGenerator[RowT] extends TRowSetColumnGetter[RowT] { } def asFloatTColumn(rows: Seq[RowT], ordinal: Int): TColumn = { - val (values, nulls) = getColumnToList[JFloat](rows, ordinal, 0.toFloat) - val doubleValues = values.asScala.map(f => JDouble.valueOf(f.toString)).asJava - TColumn.doubleVal(new TDoubleColumn(doubleValues, nulls)) + val (values, nulls) = getColumnToList[JDouble]( + rows, + ordinal, + 0.toDouble, + (row, ord) => JDouble.valueOf(getColumnAs[JFloat](row, ord).toString)) + TColumn.doubleVal(new TDoubleColumn(values, nulls)) } def asDoubleTColumn(rows: Seq[RowT], ordinal: Int): TColumn = { @@ -105,8 +106,11 @@ trait TColumnGenerator[RowT] extends TRowSetColumnGetter[RowT] { } def asByteArrayTColumn(rows: Seq[RowT], ordinal: Int): TColumn = { - val (values, nulls) = getColumnToList[Array[Byte]](rows, ordinal, defaultVal = Array[Byte]()) - val byteBufferValues = values.asScala.map(ByteBuffer.wrap).asJava - TColumn.binaryVal(new TBinaryColumn(byteBufferValues, nulls)) + val (values, nulls) = getColumnToList[ByteBuffer]( + rows, + ordinal, + defaultVal = ByteBuffer.wrap(Array[Byte]()), + (row, ord) => ByteBuffer.wrap(getColumnAs[Array[Byte]](row, ord))) + TColumn.binaryVal(new TBinaryColumn(values, nulls)) } } diff --git a/kyuubi-common/src/main/scala/org/apache/kyuubi/engine/result/TRowSetGenerator.scala b/kyuubi-common/src/main/scala/org/apache/kyuubi/engine/result/TRowSetGenerator.scala index 4f5701919eb..d5c7d382a2b 100644 --- a/kyuubi-common/src/main/scala/org/apache/kyuubi/engine/result/TRowSetGenerator.scala +++ b/kyuubi-common/src/main/scala/org/apache/kyuubi/engine/result/TRowSetGenerator.scala @@ -18,8 +18,6 @@ package org.apache.kyuubi.engine.result import java.util.{ArrayList => JArrayList} -import scala.collection.JavaConverters._ - import org.apache.kyuubi.shaded.hive.service.rpc.thrift._ trait TRowSetGenerator[SchemaT, RowT, ColumnT] @@ -42,23 +40,24 @@ trait TRowSetGenerator[SchemaT, RowT, ColumnT] } def toRowBasedSet(rows: Seq[RowT], schema: SchemaT): TRowSet = { - val tRows = rows.map { row => - var i = 0 - val columnSize = getColumnSizeFromSchemaType(schema) + val columnSize = getColumnSizeFromSchemaType(schema) + val tRows = new JArrayList[TRow](rows.size) + val rowIter = rows.iterator + while (rowIter.hasNext) { + val row = rowIter.next() val tColumnValues = new JArrayList[TColumnValue](columnSize) + var i = 0 while (i < columnSize) { - val columnValue = toTColumnValue(row, i, schema) - tColumnValues.add(columnValue) + tColumnValues.add(toTColumnValue(row, i, schema)) i += 1 } - new TRow(tColumnValues) - }.asJava + tRows.add(new TRow(tColumnValues)) + } new TRowSet(0, tRows) } def toColumnBasedSet(rows: Seq[RowT], schema: SchemaT): TRowSet = { - val rowSize = rows.length - val tRowSet = new TRowSet(0, new JArrayList[TRow](rowSize)) + val tRowSet = new TRowSet(0, new JArrayList[TRow](0)) var i = 0 val columnSize = getColumnSizeFromSchemaType(schema) val tColumns = new JArrayList[TColumn](columnSize) From 014748c358571087a7ad540285c7dbb779ff36e8 Mon Sep 17 00:00:00 2001 From: bowenliang123 Date: Tue, 4 Aug 2026 15:34:02 +0800 Subject: [PATCH 2/2] [SPARK] Build DynMethods once for HiveResult.toHiveString Each call rebuilt the DynMethods chain via reflection; hoist the bound method into a private val so only invokeChecked runs per cell. --- .../apache/kyuubi/engine/spark/schema/RowSet.scala | 14 ++++++++------ 1 file changed, 8 insertions(+), 6 deletions(-) diff --git a/externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/schema/RowSet.scala b/externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/schema/RowSet.scala index 47e6351b2d9..cf36e1ec86d 100644 --- a/externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/schema/RowSet.scala +++ b/externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/schema/RowSet.scala @@ -37,11 +37,7 @@ object RowSet { .buildChecked(HiveResult) .invokeChecked[BinaryFormatter]() - def toHiveString( - valueAndType: (Any, DataType), - nested: JBoolean = false, - timeFormatters: TimeFormatters, - binaryFormatter: BinaryFormatter): String = + private val toHiveStringMethod = DynMethods.builder("toHiveString") .impl( // for Spark 3.5 and before HiveResult.getClass, @@ -55,5 +51,11 @@ object RowSet { classOf[TimeFormatters], classOf[BinaryFormatter]) .buildChecked(HiveResult) - .invokeChecked[String](valueAndType, nested, timeFormatters, binaryFormatter) + + def toHiveString( + valueAndType: (Any, DataType), + nested: JBoolean = false, + timeFormatters: TimeFormatters, + binaryFormatter: BinaryFormatter): String = + toHiveStringMethod.invokeChecked[String](valueAndType, nested, timeFormatters, binaryFormatter) }