ARROW-11362:[Rust][DataFusion] Use iterator APIs in to_array_of_size to improve performance

This function `to_array_of_size` is about 8.3% of total instructions in the db-benchmark (aggregation) queries.

This uses the PR #9293

The case of converting an int32 to an array improved by ~5x according to the microbenchmark:

```
to_array_of_size 100000 time:   [55.501 us 55.627 us 55.809 us]
                        change: [-82.457% -82.384% -82.299%] (p = 0.00 < 0.05)
                        Performance has improved.
Found 3 outliers among 100 measurements (3.00%)
  2 (2.00%) high mild
  1 (1.00%) high severe
```

And on TCPH query 1 (SF=1, 16 partitions).

PR:

```
Query 1 iteration 0 took 90.8 ms
Query 1 iteration 1 took 106.6 ms
Query 1 iteration 2 took 101.1 ms
Query 1 iteration 3 took 101.5 ms
Query 1 iteration 4 took 96.9 ms
Query 1 iteration 5 took 100.3 ms
Query 1 iteration 6 took 99.6 ms
Query 1 iteration 7 took 100.4 ms
Query 1 iteration 8 took 104.2 ms
Query 1 iteration 9 took 100.3 ms
Query 1 avg time: 100.18 ms
```
Master:
```
Query 1 iteration 0 took 121.1 ms
Query 1 iteration 1 took 123.4 ms
Query 1 iteration 2 took 121.0 ms
Query 1 iteration 3 took 121.0 ms
Query 1 iteration 4 took 123.0 ms
Query 1 iteration 5 took 121.7 ms
Query 1 iteration 6 took 121.7 ms
Query 1 iteration 7 took 120.2 ms
Query 1 iteration 8 took 119.7 ms
Query 1 iteration 9 took 121.4 ms
Query 1 avg time: 121.43 ms
```

Closes #9305 from Dandandan/to_array_of_size_perf

Lead-authored-by: Heres, Daniel <danielheres@gmail.com>
Co-authored-by: Jorge C. Leitao <jorgecarleitao@gmail.com>
Co-authored-by: Andrew Lamb <andrew@nerdnetworks.org>
Co-authored-by: Sutou Kouhei <kou@clear-code.com>
Co-authored-by: Neville Dipale <nevilledips@gmail.com>
Co-authored-by: Dmitry Patsura <zaets28rus@gmail.com>
Co-authored-by: Yibo Cai <yibo.cai@arm.com>
Co-authored-by: Daniël Heres <danielheres@gmail.com>
Co-authored-by: Kenta Murata <mrkn@mrkn.jp>
Co-authored-by: Mahmut Bulut <vertexclique@gmail.com>
Co-authored-by: Yordan Pavlov <yordan.pavlov@outlook.com>
Co-authored-by: Max Burke <max@urbanlogiq.com>
Co-authored-by: Ryan Jennings <ryan@ryanj.net>
Co-authored-by: Benjamin Kietzman <bengilgit@gmail.com>
Co-authored-by: Jörn Horstmann <joern.horstmann@signavio.com>
Co-authored-by: Krisztián Szűcs <szucs.krisztian@gmail.com>
Co-authored-by: Joris Van den Bossche <jorisvandenbossche@gmail.com>
Co-authored-by: Johannes Müller <JohannesMueller@fico.com>
Co-authored-by: mqy <meng.qingyou@gmail.com>
Co-authored-by: Maarten A. Breddels <maartenbreddels@gmail.com>
Co-authored-by: Antoine Pitrou <antoine@python.org>
Co-authored-by: Matt Brubeck <mbrubeck@limpet.net>
Co-authored-by: Weston Pace <weston.pace@gmail.com>
Signed-off-by: Jorge C. Leitao <jorgecarleitao@gmail.com>

Author: 22 people

rust/datafusion/Cargo.toml

@@ -88,3 +88,7 @@ harness = false
 [[bench]]
 name = "filter_query_sql"
 harness = false
+
+[[bench]]
+name = "scalar"
+harness = false

rust/datafusion/benches/scalar.rs

@@ -0,0 +1,30 @@
+// 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.
+
+use criterion::{criterion_group, criterion_main, Criterion};
+use datafusion::scalar::ScalarValue;
+
+fn criterion_benchmark(c: &mut Criterion) {
+    c.bench_function("to_array_of_size 100000", |b| {
+        let scalar = ScalarValue::Int32(Some(100));
+
+        b.iter(|| assert_eq!(scalar.to_array_of_size(100000).null_count(), 0))
+    });
+}
+
+criterion_group!(benches, criterion_benchmark);
+criterion_main!(benches);

rust/datafusion/src/scalar.rs

@@ -17,7 +17,7 @@
 
 //! This module provides ScalarValue, an enum that can be used for storage of single elements
 
-use std::{convert::TryFrom, fmt, sync::Arc};
+use std::{convert::TryFrom, fmt, iter::repeat, sync::Arc};
 
 use arrow::array::{
     Int16Builder, Int32Builder, Int64Builder, Int8Builder, ListBuilder,
@@ -205,28 +205,104 @@ impl ScalarValue {
             ScalarValue::Boolean(e) => {
                 Arc::new(BooleanArray::from(vec![*e; size])) as ArrayRef
             }
-            ScalarValue::Float64(e) => {
-                Arc::new(Float64Array::from(vec![*e; size])) as ArrayRef
-            }
-            ScalarValue::Float32(e) => Arc::new(Float32Array::from(vec![*e; size])),
-            ScalarValue::Int8(e) => Arc::new(Int8Array::from(vec![*e; size])),
-            ScalarValue::Int16(e) => Arc::new(Int16Array::from(vec![*e; size])),
-            ScalarValue::Int32(e) => Arc::new(Int32Array::from(vec![*e; size])),
-            ScalarValue::Int64(e) => Arc::new(Int64Array::from(vec![*e; size])),
-            ScalarValue::UInt8(e) => Arc::new(UInt8Array::from(vec![*e; size])),
-            ScalarValue::UInt16(e) => Arc::new(UInt16Array::from(vec![*e; size])),
-            ScalarValue::UInt32(e) => Arc::new(UInt32Array::from(vec![*e; size])),
-            ScalarValue::UInt64(e) => Arc::new(UInt64Array::from(vec![*e; size])),
-            ScalarValue::TimeMicrosecond(e) => {
-                Arc::new(TimestampMicrosecondArray::from(vec![*e]))
-            }
-            ScalarValue::TimeNanosecond(e) => {
-                Arc::new(TimestampNanosecondArray::from_opt_vec(vec![*e], None))
-            }
-            ScalarValue::Utf8(e) => Arc::new(StringArray::from(vec![e.as_deref(); size])),
-            ScalarValue::LargeUtf8(e) => {
-                Arc::new(LargeStringArray::from(vec![e.as_deref(); size]))
-            }
+            ScalarValue::Float64(e) => match e {
+                Some(value) => {
+                    Arc::new(Float64Array::from_iter_values(repeat(*value).take(size)))
+                }
+                None => Arc::new(repeat(None).take(size).collect::<Float64Array>()),
+            },
+            ScalarValue::Float32(e) => match e {
+                Some(value) => {
+                    Arc::new(Float32Array::from_iter_values(repeat(*value).take(size)))
+                }
+                None => Arc::new(repeat(None).take(size).collect::<Float32Array>()),
+            },
+            ScalarValue::Int8(e) => match e {
+                Some(value) => {
+                    Arc::new(Int8Array::from_iter_values(repeat(*value).take(size)))
+                }
+                None => Arc::new(repeat(None).take(size).collect::<Int8Array>()),
+            },
+            ScalarValue::Int16(e) => match e {
+                Some(value) => {
+                    Arc::new(Int16Array::from_iter_values(repeat(*value).take(size)))
+                }
+                None => Arc::new(repeat(None).take(size).collect::<Int16Array>()),
+            },
+            ScalarValue::Int32(e) => match e {
+                Some(value) => {
+                    Arc::new(Int32Array::from_iter_values(repeat(*value).take(size)))
+                }
+                None => Arc::new(repeat(None).take(size).collect::<Int32Array>()),
+            },
+            ScalarValue::Int64(e) => match e {
+                Some(value) => {
+                    Arc::new(Int64Array::from_iter_values(repeat(*value).take(size)))
+                }
+                None => Arc::new(repeat(None).take(size).collect::<Int64Array>()),
+            },
+            ScalarValue::UInt8(e) => match e {
+                Some(value) => {
+                    Arc::new(UInt8Array::from_iter_values(repeat(*value).take(size)))
+                }
+                None => Arc::new(repeat(None).take(size).collect::<UInt8Array>()),
+            },
+            ScalarValue::UInt16(e) => match e {
+                Some(value) => {
+                    Arc::new(UInt16Array::from_iter_values(repeat(*value).take(size)))
+                }
+                None => Arc::new(repeat(None).take(size).collect::<UInt16Array>()),
+            },
+            ScalarValue::UInt32(e) => match e {
+                Some(value) => {
+                    Arc::new(UInt32Array::from_iter_values(repeat(*value).take(size)))
+                }
+                None => Arc::new(repeat(None).take(size).collect::<UInt32Array>()),
+            },
+            ScalarValue::UInt64(e) => match e {
+                Some(value) => {
+                    Arc::new(UInt64Array::from_iter_values(repeat(*value).take(size)))
+                }
+                None => Arc::new(repeat(None).take(size).collect::<UInt64Array>()),
+            },
+            ScalarValue::TimeMicrosecond(e) => match e {
+                Some(value) => Arc::new(TimestampMicrosecondArray::from_iter_values(
+                    repeat(*value).take(size),
+                )),
+                None => Arc::new(
+                    repeat(None)
+                        .take(size)
+                        .collect::<TimestampMicrosecondArray>(),
+                ),
+            },
+            ScalarValue::TimeNanosecond(e) => match e {
+                Some(value) => Arc::new(TimestampNanosecondArray::from_iter_values(
+                    repeat(*value).take(size),
+                )),
+                None => Arc::new(
+                    repeat(None)
+                        .take(size)
+                        .collect::<TimestampNanosecondArray>(),
+                ),
+            },
+            ScalarValue::Utf8(e) => match e {
+                Some(value) => {
+                    Arc::new(StringArray::from_iter_values(repeat(value).take(size)))
+                }
+                None => {
+                    Arc::new(repeat(None::<&str>).take(size).collect::<StringArray>())
+                }
+            },
+            ScalarValue::LargeUtf8(e) => match e {
+                Some(value) => {
+                    Arc::new(LargeStringArray::from_iter_values(repeat(value).take(size)))
+                }
+                None => Arc::new(
+                    repeat(None::<&str>)
+                        .take(size)
+                        .collect::<LargeStringArray>(),
+                ),
+            },
             ScalarValue::List(values, data_type) => Arc::new(match data_type {
                 DataType::Int8 => build_list!(Int8Builder, Int8, values, size),
                 DataType::Int16 => build_list!(Int16Builder, Int16, values, size),
@@ -238,7 +314,12 @@ impl ScalarValue {
                 DataType::UInt64 => build_list!(UInt64Builder, UInt64, values, size),
                 _ => panic!("Unexpected DataType for list"),
             }),
-            ScalarValue::Date32(e) => Arc::new(Date32Array::from(vec![*e; size])),
+            ScalarValue::Date32(e) => match e {
+                Some(value) => {
+                    Arc::new(Date32Array::from_iter_values(repeat(*value).take(size)))
+                }
+                None => Arc::new(repeat(None).take(size).collect::<Date32Array>()),
+            },
         }
     }