Support FixedSizedBinaryArray Parquet Data Page Statistics (#11200)

* failing add data page stats for fixed size binary array

* fix

* failing fix

* fix

Author: dharanad

datafusion/core/src/datasource/physical_plan/parquet/statistics.rs

@@ -726,6 +726,20 @@ make_data_page_stats_iterator!(
     ByteArray
 );
 
+make_data_page_stats_iterator!(
+    MaxFixedLenByteArrayDataPageStatsIterator,
+    |x: &PageIndex<FixedLenByteArray>| { x.max.clone() },
+    Index::FIXED_LEN_BYTE_ARRAY,
+    FixedLenByteArray
+);
+
+make_data_page_stats_iterator!(
+    MinFixedLenByteArrayDataPageStatsIterator,
+    |x: &PageIndex<FixedLenByteArray>| { x.min.clone() },
+    Index::FIXED_LEN_BYTE_ARRAY,
+    FixedLenByteArray
+);
+
 macro_rules! get_data_page_statistics {
     ($stat_type_prefix: ident, $data_type: ident, $iterator: ident) => {
         paste! {
@@ -903,7 +917,19 @@ macro_rules! get_data_page_statistics {
                             new_empty_array(&DataType::Time64(unit.clone()))
                         }
                     })
-                }
+                },
+               Some(DataType::FixedSizeBinary(size)) => {
+                    Ok(Arc::new(
+                        FixedSizeBinaryArray::try_from_iter(
+                            [<$stat_type_prefix FixedLenByteArrayDataPageStatsIterator>]::new($iterator)
+                            .flat_map(|x| x.into_iter())
+                            .filter_map(|x| x)
+                        ).unwrap_or_else(|e| {
+                            log::debug!("FixedSizeBinary statistics is invalid: {}", e);
+                            FixedSizeBinaryArray::new(*size, vec![].into(), None)
+                        })
+                    ))
+                },
                 _ => unimplemented!()
             }
         }

datafusion/core/tests/parquet/arrow_statistics.rs

@@ -1861,7 +1861,7 @@ async fn test_byte() {
         expected_null_counts: UInt64Array::from(vec![0, 0, 0]),
         expected_row_counts: Some(UInt64Array::from(vec![5, 5, 5])),
         column_name: "service_fixedsize",
-        check: Check::RowGroup,
+        check: Check::Both,
     }
     .run();