From a5284b861e04cb9ff19feeb330df36f89e89b62e Mon Sep 17 00:00:00 2001
From: NGA-TRAN <nga-tran@live.com>
Date: Tue, 21 May 2024 15:11:27 -0400
Subject: [PATCH 1/7] read statistics from parquet without statistics

---
 .../core/tests/parquet/arrow_statistics.rs    | 37 ++++++++++++++++++-
 1 file changed, 35 insertions(+), 2 deletions(-)

diff --git a/datafusion/core/tests/parquet/arrow_statistics.rs b/datafusion/core/tests/parquet/arrow_statistics.rs
index 272afea7b28a7..80bba6db21ba9 100644
--- a/datafusion/core/tests/parquet/arrow_statistics.rs
+++ b/datafusion/core/tests/parquet/arrow_statistics.rs
@@ -73,6 +73,22 @@ pub fn parquet_file_one_column(
     no_null_values_start: i64,
     no_null_values_end: i64,
     row_per_group: usize,
+) -> ParquetRecordBatchReaderBuilder<File> {
+    parquet_file_one_column_stats(
+        num_null,
+        no_null_values_start,
+        no_null_values_end,
+        row_per_group,
+        EnabledStatistics::Chunk,
+    )
+}
+
+pub fn parquet_file_one_column_stats(
+    num_null: usize,
+    no_null_values_start: i64,
+    no_null_values_end: i64,
+    row_per_group: usize,
+    enable_stats: EnabledStatistics,
 ) -> ParquetRecordBatchReaderBuilder<File> {
     let mut output_file = tempfile::Builder::new()
         .prefix("parquert_statistics_test")
@@ -82,7 +98,7 @@ pub fn parquet_file_one_column(
 
     let props = WriterProperties::builder()
         .set_max_row_group_size(row_per_group)
-        .set_statistics_enabled(EnabledStatistics::Chunk)
+        .set_statistics_enabled(enable_stats)
         .build();
 
     let batches = vec![make_int64_batches_with_null(
@@ -638,7 +654,24 @@ async fn test_dates_64_diff_rg_sizes() {
 // ByteArray,
 // PeriodsInColumnNames,
 // WithNullValuesPageLevel,
-// WITHOUT Stats
+
+////// Files with missing statistics ///////
+
+#[tokio::test]
+async fn test_missing_statistics() {
+    let row_per_group = 5;
+    let reader =
+        parquet_file_one_column_stats(0, 4, 7, row_per_group, EnabledStatistics::None);
+
+    Test {
+        reader,
+        expected_min: Arc::new(Int64Array::from(vec![None])),
+        expected_max: Arc::new(Int64Array::from(vec![None])),
+        expected_null_counts: UInt64Array::from(vec![None]),
+        expected_row_counts: UInt64Array::from(vec![3]), // stil has row count statistics
+    }
+    .run("i64");
+}
 
 /////// NEGATIVE TESTS ///////
 // column not found

From f9e4d6bde0268457ed121e3382881632aa155cac Mon Sep 17 00:00:00 2001
From: NGA-TRAN <nga-tran@live.com>
Date: Tue, 21 May 2024 17:03:22 -0400
Subject: [PATCH 2/7] tests for boolean and struct array

---
 .../core/tests/parquet/arrow_statistics.rs    | 47 ++++++++++++-
 datafusion/core/tests/parquet/mod.rs          | 70 ++++++++++++++++++-
 2 files changed, 113 insertions(+), 4 deletions(-)

diff --git a/datafusion/core/tests/parquet/arrow_statistics.rs b/datafusion/core/tests/parquet/arrow_statistics.rs
index 80bba6db21ba9..2045dde1f6efd 100644
--- a/datafusion/core/tests/parquet/arrow_statistics.rs
+++ b/datafusion/core/tests/parquet/arrow_statistics.rs
@@ -22,8 +22,8 @@ use std::fs::File;
 use std::sync::Arc;
 
 use arrow_array::{
-    make_array, Array, ArrayRef, Int16Array, Int32Array, Int64Array, Int8Array,
-    RecordBatch, UInt64Array,
+    make_array, Array, ArrayRef, BooleanArray, Int16Array, Int32Array, Int64Array,
+    Int8Array, RecordBatch, UInt64Array,
 };
 use arrow_schema::{DataType, Field, Schema};
 use datafusion::datasource::physical_plan::parquet::{
@@ -33,7 +33,7 @@ use parquet::arrow::arrow_reader::{ArrowReaderBuilder, ParquetRecordBatchReaderB
 use parquet::arrow::ArrowWriter;
 use parquet::file::properties::{EnabledStatistics, WriterProperties};
 
-use crate::parquet::Scenario;
+use crate::parquet::{struct_array, Scenario};
 
 use super::make_test_file_rg;
 
@@ -655,6 +655,47 @@ async fn test_dates_64_diff_rg_sizes() {
 // PeriodsInColumnNames,
 // WithNullValuesPageLevel,
 
+// TODO:
+// f64::NAN
+
+// Boolean
+#[tokio::test]
+async fn test_boolean() {
+    let row_per_group = 5;
+    // This creates a parquet files of 1 column named "bool"
+    // The file is created by 2 record batches each has 5 rows --> 2 row groups
+    let reader = parquet_file_many_columns(Scenario::Boolean, row_per_group).await;
+
+    Test {
+        reader,
+        expected_min: Arc::new(BooleanArray::from(vec![false, false])),
+        expected_max: Arc::new(BooleanArray::from(vec![true, false])),
+        expected_null_counts: UInt64Array::from(vec![1, 0]),
+        expected_row_counts: UInt64Array::from(vec![5, 5]),
+    }
+    .run("bool");
+}
+
+// struct array
+// BUG
+// todo: open ticket
+#[ignore]
+#[tokio::test]
+async fn test_struct() {
+    let row_per_group = 5;
+    // This creates a parquet files of 1 column named "struct"
+    // The file is created by 1 record batch with 3 rows in the struct array
+    let reader = parquet_file_many_columns(Scenario::StructArray, row_per_group).await;
+
+    Test {
+        reader,
+        expected_min: Arc::new(struct_array(vec![(Some(1), Some(6.0), Some(12.0))])),
+        expected_max: Arc::new(struct_array(vec![(Some(2), Some(8.5), Some(14.0))])),
+        expected_null_counts: UInt64Array::from(vec![0]),
+        expected_row_counts: UInt64Array::from(vec![3]),
+    }
+    .run("struct");
+}
 ////// Files with missing statistics ///////
 
 #[tokio::test]
diff --git a/datafusion/core/tests/parquet/mod.rs b/datafusion/core/tests/parquet/mod.rs
index 6e3f366b4373b..b490f9e91422a 100644
--- a/datafusion/core/tests/parquet/mod.rs
+++ b/datafusion/core/tests/parquet/mod.rs
@@ -28,7 +28,7 @@ use arrow::{
     record_batch::RecordBatch,
     util::pretty::pretty_format_batches,
 };
-use arrow_array::make_array;
+use arrow_array::{make_array, BooleanArray, Float32Array, StructArray};
 use chrono::{Datelike, Duration, TimeDelta};
 use datafusion::{
     datasource::{physical_plan::ParquetExec, provider_as_source, TableProvider},
@@ -65,6 +65,7 @@ fn init() {
 
 /// What data to use
 enum Scenario {
+    Boolean,
     Timestamps,
     Dates,
     Int,
@@ -81,6 +82,7 @@ enum Scenario {
     PeriodsInColumnNames,
     WithNullValues,
     WithNullValuesPageLevel,
+    StructArray,
 }
 
 enum Unit {
@@ -312,6 +314,16 @@ impl ContextWithParquet {
     }
 }
 
+fn make_boolean_batch(v: Vec<Option<bool>>) -> RecordBatch {
+    let schema = Arc::new(Schema::new(vec![Field::new(
+        "bool",
+        DataType::Boolean,
+        true,
+    )]));
+    let array = Arc::new(BooleanArray::from(v)) as ArrayRef;
+    RecordBatch::try_new(schema, vec![array.clone()]).unwrap()
+}
+
 /// Return record batch with a few rows of data for all of the supported timestamp types
 /// values with the specified offset
 ///
@@ -699,6 +711,24 @@ fn make_int_batches_with_null(
 
 fn create_data_batch(scenario: Scenario) -> Vec<RecordBatch> {
     match scenario {
+        Scenario::Boolean => {
+            vec![
+                make_boolean_batch(vec![
+                    Some(true),
+                    Some(false),
+                    Some(true),
+                    Some(false),
+                    None,
+                ]),
+                make_boolean_batch(vec![
+                    Some(false),
+                    Some(false),
+                    Some(false),
+                    Some(false),
+                    Some(false),
+                ]),
+            ]
+        }
         Scenario::Timestamps => {
             vec![
                 make_timestamp_batch(TimeDelta::try_seconds(0).unwrap()),
@@ -881,6 +911,20 @@ fn create_data_batch(scenario: Scenario) -> Vec<RecordBatch> {
                 make_int_batches_with_null(5, 1, 6),
             ]
         }
+        Scenario::StructArray => {
+            let struct_array_data = struct_array(vec![
+                (Some(1), Some(6.0), Some(12.0)),
+                (Some(2), Some(8.5), None),
+                (None, Some(8.5), Some(14.0)),
+            ]);
+
+            let schema = Arc::new(Schema::new(vec![Field::new(
+                "struct",
+                struct_array_data.data_type().clone(),
+                true,
+            )]));
+            vec![RecordBatch::try_new(schema, vec![struct_array_data]).unwrap()]
+        }
     }
 }
 
@@ -936,3 +980,27 @@ async fn make_test_file_page(scenario: Scenario, row_per_page: usize) -> NamedTe
     writer.close().unwrap();
     output_file
 }
+
+// returns a struct array with columns "int32_col", "float32_col" and "float64_col" with the specified values
+fn struct_array(input: Vec<(Option<i32>, Option<f32>, Option<f64>)>) -> ArrayRef {
+    let int_32: Int32Array = input.iter().map(|(i, _, _)| i).collect();
+    let float_32: Float32Array = input.iter().map(|(_, f, _)| f).collect();
+    let float_64: Float64Array = input.iter().map(|(_, _, f)| f).collect();
+
+    let nullable = true;
+    let struct_array = StructArray::from(vec![
+        (
+            Arc::new(Field::new("int32_col", DataType::Int32, nullable)),
+            Arc::new(int_32) as ArrayRef,
+        ),
+        (
+            Arc::new(Field::new("float32_col", DataType::Float32, nullable)),
+            Arc::new(float_32) as ArrayRef,
+        ),
+        (
+            Arc::new(Field::new("float64_col", DataType::Float64, nullable)),
+            Arc::new(float_64) as ArrayRef,
+        ),
+    ]);
+    Arc::new(struct_array)
+}

From 973cb6b27b3113ba0ac43627f966894060ecf8e8 Mon Sep 17 00:00:00 2001
From: NGA-TRAN <nga-tran@live.com>
Date: Tue, 21 May 2024 17:22:59 -0400
Subject: [PATCH 3/7] link to a the struct array bug

---
 datafusion/core/tests/parquet/arrow_statistics.rs | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/datafusion/core/tests/parquet/arrow_statistics.rs b/datafusion/core/tests/parquet/arrow_statistics.rs
index 07ec5c1c7a91c..24883a3fd9b92 100644
--- a/datafusion/core/tests/parquet/arrow_statistics.rs
+++ b/datafusion/core/tests/parquet/arrow_statistics.rs
@@ -939,7 +939,8 @@ async fn test_boolean() {
 
 // struct array
 // BUG
-// todo: open ticket
+// https://github.com/apache/datafusion/issues/10609
+// Note that: since I have not worked on struct array before, there may be a bug in the test code rather than the real bug in the code
 #[ignore]
 #[tokio::test]
 async fn test_struct() {

From 9b77a2a786134f3235260d15ad3919279088188a Mon Sep 17 00:00:00 2001
From: Andrew Lamb <andrew@nerdnetworks.org>
Date: Wed, 22 May 2024 14:02:31 -0400
Subject: [PATCH 4/7] Make builder style API

---
 .../core/tests/parquet/arrow_statistics.rs    | 34 +++++++++++++++++--
 1 file changed, 32 insertions(+), 2 deletions(-)

diff --git a/datafusion/core/tests/parquet/arrow_statistics.rs b/datafusion/core/tests/parquet/arrow_statistics.rs
index 24883a3fd9b92..bce31e956a3ec 100644
--- a/datafusion/core/tests/parquet/arrow_statistics.rs
+++ b/datafusion/core/tests/parquet/arrow_statistics.rs
@@ -144,7 +144,33 @@ pub async fn parquet_file_many_columns(
     ArrowReaderBuilder::try_new(file).unwrap()
 }
 
+/// Defines what data to create in a parquet file
+struct TestReader {
+    /// What data to create in the parquet file
+    scenario: Scenario,
+    /// Number of rows per row group
+    row_per_group: usize,
+}
+
+impl TestReader {
+    /// Create a parquet file with the specified data, and return a
+    /// ParquetRecordBatchReaderBuilder opened to that file.
+    async fn build(self) -> ParquetRecordBatchReaderBuilder<File> {
+        let TestReader {
+            scenario,
+            row_per_group,
+        } = self;
+        let file = make_test_file_rg(scenario, row_per_group).await;
+
+        // open the file & get the reader
+        let file = file.reopen().unwrap();
+        ArrowReaderBuilder::try_new(file).unwrap()
+    }
+}
+
+/// Defines a test case for statistics extraction
 struct Test {
+    /// The parquet file reader
     reader: ParquetRecordBatchReaderBuilder<File>,
     expected_min: ArrayRef,
     expected_max: ArrayRef,
@@ -312,9 +338,13 @@ async fn test_two_row_groups_with_all_nulls_in_one() {
 // Four different integer types
 #[tokio::test]
 async fn test_int_64() {
-    let row_per_group = 5;
     // This creates a parquet files of 4 columns named "i8", "i16", "i32", "i64"
-    let reader = parquet_file_many_columns(Scenario::Int, row_per_group).await;
+    let reader = TestReader {
+        scenario: Scenario::Int,
+        row_per_group: 5,
+    }
+    .build()
+    .await;
 
     Test {
         reader,

From c7c9f5ac712c3cd6133c2914e1f959d44261d833 Mon Sep 17 00:00:00 2001
From: Andrew Lamb <andrew@nerdnetworks.org>
Date: Wed, 22 May 2024 14:11:09 -0400
Subject: [PATCH 5/7] Port to builder style

---
 .../core/tests/parquet/arrow_statistics.rs    | 234 +++++++++++++-----
 1 file changed, 178 insertions(+), 56 deletions(-)

diff --git a/datafusion/core/tests/parquet/arrow_statistics.rs b/datafusion/core/tests/parquet/arrow_statistics.rs
index bce31e956a3ec..441b395764f72 100644
--- a/datafusion/core/tests/parquet/arrow_statistics.rs
+++ b/datafusion/core/tests/parquet/arrow_statistics.rs
@@ -362,10 +362,13 @@ async fn test_int_64() {
 
 #[tokio::test]
 async fn test_int_32() {
-    let row_per_group = 5;
     // This creates a parquet files of 4 columns named "i8", "i16", "i32", "i64"
-    let reader = parquet_file_many_columns(Scenario::Int, row_per_group).await;
-
+    let reader = TestReader {
+        scenario: Scenario::Int,
+        row_per_group: 5,
+    }
+    .build()
+    .await;
     Test {
         reader,
         // mins are [-5, -4, 0, 5]
@@ -388,9 +391,13 @@ async fn test_int_32() {
 #[ignore]
 #[tokio::test]
 async fn test_int_16() {
-    let row_per_group = 5;
     // This creates a parquet files of 4 columns named "i8", "i16", "i32", "i64"
-    let reader = parquet_file_many_columns(Scenario::Int, row_per_group).await;
+    let reader = TestReader {
+        scenario: Scenario::Int,
+        row_per_group: 5,
+    }
+    .build()
+    .await;
 
     Test {
         reader,
@@ -421,9 +428,13 @@ async fn test_int_16() {
 #[ignore]
 #[tokio::test]
 async fn test_int_8() {
-    let row_per_group = 5;
     // This creates a parquet files of 4 columns named "i8", "i16", "i32", "i64"
-    let reader = parquet_file_many_columns(Scenario::Int, row_per_group).await;
+    let reader = TestReader {
+        scenario: Scenario::Int,
+        row_per_group: 5,
+    }
+    .build()
+    .await;
 
     Test {
         reader,
@@ -443,8 +454,6 @@ async fn test_int_8() {
 // timestamp
 #[tokio::test]
 async fn test_timestamp() {
-    let row_per_group = 5;
-
     // This creates a parquet files of 5 columns named "nanos", "micros", "millis", "seconds", "names"
     // "nanos" --> TimestampNanosecondArray
     // "micros" --> TimestampMicrosecondArray
@@ -454,7 +463,13 @@ async fn test_timestamp() {
     //
     // The file is created by 4 record batches, each has 5 rowws.
     // Since the row group isze is set to 5, those 4 batches will go into 4 row groups
-    let reader = parquet_file_many_columns(Scenario::Timestamps, row_per_group).await;
+    // This creates a parquet files of 4 columns named "i8", "i16", "i32", "i64"
+    let reader = TestReader {
+        scenario: Scenario::Timestamps,
+        row_per_group: 5,
+    }
+    .build()
+    .await;
 
     Test {
         reader,
@@ -480,7 +495,12 @@ async fn test_timestamp() {
     .run("nanos");
 
     // micros
-    let reader = parquet_file_many_columns(Scenario::Timestamps, row_per_group).await;
+    let reader = TestReader {
+        scenario: Scenario::Timestamps,
+        row_per_group: 5,
+    }
+    .build()
+    .await;
     Test {
         reader,
         expected_min: Arc::new(Int64Array::from(vec![
@@ -501,7 +521,12 @@ async fn test_timestamp() {
     .run("micros");
 
     // millis
-    let reader = parquet_file_many_columns(Scenario::Timestamps, row_per_group).await;
+    let reader = TestReader {
+        scenario: Scenario::Timestamps,
+        row_per_group: 5,
+    }
+    .build()
+    .await;
     Test {
         reader,
         expected_min: Arc::new(Int64Array::from(vec![
@@ -522,7 +547,12 @@ async fn test_timestamp() {
     .run("millis");
 
     // seconds
-    let reader = parquet_file_many_columns(Scenario::Timestamps, row_per_group).await;
+    let reader = TestReader {
+        scenario: Scenario::Timestamps,
+        row_per_group: 5,
+    }
+    .build()
+    .await;
     Test {
         reader,
         expected_min: Arc::new(Int64Array::from(vec![
@@ -540,8 +570,6 @@ async fn test_timestamp() {
 // timestamp with different row group sizes
 #[tokio::test]
 async fn test_timestamp_diff_rg_sizes() {
-    let row_per_group = 8;
-
     // This creates a parquet files of 5 columns named "nanos", "micros", "millis", "seconds", "names"
     // "nanos" --> TimestampNanosecondArray
     // "micros" --> TimestampMicrosecondArray
@@ -550,7 +578,12 @@ async fn test_timestamp_diff_rg_sizes() {
     // "names" --> StringArray
     //
     // The file is created by 4 record batches (each has a null row), each has 5 rows but then will be split into 3 row groups with size 8, 8, 4
-    let reader = parquet_file_many_columns(Scenario::Timestamps, row_per_group).await;
+    let reader = TestReader {
+        scenario: Scenario::Timestamps,
+        row_per_group: 8, // note that the row group size is 8
+    }
+    .build()
+    .await;
 
     Test {
         reader,
@@ -574,7 +607,12 @@ async fn test_timestamp_diff_rg_sizes() {
     .run("nanos");
 
     // micros
-    let reader = parquet_file_many_columns(Scenario::Timestamps, row_per_group).await;
+    let reader = TestReader {
+        scenario: Scenario::Timestamps,
+        row_per_group: 8, // note that the row group size is 8
+    }
+    .build()
+    .await;
     Test {
         reader,
         expected_min: Arc::new(Int64Array::from(vec![
@@ -593,7 +631,12 @@ async fn test_timestamp_diff_rg_sizes() {
     .run("micros");
 
     // millis
-    let reader = parquet_file_many_columns(Scenario::Timestamps, row_per_group).await;
+    let reader = TestReader {
+        scenario: Scenario::Timestamps,
+        row_per_group: 8, // note that the row group size is 8
+    }
+    .build()
+    .await;
     Test {
         reader,
         expected_min: Arc::new(Int64Array::from(vec![
@@ -612,7 +655,12 @@ async fn test_timestamp_diff_rg_sizes() {
     .run("millis");
 
     // seconds
-    let reader = parquet_file_many_columns(Scenario::Timestamps, row_per_group).await;
+    let reader = TestReader {
+        scenario: Scenario::Timestamps,
+        row_per_group: 8, // note that the row group size is 8
+    }
+    .build()
+    .await;
     Test {
         reader,
         expected_min: Arc::new(Int64Array::from(vec![
@@ -632,15 +680,18 @@ async fn test_timestamp_diff_rg_sizes() {
 //  https://github.com/apache/datafusion/issues/10587
 #[tokio::test]
 async fn test_dates_32_diff_rg_sizes() {
-    let row_per_group = 13;
-
     // This creates a parquet files of 3 columns named "date32", "date64", "names"
     // "date32" --> Date32Array
     // "date64" --> Date64Array
     // "names" --> StringArray
     //
     // The file is created by 4 record batches (each has a null row), each has 5 rows but then will be split into 2 row groups with size 13, 7
-    let reader = parquet_file_many_columns(Scenario::Dates, row_per_group).await;
+    let reader = TestReader {
+        scenario: Scenario::Dates,
+        row_per_group: 13,
+    }
+    .build()
+    .await;
 
     Test {
         reader,
@@ -662,9 +713,13 @@ async fn test_dates_32_diff_rg_sizes() {
 #[ignore]
 #[tokio::test]
 async fn test_dates_64_diff_rg_sizes() {
-    let row_per_group = 13;
     // The file is created by 4 record batches (each has a null row), each has 5 rows but then will be split into 2 row groups with size 13, 7
-    let reader = parquet_file_many_columns(Scenario::Dates, row_per_group).await;
+    let reader = TestReader {
+        scenario: Scenario::Dates,
+        row_per_group: 13,
+    }
+    .build()
+    .await;
     Test {
         reader,
         expected_min: Arc::new(Int64Array::from(vec![18262, 18565])), // panic here because the actual data is Int32Array
@@ -679,8 +734,6 @@ async fn test_dates_64_diff_rg_sizes() {
 // https://github.com/apache/datafusion/issues/10604
 #[tokio::test]
 async fn test_uint() {
-    let row_per_group = 4;
-
     // This creates a parquet files of 4 columns named "u8", "u16", "u32", "u64"
     // "u8" --> UInt8Array
     // "u16" --> UInt16Array
@@ -688,7 +741,12 @@ async fn test_uint() {
     // "u64" --> UInt64Array
 
     // The file is created by 4 record batches (each has a null row), each has 5 rows but then will be split into 5 row groups with size 4
-    let reader = parquet_file_many_columns(Scenario::UInt, row_per_group).await;
+    let reader = TestReader {
+        scenario: Scenario::UInt,
+        row_per_group: 4,
+    }
+    .build()
+    .await;
 
     // u8
     // BUG: expect UInt8Array but returns Int32Array
@@ -703,7 +761,12 @@ async fn test_uint() {
 
     // u16
     // BUG: expect UInt16Array but returns Int32Array
-    let reader = parquet_file_many_columns(Scenario::UInt, row_per_group).await;
+    let reader = TestReader {
+        scenario: Scenario::UInt,
+        row_per_group: 4,
+    }
+    .build()
+    .await;
     Test {
         reader,
         expected_min: Arc::new(Int32Array::from(vec![0, 1, 4, 7, 251])), // shoudld be UInt16Array
@@ -715,7 +778,12 @@ async fn test_uint() {
 
     // u32
     // BUG: expect UInt32Array but returns Int32Array
-    let reader = parquet_file_many_columns(Scenario::UInt, row_per_group).await;
+    let reader = TestReader {
+        scenario: Scenario::UInt,
+        row_per_group: 4,
+    }
+    .build()
+    .await;
     Test {
         reader,
         expected_min: Arc::new(Int32Array::from(vec![0, 1, 4, 7, 251])), // shoudld be UInt32Array
@@ -727,7 +795,12 @@ async fn test_uint() {
 
     // u64
     // BUG: expect UInt64rray but returns Int64Array
-    let reader = parquet_file_many_columns(Scenario::UInt, row_per_group).await;
+    let reader = TestReader {
+        scenario: Scenario::UInt,
+        row_per_group: 4,
+    }
+    .build()
+    .await;
     Test {
         reader,
         expected_min: Arc::new(Int64Array::from(vec![0, 1, 4, 7, 251])), // shoudld be UInt64Array
@@ -740,10 +813,14 @@ async fn test_uint() {
 
 #[tokio::test]
 async fn test_int32_range() {
-    let row_per_group = 5;
     // This creates a parquet file of 1 column "i"
     // file has 2 record batches, each has 2 rows. They will be saved into one row group
-    let reader = parquet_file_many_columns(Scenario::Int32Range, row_per_group).await;
+    let reader = TestReader {
+        scenario: Scenario::Int32Range,
+        row_per_group: 5,
+    }
+    .build()
+    .await;
 
     Test {
         reader,
@@ -759,15 +836,19 @@ async fn test_int32_range() {
 // https://github.com/apache/datafusion/issues/10604
 #[tokio::test]
 async fn test_uint32_range() {
-    let row_per_group = 5;
     // This creates a parquet file of 1 column "u"
     // file has 2 record batches, each has 2 rows. They will be saved into one row group
-    let reader = parquet_file_many_columns(Scenario::UInt32Range, row_per_group).await;
+    let reader = TestReader {
+        scenario: Scenario::UInt32Range,
+        row_per_group: 5,
+    }
+    .build()
+    .await;
 
     Test {
         reader,
-        expected_min: Arc::new(Int32Array::from(vec![0])), // shoudld be UInt32Array
-        expected_max: Arc::new(Int32Array::from(vec![300000])), // shoudld be UInt32Array
+        expected_min: Arc::new(Int32Array::from(vec![0])), // should be UInt32Array
+        expected_max: Arc::new(Int32Array::from(vec![300000])), // should be UInt32Array
         expected_null_counts: UInt64Array::from(vec![0]),
         expected_row_counts: UInt64Array::from(vec![4]),
     }
@@ -776,10 +857,14 @@ async fn test_uint32_range() {
 
 #[tokio::test]
 async fn test_float64() {
-    let row_per_group = 5;
     // This creates a parquet file of 1 column "f"
     // file has 4 record batches, each has 5 rows. They will be saved into 4 row groups
-    let reader = parquet_file_many_columns(Scenario::Float64, row_per_group).await;
+    let reader = TestReader {
+        scenario: Scenario::Float64,
+        row_per_group: 5,
+    }
+    .build()
+    .await;
 
     Test {
         reader,
@@ -793,10 +878,14 @@ async fn test_float64() {
 
 #[tokio::test]
 async fn test_decimal() {
-    let row_per_group = 5;
     // This creates a parquet file of 1 column "decimal_col" with decimal data type and precicion 9, scale 2
     // file has 3 record batches, each has 5 rows. They will be saved into 3 row groups
-    let reader = parquet_file_many_columns(Scenario::Decimal, row_per_group).await;
+    let reader = TestReader {
+        scenario: Scenario::Decimal,
+        row_per_group: 5,
+    }
+    .build()
+    .await;
 
     Test {
         reader,
@@ -820,8 +909,6 @@ async fn test_decimal() {
 // https://github.com/apache/datafusion/issues/10605
 #[tokio::test]
 async fn test_byte() {
-    let row_per_group = 5;
-
     // This creates a parquet file of 4 columns
     // "name"
     // "service_string"
@@ -829,7 +916,12 @@ async fn test_byte() {
     // "service_fixedsize"
 
     // file has 3 record batches, each has 5 rows. They will be saved into 3 row groups
-    let reader = parquet_file_many_columns(Scenario::ByteArray, row_per_group).await;
+    let reader = TestReader {
+        scenario: Scenario::ByteArray,
+        row_per_group: 5,
+    }
+    .build()
+    .await;
 
     // column "name"
     Test {
@@ -850,7 +942,12 @@ async fn test_byte() {
     .run("name");
 
     // column "service_string"
-    let reader = parquet_file_many_columns(Scenario::ByteArray, row_per_group).await;
+    let reader = TestReader {
+        scenario: Scenario::ByteArray,
+        row_per_group: 5,
+    }
+    .build()
+    .await;
     Test {
         reader,
         expected_min: Arc::new(StringArray::from(vec![
@@ -869,7 +966,12 @@ async fn test_byte() {
     .run("service_string");
 
     // column "service_binary"
-    let reader = parquet_file_many_columns(Scenario::ByteArray, row_per_group).await;
+    let reader = TestReader {
+        scenario: Scenario::ByteArray,
+        row_per_group: 5,
+    }
+    .build()
+    .await;
     Test {
         reader,
         expected_min: Arc::new(StringArray::from(vec![
@@ -892,7 +994,12 @@ async fn test_byte() {
     let min_input = vec![vec![102, 101, 49], vec![98, 101, 49], vec![98, 101, 52]];
     // b"fe5", b"fe6", b"be8"
     let max_input = vec![vec![102, 101, 55], vec![102, 101, 54], vec![98, 101, 56]];
-    let reader = parquet_file_many_columns(Scenario::ByteArray, row_per_group).await;
+    let reader = TestReader {
+        scenario: Scenario::ByteArray,
+        row_per_group: 5,
+    }
+    .build()
+    .await;
     Test {
         reader,
         expected_min: Arc::new(
@@ -910,11 +1017,14 @@ async fn test_byte() {
 // PeriodsInColumnNames
 #[tokio::test]
 async fn test_period_in_column_names() {
-    let row_per_group = 5;
     // This creates a parquet file of 2 columns "name" and "service.name"
     // file has 3 record batches, each has 5 rows. They will be saved into 3 row groups
-    let reader =
-        parquet_file_many_columns(Scenario::PeriodsInColumnNames, row_per_group).await;
+    let reader = TestReader {
+        scenario: Scenario::PeriodsInColumnNames,
+        row_per_group: 5,
+    }
+    .build()
+    .await;
 
     // column "name"
     Test {
@@ -935,8 +1045,13 @@ async fn test_period_in_column_names() {
     .run("name");
 
     // column "service.name"
-    let reader =
-        parquet_file_many_columns(Scenario::PeriodsInColumnNames, row_per_group).await;
+    let reader = TestReader {
+        scenario: Scenario::PeriodsInColumnNames,
+        row_per_group: 5,
+    }
+    .build()
+    .await;
+
     Test {
         reader,
         expected_min: Arc::new(StringArray::from(vec!["frontend", "backend", "backend"])),
@@ -952,10 +1067,14 @@ async fn test_period_in_column_names() {
 // Boolean
 #[tokio::test]
 async fn test_boolean() {
-    let row_per_group = 5;
     // This creates a parquet files of 1 column named "bool"
     // The file is created by 2 record batches each has 5 rows --> 2 row groups
-    let reader = parquet_file_many_columns(Scenario::Boolean, row_per_group).await;
+    let reader = TestReader {
+        scenario: Scenario::Boolean,
+        row_per_group: 5,
+    }
+    .build()
+    .await;
 
     Test {
         reader,
@@ -974,11 +1093,14 @@ async fn test_boolean() {
 #[ignore]
 #[tokio::test]
 async fn test_struct() {
-    let row_per_group = 5;
     // This creates a parquet files of 1 column named "struct"
     // The file is created by 1 record batch with 3 rows in the struct array
-    let reader = parquet_file_many_columns(Scenario::StructArray, row_per_group).await;
-
+    let reader = TestReader {
+        scenario: Scenario::StructArray,
+        row_per_group: 5,
+    }
+    .build()
+    .await;
     Test {
         reader,
         expected_min: Arc::new(struct_array(vec![(Some(1), Some(6.0), Some(12.0))])),

From bee14d5e3ef85a4aaa9498831aa4c97b4916a0f7 Mon Sep 17 00:00:00 2001
From: Andrew Lamb <andrew@nerdnetworks.org>
Date: Wed, 22 May 2024 14:18:31 -0400
Subject: [PATCH 6/7] simlify

---
 .../core/tests/parquet/arrow_statistics.rs    | 234 +++++-------------
 datafusion/core/tests/parquet/mod.rs          |   1 +
 2 files changed, 57 insertions(+), 178 deletions(-)

diff --git a/datafusion/core/tests/parquet/arrow_statistics.rs b/datafusion/core/tests/parquet/arrow_statistics.rs
index 441b395764f72..7914faaf444a1 100644
--- a/datafusion/core/tests/parquet/arrow_statistics.rs
+++ b/datafusion/core/tests/parquet/arrow_statistics.rs
@@ -130,21 +130,8 @@ pub fn parquet_file_one_column_stats(
     ArrowReaderBuilder::try_new(file).unwrap()
 }
 
-// Create a parquet file with many columns each has different data type
-//  - Data types are specified by the given scenario
-//  - Row group sizes are withe the same or different depending on the provided row_per_group & data created in the scenario
-pub async fn parquet_file_many_columns(
-    scenario: super::Scenario,
-    row_per_group: usize,
-) -> ParquetRecordBatchReaderBuilder<File> {
-    let file = make_test_file_rg(scenario, row_per_group).await;
-
-    // open the file & get the reader
-    let file = file.reopen().unwrap();
-    ArrowReaderBuilder::try_new(file).unwrap()
-}
-
 /// Defines what data to create in a parquet file
+#[derive(Debug, Clone, Copy)]
 struct TestReader {
     /// What data to create in the parquet file
     scenario: Scenario,
@@ -342,12 +329,10 @@ async fn test_int_64() {
     let reader = TestReader {
         scenario: Scenario::Int,
         row_per_group: 5,
-    }
-    .build()
-    .await;
+    };
 
     Test {
-        reader,
+        reader: reader.build().await,
         // mins are [-5, -4, 0, 5]
         expected_min: Arc::new(Int64Array::from(vec![-5, -4, 0, 5])),
         // maxes are [-1, 0, 4, 9]
@@ -366,11 +351,10 @@ async fn test_int_32() {
     let reader = TestReader {
         scenario: Scenario::Int,
         row_per_group: 5,
-    }
-    .build()
-    .await;
+    };
+
     Test {
-        reader,
+        reader: reader.build().await,
         // mins are [-5, -4, 0, 5]
         expected_min: Arc::new(Int32Array::from(vec![-5, -4, 0, 5])),
         // maxes are [-1, 0, 4, 9]
@@ -395,12 +379,10 @@ async fn test_int_16() {
     let reader = TestReader {
         scenario: Scenario::Int,
         row_per_group: 5,
-    }
-    .build()
-    .await;
+    };
 
     Test {
-        reader,
+        reader: reader.build().await,
         // mins are [-5, -4, 0, 5]
         // BUG: not sure why this returns same data but in Int32Array type even though I debugged and the columns name is "i16" an its data is Int16
         // My debugging tells me the bug is either at:
@@ -432,12 +414,10 @@ async fn test_int_8() {
     let reader = TestReader {
         scenario: Scenario::Int,
         row_per_group: 5,
-    }
-    .build()
-    .await;
+    };
 
     Test {
-        reader,
+        reader: reader.build().await,
         // mins are [-5, -4, 0, 5]
         // BUG: not sure why this returns same data but in Int32Array even though I debugged and the columns name is "i8" an its data is Int8
         expected_min: Arc::new(Int8Array::from(vec![-5, -4, 0, 5])), // panic here because the actual data is Int32Array
@@ -467,12 +447,10 @@ async fn test_timestamp() {
     let reader = TestReader {
         scenario: Scenario::Timestamps,
         row_per_group: 5,
-    }
-    .build()
-    .await;
+    };
 
     Test {
-        reader,
+        reader: reader.build().await,
         // mins are [1577840461000000000, 1577840471000000000, 1577841061000000000, 1578704461000000000,]
         expected_min: Arc::new(Int64Array::from(vec![
             1577840461000000000,
@@ -495,14 +473,9 @@ async fn test_timestamp() {
     .run("nanos");
 
     // micros
-    let reader = TestReader {
-        scenario: Scenario::Timestamps,
-        row_per_group: 5,
-    }
-    .build()
-    .await;
+
     Test {
-        reader,
+        reader: reader.build().await,
         expected_min: Arc::new(Int64Array::from(vec![
             1577840461000000,
             1577840471000000,
@@ -521,14 +494,8 @@ async fn test_timestamp() {
     .run("micros");
 
     // millis
-    let reader = TestReader {
-        scenario: Scenario::Timestamps,
-        row_per_group: 5,
-    }
-    .build()
-    .await;
     Test {
-        reader,
+        reader: reader.build().await,
         expected_min: Arc::new(Int64Array::from(vec![
             1577840461000,
             1577840471000,
@@ -547,14 +514,8 @@ async fn test_timestamp() {
     .run("millis");
 
     // seconds
-    let reader = TestReader {
-        scenario: Scenario::Timestamps,
-        row_per_group: 5,
-    }
-    .build()
-    .await;
     Test {
-        reader,
+        reader: reader.build().await,
         expected_min: Arc::new(Int64Array::from(vec![
             1577840461, 1577840471, 1577841061, 1578704461,
         ])),
@@ -581,12 +542,10 @@ async fn test_timestamp_diff_rg_sizes() {
     let reader = TestReader {
         scenario: Scenario::Timestamps,
         row_per_group: 8, // note that the row group size is 8
-    }
-    .build()
-    .await;
+    };
 
     Test {
-        reader,
+        reader: reader.build().await,
         // mins are [1577840461000000000, 1577841061000000000, 1578704521000000000]
         expected_min: Arc::new(Int64Array::from(vec![
             1577840461000000000,
@@ -607,14 +566,8 @@ async fn test_timestamp_diff_rg_sizes() {
     .run("nanos");
 
     // micros
-    let reader = TestReader {
-        scenario: Scenario::Timestamps,
-        row_per_group: 8, // note that the row group size is 8
-    }
-    .build()
-    .await;
     Test {
-        reader,
+        reader: reader.build().await,
         expected_min: Arc::new(Int64Array::from(vec![
             1577840461000000,
             1577841061000000,
@@ -631,14 +584,8 @@ async fn test_timestamp_diff_rg_sizes() {
     .run("micros");
 
     // millis
-    let reader = TestReader {
-        scenario: Scenario::Timestamps,
-        row_per_group: 8, // note that the row group size is 8
-    }
-    .build()
-    .await;
     Test {
-        reader,
+        reader: reader.build().await,
         expected_min: Arc::new(Int64Array::from(vec![
             1577840461000,
             1577841061000,
@@ -655,14 +602,8 @@ async fn test_timestamp_diff_rg_sizes() {
     .run("millis");
 
     // seconds
-    let reader = TestReader {
-        scenario: Scenario::Timestamps,
-        row_per_group: 8, // note that the row group size is 8
-    }
-    .build()
-    .await;
     Test {
-        reader,
+        reader: reader.build().await,
         expected_min: Arc::new(Int64Array::from(vec![
             1577840461, 1577841061, 1578704521,
         ])),
@@ -689,12 +630,9 @@ async fn test_dates_32_diff_rg_sizes() {
     let reader = TestReader {
         scenario: Scenario::Dates,
         row_per_group: 13,
-    }
-    .build()
-    .await;
-
+    };
     Test {
-        reader,
+        reader: reader.build().await,
         // mins are [18262, 18565,]
         expected_min: Arc::new(Int32Array::from(vec![18262, 18565])),
         // maxes are [18564, 21865,]
@@ -717,11 +655,9 @@ async fn test_dates_64_diff_rg_sizes() {
     let reader = TestReader {
         scenario: Scenario::Dates,
         row_per_group: 13,
-    }
-    .build()
-    .await;
+    };
     Test {
-        reader,
+        reader: reader.build().await,
         expected_min: Arc::new(Int64Array::from(vec![18262, 18565])), // panic here because the actual data is Int32Array
         expected_max: Arc::new(Int64Array::from(vec![18564, 21865])),
         expected_null_counts: UInt64Array::from(vec![2, 2]),
@@ -744,14 +680,12 @@ async fn test_uint() {
     let reader = TestReader {
         scenario: Scenario::UInt,
         row_per_group: 4,
-    }
-    .build()
-    .await;
+    };
 
     // u8
     // BUG: expect UInt8Array but returns Int32Array
     Test {
-        reader,
+        reader: reader.build().await,
         expected_min: Arc::new(Int32Array::from(vec![0, 1, 4, 7, 251])), // shoudld be UInt8Array
         expected_max: Arc::new(Int32Array::from(vec![3, 4, 6, 250, 254])), // shoudld be UInt8Array
         expected_null_counts: UInt64Array::from(vec![0, 0, 0, 0, 0]),
@@ -761,14 +695,8 @@ async fn test_uint() {
 
     // u16
     // BUG: expect UInt16Array but returns Int32Array
-    let reader = TestReader {
-        scenario: Scenario::UInt,
-        row_per_group: 4,
-    }
-    .build()
-    .await;
     Test {
-        reader,
+        reader: reader.build().await,
         expected_min: Arc::new(Int32Array::from(vec![0, 1, 4, 7, 251])), // shoudld be UInt16Array
         expected_max: Arc::new(Int32Array::from(vec![3, 4, 6, 250, 254])), // shoudld be UInt16Array
         expected_null_counts: UInt64Array::from(vec![0, 0, 0, 0, 0]),
@@ -778,14 +706,8 @@ async fn test_uint() {
 
     // u32
     // BUG: expect UInt32Array but returns Int32Array
-    let reader = TestReader {
-        scenario: Scenario::UInt,
-        row_per_group: 4,
-    }
-    .build()
-    .await;
     Test {
-        reader,
+        reader: reader.build().await,
         expected_min: Arc::new(Int32Array::from(vec![0, 1, 4, 7, 251])), // shoudld be UInt32Array
         expected_max: Arc::new(Int32Array::from(vec![3, 4, 6, 250, 254])), // shoudld be UInt32Array
         expected_null_counts: UInt64Array::from(vec![0, 0, 0, 0, 0]),
@@ -795,14 +717,8 @@ async fn test_uint() {
 
     // u64
     // BUG: expect UInt64rray but returns Int64Array
-    let reader = TestReader {
-        scenario: Scenario::UInt,
-        row_per_group: 4,
-    }
-    .build()
-    .await;
     Test {
-        reader,
+        reader: reader.build().await,
         expected_min: Arc::new(Int64Array::from(vec![0, 1, 4, 7, 251])), // shoudld be UInt64Array
         expected_max: Arc::new(Int64Array::from(vec![3, 4, 6, 250, 254])), // shoudld be UInt64Array
         expected_null_counts: UInt64Array::from(vec![0, 0, 0, 0, 0]),
@@ -818,12 +734,10 @@ async fn test_int32_range() {
     let reader = TestReader {
         scenario: Scenario::Int32Range,
         row_per_group: 5,
-    }
-    .build()
-    .await;
+    };
 
     Test {
-        reader,
+        reader: reader.build().await,
         expected_min: Arc::new(Int32Array::from(vec![0])),
         expected_max: Arc::new(Int32Array::from(vec![300000])),
         expected_null_counts: UInt64Array::from(vec![0]),
@@ -841,12 +755,10 @@ async fn test_uint32_range() {
     let reader = TestReader {
         scenario: Scenario::UInt32Range,
         row_per_group: 5,
-    }
-    .build()
-    .await;
+    };
 
     Test {
-        reader,
+        reader: reader.build().await,
         expected_min: Arc::new(Int32Array::from(vec![0])), // should be UInt32Array
         expected_max: Arc::new(Int32Array::from(vec![300000])), // should be UInt32Array
         expected_null_counts: UInt64Array::from(vec![0]),
@@ -862,12 +774,10 @@ async fn test_float64() {
     let reader = TestReader {
         scenario: Scenario::Float64,
         row_per_group: 5,
-    }
-    .build()
-    .await;
+    };
 
     Test {
-        reader,
+        reader: reader.build().await,
         expected_min: Arc::new(Float64Array::from(vec![-5.0, -4.0, -0.0, 5.0])),
         expected_max: Arc::new(Float64Array::from(vec![-1.0, 0.0, 4.0, 9.0])),
         expected_null_counts: UInt64Array::from(vec![0, 0, 0, 0]),
@@ -883,12 +793,10 @@ async fn test_decimal() {
     let reader = TestReader {
         scenario: Scenario::Decimal,
         row_per_group: 5,
-    }
-    .build()
-    .await;
+    };
 
     Test {
-        reader,
+        reader: reader.build().await,
         expected_min: Arc::new(
             Decimal128Array::from(vec![100, -500, 2000])
                 .with_precision_and_scale(9, 2)
@@ -919,13 +827,11 @@ async fn test_byte() {
     let reader = TestReader {
         scenario: Scenario::ByteArray,
         row_per_group: 5,
-    }
-    .build()
-    .await;
+    };
 
     // column "name"
     Test {
-        reader,
+        reader: reader.build().await,
         expected_min: Arc::new(StringArray::from(vec![
             "all frontends",
             "mixed",
@@ -942,14 +848,8 @@ async fn test_byte() {
     .run("name");
 
     // column "service_string"
-    let reader = TestReader {
-        scenario: Scenario::ByteArray,
-        row_per_group: 5,
-    }
-    .build()
-    .await;
     Test {
-        reader,
+        reader: reader.build().await,
         expected_min: Arc::new(StringArray::from(vec![
             "frontend five",
             "backend one",
@@ -966,14 +866,8 @@ async fn test_byte() {
     .run("service_string");
 
     // column "service_binary"
-    let reader = TestReader {
-        scenario: Scenario::ByteArray,
-        row_per_group: 5,
-    }
-    .build()
-    .await;
     Test {
-        reader,
+        reader: reader.build().await,
         expected_min: Arc::new(StringArray::from(vec![
             "frontend five",
             "backend one",
@@ -994,14 +888,9 @@ async fn test_byte() {
     let min_input = vec![vec![102, 101, 49], vec![98, 101, 49], vec![98, 101, 52]];
     // b"fe5", b"fe6", b"be8"
     let max_input = vec![vec![102, 101, 55], vec![102, 101, 54], vec![98, 101, 56]];
-    let reader = TestReader {
-        scenario: Scenario::ByteArray,
-        row_per_group: 5,
-    }
-    .build()
-    .await;
+
     Test {
-        reader,
+        reader: reader.build().await,
         expected_min: Arc::new(
             FixedSizeBinaryArray::try_from_iter(min_input.into_iter()).unwrap(),
         ),
@@ -1022,13 +911,11 @@ async fn test_period_in_column_names() {
     let reader = TestReader {
         scenario: Scenario::PeriodsInColumnNames,
         row_per_group: 5,
-    }
-    .build()
-    .await;
+    };
 
     // column "name"
     Test {
-        reader,
+        reader: reader.build().await,
         expected_min: Arc::new(StringArray::from(vec![
             "HTTP GET / DISPATCH",
             "HTTP PUT / DISPATCH",
@@ -1045,15 +932,8 @@ async fn test_period_in_column_names() {
     .run("name");
 
     // column "service.name"
-    let reader = TestReader {
-        scenario: Scenario::PeriodsInColumnNames,
-        row_per_group: 5,
-    }
-    .build()
-    .await;
-
     Test {
-        reader,
+        reader: reader.build().await,
         expected_min: Arc::new(StringArray::from(vec!["frontend", "backend", "backend"])),
         expected_max: Arc::new(StringArray::from(vec![
             "frontend", "frontend", "backend",
@@ -1072,12 +952,10 @@ async fn test_boolean() {
     let reader = TestReader {
         scenario: Scenario::Boolean,
         row_per_group: 5,
-    }
-    .build()
-    .await;
+    };
 
     Test {
-        reader,
+        reader: reader.build().await,
         expected_min: Arc::new(BooleanArray::from(vec![false, false])),
         expected_max: Arc::new(BooleanArray::from(vec![true, false])),
         expected_null_counts: UInt64Array::from(vec![1, 0]),
@@ -1098,11 +976,9 @@ async fn test_struct() {
     let reader = TestReader {
         scenario: Scenario::StructArray,
         row_per_group: 5,
-    }
-    .build()
-    .await;
+    };
     Test {
-        reader,
+        reader: reader.build().await,
         expected_min: Arc::new(struct_array(vec![(Some(1), Some(6.0), Some(12.0))])),
         expected_max: Arc::new(struct_array(vec![(Some(2), Some(8.5), Some(14.0))])),
         expected_null_counts: UInt64Array::from(vec![0]),
@@ -1132,10 +1008,12 @@ async fn test_missing_statistics() {
 // column not found
 #[tokio::test]
 async fn test_column_not_found() {
-    let row_per_group = 5;
-    let reader = parquet_file_many_columns(Scenario::Dates, row_per_group).await;
+    let reader = TestReader {
+        scenario: Scenario::Dates,
+        row_per_group: 5,
+    };
     Test {
-        reader,
+        reader: reader.build().await,
         expected_min: Arc::new(Int64Array::from(vec![18262, 18565])),
         expected_max: Arc::new(Int64Array::from(vec![18564, 21865])),
         expected_null_counts: UInt64Array::from(vec![2, 2]),
diff --git a/datafusion/core/tests/parquet/mod.rs b/datafusion/core/tests/parquet/mod.rs
index 77f36a48927ba..c5d0ad60bc10f 100644
--- a/datafusion/core/tests/parquet/mod.rs
+++ b/datafusion/core/tests/parquet/mod.rs
@@ -64,6 +64,7 @@ fn init() {
 // ----------------------
 
 /// What data to use
+#[derive(Debug, Clone, Copy)]
 enum Scenario {
     Boolean,
     Timestamps,

From e127fc40c3c98c029c2c007f6332fe352e3eb0ab Mon Sep 17 00:00:00 2001
From: Andrew Lamb <andrew@nerdnetworks.org>
Date: Wed, 22 May 2024 14:28:57 -0400
Subject: [PATCH 7/7] Add column name to test

---
 .../core/tests/parquet/arrow_statistics.rs    | 125 ++++++++++++------
 1 file changed, 83 insertions(+), 42 deletions(-)

diff --git a/datafusion/core/tests/parquet/arrow_statistics.rs b/datafusion/core/tests/parquet/arrow_statistics.rs
index 7914faaf444a1..36fffe5ac4d95 100644
--- a/datafusion/core/tests/parquet/arrow_statistics.rs
+++ b/datafusion/core/tests/parquet/arrow_statistics.rs
@@ -163,20 +163,23 @@ struct Test {
     expected_max: ArrayRef,
     expected_null_counts: UInt64Array,
     expected_row_counts: UInt64Array,
+    /// Which column to extract statistics from
+    column_name: &'static str,
 }
 
 impl Test {
-    fn run(self, col_name: &str) {
+    fn run(self) {
         let Self {
             reader,
             expected_min,
             expected_max,
             expected_null_counts,
             expected_row_counts,
+            column_name,
         } = self;
 
         let min = StatisticsConverter::try_new(
-            col_name,
+            column_name,
             RequestedStatistics::Min,
             reader.schema(),
         )
@@ -187,7 +190,7 @@ impl Test {
         assert_eq!(&min, &expected_min, "Mismatch with expected minimums");
 
         let max = StatisticsConverter::try_new(
-            col_name,
+            column_name,
             RequestedStatistics::Max,
             reader.schema(),
         )
@@ -197,7 +200,7 @@ impl Test {
         assert_eq!(&max, &expected_max, "Mismatch with expected maximum");
 
         let null_counts = StatisticsConverter::try_new(
-            col_name,
+            column_name,
             RequestedStatistics::NullCount,
             reader.schema(),
         )
@@ -217,17 +220,19 @@ impl Test {
         );
     }
 
-    fn run_col_not_found(self, col_name: &str) {
+    /// Run the test and expect a column not found error
+    fn run_col_not_found(self) {
         let Self {
             reader,
             expected_min: _,
             expected_max: _,
             expected_null_counts: _,
             expected_row_counts: _,
+            column_name,
         } = self;
 
         let min = StatisticsConverter::try_new(
-            col_name,
+            column_name,
             RequestedStatistics::Min,
             reader.schema(),
         );
@@ -256,8 +261,9 @@ async fn test_one_row_group_without_null() {
         expected_null_counts: UInt64Array::from(vec![0]),
         // 3 rows
         expected_row_counts: UInt64Array::from(vec![3]),
+        column_name: "i64",
     }
-    .run("i64")
+    .run()
 }
 
 #[tokio::test]
@@ -275,8 +281,9 @@ async fn test_one_row_group_with_null_and_negative() {
         expected_null_counts: UInt64Array::from(vec![2]),
         // 8 rows
         expected_row_counts: UInt64Array::from(vec![8]),
+        column_name: "i64",
     }
-    .run("i64")
+    .run()
 }
 
 #[tokio::test]
@@ -294,8 +301,9 @@ async fn test_two_row_group_with_null() {
         expected_null_counts: UInt64Array::from(vec![0, 2]),
         // row counts are [10, 5]
         expected_row_counts: UInt64Array::from(vec![10, 5]),
+        column_name: "i64",
     }
-    .run("i64")
+    .run()
 }
 
 #[tokio::test]
@@ -313,8 +321,9 @@ async fn test_two_row_groups_with_all_nulls_in_one() {
         expected_null_counts: UInt64Array::from(vec![1, 3]),
         // row counts are [5, 3]
         expected_row_counts: UInt64Array::from(vec![5, 3]),
+        column_name: "i64",
     }
-    .run("i64")
+    .run()
 }
 
 /////////////// MORE GENERAL TESTS //////////////////////
@@ -341,8 +350,9 @@ async fn test_int_64() {
         expected_null_counts: UInt64Array::from(vec![0, 0, 0, 0]),
         // row counts are [5, 5, 5, 5]
         expected_row_counts: UInt64Array::from(vec![5, 5, 5, 5]),
+        column_name: "i64",
     }
-    .run("i64");
+    .run();
 }
 
 #[tokio::test]
@@ -363,8 +373,9 @@ async fn test_int_32() {
         expected_null_counts: UInt64Array::from(vec![0, 0, 0, 0]),
         // row counts are [5, 5, 5, 5]
         expected_row_counts: UInt64Array::from(vec![5, 5, 5, 5]),
+        column_name: "i32",
     }
-    .run("i32");
+    .run();
 }
 
 // BUG: ignore this test for now
@@ -401,8 +412,9 @@ async fn test_int_16() {
         expected_null_counts: UInt64Array::from(vec![0, 0, 0, 0]),
         // row counts are [5, 5, 5, 5]
         expected_row_counts: UInt64Array::from(vec![5, 5, 5, 5]),
+        column_name: "i16",
     }
-    .run("i16");
+    .run();
 }
 
 // BUG (same as above): ignore this test for now
@@ -427,8 +439,9 @@ async fn test_int_8() {
         expected_null_counts: UInt64Array::from(vec![0, 0, 0, 0]),
         // row counts are [5, 5, 5, 5]
         expected_row_counts: UInt64Array::from(vec![5, 5, 5, 5]),
+        column_name: "i8",
     }
-    .run("i8");
+    .run();
 }
 
 // timestamp
@@ -469,8 +482,9 @@ async fn test_timestamp() {
         expected_null_counts: UInt64Array::from(vec![1, 1, 1, 1]),
         // row counts are [5, 5, 5, 5]
         expected_row_counts: UInt64Array::from(vec![5, 5, 5, 5]),
+        column_name: "nanos",
     }
-    .run("nanos");
+    .run();
 
     // micros
 
@@ -490,8 +504,9 @@ async fn test_timestamp() {
         ])),
         expected_null_counts: UInt64Array::from(vec![1, 1, 1, 1]),
         expected_row_counts: UInt64Array::from(vec![5, 5, 5, 5]),
+        column_name: "micros",
     }
-    .run("micros");
+    .run();
 
     // millis
     Test {
@@ -510,8 +525,9 @@ async fn test_timestamp() {
         ])),
         expected_null_counts: UInt64Array::from(vec![1, 1, 1, 1]),
         expected_row_counts: UInt64Array::from(vec![5, 5, 5, 5]),
+        column_name: "millis",
     }
-    .run("millis");
+    .run();
 
     // seconds
     Test {
@@ -524,8 +540,9 @@ async fn test_timestamp() {
         ])),
         expected_null_counts: UInt64Array::from(vec![1, 1, 1, 1]),
         expected_row_counts: UInt64Array::from(vec![5, 5, 5, 5]),
+        column_name: "seconds",
     }
-    .run("seconds");
+    .run();
 }
 
 // timestamp with different row group sizes
@@ -562,8 +579,9 @@ async fn test_timestamp_diff_rg_sizes() {
         expected_null_counts: UInt64Array::from(vec![1, 2, 1]),
         // row counts are [8, 8, 4]
         expected_row_counts: UInt64Array::from(vec![8, 8, 4]),
+        column_name: "nanos",
     }
-    .run("nanos");
+    .run();
 
     // micros
     Test {
@@ -580,8 +598,9 @@ async fn test_timestamp_diff_rg_sizes() {
         ])),
         expected_null_counts: UInt64Array::from(vec![1, 2, 1]),
         expected_row_counts: UInt64Array::from(vec![8, 8, 4]),
+        column_name: "micros",
     }
-    .run("micros");
+    .run();
 
     // millis
     Test {
@@ -598,8 +617,9 @@ async fn test_timestamp_diff_rg_sizes() {
         ])),
         expected_null_counts: UInt64Array::from(vec![1, 2, 1]),
         expected_row_counts: UInt64Array::from(vec![8, 8, 4]),
+        column_name: "millis",
     }
-    .run("millis");
+    .run();
 
     // seconds
     Test {
@@ -612,8 +632,9 @@ async fn test_timestamp_diff_rg_sizes() {
         ])),
         expected_null_counts: UInt64Array::from(vec![1, 2, 1]),
         expected_row_counts: UInt64Array::from(vec![8, 8, 4]),
+        column_name: "seconds",
     }
-    .run("seconds");
+    .run();
 }
 
 // date with different row group sizes
@@ -641,8 +662,9 @@ async fn test_dates_32_diff_rg_sizes() {
         expected_null_counts: UInt64Array::from(vec![2, 2]),
         // row counts are [13, 7]
         expected_row_counts: UInt64Array::from(vec![13, 7]),
+        column_name: "date32",
     }
-    .run("date32");
+    .run();
 }
 
 // BUG: same as above. Expect to return Date64Array but returns Int32Array
@@ -662,8 +684,9 @@ async fn test_dates_64_diff_rg_sizes() {
         expected_max: Arc::new(Int64Array::from(vec![18564, 21865])),
         expected_null_counts: UInt64Array::from(vec![2, 2]),
         expected_row_counts: UInt64Array::from(vec![13, 7]),
+        column_name: "date64",
     }
-    .run("date64");
+    .run();
 }
 
 // BUG:
@@ -690,8 +713,9 @@ async fn test_uint() {
         expected_max: Arc::new(Int32Array::from(vec![3, 4, 6, 250, 254])), // shoudld be UInt8Array
         expected_null_counts: UInt64Array::from(vec![0, 0, 0, 0, 0]),
         expected_row_counts: UInt64Array::from(vec![4, 4, 4, 4, 4]),
+        column_name: "u8",
     }
-    .run("u8");
+    .run();
 
     // u16
     // BUG: expect UInt16Array but returns Int32Array
@@ -701,8 +725,9 @@ async fn test_uint() {
         expected_max: Arc::new(Int32Array::from(vec![3, 4, 6, 250, 254])), // shoudld be UInt16Array
         expected_null_counts: UInt64Array::from(vec![0, 0, 0, 0, 0]),
         expected_row_counts: UInt64Array::from(vec![4, 4, 4, 4, 4]),
+        column_name: "u16",
     }
-    .run("u16");
+    .run();
 
     // u32
     // BUG: expect UInt32Array but returns Int32Array
@@ -712,8 +737,9 @@ async fn test_uint() {
         expected_max: Arc::new(Int32Array::from(vec![3, 4, 6, 250, 254])), // shoudld be UInt32Array
         expected_null_counts: UInt64Array::from(vec![0, 0, 0, 0, 0]),
         expected_row_counts: UInt64Array::from(vec![4, 4, 4, 4, 4]),
+        column_name: "u32",
     }
-    .run("u32");
+    .run();
 
     // u64
     // BUG: expect UInt64rray but returns Int64Array
@@ -723,8 +749,9 @@ async fn test_uint() {
         expected_max: Arc::new(Int64Array::from(vec![3, 4, 6, 250, 254])), // shoudld be UInt64Array
         expected_null_counts: UInt64Array::from(vec![0, 0, 0, 0, 0]),
         expected_row_counts: UInt64Array::from(vec![4, 4, 4, 4, 4]),
+        column_name: "u64",
     }
-    .run("u64");
+    .run();
 }
 
 #[tokio::test]
@@ -742,8 +769,9 @@ async fn test_int32_range() {
         expected_max: Arc::new(Int32Array::from(vec![300000])),
         expected_null_counts: UInt64Array::from(vec![0]),
         expected_row_counts: UInt64Array::from(vec![4]),
+        column_name: "i",
     }
-    .run("i");
+    .run();
 }
 
 // BUG: not convert UInt32Array to Int32Array
@@ -763,8 +791,9 @@ async fn test_uint32_range() {
         expected_max: Arc::new(Int32Array::from(vec![300000])), // should be UInt32Array
         expected_null_counts: UInt64Array::from(vec![0]),
         expected_row_counts: UInt64Array::from(vec![4]),
+        column_name: "u",
     }
-    .run("u");
+    .run();
 }
 
 #[tokio::test]
@@ -782,8 +811,9 @@ async fn test_float64() {
         expected_max: Arc::new(Float64Array::from(vec![-1.0, 0.0, 4.0, 9.0])),
         expected_null_counts: UInt64Array::from(vec![0, 0, 0, 0]),
         expected_row_counts: UInt64Array::from(vec![5, 5, 5, 5]),
+        column_name: "f",
     }
-    .run("f");
+    .run();
 }
 
 #[tokio::test]
@@ -809,8 +839,9 @@ async fn test_decimal() {
         ),
         expected_null_counts: UInt64Array::from(vec![0, 0, 0]),
         expected_row_counts: UInt64Array::from(vec![5, 5, 5]),
+        column_name: "decimal_col",
     }
-    .run("decimal_col");
+    .run();
 }
 
 // BUG: not convert BinaryArray to StringArray
@@ -844,8 +875,9 @@ async fn test_byte() {
         ])),
         expected_null_counts: UInt64Array::from(vec![0, 0, 0]),
         expected_row_counts: UInt64Array::from(vec![5, 5, 5]),
+        column_name: "name",
     }
-    .run("name");
+    .run();
 
     // column "service_string"
     Test {
@@ -862,8 +894,9 @@ async fn test_byte() {
         ])),
         expected_null_counts: UInt64Array::from(vec![0, 0, 0]),
         expected_row_counts: UInt64Array::from(vec![5, 5, 5]),
+        column_name: "service_string",
     }
-    .run("service_string");
+    .run();
 
     // column "service_binary"
     Test {
@@ -880,8 +913,9 @@ async fn test_byte() {
         ])), // Shuld be BinaryArray
         expected_null_counts: UInt64Array::from(vec![0, 0, 0]),
         expected_row_counts: UInt64Array::from(vec![5, 5, 5]),
+        column_name: "service_binary",
     }
-    .run("service_binary");
+    .run();
 
     // column "service_fixedsize"
     // b"fe1", b"be1", b"be4"
@@ -899,8 +933,9 @@ async fn test_byte() {
         ),
         expected_null_counts: UInt64Array::from(vec![0, 0, 0]),
         expected_row_counts: UInt64Array::from(vec![5, 5, 5]),
+        column_name: "service_fixedsize",
     }
-    .run("service_fixedsize");
+    .run();
 }
 
 // PeriodsInColumnNames
@@ -928,8 +963,9 @@ async fn test_period_in_column_names() {
         ])),
         expected_null_counts: UInt64Array::from(vec![0, 0, 0]),
         expected_row_counts: UInt64Array::from(vec![5, 5, 5]),
+        column_name: "name",
     }
-    .run("name");
+    .run();
 
     // column "service.name"
     Test {
@@ -940,8 +976,9 @@ async fn test_period_in_column_names() {
         ])),
         expected_null_counts: UInt64Array::from(vec![0, 0, 0]),
         expected_row_counts: UInt64Array::from(vec![5, 5, 5]),
+        column_name: "service.name",
     }
-    .run("service.name");
+    .run();
 }
 
 // Boolean
@@ -960,8 +997,9 @@ async fn test_boolean() {
         expected_max: Arc::new(BooleanArray::from(vec![true, false])),
         expected_null_counts: UInt64Array::from(vec![1, 0]),
         expected_row_counts: UInt64Array::from(vec![5, 5]),
+        column_name: "bool",
     }
-    .run("bool");
+    .run();
 }
 
 // struct array
@@ -983,8 +1021,9 @@ async fn test_struct() {
         expected_max: Arc::new(struct_array(vec![(Some(2), Some(8.5), Some(14.0))])),
         expected_null_counts: UInt64Array::from(vec![0]),
         expected_row_counts: UInt64Array::from(vec![3]),
+        column_name: "struct",
     }
-    .run("struct");
+    .run();
 }
 ////// Files with missing statistics ///////
 
@@ -1000,8 +1039,9 @@ async fn test_missing_statistics() {
         expected_max: Arc::new(Int64Array::from(vec![None])),
         expected_null_counts: UInt64Array::from(vec![None]),
         expected_row_counts: UInt64Array::from(vec![3]), // stil has row count statistics
+        column_name: "i64",
     }
-    .run("i64");
+    .run();
 }
 
 /////// NEGATIVE TESTS ///////
@@ -1018,6 +1058,7 @@ async fn test_column_not_found() {
         expected_max: Arc::new(Int64Array::from(vec![18564, 21865])),
         expected_null_counts: UInt64Array::from(vec![2, 2]),
         expected_row_counts: UInt64Array::from(vec![13, 7]),
+        column_name: "not_a_column",
     }
-    .run_col_not_found("not_a_column");
+    .run_col_not_found();
 }
