You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "ding-young (via GitHub)" <gi...@apache.org> on 2025/06/23 11:36:25 UTC
[PR] Add microbenchmark for spilling with compression [datafusion]
ding-young opened a new pull request, #16512:
URL: https://github.com/apache/datafusion/pull/16512
## Which issue does this PR close?
<!--
We generally require a GitHub issue to be filed for all bug fixes and enhancements and this helps us generate change logs for our releases. You can link an issue to this PR using the GitHub syntax. For example `Closes #123` indicates that this PR will close issue #123.
-->
- Related to #16367
## Rationale for this change
<!--
Why are you proposing this change? If this is already explained clearly in the issue then this section is not needed.
Explaining clearly why changes are proposed helps reviewers understand your changes and offer better suggestions for fixes.
-->
## What changes are included in this PR?
This pr adds some microbenchmarks to compare performance characteristics between different compression codecs.
To make benchmark more realistic, I ran TPC-H SF10 with limited memory, and made `RecordBatches` that resemble the data it spills on `AggregateExec`.
## TODO
Most TPC-H queries spills on `AggregateExec` or fails on `HashJoin` (since by default tpc-h benchmark sets `prefer_hash_join` as true), while `SortExec` is usually at the end of execution pipeline without less memory pressure. Therefore, I need to add other benchmark cases based on `sort_tpch` to mimick RecordBatches that are spilled during `SortExec`.
<!--
There is no need to duplicate the description in the issue here but it is sometimes worth providing a summary of the individual changes in this PR.
-->
## Are these changes tested?
<!--
We typically require tests for all PRs in order to:
1. Prevent the code from being accidentally broken by subsequent changes
2. Serve as another way to document the expected behavior of the code
If tests are not included in your PR, please explain why (for example, are they covered by existing tests)?
-->
## Are there any user-facing changes?
<!--
If there are user-facing changes then we may require documentation to be updated before approving the PR.
-->
<!--
If there are any breaking changes to public APIs, please add the `api change` label.
-->
--
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-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add microbenchmark for spilling with compression [datafusion]
Posted by "ding-young (via GitHub)" <gi...@apache.org>.
ding-young commented on PR #16512:
URL: https://github.com/apache/datafusion/pull/16512#issuecomment-2996151512
To run bench, `cargo bench --bench spill_io`
### Q2
- spill_compression/q2/uncompressed
time: [51.207 ms 51.521 ms 51.841 ms]
[q2 | Uncompressed] mem: 9.4 MB| disk: 9.5 MB| compression ratio: 0.990x
- spill_compression/q2/zstd
time: [146.99 ms 147.36 ms 147.70 ms]
[q2 | Zstd] mem: 9.4 MB| disk: 1546.1 KB| compression ratio: 6.215x
- spill_compression/q2/lz4_frame
time: [97.601 ms 97.942 ms 98.273 ms]
[q2 | Lz4Frame] mem: 9.4 MB| disk: 3.2 MB| compression ratio: 2.922x
### Q16
- spill_compression/q16/uncompressed
time: [75.849 ms 78.053 ms 80.220 ms]
[q16 | Uncompressed] mem: 23.5 MB| disk: 19.5 MB| compression ratio: 1.209x
- spill_compression/q16/zstd
time: [235.68 ms 236.48 ms 237.19 ms]
[q16 | Zstd] mem: 23.5 MB| disk: 4.4 MB| compression ratio: 5.373x
- spill_compression/q16/lz4_frame
time: [142.50 ms 145.18 ms 147.88 ms]
[q16 | Lz4Frame] mem: 23.5 MB| disk: 7.8 MB| compression ratio: 3.007x
### Q20
- spill_compression/q20/uncompressed
time: [63.979 ms 64.233 ms 64.482 ms]
[q20 | Uncompressed] mem: 12.5 MB| disk: 12.7 MB| compression ratio: 0.989x
- spill_compression/q20/zstd
time: [189.99 ms 190.57 ms 191.14 ms]
[q20 | Zstd] mem: 12.5 MB| disk: 2.4 MB| compression ratio: 5.282x
- spill_compression/q20/lz4_frame
time: [123.13 ms 123.43 ms 123.72 ms]
[q20 | Lz4Frame] mem: 12.5 MB| disk: 4.8 MB| compression ratio: 2.629x
--
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-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add microbenchmark for spilling with compression [datafusion]
Posted by "2010YOUY01 (via GitHub)" <gi...@apache.org>.
2010YOUY01 merged PR #16512:
URL: https://github.com/apache/datafusion/pull/16512
--
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-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add microbenchmark for spilling with compression [datafusion]
Posted by "2010YOUY01 (via GitHub)" <gi...@apache.org>.
2010YOUY01 commented on code in PR #16512:
URL: https://github.com/apache/datafusion/pull/16512#discussion_r2168056666
##########
datafusion/physical-plan/benches/spill_io.rs:
##########
@@ -119,5 +126,414 @@ fn bench_spill_io(c: &mut Criterion) {
group.finish();
}
-criterion_group!(benches, bench_spill_io);
+// Generate `num_batches` RecordBatches mimicking TPC-H Q2's partial aggregate result:
+// GROUP BY ps_partkey -> MIN(ps_supplycost)
+fn create_q2_like_batches(num_batches: usize) -> (Arc<Schema>, Vec<RecordBatch>) {
+ // use fixed seed
+ let seed = 2;
+ let mut rng = rand::rngs::StdRng::seed_from_u64(seed);
+ let mut batches = Vec::with_capacity(num_batches);
+
+ let mut current_key = 400000_i64;
+
+ let schema = Arc::new(Schema::new(vec![
+ Field::new("ps_partkey", DataType::Int64, false),
+ Field::new("min_ps_supplycost", DataType::Decimal128(15, 2), true),
+ ]));
+
+ for _ in 0..num_batches {
+ let mut partkey_builder = Int64Builder::new();
+ let mut cost_builder = Decimal128Builder::new()
+ .with_precision_and_scale(15, 2)
+ .unwrap();
+
+ for _ in 0..8192 {
Review Comment:
I think we should make this `8192` a global variable or function arg
##########
datafusion/physical-plan/benches/spill_io.rs:
##########
@@ -119,5 +126,414 @@ fn bench_spill_io(c: &mut Criterion) {
group.finish();
}
-criterion_group!(benches, bench_spill_io);
+// Generate `num_batches` RecordBatches mimicking TPC-H Q2's partial aggregate result:
+// GROUP BY ps_partkey -> MIN(ps_supplycost)
+fn create_q2_like_batches(num_batches: usize) -> (Arc<Schema>, Vec<RecordBatch>) {
+ // use fixed seed
+ let seed = 2;
+ let mut rng = rand::rngs::StdRng::seed_from_u64(seed);
+ let mut batches = Vec::with_capacity(num_batches);
+
+ let mut current_key = 400000_i64;
+
+ let schema = Arc::new(Schema::new(vec![
+ Field::new("ps_partkey", DataType::Int64, false),
+ Field::new("min_ps_supplycost", DataType::Decimal128(15, 2), true),
+ ]));
+
+ for _ in 0..num_batches {
+ let mut partkey_builder = Int64Builder::new();
+ let mut cost_builder = Decimal128Builder::new()
+ .with_precision_and_scale(15, 2)
+ .unwrap();
+
+ for _ in 0..8192 {
+ // Occasionally skip a few partkey values to simulate sparsity
+ let jump = if rng.random_bool(0.05) {
+ rng.random_range(2..10)
+ } else {
+ 1
+ };
+ current_key += jump;
+
+ let supply_cost = rng.random_range(10_00..100_000) as i128;
+
+ partkey_builder.append_value(current_key);
+ cost_builder.append_value(supply_cost);
+ }
+
+ let batch = RecordBatch::try_new(
+ Arc::clone(&schema),
+ vec![
+ Arc::new(partkey_builder.finish()),
+ Arc::new(cost_builder.finish()),
+ ],
+ )
+ .unwrap();
+
+ batches.push(batch);
+ }
+
+ (schema, batches)
+}
+
+/// Generate `num_batches` RecordBatches mimicking TPC-H Q16's partial aggregate result:
+/// GROUP BY (p_brand, p_type, p_size) -> COUNT(DISTINCT ps_suppkey)
+pub fn create_q16_like_batches(num_batches: usize) -> (Arc<Schema>, Vec<RecordBatch>) {
+ let seed = 16;
+ let mut rng = rand::rngs::StdRng::seed_from_u64(seed);
+ let mut batches = Vec::with_capacity(num_batches);
+
+ let schema = Arc::new(Schema::new(vec![
+ Field::new("p_brand", DataType::Utf8, false),
+ Field::new("p_type", DataType::Utf8, false),
+ Field::new("p_size", DataType::Int32, false),
+ Field::new("alias1", DataType::Int64, false), // COUNT(DISTINCT ps_suppkey)
+ ]));
+
+ // Representative string pools
+ let brands = ["Brand#32", "Brand#33", "Brand#41", "Brand#42", "Brand#55"];
+ let types = [
+ "PROMO ANODIZED NICKEL",
+ "STANDARD BRUSHED NICKEL",
+ "PROMO POLISHED COPPER",
+ "ECONOMY ANODIZED BRASS",
+ "LARGE BURNISHED COPPER",
+ "STANDARD POLISHED TIN",
+ "SMALL PLATED STEEL",
+ "MEDIUM POLISHED COPPER",
+ ];
+ let sizes = [3, 9, 14, 19, 23, 36, 45, 49];
+
+ for _ in 0..num_batches {
+ let mut brand_builder = StringBuilder::new();
+ let mut type_builder = StringBuilder::new();
+ let mut size_builder = Int32Builder::new();
+ let mut count_builder = Int64Builder::new();
+
+ for _ in 0..8192 {
+ let brand = brands[rng.random_range(0..brands.len())];
+ let ptype = types[rng.random_range(0..types.len())];
+ let size = sizes[rng.random_range(0..sizes.len())];
+ let count = rng.random_range(1000..100_000);
+
+ brand_builder.append_value(brand);
+ type_builder.append_value(ptype);
+ size_builder.append_value(size);
+ count_builder.append_value(count);
+ }
+
+ let batch = RecordBatch::try_new(
+ Arc::clone(&schema),
+ vec![
+ Arc::new(brand_builder.finish()),
+ Arc::new(type_builder.finish()),
+ Arc::new(size_builder.finish()),
+ Arc::new(count_builder.finish()),
+ ],
+ )
+ .unwrap();
+
+ batches.push(batch);
+ }
+
+ (schema, batches)
+}
+
+// Generate `num_batches` RecordBatches mimicking TPC-H Q20's partial aggregate result:
+// GROUP BY (l_partkey, l_suppkey) -> SUM(l_quantity)
+fn create_q20_like_batches(num_batches: usize) -> (Arc<Schema>, Vec<RecordBatch>) {
+ let seed = 20;
+ let mut rng = rand::rngs::StdRng::seed_from_u64(seed);
+ let mut batches = Vec::with_capacity(num_batches);
+
+ let mut current_partkey = 400000_i64;
+
+ let schema = Arc::new(Schema::new(vec![
+ Field::new("l_partkey", DataType::Int64, false),
+ Field::new("l_suppkey", DataType::Int64, false),
+ Field::new("sum_l_quantity", DataType::Decimal128(25, 2), true),
+ ]));
+
+ for _ in 0..num_batches {
+ let mut partkey_builder = Int64Builder::new();
+ let mut suppkey_builder = Int64Builder::new();
+ let mut quantity_builder = Decimal128Builder::new()
+ .with_precision_and_scale(25, 2)
+ .unwrap();
+
+ for _ in 0..8192 {
+ // Occasionally skip a few partkey values to simulate sparsity
+ let partkey_jump = if rng.random_bool(0.03) {
+ rng.random_range(2..6)
+ } else {
+ 1
+ };
+ current_partkey += partkey_jump;
+
+ let suppkey = rng.random_range(10_000..99_999);
+ let quantity = rng.random_range(500..20_000) as i128;
+
+ partkey_builder.append_value(current_partkey);
+ suppkey_builder.append_value(suppkey);
+ quantity_builder.append_value(quantity);
+ }
+
+ let batch = RecordBatch::try_new(
+ Arc::clone(&schema),
+ vec![
+ Arc::new(partkey_builder.finish()),
+ Arc::new(suppkey_builder.finish()),
+ Arc::new(quantity_builder.finish()),
+ ],
+ )
+ .unwrap();
+
+ batches.push(batch);
+ }
+
+ (schema, batches)
+}
+
+/// Genereate `num_batches` wide RecordBatches resembling sort-tpch Q10 for benchmarking.
+/// This includes multiple numeric, date, and Utf8View columns (15 total).
+pub fn create_wide_batches(num_batches: usize) -> (Arc<Schema>, Vec<RecordBatch>) {
+ let seed = 10;
+ let mut rng = rand::rngs::StdRng::seed_from_u64(seed);
+ let mut batches = Vec::with_capacity(num_batches);
+
+ let schema = Arc::new(Schema::new(vec![
+ Field::new("l_linenumber", DataType::Int32, false),
+ Field::new("l_suppkey", DataType::Int64, false),
+ Field::new("l_orderkey", DataType::Int64, false),
+ Field::new("l_partkey", DataType::Int64, false),
+ Field::new("l_quantity", DataType::Decimal128(15, 2), false),
+ Field::new("l_extendedprice", DataType::Decimal128(15, 2), false),
+ Field::new("l_discount", DataType::Decimal128(15, 2), false),
+ Field::new("l_tax", DataType::Decimal128(15, 2), false),
+ Field::new("l_returnflag", DataType::Utf8, false),
+ Field::new("l_linestatus", DataType::Utf8, false),
+ Field::new("l_shipdate", DataType::Date32, false),
+ Field::new("l_commitdate", DataType::Date32, false),
+ Field::new("l_receiptdate", DataType::Date32, false),
+ Field::new("l_shipinstruct", DataType::Utf8, false),
+ Field::new("l_shipmode", DataType::Utf8, false),
+ ]));
+
+ for _ in 0..num_batches {
+ let mut linenum = Int32Builder::new();
+ let mut suppkey = Int64Builder::new();
+ let mut orderkey = Int64Builder::new();
+ let mut partkey = Int64Builder::new();
+ let mut quantity = Decimal128Builder::new()
+ .with_precision_and_scale(15, 2)
+ .unwrap();
+ let mut extprice = Decimal128Builder::new()
+ .with_precision_and_scale(15, 2)
+ .unwrap();
+ let mut discount = Decimal128Builder::new()
+ .with_precision_and_scale(15, 2)
+ .unwrap();
+ let mut tax = Decimal128Builder::new()
+ .with_precision_and_scale(15, 2)
+ .unwrap();
+ let mut retflag = StringBuilder::new();
+ let mut linestatus = StringBuilder::new();
+ let mut shipdate = Date32Builder::new();
+ let mut commitdate = Date32Builder::new();
+ let mut receiptdate = Date32Builder::new();
+ let mut shipinstruct = StringBuilder::new();
+ let mut shipmode = StringBuilder::new();
+
+ let return_flags = ["A", "N", "R"];
+ let statuses = ["F", "O"];
+ let instructs = ["DELIVER IN PERSON", "COLLECT COD", "NONE"];
+ let modes = ["TRUCK", "MAIL", "SHIP", "RAIL", "AIR"];
+
+ for i in 0..8192 {
+ linenum.append_value(i % 7);
+ suppkey.append_value(rng.random_range(0..100_000));
+ orderkey.append_value(1_000_000 + i as i64);
+ partkey.append_value(rng.random_range(0..200_000));
+
+ quantity.append_value(rng.random_range(100..10000) as i128);
+ extprice.append_value(rng.random_range(1_000..1_000_000) as i128);
+ discount.append_value(rng.random_range(0..10000) as i128);
+ tax.append_value(rng.random_range(0..5000) as i128);
+
+ retflag.append_value(return_flags[rng.random_range(0..return_flags.len())]);
+ linestatus.append_value(statuses[rng.random_range(0..statuses.len())]);
+
+ let base_date = 10_000;
+ shipdate.append_value(base_date + (i % 1000));
+ commitdate.append_value(base_date + (i % 1000) + 1);
+ receiptdate.append_value(base_date + (i % 1000) + 2);
+
+ shipinstruct.append_value(instructs[rng.random_range(0..instructs.len())]);
+ shipmode.append_value(modes[rng.random_range(0..modes.len())]);
+ }
+
+ let batch = RecordBatch::try_new(
+ Arc::clone(&schema),
+ vec![
+ Arc::new(linenum.finish()),
+ Arc::new(suppkey.finish()),
+ Arc::new(orderkey.finish()),
+ Arc::new(partkey.finish()),
+ Arc::new(quantity.finish()),
+ Arc::new(extprice.finish()),
+ Arc::new(discount.finish()),
+ Arc::new(tax.finish()),
+ Arc::new(retflag.finish()),
+ Arc::new(linestatus.finish()),
+ Arc::new(shipdate.finish()),
+ Arc::new(commitdate.finish()),
+ Arc::new(receiptdate.finish()),
+ Arc::new(shipinstruct.finish()),
+ Arc::new(shipmode.finish()),
+ ],
+ )
+ .unwrap();
+ batches.push(batch);
+ }
+ (schema, batches)
+}
+
+// Benchmarks spill write + read performance across multiple compression codecs
+// using realistic input data inspired by TPC-H aggregate spill scenarios.
+//
+// This function prepares synthetic RecordBatches that mimic the schema and distribution
+// of intermediate aggregate results from representative TPC-H queries (Q2, Q16, Q20) and sort-tpch Q10
+// For each dataset:
+// - It evaluates spill performance under different compression codecs (e.g., Uncompressed, Zstd, LZ4).
+// - It measures end-to-end spill write + read performance using Criterion.
+// - It prints the observed memory-to-disk compression ratio for each codec.
+//
+// This helps evaluate the tradeoffs between compression ratio and runtime overhead for various codecs.
+fn bench_spill_compression(c: &mut Criterion) {
+ let env = Arc::new(RuntimeEnv::default());
+ let mut group = c.benchmark_group("spill_compression");
+ let rt = Runtime::new().unwrap();
+ let compressions = vec![
+ SpillCompression::Uncompressed,
+ SpillCompression::Zstd,
+ SpillCompression::Lz4Frame,
+ ];
+
+ // Modify this value to change data volume. Note that each batch contains 8192 rows.
+ let num_batches = 50;
+
+ // Q2 [Int64, Decimal128]
+ let (schema, batches) = create_q2_like_batches(50);
Review Comment:
```suggestion
let (schema, batches) = create_q2_like_batches(num_batches);
```
and same for the below two functions
##########
datafusion/physical-plan/benches/spill_io.rs:
##########
@@ -119,5 +126,414 @@ fn bench_spill_io(c: &mut Criterion) {
group.finish();
}
-criterion_group!(benches, bench_spill_io);
+// Generate `num_batches` RecordBatches mimicking TPC-H Q2's partial aggregate result:
+// GROUP BY ps_partkey -> MIN(ps_supplycost)
+fn create_q2_like_batches(num_batches: usize) -> (Arc<Schema>, Vec<RecordBatch>) {
+ // use fixed seed
+ let seed = 2;
+ let mut rng = rand::rngs::StdRng::seed_from_u64(seed);
+ let mut batches = Vec::with_capacity(num_batches);
+
+ let mut current_key = 400000_i64;
+
+ let schema = Arc::new(Schema::new(vec![
+ Field::new("ps_partkey", DataType::Int64, false),
+ Field::new("min_ps_supplycost", DataType::Decimal128(15, 2), true),
+ ]));
+
+ for _ in 0..num_batches {
+ let mut partkey_builder = Int64Builder::new();
+ let mut cost_builder = Decimal128Builder::new()
+ .with_precision_and_scale(15, 2)
+ .unwrap();
+
+ for _ in 0..8192 {
+ // Occasionally skip a few partkey values to simulate sparsity
+ let jump = if rng.random_bool(0.05) {
+ rng.random_range(2..10)
+ } else {
+ 1
+ };
+ current_key += jump;
+
+ let supply_cost = rng.random_range(10_00..100_000) as i128;
+
+ partkey_builder.append_value(current_key);
+ cost_builder.append_value(supply_cost);
+ }
+
+ let batch = RecordBatch::try_new(
+ Arc::clone(&schema),
+ vec![
+ Arc::new(partkey_builder.finish()),
+ Arc::new(cost_builder.finish()),
+ ],
+ )
+ .unwrap();
+
+ batches.push(batch);
+ }
+
+ (schema, batches)
+}
+
+/// Generate `num_batches` RecordBatches mimicking TPC-H Q16's partial aggregate result:
+/// GROUP BY (p_brand, p_type, p_size) -> COUNT(DISTINCT ps_suppkey)
+pub fn create_q16_like_batches(num_batches: usize) -> (Arc<Schema>, Vec<RecordBatch>) {
+ let seed = 16;
+ let mut rng = rand::rngs::StdRng::seed_from_u64(seed);
+ let mut batches = Vec::with_capacity(num_batches);
+
+ let schema = Arc::new(Schema::new(vec![
+ Field::new("p_brand", DataType::Utf8, false),
+ Field::new("p_type", DataType::Utf8, false),
+ Field::new("p_size", DataType::Int32, false),
+ Field::new("alias1", DataType::Int64, false), // COUNT(DISTINCT ps_suppkey)
+ ]));
+
+ // Representative string pools
+ let brands = ["Brand#32", "Brand#33", "Brand#41", "Brand#42", "Brand#55"];
+ let types = [
+ "PROMO ANODIZED NICKEL",
+ "STANDARD BRUSHED NICKEL",
+ "PROMO POLISHED COPPER",
+ "ECONOMY ANODIZED BRASS",
+ "LARGE BURNISHED COPPER",
+ "STANDARD POLISHED TIN",
+ "SMALL PLATED STEEL",
+ "MEDIUM POLISHED COPPER",
+ ];
+ let sizes = [3, 9, 14, 19, 23, 36, 45, 49];
+
+ for _ in 0..num_batches {
+ let mut brand_builder = StringBuilder::new();
+ let mut type_builder = StringBuilder::new();
+ let mut size_builder = Int32Builder::new();
+ let mut count_builder = Int64Builder::new();
+
+ for _ in 0..8192 {
+ let brand = brands[rng.random_range(0..brands.len())];
+ let ptype = types[rng.random_range(0..types.len())];
+ let size = sizes[rng.random_range(0..sizes.len())];
+ let count = rng.random_range(1000..100_000);
+
+ brand_builder.append_value(brand);
+ type_builder.append_value(ptype);
+ size_builder.append_value(size);
+ count_builder.append_value(count);
+ }
+
+ let batch = RecordBatch::try_new(
+ Arc::clone(&schema),
+ vec![
+ Arc::new(brand_builder.finish()),
+ Arc::new(type_builder.finish()),
+ Arc::new(size_builder.finish()),
+ Arc::new(count_builder.finish()),
+ ],
+ )
+ .unwrap();
+
+ batches.push(batch);
+ }
+
+ (schema, batches)
+}
+
+// Generate `num_batches` RecordBatches mimicking TPC-H Q20's partial aggregate result:
+// GROUP BY (l_partkey, l_suppkey) -> SUM(l_quantity)
+fn create_q20_like_batches(num_batches: usize) -> (Arc<Schema>, Vec<RecordBatch>) {
+ let seed = 20;
+ let mut rng = rand::rngs::StdRng::seed_from_u64(seed);
+ let mut batches = Vec::with_capacity(num_batches);
+
+ let mut current_partkey = 400000_i64;
+
+ let schema = Arc::new(Schema::new(vec![
+ Field::new("l_partkey", DataType::Int64, false),
+ Field::new("l_suppkey", DataType::Int64, false),
+ Field::new("sum_l_quantity", DataType::Decimal128(25, 2), true),
+ ]));
+
+ for _ in 0..num_batches {
+ let mut partkey_builder = Int64Builder::new();
+ let mut suppkey_builder = Int64Builder::new();
+ let mut quantity_builder = Decimal128Builder::new()
+ .with_precision_and_scale(25, 2)
+ .unwrap();
+
+ for _ in 0..8192 {
+ // Occasionally skip a few partkey values to simulate sparsity
+ let partkey_jump = if rng.random_bool(0.03) {
+ rng.random_range(2..6)
+ } else {
+ 1
+ };
+ current_partkey += partkey_jump;
+
+ let suppkey = rng.random_range(10_000..99_999);
+ let quantity = rng.random_range(500..20_000) as i128;
+
+ partkey_builder.append_value(current_partkey);
+ suppkey_builder.append_value(suppkey);
+ quantity_builder.append_value(quantity);
+ }
+
+ let batch = RecordBatch::try_new(
+ Arc::clone(&schema),
+ vec![
+ Arc::new(partkey_builder.finish()),
+ Arc::new(suppkey_builder.finish()),
+ Arc::new(quantity_builder.finish()),
+ ],
+ )
+ .unwrap();
+
+ batches.push(batch);
+ }
+
+ (schema, batches)
+}
+
+/// Genereate `num_batches` wide RecordBatches resembling sort-tpch Q10 for benchmarking.
+/// This includes multiple numeric, date, and Utf8View columns (15 total).
+pub fn create_wide_batches(num_batches: usize) -> (Arc<Schema>, Vec<RecordBatch>) {
+ let seed = 10;
+ let mut rng = rand::rngs::StdRng::seed_from_u64(seed);
+ let mut batches = Vec::with_capacity(num_batches);
+
+ let schema = Arc::new(Schema::new(vec![
+ Field::new("l_linenumber", DataType::Int32, false),
+ Field::new("l_suppkey", DataType::Int64, false),
+ Field::new("l_orderkey", DataType::Int64, false),
+ Field::new("l_partkey", DataType::Int64, false),
+ Field::new("l_quantity", DataType::Decimal128(15, 2), false),
+ Field::new("l_extendedprice", DataType::Decimal128(15, 2), false),
+ Field::new("l_discount", DataType::Decimal128(15, 2), false),
+ Field::new("l_tax", DataType::Decimal128(15, 2), false),
+ Field::new("l_returnflag", DataType::Utf8, false),
+ Field::new("l_linestatus", DataType::Utf8, false),
+ Field::new("l_shipdate", DataType::Date32, false),
+ Field::new("l_commitdate", DataType::Date32, false),
+ Field::new("l_receiptdate", DataType::Date32, false),
+ Field::new("l_shipinstruct", DataType::Utf8, false),
+ Field::new("l_shipmode", DataType::Utf8, false),
+ ]));
+
+ for _ in 0..num_batches {
+ let mut linenum = Int32Builder::new();
+ let mut suppkey = Int64Builder::new();
+ let mut orderkey = Int64Builder::new();
+ let mut partkey = Int64Builder::new();
+ let mut quantity = Decimal128Builder::new()
+ .with_precision_and_scale(15, 2)
+ .unwrap();
+ let mut extprice = Decimal128Builder::new()
+ .with_precision_and_scale(15, 2)
+ .unwrap();
+ let mut discount = Decimal128Builder::new()
+ .with_precision_and_scale(15, 2)
+ .unwrap();
+ let mut tax = Decimal128Builder::new()
+ .with_precision_and_scale(15, 2)
+ .unwrap();
+ let mut retflag = StringBuilder::new();
+ let mut linestatus = StringBuilder::new();
+ let mut shipdate = Date32Builder::new();
+ let mut commitdate = Date32Builder::new();
+ let mut receiptdate = Date32Builder::new();
+ let mut shipinstruct = StringBuilder::new();
+ let mut shipmode = StringBuilder::new();
+
+ let return_flags = ["A", "N", "R"];
+ let statuses = ["F", "O"];
+ let instructs = ["DELIVER IN PERSON", "COLLECT COD", "NONE"];
+ let modes = ["TRUCK", "MAIL", "SHIP", "RAIL", "AIR"];
+
+ for i in 0..8192 {
+ linenum.append_value(i % 7);
+ suppkey.append_value(rng.random_range(0..100_000));
+ orderkey.append_value(1_000_000 + i as i64);
+ partkey.append_value(rng.random_range(0..200_000));
+
+ quantity.append_value(rng.random_range(100..10000) as i128);
+ extprice.append_value(rng.random_range(1_000..1_000_000) as i128);
+ discount.append_value(rng.random_range(0..10000) as i128);
+ tax.append_value(rng.random_range(0..5000) as i128);
+
+ retflag.append_value(return_flags[rng.random_range(0..return_flags.len())]);
+ linestatus.append_value(statuses[rng.random_range(0..statuses.len())]);
+
+ let base_date = 10_000;
+ shipdate.append_value(base_date + (i % 1000));
+ commitdate.append_value(base_date + (i % 1000) + 1);
+ receiptdate.append_value(base_date + (i % 1000) + 2);
+
+ shipinstruct.append_value(instructs[rng.random_range(0..instructs.len())]);
+ shipmode.append_value(modes[rng.random_range(0..modes.len())]);
+ }
+
+ let batch = RecordBatch::try_new(
+ Arc::clone(&schema),
+ vec![
+ Arc::new(linenum.finish()),
+ Arc::new(suppkey.finish()),
+ Arc::new(orderkey.finish()),
+ Arc::new(partkey.finish()),
+ Arc::new(quantity.finish()),
+ Arc::new(extprice.finish()),
+ Arc::new(discount.finish()),
+ Arc::new(tax.finish()),
+ Arc::new(retflag.finish()),
+ Arc::new(linestatus.finish()),
+ Arc::new(shipdate.finish()),
+ Arc::new(commitdate.finish()),
+ Arc::new(receiptdate.finish()),
+ Arc::new(shipinstruct.finish()),
+ Arc::new(shipmode.finish()),
+ ],
+ )
+ .unwrap();
+ batches.push(batch);
+ }
+ (schema, batches)
+}
+
+// Benchmarks spill write + read performance across multiple compression codecs
+// using realistic input data inspired by TPC-H aggregate spill scenarios.
+//
+// This function prepares synthetic RecordBatches that mimic the schema and distribution
+// of intermediate aggregate results from representative TPC-H queries (Q2, Q16, Q20) and sort-tpch Q10
Review Comment:
Maybe we can also copy the schemas of each workload in implementation comments to here.
--
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-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add microbenchmark for spilling with compression [datafusion]
Posted by "ding-young (via GitHub)" <gi...@apache.org>.
ding-young commented on code in PR #16512:
URL: https://github.com/apache/datafusion/pull/16512#discussion_r2174808708
##########
datafusion/physical-plan/benches/spill_io.rs:
##########
@@ -119,5 +127,450 @@ fn bench_spill_io(c: &mut Criterion) {
group.finish();
}
-criterion_group!(benches, bench_spill_io);
+// Generate `num_batches` RecordBatches mimicking TPC-H Q2's partial aggregate result:
+// GROUP BY ps_partkey -> MIN(ps_supplycost)
+fn create_q2_like_batches(
+ num_batches: usize,
+ num_rows: usize,
+) -> (Arc<Schema>, Vec<RecordBatch>) {
+ // use fixed seed
+ let seed = 2;
+ let mut rng = rand::rngs::StdRng::seed_from_u64(seed);
+ let mut batches = Vec::with_capacity(num_batches);
+
+ let mut current_key = 400000_i64;
+
+ let schema = Arc::new(Schema::new(vec![
+ Field::new("ps_partkey", DataType::Int64, false),
+ Field::new("min_ps_supplycost", DataType::Decimal128(15, 2), true),
+ ]));
+
+ for _ in 0..num_batches {
+ let mut partkey_builder = Int64Builder::new();
+ let mut cost_builder = Decimal128Builder::new()
+ .with_precision_and_scale(15, 2)
+ .unwrap();
+
+ for _ in 0..num_rows {
+ // Occasionally skip a few partkey values to simulate sparsity
+ let jump = if rng.random_bool(0.05) {
+ rng.random_range(2..10)
+ } else {
+ 1
+ };
+ current_key += jump;
+
+ let supply_cost = rng.random_range(10_00..100_000) as i128;
+
+ partkey_builder.append_value(current_key);
+ cost_builder.append_value(supply_cost);
+ }
+
+ let batch = RecordBatch::try_new(
+ Arc::clone(&schema),
+ vec![
+ Arc::new(partkey_builder.finish()),
+ Arc::new(cost_builder.finish()),
+ ],
+ )
+ .unwrap();
+
+ batches.push(batch);
+ }
+
+ (schema, batches)
+}
+
+/// Generate `num_batches` RecordBatches mimicking TPC-H Q16's partial aggregate result:
+/// GROUP BY (p_brand, p_type, p_size) -> COUNT(DISTINCT ps_suppkey)
+pub fn create_q16_like_batches(
+ num_batches: usize,
+ num_rows: usize,
+) -> (Arc<Schema>, Vec<RecordBatch>) {
+ let seed = 16;
+ let mut rng = rand::rngs::StdRng::seed_from_u64(seed);
+ let mut batches = Vec::with_capacity(num_batches);
+
+ let schema = Arc::new(Schema::new(vec![
+ Field::new("p_brand", DataType::Utf8, false),
+ Field::new("p_type", DataType::Utf8, false),
+ Field::new("p_size", DataType::Int32, false),
+ Field::new("alias1", DataType::Int64, false), // COUNT(DISTINCT ps_suppkey)
+ ]));
+
+ // Representative string pools
+ let brands = ["Brand#32", "Brand#33", "Brand#41", "Brand#42", "Brand#55"];
+ let types = [
+ "PROMO ANODIZED NICKEL",
+ "STANDARD BRUSHED NICKEL",
+ "PROMO POLISHED COPPER",
+ "ECONOMY ANODIZED BRASS",
+ "LARGE BURNISHED COPPER",
+ "STANDARD POLISHED TIN",
+ "SMALL PLATED STEEL",
+ "MEDIUM POLISHED COPPER",
+ ];
+ let sizes = [3, 9, 14, 19, 23, 36, 45, 49];
+
+ for _ in 0..num_batches {
+ let mut brand_builder = StringBuilder::new();
+ let mut type_builder = StringBuilder::new();
+ let mut size_builder = Int32Builder::new();
+ let mut count_builder = Int64Builder::new();
+
+ for _ in 0..num_rows {
+ let brand = brands[rng.random_range(0..brands.len())];
+ let ptype = types[rng.random_range(0..types.len())];
+ let size = sizes[rng.random_range(0..sizes.len())];
+ let count = rng.random_range(1000..100_000);
+
+ brand_builder.append_value(brand);
+ type_builder.append_value(ptype);
+ size_builder.append_value(size);
+ count_builder.append_value(count);
+ }
+
+ let batch = RecordBatch::try_new(
+ Arc::clone(&schema),
+ vec![
+ Arc::new(brand_builder.finish()),
+ Arc::new(type_builder.finish()),
+ Arc::new(size_builder.finish()),
+ Arc::new(count_builder.finish()),
+ ],
+ )
+ .unwrap();
+
+ batches.push(batch);
+ }
+
+ (schema, batches)
+}
+
+// Generate `num_batches` RecordBatches mimicking TPC-H Q20's partial aggregate result:
+// GROUP BY (l_partkey, l_suppkey) -> SUM(l_quantity)
+fn create_q20_like_batches(
+ num_batches: usize,
+ num_rows: usize,
+) -> (Arc<Schema>, Vec<RecordBatch>) {
+ let seed = 20;
+ let mut rng = rand::rngs::StdRng::seed_from_u64(seed);
+ let mut batches = Vec::with_capacity(num_batches);
+
+ let mut current_partkey = 400000_i64;
+
+ let schema = Arc::new(Schema::new(vec![
+ Field::new("l_partkey", DataType::Int64, false),
+ Field::new("l_suppkey", DataType::Int64, false),
+ Field::new("sum_l_quantity", DataType::Decimal128(25, 2), true),
+ ]));
+
+ for _ in 0..num_batches {
+ let mut partkey_builder = Int64Builder::new();
+ let mut suppkey_builder = Int64Builder::new();
+ let mut quantity_builder = Decimal128Builder::new()
+ .with_precision_and_scale(25, 2)
+ .unwrap();
+
+ for _ in 0..num_rows {
+ // Occasionally skip a few partkey values to simulate sparsity
+ let partkey_jump = if rng.random_bool(0.03) {
+ rng.random_range(2..6)
+ } else {
+ 1
+ };
+ current_partkey += partkey_jump;
+
+ let suppkey = rng.random_range(10_000..99_999);
+ let quantity = rng.random_range(500..20_000) as i128;
+
+ partkey_builder.append_value(current_partkey);
+ suppkey_builder.append_value(suppkey);
+ quantity_builder.append_value(quantity);
+ }
+
+ let batch = RecordBatch::try_new(
+ Arc::clone(&schema),
+ vec![
+ Arc::new(partkey_builder.finish()),
+ Arc::new(suppkey_builder.finish()),
+ Arc::new(quantity_builder.finish()),
+ ],
+ )
+ .unwrap();
+
+ batches.push(batch);
+ }
+
+ (schema, batches)
+}
+
+/// Genereate `num_batches` wide RecordBatches resembling sort-tpch Q10 for benchmarking.
+/// This includes multiple numeric, date, and Utf8View columns (15 total).
+pub fn create_wide_batches(
+ num_batches: usize,
+ num_rows: usize,
+) -> (Arc<Schema>, Vec<RecordBatch>) {
+ let seed = 10;
+ let mut rng = rand::rngs::StdRng::seed_from_u64(seed);
+ let mut batches = Vec::with_capacity(num_batches);
+
+ let schema = Arc::new(Schema::new(vec![
+ Field::new("l_linenumber", DataType::Int32, false),
+ Field::new("l_suppkey", DataType::Int64, false),
+ Field::new("l_orderkey", DataType::Int64, false),
+ Field::new("l_partkey", DataType::Int64, false),
+ Field::new("l_quantity", DataType::Decimal128(15, 2), false),
+ Field::new("l_extendedprice", DataType::Decimal128(15, 2), false),
+ Field::new("l_discount", DataType::Decimal128(15, 2), false),
+ Field::new("l_tax", DataType::Decimal128(15, 2), false),
+ Field::new("l_returnflag", DataType::Utf8, false),
+ Field::new("l_linestatus", DataType::Utf8, false),
+ Field::new("l_shipdate", DataType::Date32, false),
+ Field::new("l_commitdate", DataType::Date32, false),
+ Field::new("l_receiptdate", DataType::Date32, false),
+ Field::new("l_shipinstruct", DataType::Utf8, false),
+ Field::new("l_shipmode", DataType::Utf8, false),
+ ]));
+
+ for _ in 0..num_batches {
+ let mut linenum = Int32Builder::new();
+ let mut suppkey = Int64Builder::new();
+ let mut orderkey = Int64Builder::new();
+ let mut partkey = Int64Builder::new();
+ let mut quantity = Decimal128Builder::new()
+ .with_precision_and_scale(15, 2)
+ .unwrap();
+ let mut extprice = Decimal128Builder::new()
+ .with_precision_and_scale(15, 2)
+ .unwrap();
+ let mut discount = Decimal128Builder::new()
+ .with_precision_and_scale(15, 2)
+ .unwrap();
+ let mut tax = Decimal128Builder::new()
+ .with_precision_and_scale(15, 2)
+ .unwrap();
+ let mut retflag = StringBuilder::new();
+ let mut linestatus = StringBuilder::new();
+ let mut shipdate = Date32Builder::new();
+ let mut commitdate = Date32Builder::new();
+ let mut receiptdate = Date32Builder::new();
+ let mut shipinstruct = StringBuilder::new();
+ let mut shipmode = StringBuilder::new();
+
+ let return_flags = ["A", "N", "R"];
+ let statuses = ["F", "O"];
+ let instructs = ["DELIVER IN PERSON", "COLLECT COD", "NONE"];
+ let modes = ["TRUCK", "MAIL", "SHIP", "RAIL", "AIR"];
+
+ for i in 0..num_rows {
+ linenum.append_value((i % 7) as i32);
+ suppkey.append_value(rng.random_range(0..100_000));
+ orderkey.append_value(1_000_000 + i as i64);
+ partkey.append_value(rng.random_range(0..200_000));
+
+ quantity.append_value(rng.random_range(100..10000) as i128);
+ extprice.append_value(rng.random_range(1_000..1_000_000) as i128);
+ discount.append_value(rng.random_range(0..10000) as i128);
+ tax.append_value(rng.random_range(0..5000) as i128);
+
+ retflag.append_value(return_flags[rng.random_range(0..return_flags.len())]);
+ linestatus.append_value(statuses[rng.random_range(0..statuses.len())]);
+
+ let base_date = 10_000;
+ shipdate.append_value(base_date + (i % 1000) as i32);
+ commitdate.append_value(base_date + (i % 1000) as i32 + 1);
+ receiptdate.append_value(base_date + (i % 1000) as i32 + 2);
+
+ shipinstruct.append_value(instructs[rng.random_range(0..instructs.len())]);
+ shipmode.append_value(modes[rng.random_range(0..modes.len())]);
+ }
+
+ let batch = RecordBatch::try_new(
+ Arc::clone(&schema),
+ vec![
+ Arc::new(linenum.finish()),
+ Arc::new(suppkey.finish()),
+ Arc::new(orderkey.finish()),
+ Arc::new(partkey.finish()),
+ Arc::new(quantity.finish()),
+ Arc::new(extprice.finish()),
+ Arc::new(discount.finish()),
+ Arc::new(tax.finish()),
+ Arc::new(retflag.finish()),
+ Arc::new(linestatus.finish()),
+ Arc::new(shipdate.finish()),
+ Arc::new(commitdate.finish()),
+ Arc::new(receiptdate.finish()),
+ Arc::new(shipinstruct.finish()),
+ Arc::new(shipmode.finish()),
+ ],
+ )
+ .unwrap();
+ batches.push(batch);
+ }
+ (schema, batches)
+}
+
+// Benchmarks spill write + read performance across multiple compression codecs
+// using realistic input data inspired by TPC-H aggregate spill scenarios.
+//
+// This function prepares synthetic RecordBatches that mimic the schema and distribution
+// of intermediate aggregate results from representative TPC-H queries (Q2, Q16, Q20) and sort-tpch Q10.
+// The schemas of these batches are:
+// Q2 [Int64, Decimal128]
+// Q16 [Utf8, Utf8, Int32, Int64]
+// Q20 [Int64, Int64, Decimal128]
+// sort-tpch Q10 (wide batch) [Int32, Int64 * 3, Decimal128 * 4, Date * 3, Utf8 * 4]
+// For each dataset:
+// - It evaluates spill performance under different compression codecs (e.g., Uncompressed, Zstd, LZ4).
+// - It measures end-to-end spill write + read performance using Criterion.
+// - It prints the observed memory-to-disk compression ratio for each codec.
+//
+// This helps evaluate the tradeoffs between compression ratio and runtime overhead for various codecs.
+fn bench_spill_compression(c: &mut Criterion) {
+ let env = Arc::new(RuntimeEnv::default());
+ let mut group = c.benchmark_group("spill_compression");
+ let rt = Runtime::new().unwrap();
+ let compressions = vec![
+ SpillCompression::Uncompressed,
+ SpillCompression::Zstd,
+ SpillCompression::Lz4Frame,
+ ];
+
+ // Modify these values to change data volume. Note that each batch contains `num_rows` rows.
+ let num_batches = 50;
+ let num_rows = 8192;
+
+ // Q2 [Int64, Decimal128]
+ let (schema, batches) = create_q2_like_batches(num_batches, num_rows);
+ benchmark_spill_batches_for_all_codec(
+ &mut group,
+ "q2",
+ batches,
+ &compressions,
+ &rt,
+ env.clone(),
+ schema,
+ );
+ // Q16 [Utf8, Utf8, Int32, Int64]
+ let (schema, batches) = create_q16_like_batches(num_batches, num_rows);
+ benchmark_spill_batches_for_all_codec(
+ &mut group,
+ "q16",
+ batches,
+ &compressions,
+ &rt,
+ env.clone(),
+ schema,
+ );
+ // Q20 [Int64, Int64, Decimal128]
+ let (schema, batches) = create_q20_like_batches(num_batches, num_rows);
+ benchmark_spill_batches_for_all_codec(
+ &mut group,
+ "q20",
+ batches,
+ &compressions,
+ &rt,
+ env.clone(),
+ schema,
+ );
+ // sort-tpch Q10 (wide batch) [Int32, Int64 * 3, Decimal128 * 4, Date * 3, Utf8 * 4]
+ let (schema, batches) = create_wide_batches(num_batches, num_rows);
+ benchmark_spill_batches_for_all_codec(
+ &mut group,
+ "wide",
+ batches,
+ &compressions,
+ &rt,
+ env,
+ schema,
+ );
+ group.finish();
+}
+
+fn benchmark_spill_batches_for_all_codec(
+ group: &mut BenchmarkGroup<'_, WallTime>,
+ batch_label: &str,
+ batches: Vec<RecordBatch>,
+ compressions: &[SpillCompression],
+ rt: &Runtime,
+ env: Arc<RuntimeEnv>,
+ schema: Arc<Schema>,
+) {
+ let mem_bytes: usize = batches.iter().map(|b| b.get_array_memory_size()).sum();
+
+ for &compression in compressions {
+ let metrics = SpillMetrics::new(&ExecutionPlanMetricsSet::new(), 0);
+ let spill_manager =
+ SpillManager::new(Arc::clone(&env), metrics.clone(), Arc::clone(&schema))
+ .with_compression_type(compression);
+
+ let bench_id = BenchmarkId::new(batch_label, compression.to_string());
+ group.bench_with_input(bench_id, &spill_manager, |b, spill_manager| {
+ b.iter_batched(
+ || batches.clone(),
+ |batches| {
+ rt.block_on(async {
+ let spill_file = spill_manager
+ .spill_record_batch_and_finish(
+ &batches,
+ &format!("{batch_label}_{compression}"),
+ )
+ .unwrap()
+ .unwrap();
+ let stream =
+ spill_manager.read_spill_as_stream(spill_file).unwrap();
+ let _ = collect(stream).await.unwrap();
+ })
+ },
+ BatchSize::LargeInput,
+ )
+ });
+
+ // Run Spilling Read & Write once more to read file size & calculate bandwidth
+ let start = Instant::now();
+
+ let spill_file = spill_manager
+ .spill_record_batch_and_finish(
+ &batches,
+ &format!("{batch_label}_{compression}"),
+ )
+ .unwrap()
+ .unwrap();
+
+ // calculate write_throughput (includes both compression and I/O time) based on in memory batch size
+ let write_time = start.elapsed();
+ let write_throughput = (mem_bytes as u128 / write_time.as_millis().max(1)) * 1000;
+
+ // calculate compression ratio
+ let disk_bytes = std::fs::metadata(spill_file.path())
+ .expect("metadata read fail")
+ .len() as usize;
+ let ratio = mem_bytes as f64 / disk_bytes.max(1) as f64;
+
+ // calculate read_throughput (includes both compression and I/O time) based on in memory batch size
+ let rt = Runtime::new().unwrap();
+ let start = Instant::now();
+ rt.block_on(async {
+ let stream = spill_manager.read_spill_as_stream(spill_file).unwrap();
+ let _ = collect(stream).await.unwrap();
+ });
+ let read_time = start.elapsed();
+ let read_throughput = (mem_bytes as u128 / read_time.as_millis().max(1)) * 1000;
Review Comment:
I added separate measurements for write and read throughput after the criterion benchmark, by running the spilling logic once more (just like how compression ratio is calculated). This avoids the risk of Instant::now() adding latency inside the closure. Also, since the timing includes both I/O and (de)compression time, I used the term `throughput` instead of `bandwidth`.
--
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-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add microbenchmark for spilling with compression [datafusion]
Posted by "ding-young (via GitHub)" <gi...@apache.org>.
ding-young commented on PR #16512:
URL: https://github.com/apache/datafusion/pull/16512#issuecomment-3018735400
On my machine, avg bandwidth (throughput) is
TBD...
And, when I ran `strace -c -e trace=write,read cargo bench --bench spill_io compression` only for plain encoding,
it seems like there are too many `read` system calls compared to `write`.
--
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-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add microbenchmark for spilling with compression [datafusion]
Posted by "ding-young (via GitHub)" <gi...@apache.org>.
ding-young commented on PR #16512:
URL: https://github.com/apache/datafusion/pull/16512#issuecomment-3000078588
As expected, although `lz4_frame` has a lower compression ratio than `zstd`, it runs faster, making it a reasonable tradeoff. However, since it's roughly 2x slower than the uncompressed case, we may need to explore optimization strategies further.
If there are other interesting directions for benchmarking such as measuring write and read times separately, or isolating compression and decompression overhead, or testing on different data types, I’d love to hear suggestions.
--
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-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add microbenchmark for spilling with compression [datafusion]
Posted by "ding-young (via GitHub)" <gi...@apache.org>.
ding-young commented on PR #16512:
URL: https://github.com/apache/datafusion/pull/16512#issuecomment-3018158695
@2010YOUY01 Thank you for detailed review! The bandwidth result is quite interesting..
Btw, I'd like to measure the bandwidth, but it looks like Criterion doesn’t expose the measured time as a variable. I was wondering if you used something like `Instant::now()` inside the `iter_batched` closure to manually record the elapsed time.
--
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-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add microbenchmark for spilling with compression [datafusion]
Posted by "2010YOUY01 (via GitHub)" <gi...@apache.org>.
2010YOUY01 commented on PR #16512:
URL: https://github.com/apache/datafusion/pull/16512#issuecomment-3018229716
> @2010YOUY01 Thank you for detailed review! The bandwidth result is quite interesting.. Btw, I'd like to measure the bandwidth, but it looks like Criterion doesn’t expose the measured time as a variable. I was wondering if you used something like `Instant::now()` inside the `iter_batched` closure to manually record the elapsed time.
I calculated it manually, I think this `Instant::now()` approach is reasonable.
--
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-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add microbenchmark for spilling with compression [datafusion]
Posted by "2010YOUY01 (via GitHub)" <gi...@apache.org>.
2010YOUY01 commented on PR #16512:
URL: https://github.com/apache/datafusion/pull/16512#issuecomment-3025966862
> On my machine, avg bandwidth (throughput) for Q2 is
>
> (uncompressed) plain: (w) 938.4 MB/s (r) 1042.7 MB/s zstd: (w) 234.6 MB/s (r) 276.0 MB/s lz4_frame: (w) 521.3 MB/s (r) 446.9 MB/s
>
> Considering my SSD bandwidth is around 3.2GB/s, as you pointed out, it seems that there's room for optimization.
>
> Plus, when I ran `strace -c -e trace=write,read cargo bench --bench spill_io compression` only for plain encoding, it seems like there are too many `read` system calls compared to `write`. We may need to investigate this more or find ways to reduce the number of `read` calls. 
Also, this read and write throughput ratio for lz4 and zstd also not look normal 🤔
Based on the compression benchmarks, the decompression speed should be several times faster than compression:
https://github.com/lz4/lz4
https://github.com/facebook/zstd
--
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-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org