andygrove commented on code in PR #7065:
URL: https://github.com/apache/datafusion-comet/pull/7065#discussion_r4308022953
##########
spark/src/main/java/org/apache/spark/shuffle/sort/SpillSorter.java:
##########
@@ -235,135 +238,83 @@ public int numRecords() {
return this.inMemSorter.numRecords();
}
+ /**
+ * Sorts the buffered records by partition and writes them to the spill file
set by {@link
+ * #setSpillInfo}, every partition in one native call, recording each
partition's length and
+ * checksum.
+ */
public void writeSortedFileNative(boolean isLastFile, boolean
tracingEnabled) throws IOException {
Review Comment:
Renamed to `writeSortedPartitions` in fc7595251e, with the two callers in
`CometShuffleExternalSorter`, `SpillSorterSuite` and `jvm_shuffle.md`. A grep
for `writeSortedFileNative` now only finds the per-partition path.
##########
native/shuffle/src/spark_unsafe/row.rs:
##########
@@ -1357,6 +1356,170 @@ fn make_builders(
Ok(builder)
}
+/// Bits of Spark's `PackedRecordPointer`: the partition id in the upper 24
bits, then the page
+/// number in 13 bits and the offset in the page in the lower 27.
+const PACKED_PARTITION_ID_SHIFT: u32 = 40;
+const PACKED_PAGE_NUMBER_SHIFT: u32 = 27;
+const PACKED_PAGE_NUMBER_MASK: u64 = (1 << 13) - 1;
+const PACKED_OFFSET_MASK: u64 = (1 << 27) - 1;
+
+/// Bytes `UnsafeRowSerializer` writes before each row for the row's size.
+const SERIALIZED_ROW_SIZE_BYTES: i64 = 4;
+
+/// Encoded blocks are buffered until they reach this many bytes, so that a
spill of many small
+/// partitions is not written one partition at a time.
+const OUTPUT_BUFFER_BYTES: usize = 1 << 20;
Review Comment:
Added to the doc comment on `buffer` in fc7595251e: it is native memory that
no pool reserves, like the builders, and its contents are written out as soon
as they reach `OUTPUT_BUFFER_BYTES`, so they stay under that size plus one
block. It also says the `Vec` keeps the capacity it grew to, since `clear()`
does not release it.
##########
spark/src/test/scala/org/apache/spark/sql/benchmark/CometColumnarShuffleWriteBenchmark.scala:
##########
@@ -0,0 +1,119 @@
+/*
+ * 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.sql.benchmark
+
+import org.apache.spark.benchmark.Benchmark
+import org.apache.spark.sql.DataFrame
+import org.apache.spark.sql.comet.execution.shuffle.{CometColumnarShuffle,
CometShuffleExchangeExec}
+import org.apache.spark.sql.functions.col
+import org.apache.spark.sql.internal.SQLConf
+
+import org.apache.comet.CometConf
+
+/**
+ * Measures Comet's JVM columnar shuffle at partition counts above Spark's
+ * `spark.shuffle.sort.bypassMergeThreshold` (200), where the sort-based
writer is used. That
+ * writer sorts its buffered rows by partition and writes every partition
holding a row, once per
+ * spill and once more at the end, so its cost per partition is paid
`partitions x (spills + 1)`
+ * times. Each case hash-partitions the same rows and discards the result.
+ *
+ * The partition count is set through `spark.sql.shuffle.partitions` rather
than `repartition(n,
+ * ...)`, so that AQE coalesces the read side into a few tasks and the
measurement is not
+ * dominated by scheduling one reduce task per partition.
+ *
+ * `spark.comet.shuffle.jvm.spillThreshold` makes Comet spill a fixed number
of rows at a time. It
+ * does not apply to Spark's shuffle, which does not spill in these cases.
+ *
+ * To run this benchmark:
+ * {{{
+ * SPARK_GENERATE_BENCHMARK_FILES=1 make
benchmark-org.apache.spark.sql.benchmark.CometColumnarShuffleWriteBenchmark
+ * }}}
+ * Results will be written to
"spark/benchmarks/CometColumnarShuffleWriteBenchmark-**results.txt".
+ */
+object CometColumnarShuffleWriteBenchmark extends CometBenchmarkBase {
Review Comment:
Yes, the session. `CometShuffleBenchmark` runs `local[5]` and sets a spill
threshold of 30000 for the whole session, and the master cannot change per
group. These cases need `local[1]`, so the map tasks run one at a time and the
time follows their cost, and the default threshold, so the no-spill case does
not spill. I added that to the scaladoc in fc7595251e rather than moving the
group.
--
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]
