kazuyukitanimura commented on code in PR #1190:
URL: https://github.com/apache/datafusion-comet/pull/1190#discussion_r1911623676
##########
native/core/benches/shuffle_writer.rs:
##########
@@ -31,67 +31,54 @@ use std::sync::Arc;
use tokio::runtime::Runtime;
fn criterion_benchmark(c: &mut Criterion) {
+ let batch = create_batch(8192, true);
let mut group = c.benchmark_group("shuffle_writer");
- group.bench_function("shuffle_writer: encode (no compression))", |b| {
- let batch = create_batch(8192, true);
- let mut buffer = vec![];
- let ipc_time = Time::default();
- b.iter(|| {
- buffer.clear();
- let mut cursor = Cursor::new(&mut buffer);
- write_ipc_compressed(&batch, &mut cursor, &CompressionCodec::None,
&ipc_time)
- });
- });
- group.bench_function("shuffle_writer: encode and compress (snappy)", |b| {
- let batch = create_batch(8192, true);
- let mut buffer = vec![];
- let ipc_time = Time::default();
- b.iter(|| {
- buffer.clear();
- let mut cursor = Cursor::new(&mut buffer);
- write_ipc_compressed(&batch, &mut cursor,
&CompressionCodec::Snappy, &ipc_time)
- });
- });
- group.bench_function("shuffle_writer: encode and compress (lz4)", |b| {
- let batch = create_batch(8192, true);
- let mut buffer = vec![];
- let ipc_time = Time::default();
- b.iter(|| {
- buffer.clear();
- let mut cursor = Cursor::new(&mut buffer);
- write_ipc_compressed(&batch, &mut cursor,
&CompressionCodec::Lz4Frame, &ipc_time)
- });
- });
- group.bench_function("shuffle_writer: encode and compress (zstd level 1)",
|b| {
- let batch = create_batch(8192, true);
- let mut buffer = vec![];
- let ipc_time = Time::default();
- b.iter(|| {
- buffer.clear();
- let mut cursor = Cursor::new(&mut buffer);
- write_ipc_compressed(&batch, &mut cursor,
&CompressionCodec::Zstd(1), &ipc_time)
- });
- });
- group.bench_function("shuffle_writer: encode and compress (zstd level 6)",
|b| {
- let batch = create_batch(8192, true);
- let mut buffer = vec![];
- let ipc_time = Time::default();
- b.iter(|| {
- buffer.clear();
- let mut cursor = Cursor::new(&mut buffer);
- write_ipc_compressed(&batch, &mut cursor,
&CompressionCodec::Zstd(6), &ipc_time)
- });
- });
- group.bench_function("shuffle_writer: end to end", |b| {
- let ctx = SessionContext::new();
- let exec = create_shuffle_writer_exec(CompressionCodec::Zstd(1));
- b.iter(|| {
- let task_ctx = ctx.task_ctx();
- let stream = exec.execute(0, task_ctx).unwrap();
- let rt = Runtime::new().unwrap();
- criterion::black_box(rt.block_on(collect(stream)).unwrap());
- });
- });
+ for compression_codec in &[
+ CompressionCodec::None,
+ CompressionCodec::Lz4Frame,
+ CompressionCodec::Snappy,
+ CompressionCodec::Zstd(1),
+ CompressionCodec::Zstd(6),
+ ] {
+ for enable_fast_encoding in [true, false] {
+ let name = format!("shuffle_writer: write encoded
(enable_fast_encoding={enable_fast_encoding},
compression={compression_codec:?})");
+ group.bench_function(name, |b| {
+ let mut buffer = vec![];
+ let ipc_time = Time::default();
+ let w = ShuffleBlockWriter::try_new(
+ &batch.schema(),
+ enable_fast_encoding,
+ compression_codec.clone(),
+ )
+ .unwrap();
+ b.iter(|| {
+ buffer.clear();
+ let mut cursor = Cursor::new(&mut buffer);
+ w.write_batch(&batch, &mut cursor, &ipc_time).unwrap();
+ });
+ });
+ }
+ }
+
+ for compression_codec in [
+ CompressionCodec::None,
+ CompressionCodec::Lz4Frame,
+ CompressionCodec::Zstd(1),
Review Comment:
nit/optional: snappy and zstd(6) here as well?
--
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: github-unsubscr...@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscr...@datafusion.apache.org
For additional commands, e-mail: github-h...@datafusion.apache.org
