andygrove commented on code in PR #5822: URL: https://github.com/apache/datafusion-comet/pull/5822#discussion_r3973830990
########## native/spark-expr/benches/common/matched_maps.rs: ########## @@ -0,0 +1,216 @@ +// 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. + +//! Matched map inputs from CometShuffleBenchmark (PR #5788). + +use arrow::array::{ + Array, ArrayRef, Int32Array, Int32Builder, MapArray, MapBuilder, MapFieldNames, StringBuilder, + StructArray, +}; +use arrow::datatypes::Field; +use criterion::{BenchmarkId, Criterion, Throughput}; +use datafusion::physical_plan::ColumnarValue; +use datafusion_comet_spark_expr::{murmur3::create_murmur3_hashes, spark_map_sort}; +use std::{hint::black_box, sync::Arc}; + +pub const ROWS: usize = 8192; + +// Scala Long arithmetic wraps; pmod is applied to the signed result of mix64. +fn c1(row: usize) -> i32 { + let mut z = (row as u64).wrapping_add(0x9e3779b97f4a7c15); + z = (z ^ (z >> 30)).wrapping_mul(0xbf58476d1ce4e5b9); + z = (z ^ (z >> 27)).wrapping_mul(0x94d049bb133111eb); + ((z ^ (z >> 31)) as i64).rem_euclid(1_000_000) as i32 +} + +fn maps(max_entries: usize, reversed: bool) -> ArrayRef { + let mut builder = MapBuilder::new( + Some(MapFieldNames { + entry: "entries".into(), + key: "key".into(), + value: "value".into(), + }), + StringBuilder::new(), + Int32Builder::new(), + ); + for row in 0..ROWS { + let value = c1(row); + let count = if max_entries == 1 { + 1 + } else { + 2 + value as usize % (max_entries - 1) + }; + for idx in 0..count { + // Singleton is MAP(CAST(c1 AS STRING), c1); larger maps use sequence(1, count). + let x = if max_entries == 1 { + 0 + } else if reversed { + count - idx + } else { + idx + 1 + }; + let entry = value + x as i32; + builder.keys().append_value(entry.to_string()); + builder.values().append_value(entry); + } + builder.append(true).unwrap(); + } + Arc::new(builder.finish()) +} + +fn normalize(args: &[ColumnarValue]) -> ArrayRef { + match spark_map_sort(args).unwrap() { + ColumnarValue::Array(array) => array, + _ => panic!("expected array"), + } +} + +fn wrap(map: ArrayRef, ints: &ArrayRef) -> ArrayRef { + Arc::new(StructArray::new( + vec![ + Arc::new(Field::new("m", map.data_type().clone(), true)), + Arc::new(Field::new("i", ints.data_type().clone(), true)), + ] + .into(), + vec![map, Arc::clone(ints)], + None, + )) +} + +fn hashes(array: &ArrayRef) -> Vec<u32> { + let mut hashes = vec![42; ROWS]; + create_murmur3_hashes(std::slice::from_ref(array), &mut hashes).unwrap(); + hashes +} + +/// Shared registration keeps every stage's data and correctness checks identical. +/// `stage` is one of hash_only, normalize_only, normalize_hash. +/// Input construction and checks are untimed. Hash storage is allocated once; resetting it +/// to seed 42 is timed. Normalization output allocation/drop, and struct reconstruction in +/// normalize_hash, are timed. normalize_only measures mapsort itself (no struct wrapper). +pub fn bench_maps(c: &mut Criterion, stage: &str) { + assert!(["hash_only", "normalize_only", "normalize_hash"].contains(&stage)); + let ints: ArrayRef = Arc::new(Int32Array::from_iter_values((0..ROWS).map(c1))); + let mut group = c.benchmark_group(format!("matched_maps/{stage}")); + group.throughput(Throughput::Elements(ROWS as u64)); + for max_entries in [1, 10, 50] { + let forward = maps(max_entries, false); + let reversed = maps(max_entries, true); + let normalized = normalize(&[ColumnarValue::Array(Arc::clone(&forward))]); + let normalized_reverse = normalize(&[ColumnarValue::Array(Arc::clone(&reversed))]); + assert_eq!(normalized.to_data(), normalized_reverse.to_data()); + // Independently check lexical ordering and key/value alignment; numeric order does + // not in general imply string order for unpadded decimal keys. + let expected = normalized.as_any().downcast_ref::<MapArray>().unwrap(); + let keys = expected + .keys() + .as_any() + .downcast_ref::<arrow::array::StringArray>() + .unwrap(); + let values = expected + .values() + .as_any() + .downcast_ref::<Int32Array>() + .unwrap(); + assert_eq!(expected.len(), ROWS); + for (row, offsets) in expected.value_offsets().windows(2).enumerate() { + let base = c1(row); + let count = if max_entries == 1 { + 1 + } else { + 2 + base as usize % (max_entries - 1) + }; + assert_eq!((offsets[1] - offsets[0]) as usize, count); + let mut actual_values: Vec<_> = (offsets[0] as usize..offsets[1] as usize) + .map(|i| values.value(i)) + .collect(); + actual_values.sort_unstable(); + let start = if max_entries == 1 { base } else { base + 1 }; + assert_eq!( + actual_values, + (start..start + count as i32).collect::<Vec<_>>() + ); + for i in offsets[0] as usize..offsets[1] as usize { + assert_eq!(keys.value(i), values.value(i).to_string()); + if i > offsets[0] as usize { + assert!(keys.value(i - 1) < keys.value(i)); + } + } + } + assert_eq!(hashes(&normalized), hashes(&normalized_reverse)); + assert_eq!( + hashes(&wrap(Arc::clone(&normalized), &ints)), + hashes(&wrap(normalized_reverse, &ints)) + ); Review Comment: Line 115 already asserts the two normalized arrays have equal `ArrayData`, which makes these two unable to fail: equal data hashes equally. The struct-versus-fields check just below is doing real work, but these are dead weight. There is a check that would fail and that #5818 explicitly asked for, though: "Raw map hashes need not be order-independent without normalization." Hashing the raw `forward` and `reversed` maps and asserting they differ demonstrates that normalization is load-bearing for the partition key rather than assuming it. I tried it on your branch and it discriminates cleanly, 8192/8192 rows differ at both 2-10 and 2-50, and 0/8192 for the singleton control, which is the right answer there since a one-entry map has no order to reverse. ########## native/spark-expr/benches/common/matched_maps.rs: ########## @@ -0,0 +1,216 @@ +// 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. + +//! Matched map inputs from CometShuffleBenchmark (PR #5788). + +use arrow::array::{ + Array, ArrayRef, Int32Array, Int32Builder, MapArray, MapBuilder, MapFieldNames, StringBuilder, + StructArray, +}; +use arrow::datatypes::Field; +use criterion::{BenchmarkId, Criterion, Throughput}; +use datafusion::physical_plan::ColumnarValue; +use datafusion_comet_spark_expr::{murmur3::create_murmur3_hashes, spark_map_sort}; +use std::{hint::black_box, sync::Arc}; + +pub const ROWS: usize = 8192; + +// Scala Long arithmetic wraps; pmod is applied to the signed result of mix64. +fn c1(row: usize) -> i32 { + let mut z = (row as u64).wrapping_add(0x9e3779b97f4a7c15); + z = (z ^ (z >> 30)).wrapping_mul(0xbf58476d1ce4e5b9); + z = (z ^ (z >> 27)).wrapping_mul(0x94d049bb133111eb); + ((z ^ (z >> 31)) as i64).rem_euclid(1_000_000) as i32 +} + +fn maps(max_entries: usize, reversed: bool) -> ArrayRef { + let mut builder = MapBuilder::new( + Some(MapFieldNames { + entry: "entries".into(), + key: "key".into(), + value: "value".into(), + }), + StringBuilder::new(), + Int32Builder::new(), + ); + for row in 0..ROWS { + let value = c1(row); + let count = if max_entries == 1 { + 1 + } else { + 2 + value as usize % (max_entries - 1) + }; + for idx in 0..count { + // Singleton is MAP(CAST(c1 AS STRING), c1); larger maps use sequence(1, count). + let x = if max_entries == 1 { + 0 + } else if reversed { + count - idx + } else { + idx + 1 + }; + let entry = value + x as i32; + builder.keys().append_value(entry.to_string()); + builder.values().append_value(entry); + } + builder.append(true).unwrap(); + } + Arc::new(builder.finish()) +} + +fn normalize(args: &[ColumnarValue]) -> ArrayRef { + match spark_map_sort(args).unwrap() { + ColumnarValue::Array(array) => array, + _ => panic!("expected array"), + } +} + +fn wrap(map: ArrayRef, ints: &ArrayRef) -> ArrayRef { + Arc::new(StructArray::new( + vec![ + Arc::new(Field::new("m", map.data_type().clone(), true)), + Arc::new(Field::new("i", ints.data_type().clone(), true)), + ] + .into(), + vec![map, Arc::clone(ints)], + None, + )) +} + +fn hashes(array: &ArrayRef) -> Vec<u32> { + let mut hashes = vec![42; ROWS]; + create_murmur3_hashes(std::slice::from_ref(array), &mut hashes).unwrap(); + hashes +} + +/// Shared registration keeps every stage's data and correctness checks identical. +/// `stage` is one of hash_only, normalize_only, normalize_hash. +/// Input construction and checks are untimed. Hash storage is allocated once; resetting it +/// to seed 42 is timed. Normalization output allocation/drop, and struct reconstruction in +/// normalize_hash, are timed. normalize_only measures mapsort itself (no struct wrapper). +pub fn bench_maps(c: &mut Criterion, stage: &str) { + assert!(["hash_only", "normalize_only", "normalize_hash"].contains(&stage)); + let ints: ArrayRef = Arc::new(Int32Array::from_iter_values((0..ROWS).map(c1))); + let mut group = c.benchmark_group(format!("matched_maps/{stage}")); + group.throughput(Throughput::Elements(ROWS as u64)); + for max_entries in [1, 10, 50] { + let forward = maps(max_entries, false); + let reversed = maps(max_entries, true); + let normalized = normalize(&[ColumnarValue::Array(Arc::clone(&forward))]); + let normalized_reverse = normalize(&[ColumnarValue::Array(Arc::clone(&reversed))]); + assert_eq!(normalized.to_data(), normalized_reverse.to_data()); + // Independently check lexical ordering and key/value alignment; numeric order does + // not in general imply string order for unpadded decimal keys. + let expected = normalized.as_any().downcast_ref::<MapArray>().unwrap(); + let keys = expected + .keys() + .as_any() + .downcast_ref::<arrow::array::StringArray>() + .unwrap(); + let values = expected + .values() + .as_any() + .downcast_ref::<Int32Array>() + .unwrap(); + assert_eq!(expected.len(), ROWS); + for (row, offsets) in expected.value_offsets().windows(2).enumerate() { + let base = c1(row); + let count = if max_entries == 1 { + 1 + } else { + 2 + base as usize % (max_entries - 1) + }; + assert_eq!((offsets[1] - offsets[0]) as usize, count); + let mut actual_values: Vec<_> = (offsets[0] as usize..offsets[1] as usize) + .map(|i| values.value(i)) + .collect(); + actual_values.sort_unstable(); + let start = if max_entries == 1 { base } else { base + 1 }; + assert_eq!( + actual_values, + (start..start + count as i32).collect::<Vec<_>>() + ); + for i in offsets[0] as usize..offsets[1] as usize { + assert_eq!(keys.value(i), values.value(i).to_string()); + if i > offsets[0] as usize { + assert!(keys.value(i - 1) < keys.value(i)); + } + } + } + assert_eq!(hashes(&normalized), hashes(&normalized_reverse)); + assert_eq!( + hashes(&wrap(Arc::clone(&normalized), &ints)), + hashes(&wrap(normalized_reverse, &ints)) + ); + let mut field_hashes = vec![42; ROWS]; + create_murmur3_hashes( + &[Arc::clone(&normalized), Arc::clone(&ints)], + &mut field_hashes, + ) + .unwrap(); + assert_eq!(field_hashes, hashes(&wrap(Arc::clone(&normalized), &ints))); + eprintln!( + "validated {stage}: max_entries={max_entries}, rows={ROWS}, entries={}", + expected.entries().len() + ); + for (order, raw) in [("forward", forward), ("reversed", reversed)] { + let args = [ColumnarValue::Array(raw)]; + for shape in ["map", "struct_map_int"] { + // mapsort is the same operation for either enclosing shape; measure once. + if stage == "normalize_only" && shape != "map" { + continue; + } + let input = if shape == "map" { + Arc::clone(&normalized) + } else { + wrap(Arc::clone(&normalized), &ints) + }; Review Comment: In `hash_only`, `input` comes from `normalized` on both passes of the `order` loop, so `hash_only/map/50/forward` and `hash_only/map/50/reversed` hash byte-identical arrays. Your description says this, and I think using it as a repeated-measurement control is a good call. The two struct/50 numbers landing 1.2% apart is a useful noise floor for the rest of the table. The problem is that the code does not say it. There is a comment four lines up for the analogous `normalize_only` shape skip, but nothing here, and the Criterion report is what outlives the PR body. Someone reading `hash_only/.../reversed` later will take order-insensitivity as a measured result rather than something true by construction. Could you either add a comment here, or rename the two labels to something that does not imply a comparison, `repeat_a` and `repeat_b` for instance? ########## native/spark-expr/benches/common/matched_maps.rs: ########## @@ -0,0 +1,216 @@ +// 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. + +//! Matched map inputs from CometShuffleBenchmark (PR #5788). + +use arrow::array::{ + Array, ArrayRef, Int32Array, Int32Builder, MapArray, MapBuilder, MapFieldNames, StringBuilder, + StructArray, +}; +use arrow::datatypes::Field; +use criterion::{BenchmarkId, Criterion, Throughput}; +use datafusion::physical_plan::ColumnarValue; +use datafusion_comet_spark_expr::{murmur3::create_murmur3_hashes, spark_map_sort}; +use std::{hint::black_box, sync::Arc}; + +pub const ROWS: usize = 8192; + +// Scala Long arithmetic wraps; pmod is applied to the signed result of mix64. +fn c1(row: usize) -> i32 { + let mut z = (row as u64).wrapping_add(0x9e3779b97f4a7c15); + z = (z ^ (z >> 30)).wrapping_mul(0xbf58476d1ce4e5b9); + z = (z ^ (z >> 27)).wrapping_mul(0x94d049bb133111eb); + ((z ^ (z >> 31)) as i64).rem_euclid(1_000_000) as i32 +} + +fn maps(max_entries: usize, reversed: bool) -> ArrayRef { + let mut builder = MapBuilder::new( + Some(MapFieldNames { + entry: "entries".into(), + key: "key".into(), + value: "value".into(), + }), + StringBuilder::new(), + Int32Builder::new(), + ); + for row in 0..ROWS { + let value = c1(row); + let count = if max_entries == 1 { + 1 + } else { + 2 + value as usize % (max_entries - 1) + }; + for idx in 0..count { + // Singleton is MAP(CAST(c1 AS STRING), c1); larger maps use sequence(1, count). + let x = if max_entries == 1 { + 0 + } else if reversed { + count - idx + } else { + idx + 1 + }; + let entry = value + x as i32; + builder.keys().append_value(entry.to_string()); + builder.values().append_value(entry); + } + builder.append(true).unwrap(); + } + Arc::new(builder.finish()) +} + +fn normalize(args: &[ColumnarValue]) -> ArrayRef { + match spark_map_sort(args).unwrap() { + ColumnarValue::Array(array) => array, + _ => panic!("expected array"), + } +} + +fn wrap(map: ArrayRef, ints: &ArrayRef) -> ArrayRef { + Arc::new(StructArray::new( + vec![ + Arc::new(Field::new("m", map.data_type().clone(), true)), + Arc::new(Field::new("i", ints.data_type().clone(), true)), + ] + .into(), + vec![map, Arc::clone(ints)], + None, + )) +} + +fn hashes(array: &ArrayRef) -> Vec<u32> { + let mut hashes = vec![42; ROWS]; + create_murmur3_hashes(std::slice::from_ref(array), &mut hashes).unwrap(); + hashes +} + +/// Shared registration keeps every stage's data and correctness checks identical. +/// `stage` is one of hash_only, normalize_only, normalize_hash. +/// Input construction and checks are untimed. Hash storage is allocated once; resetting it +/// to seed 42 is timed. Normalization output allocation/drop, and struct reconstruction in +/// normalize_hash, are timed. normalize_only measures mapsort itself (no struct wrapper). +pub fn bench_maps(c: &mut Criterion, stage: &str) { + assert!(["hash_only", "normalize_only", "normalize_hash"].contains(&stage)); Review Comment: `stage` is a `&str` validated by a runtime `assert!` and then compared three times, twice inside `b.iter`. All three call sites pass literals. A small enum would move the check to compile time, make the `normalize_only`-skips-struct rule exhaustive rather than a string compare, and keep the comparisons out of the timed closure. Same idea for `shape`, though that one is smaller. ########## native/spark-expr/benches/common/matched_maps.rs: ########## @@ -0,0 +1,216 @@ +// 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. + +//! Matched map inputs from CometShuffleBenchmark (PR #5788). + +use arrow::array::{ + Array, ArrayRef, Int32Array, Int32Builder, MapArray, MapBuilder, MapFieldNames, StringBuilder, + StructArray, +}; +use arrow::datatypes::Field; +use criterion::{BenchmarkId, Criterion, Throughput}; +use datafusion::physical_plan::ColumnarValue; +use datafusion_comet_spark_expr::{murmur3::create_murmur3_hashes, spark_map_sort}; +use std::{hint::black_box, sync::Arc}; + +pub const ROWS: usize = 8192; + +// Scala Long arithmetic wraps; pmod is applied to the signed result of mix64. +fn c1(row: usize) -> i32 { + let mut z = (row as u64).wrapping_add(0x9e3779b97f4a7c15); + z = (z ^ (z >> 30)).wrapping_mul(0xbf58476d1ce4e5b9); + z = (z ^ (z >> 27)).wrapping_mul(0x94d049bb133111eb); + ((z ^ (z >> 31)) as i64).rem_euclid(1_000_000) as i32 +} + +fn maps(max_entries: usize, reversed: bool) -> ArrayRef { + let mut builder = MapBuilder::new( + Some(MapFieldNames { + entry: "entries".into(), + key: "key".into(), + value: "value".into(), + }), Review Comment: `MapFieldNames { entry: "entries", key: "key", value: "value" }` is now spelled out five times under `benches/`: twice inline in `hash.rs`, once as `map_field_names()` in `map_sort.rs`, and now again here, in a file that gets compiled into `map_sort.rs` right beside that helper. Could this move to `common/mod.rs` and the other four call it? Two smaller things while you are in there. The module is `mod matched_maps_data` but the file is `matched_maps.rs`, which made it harder than it needed to be to find. And `common/mod.rs` still documents itself as "helpers shared by the cast-from-string benchmarks", pulled in via `#[path = "common/mod.rs"]`. Worth a line noting that the directory now also holds standalone modules included directly, so the next person does not have to work out which convention applies. ########## native/spark-expr/benches/common/matched_maps.rs: ########## @@ -0,0 +1,216 @@ +// 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. + +//! Matched map inputs from CometShuffleBenchmark (PR #5788). + +use arrow::array::{ + Array, ArrayRef, Int32Array, Int32Builder, MapArray, MapBuilder, MapFieldNames, StringBuilder, + StructArray, +}; +use arrow::datatypes::Field; +use criterion::{BenchmarkId, Criterion, Throughput}; +use datafusion::physical_plan::ColumnarValue; +use datafusion_comet_spark_expr::{murmur3::create_murmur3_hashes, spark_map_sort}; +use std::{hint::black_box, sync::Arc}; + +pub const ROWS: usize = 8192; + +// Scala Long arithmetic wraps; pmod is applied to the signed result of mix64. +fn c1(row: usize) -> i32 { + let mut z = (row as u64).wrapping_add(0x9e3779b97f4a7c15); + z = (z ^ (z >> 30)).wrapping_mul(0xbf58476d1ce4e5b9); + z = (z ^ (z >> 27)).wrapping_mul(0x94d049bb133111eb); + ((z ^ (z >> 31)) as i64).rem_euclid(1_000_000) as i32 +} + +fn maps(max_entries: usize, reversed: bool) -> ArrayRef { + let mut builder = MapBuilder::new( + Some(MapFieldNames { + entry: "entries".into(), + key: "key".into(), + value: "value".into(), + }), + StringBuilder::new(), + Int32Builder::new(), + ); + for row in 0..ROWS { + let value = c1(row); + let count = if max_entries == 1 { + 1 + } else { + 2 + value as usize % (max_entries - 1) + }; + for idx in 0..count { + // Singleton is MAP(CAST(c1 AS STRING), c1); larger maps use sequence(1, count). + let x = if max_entries == 1 { + 0 + } else if reversed { + count - idx + } else { + idx + 1 + }; + let entry = value + x as i32; + builder.keys().append_value(entry.to_string()); + builder.values().append_value(entry); + } + builder.append(true).unwrap(); + } + Arc::new(builder.finish()) +} + +fn normalize(args: &[ColumnarValue]) -> ArrayRef { + match spark_map_sort(args).unwrap() { + ColumnarValue::Array(array) => array, + _ => panic!("expected array"), + } +} + +fn wrap(map: ArrayRef, ints: &ArrayRef) -> ArrayRef { + Arc::new(StructArray::new( + vec![ + Arc::new(Field::new("m", map.data_type().clone(), true)), + Arc::new(Field::new("i", ints.data_type().clone(), true)), + ] + .into(), + vec![map, Arc::clone(ints)], + None, + )) +} + +fn hashes(array: &ArrayRef) -> Vec<u32> { + let mut hashes = vec![42; ROWS]; + create_murmur3_hashes(std::slice::from_ref(array), &mut hashes).unwrap(); + hashes +} + +/// Shared registration keeps every stage's data and correctness checks identical. +/// `stage` is one of hash_only, normalize_only, normalize_hash. +/// Input construction and checks are untimed. Hash storage is allocated once; resetting it +/// to seed 42 is timed. Normalization output allocation/drop, and struct reconstruction in +/// normalize_hash, are timed. normalize_only measures mapsort itself (no struct wrapper). +pub fn bench_maps(c: &mut Criterion, stage: &str) { + assert!(["hash_only", "normalize_only", "normalize_hash"].contains(&stage)); + let ints: ArrayRef = Arc::new(Int32Array::from_iter_values((0..ROWS).map(c1))); + let mut group = c.benchmark_group(format!("matched_maps/{stage}")); + group.throughput(Throughput::Elements(ROWS as u64)); + for max_entries in [1, 10, 50] { + let forward = maps(max_entries, false); + let reversed = maps(max_entries, true); + let normalized = normalize(&[ColumnarValue::Array(Arc::clone(&forward))]); + let normalized_reverse = normalize(&[ColumnarValue::Array(Arc::clone(&reversed))]); + assert_eq!(normalized.to_data(), normalized_reverse.to_data()); + // Independently check lexical ordering and key/value alignment; numeric order does + // not in general imply string order for unpadded decimal keys. + let expected = normalized.as_any().downcast_ref::<MapArray>().unwrap(); + let keys = expected + .keys() + .as_any() + .downcast_ref::<arrow::array::StringArray>() + .unwrap(); + let values = expected + .values() + .as_any() + .downcast_ref::<Int32Array>() + .unwrap(); + assert_eq!(expected.len(), ROWS); + for (row, offsets) in expected.value_offsets().windows(2).enumerate() { + let base = c1(row); + let count = if max_entries == 1 { + 1 + } else { + 2 + base as usize % (max_entries - 1) + }; + assert_eq!((offsets[1] - offsets[0]) as usize, count); + let mut actual_values: Vec<_> = (offsets[0] as usize..offsets[1] as usize) + .map(|i| values.value(i)) + .collect(); + actual_values.sort_unstable(); + let start = if max_entries == 1 { base } else { base + 1 }; + assert_eq!( + actual_values, + (start..start + count as i32).collect::<Vec<_>>() + ); + for i in offsets[0] as usize..offsets[1] as usize { + assert_eq!(keys.value(i), values.value(i).to_string()); + if i > offsets[0] as usize { + assert!(keys.value(i - 1) < keys.value(i)); + } + } + } + assert_eq!(hashes(&normalized), hashes(&normalized_reverse)); + assert_eq!( + hashes(&wrap(Arc::clone(&normalized), &ints)), + hashes(&wrap(normalized_reverse, &ints)) + ); + let mut field_hashes = vec![42; ROWS]; + create_murmur3_hashes( + &[Arc::clone(&normalized), Arc::clone(&ints)], + &mut field_hashes, + ) + .unwrap(); + assert_eq!(field_hashes, hashes(&wrap(Arc::clone(&normalized), &ints))); + eprintln!( + "validated {stage}: max_entries={max_entries}, rows={ROWS}, entries={}", + expected.entries().len() + ); Review Comment: This runs at registration time, so it prints on `--list`, on `--test`, and on any filtered `cargo bench --bench hash` run, including one that only wants the pre-existing hash cases. The setup itself is cheap, I measured `--list` on the hash binary at 0.14 s, so this is only about the stderr noise. Moving it inside the `bench_function` closure or dropping it would keep unrelated runs quiet. Related, a few lines down: `input` and `buffer` are built for every `normalize_only` case and never read, since that stage takes the other branch of the closure. Hoisting them into the `else` would make the three stages easier to follow. -- 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]
