sunchao commented on code in PR #6856:
URL: https://github.com/apache/datafusion-comet/pull/6856#discussion_r4266110522
##########
spark/src/test/scala/org/apache/spark/sql/benchmark/CometC2RIsolatedBench.scala:
##########
@@ -92,15 +148,58 @@ object CometC2RIsolatedBench {
batches.foreach(_.close())
}
+ /** Reads every field of the flat schema. */
+ private def consumeFlat(u: InternalRow): Long =
+ u.getLong(0) + u.getInt(1) + u.getDouble(2).toLong +
u.getUTF8String(3).numBytes()
+
+ /**
+ * Reads each nested column through its accessor, descending into the
struct, array and map
+ * elements, so that the cost of materialising a nested value is part of the
measurement.
+ */
+ private def consumeNested(u: InternalRow): Long = {
+ var sink = u.getLong(0)
+ if (!u.isNullAt(1)) {
+ val s = u.getStruct(1, 3)
+ sink += s.getInt(0) + s.getUTF8String(1).numBytes()
+ if (!s.isNullAt(2)) sink += s.getStruct(2, 2).getLong(1)
+ }
+ if (!u.isNullAt(2)) {
+ val a = u.getArray(2)
+ var i = 0
+ while (i < a.numElements()) {
+ if (!a.isNullAt(i)) sink += a.getInt(i)
+ i += 1
+ }
+ }
+ if (!u.isNullAt(3)) {
+ val a = u.getArray(3)
+ var i = 0
+ while (i < a.numElements()) {
+ sink += a.getUTF8String(i).numBytes()
+ i += 1
+ }
+ }
+ if (!u.isNullAt(4)) {
+ val m = u.getMap(4)
+ sink += m.numElements() + m.valueArray().getInt(0)
+ }
+ val structs = u.getArray(5)
+ var i = 0
+ while (i < structs.numElements()) {
+ val e = structs.getStruct(i, 2)
+ sink += e.getInt(0) + e.getUTF8String(1).numBytes()
+ i += 1
+ }
+ sink
+ }
+
def main(args: Array[String]): Unit = {
- val schema = new StructType()
- .add("a", LongType)
- .add("b", IntegerType)
- .add("c", DoubleType)
- .add("d", StringType)
-
- runForBatchSize(schema, 8192)
- runForBatchSize(schema, 512)
- runForBatchSize(schema, 32)
+ runForBatchSize("flat", flatSchema, 8192, flatRow, consumeFlat)
+ runForBatchSize("flat", flatSchema, 512, flatRow, consumeFlat)
+ runForBatchSize("flat", flatSchema, 32, flatRow, consumeFlat)
+
+ runForBatchSize("nested", nestedSchema, 8192, nestedRow, consumeNested)
+ runForBatchSize("nested", nestedSchema, 512, nestedRow, consumeNested)
+ runForBatchSize("nested", nestedSchema, 32, nestedRow, consumeNested)
Review Comment:
[P2] Please bound the retained fixtures for this new 32-row nested case.
`makeBatches` materializes all 1,048,576 rows into 32,768 independently
allocated batches before timing. `ArrowWriter.create` sizes top-level
fixed-width vectors, but nested vectors retain Arrow's default capacities. The
resulting fixture allocation is approximately 12.14 GiB, versus 1.54 GiB for
the existing flat 32-row case. This makes a small-batch conversion benchmark
require over 12 GiB of native fixture memory before either arm runs, preventing
it from fitting within an 8 GiB worker budget that could accommodate the
previous fixture. Right-size the nested buffers during setup, or retain a
bounded batch collection and repeat conversion to preserve the timed row count.
Evidence: A disposable Spark 4.1.3 / Arrow 18.3.0 probe compiled exact-head
`ArrowWriters.scala` and `CometArrowConverters.scala` and used the benchmark's
unchanged schema and row generators. Four 32-row nested batches retained
1,591,296 bytes at starting rows 0, 524288, and 1048448. That is 397,824 bytes
per batch, or 13,035,896,832 bytes for 32,768 batches. The equivalent flat
samples retained 201,728 bytes, totaling 1,652,555,776 bytes. A separate
bounded 256-batch run retained 101,842,944 bytes for nested fixtures versus
12,910,592 bytes for flat fixtures. All buffers were closed after measurement.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
