Repository: apache/arrow-rs
Pull request #6013 (2024-07-07 05:22:32)

feat(parquet): Implement AsyncFileWriter for `object_store::buffered::BufWriter`
# Which issue does this PR close?

<!--
We generally require a GitHub issue to be filed for all bug fixes and enhancements and this helps us generate change logs for our releases. You can link an issue to this PR using the GitHub syntax. For example `Closes #123` indicates that this PR will close issue #123.
-->

Closes https://github.com/apache/arrow-rs/issues/5766

# Rationale for this change
 
Implement `AsyncFileWriter` for `object_store::buffered::BufWriter`

# What changes are included in this PR?

Add a new impl for `AsyncFileWriter`

# Are there any user-facing changes?


New impl added.


Xuanwo: This PR is blocked by `BufWriter::put` API is not released yet.
Xuanwo: Hi, @alamb @tustvold, is it possible to run a patch release for `object_store`?
alamb: Thanks for the reminder @Xuanwo 

I filed this ticket to track the next release https://github.com/apache/arrow-rs/issues/6022

I looked at the commits that went into object_store: https://github.com/apache/arrow-rs/commits/master/object_store

As of today there do not appear to be any breaking API changes to object store and thus I think we can release a minor 0.10.1 version


Xuanwo: Hi, @alamb, Sorry for the long waiting, this PR is ready to review now.

Patch:
diff --git a/arrow-arith/src/temporal.rs b/arrow-arith/src/temporal.rs
index a665b7c06..6f8e8cc18 100644
--- a/arrow-arith/src/temporal.rs
+++ b/arrow-arith/src/temporal.rs
@@ -1596,8 +1596,6 @@ mod tests {
         ensure_returns_error(&Time64NanosecondArray::from(vec![0]));
     }
 
-    // IntervalDayTimeType week, day, hour, miute, second, mili, u, nano; invalid month, year; ignores the other part
-    // IntervalMonthDayNanoType year -> nano; days don't affect months, time doesn't affect days, time doesn't affect months (and vice versa)
     #[test]
     fn test_interval_year_month_array() {
         let input: IntervalYearMonthArray = vec![0, 5, 24].into();
@@ -1618,6 +1616,8 @@ mod tests {
         assert!(date_part(&input, DatePart::Week).is_err());
     }
 
+    // IntervalDayTimeType week, day, hour, minute, second, milli, u, nano;
+    // invalid month, year; ignores the other part
     #[test]
     fn test_interval_day_time_array() {
         let input: IntervalDayTimeArray = vec![
@@ -1693,6 +1693,8 @@ mod tests {
         assert!(date_part(&input, DatePart::Year).is_err());
     }
 
+    // IntervalMonthDayNanoType year -> nano;
+    // days don't affect months, time doesn't affect days, time doesn't affect months (and vice versa)
     #[test]
     fn test_interval_month_day_nano_array() {
         let input: IntervalMonthDayNanoArray = vec![
@@ -1728,7 +1730,7 @@ mod tests {
         assert_eq!(10, actual.value(1));
         assert_eq!(35, actual.value(2));
 
-        // Times follow from nanos, but are not affected by months or dats.
+        // Times follow from nanos, but are not affected by months or days.
         let actual = date_part(&input, DatePart::Hour).unwrap();
         let actual = actual.as_primitive::<Int32Type>();
         assert_eq!(0, actual.value(0));
diff --git a/arrow-array/src/array/primitive_array.rs b/arrow-array/src/array/primitive_array.rs
index 7ec9ff928..70a8ceaef 100644
--- a/arrow-array/src/array/primitive_array.rs
+++ b/arrow-array/src/array/primitive_array.rs
@@ -239,13 +239,13 @@ pub type Float32Array = PrimitiveArray<Float32Type>;
 /// Construction
 ///
 /// ```
-/// # use arrow_array::Float32Array;
+/// # use arrow_array::Float64Array;
 /// // Create from Vec<Option<f32>>
-/// let arr = Float32Array::from(vec![Some(1.0), None, Some(2.0)]);
+/// let arr = Float64Array::from(vec![Some(1.0), None, Some(2.0)]);
 /// // Create from Vec<f32>
-/// let arr = Float32Array::from(vec![1.0, 2.0, 3.0]);
+/// let arr = Float64Array::from(vec![1.0, 2.0, 3.0]);
 /// // Create iter/collect
-/// let arr: Float32Array = std::iter::repeat(42.0).take(10).collect();
+/// let arr: Float64Array = std::iter::repeat(42.0).take(10).collect();
 /// ```
 ///
 /// See [`PrimitiveArray`] for more information and examples
diff --git a/arrow-schema/src/ffi.rs b/arrow-schema/src/ffi.rs
index d12e237d7..b0c6a9f52 100644
--- a/arrow-schema/src/ffi.rs
+++ b/arrow-schema/src/ffi.rs
@@ -390,8 +390,10 @@ impl TryFrom<&FFI_ArrowSchema> for DataType {
             "e" => DataType::Float16,
             "f" => DataType::Float32,
             "g" => DataType::Float64,
+            "vz" => DataType::BinaryView,
             "z" => DataType::Binary,
             "Z" => DataType::LargeBinary,
+            "vu" => DataType::Utf8View,
             "u" => DataType::Utf8,
             "U" => DataType::LargeUtf8,
             "tdD" => DataType::Date32,
@@ -660,8 +662,10 @@ fn get_format_string(dtype: &DataType) -> Result<String, ArrowError> {
         DataType::Float16 => Ok("e".to_string()),
         DataType::Float32 => Ok("f".to_string()),
         DataType::Float64 => Ok("g".to_string()),
+        DataType::BinaryView => Ok("vz".to_string()),
         DataType::Binary => Ok("z".to_string()),
         DataType::LargeBinary => Ok("Z".to_string()),
+        DataType::Utf8View => Ok("vu".to_string()),
         DataType::Utf8 => Ok("u".to_string()),
         DataType::LargeUtf8 => Ok("U".to_string()),
         DataType::FixedSizeBinary(num_bytes) => Ok(format!("w:{num_bytes}")),
@@ -810,6 +814,10 @@ mod tests {
             5,
         ));
         round_trip_type(DataType::Utf8);
+        round_trip_type(DataType::Utf8View);
+        round_trip_type(DataType::BinaryView);
+        round_trip_type(DataType::Binary);
+        round_trip_type(DataType::LargeBinary);
         round_trip_type(DataType::List(Arc::new(Field::new(
             "a",
             DataType::Int16,
diff --git a/arrow-select/src/take.rs b/arrow-select/src/take.rs
index d6892eb0a..b66133ac7 100644
--- a/arrow-select/src/take.rs
+++ b/arrow-select/src/take.rs
@@ -487,11 +487,10 @@ fn take_byte_view<T: ByteViewType, IndexType: ArrowPrimitiveType>(
 ) -> Result<GenericByteViewArray<T>, ArrowError> {
     let new_views = take_native(array.views(), indices);
     let new_nulls = take_nulls(array.nulls(), indices);
-    Ok(GenericByteViewArray::new(
-        new_views,
-        array.data_buffers().to_vec(),
-        new_nulls,
-    ))
+    // Safety:  array.views was valid, and take_native copies only valid values, and verifies bounds
+    Ok(unsafe {
+        GenericByteViewArray::new_unchecked(new_views, array.data_buffers().to_vec(), new_nulls)
+    })
 }
 
 /// `take` implementation for list arrays
diff --git a/arrow/benches/aggregate_kernels.rs b/arrow/benches/aggregate_kernels.rs
index 434bb4778..6e224a48c 100644
--- a/arrow/benches/aggregate_kernels.rs
+++ b/arrow/benches/aggregate_kernels.rs
@@ -104,6 +104,9 @@ fn add_benchmark(c: &mut Criterion) {
             .bench_function("or nonnull mixed", |b| {
                 b.iter(|| bool_or(&nonnull_bools_mixed))
             })
+            .bench_function("and nonnull mixed", |b| {
+                b.iter(|| bool_and(&nonnull_bools_mixed))
+            })
             .bench_function("min nonnull false", |b| {
                 b.iter(|| min_boolean(&nonnull_bools_all_false))
             })
@@ -113,6 +116,9 @@ fn add_benchmark(c: &mut Criterion) {
             .bench_function("or nonnull false", |b| {
                 b.iter(|| bool_or(&nonnull_bools_all_false))
             })
+            .bench_function("and nonnull false", |b| {
+                b.iter(|| bool_and(&nonnull_bools_all_false))
+            })
             .bench_function("min nonnull true", |b| {
                 b.iter(|| min_boolean(&nonnull_bools_all_true))
             })
@@ -122,6 +128,9 @@ fn add_benchmark(c: &mut Criterion) {
             .bench_function("or nonnull true", |b| {
                 b.iter(|| bool_or(&nonnull_bools_all_true))
             })
+            .bench_function("and nonnull true", |b| {
+                b.iter(|| bool_and(&nonnull_bools_all_true))
+            })
             .bench_function("min nullable mixed", |b| {
                 b.iter(|| min_boolean(&nullable_bool_mixed))
             })
@@ -131,6 +140,9 @@ fn add_benchmark(c: &mut Criterion) {
             .bench_function("or nullable mixed", |b| {
                 b.iter(|| bool_or(&nullable_bool_mixed))
             })
+            .bench_function("and nullable mixed", |b| {
+                b.iter(|| bool_and(&nullable_bool_mixed))
+            })
             .bench_function("min nullable false", |b| {
                 b.iter(|| min_boolean(&nullable_bool_all_false))
             })
@@ -140,6 +152,9 @@ fn add_benchmark(c: &mut Criterion) {
             .bench_function("or nullable false", |b| {
                 b.iter(|| bool_or(&nullable_bool_all_false))
             })
+            .bench_function("and nullable false", |b| {
+                b.iter(|| bool_and(&nullable_bool_all_false))
+            })
             .bench_function("min nullable true", |b| {
                 b.iter(|| min_boolean(&nullable_bool_all_true))
             })
@@ -148,6 +163,9 @@ fn add_benchmark(c: &mut Criterion) {
             })
             .bench_function("or nullable true", |b| {
                 b.iter(|| bool_or(&nullable_bool_all_true))
+            })
+            .bench_function("and nullable true", |b| {
+                b.iter(|| bool_and(&nullable_bool_all_true))
             });
     }
 }
diff --git a/arrow/benches/filter_kernels.rs b/arrow/benches/filter_kernels.rs
index e48b53022..ae5eec386 100644
--- a/arrow/benches/filter_kernels.rs
+++ b/arrow/benches/filter_kernels.rs
@@ -204,6 +204,49 @@ fn add_benchmark(c: &mut Criterion) {
         |b| b.iter(|| bench_built_filter(&sparse_filter, &data_array)),
     );
 
+    let mut add_benchmark_for_fsb_with_length = |value_length: usize| {
+        let data_array = create_fsb_array(size, 0.0, value_length);
+        c.bench_function(
+            format!("filter fsb with value length {value_length} (kept 1/2)").as_str(),
+            |b| b.iter(|| bench_filter(&data_array, &filter_array)),
+        );
+        c.bench_function(
+            format!(
+                "filter fsb with value length {value_length} high selectivity (kept 1023/1024)"
+            )
+            .as_str(),
+            |b| b.iter(|| bench_filter(&data_array, &dense_filter_array)),
+        );
+        c.bench_function(
+            format!("filter fsb with value length {value_length} low selectivity (kept 1/1024)")
+                .as_str(),
+            |b| b.iter(|| bench_filter(&data_array, &sparse_filter_array)),
+        );
+
+        c.bench_function(
+            format!("filter context fsb with value length {value_length} (kept 1/2)").as_str(),
+            |b| b.iter(|| bench_built_filter(&filter, &filter_array)),
+        );
+        c.bench_function(
+            format!(
+                "filter context fsb with value length {value_length} high selectivity (kept 1023/1024)"
+            )
+            .as_str(),
+            |b| b.iter(|| bench_built_filter(&filter, &dense_filter_array)),
+        );
+        c.bench_function(
+            format!(
+                "filter context fsb with value length {value_length} low selectivity (kept 1/1024)"
+            )
+            .as_str(),
+            |b| b.iter(|| bench_built_filter(&filter, &sparse_filter_array)),
+        );
+    };
+
+    add_benchmark_for_fsb_with_length(5);
+    add_benchmark_for_fsb_with_length(20);
+    add_benchmark_for_fsb_with_length(50);
+
     let data_array = create_primitive_array::<Float32Type>(size, 0.0);
 
     let field = Field::new("c1", data_array.data_type().clone(), true);
diff --git a/arrow/benches/take_kernels.rs b/arrow/benches/take_kernels.rs
index 9c3f1eb40..77ec54c97 100644
--- a/arrow/benches/take_kernels.rs
+++ b/arrow/benches/take_kernels.rs
@@ -149,6 +149,42 @@ fn add_benchmark(c: &mut Criterion) {
         b.iter(|| bench_take(&values, &indices))
     });
 
+    let values = create_string_view_array(512, 0.0);
+    let indices = create_random_index(512, 0.0);
+    c.bench_function("take stringview 512", |b| {
+        b.iter(|| bench_take(&values, &indices))
+    });
+
+    let values = create_string_view_array(1024, 0.0);
+    let indices = create_random_index(1024, 0.0);
+    c.bench_function("take stringview 1024", |b| {
+        b.iter(|| bench_take(&values, &indices))
+    });
+
+    let values = create_string_view_array(512, 0.0);
+    let indices = create_random_index(512, 0.5);
+    c.bench_function("take stringview null indices 512", |b| {
+        b.iter(|| bench_take(&values, &indices))
+    });
+
+    let values = create_string_view_array(1024, 0.0);
+    let indices = create_random_index(1024, 0.5);
+    c.bench_function("take stringview null indices 1024", |b| {
+        b.iter(|| bench_take(&values, &indices))
+    });
+
+    let values = create_string_view_array(1024, 0.5);
+    let indices = create_random_index(1024, 0.0);
+    c.bench_function("take stringview null values 1024", |b| {
+        b.iter(|| bench_take(&values, &indices))
+    });
+
+    let values = create_string_view_array(1024, 0.5);
+    let indices = create_random_index(1024, 0.5);
+    c.bench_function("take stringview null values null indices 1024", |b| {
+        b.iter(|| bench_take(&values, &indices))
+    });
+
     let values = create_primitive_run_array::<Int32Type, Int32Type>(1024, 512);
     let indices = create_random_index(1024, 0.0);
     c.bench_function(
diff --git a/arrow/src/util/bench_util.rs b/arrow/src/util/bench_util.rs
index ac7f86d56..2561c925a 100644
--- a/arrow/src/util/bench_util.rs
+++ b/arrow/src/util/bench_util.rs
@@ -160,6 +160,34 @@ pub fn create_string_array_with_len<Offset: OffsetSizeTrait>(
         .collect()
 }
 
+/// Creates a random (but fixed-seeded) string view array of a given size and null density.
+///
+/// See `create_string_array` above for more details.
+pub fn create_string_view_array(size: usize, null_density: f32) -> StringViewArray {
+    create_string_view_array_with_max_len(size, null_density, 400)
+}
+
+/// Creates a random (but fixed-seeded) array of rand size with a given max size, null density and length
+fn create_string_view_array_with_max_len(
+    size: usize,
+    null_density: f32,
+    max_str_len: usize,
+) -> StringViewArray {
+    let rng = &mut seedable_rng();
+    (0..size)
+        .map(|_| {
+            if rng.gen::<f32>() < null_density {
+                None
+            } else {
+                let str_len = rng.gen_range(0..max_str_len);
+                let value = rng.sample_iter(&Alphanumeric).take(str_len).collect();
+                let value = String::from_utf8(value).unwrap();
+                Some(value)
+            }
+        })
+        .collect()
+}
+
 /// Creates a random (but fixed-seeded) array of a given size, null density and length
 pub fn create_string_view_array_with_len(
     size: usize,
diff --git a/object_store/Cargo.toml b/object_store/Cargo.toml
index 4e845e5ca..71e0bcb86 100644
--- a/object_store/Cargo.toml
+++ b/object_store/Cargo.toml
@@ -38,7 +38,7 @@ humantime = "2.1"
 itertools = "0.13.0"
 parking_lot = { version = "0.12" }
 percent-encoding = "2.1"
-snafu = "0.7"
+snafu = { version = "0.8", default-features = false, features = ["std", "rust_1_61"] }
 tracing = { version = "0.1" }
 url = "2.2"
 walkdir = "2"
diff --git a/object_store/src/client/get.rs b/object_store/src/client/get.rs
index b45eaa143..0fef5785c 100644
--- a/object_store/src/client/get.rs
+++ b/object_store/src/client/get.rs
@@ -103,7 +103,7 @@ enum GetResultError {
         source: crate::client::header::Error,
     },
 
-    #[snafu(context(false))]
+    #[snafu(transparent)]
     InvalidRangeRequest {
         source: crate::util::InvalidGetRange,
     },
@@ -386,7 +386,7 @@ mod tests {
         let err = get_result::<TestClient>(&path, Some(get_range.clone()), resp).unwrap_err();
         assert_eq!(
             err.to_string(),
-            "InvalidRangeRequest: Wanted range starting at 2, but object was only 2 bytes long"
+            "Wanted range starting at 2, but object was only 2 bytes long"
         );
 
         let resp = make_response(
diff --git a/object_store/src/lib.rs b/object_store/src/lib.rs
index 904cb670e..7699477b9 100644
--- a/object_store/src/lib.rs
+++ b/object_store/src/lib.rs
@@ -1225,6 +1225,7 @@ pub type Result<T, E = Error> = std::result::Result<T, E>;
 /// A specialized `Error` for object store-related errors
 #[derive(Debug, Snafu)]
 #[allow(missing_docs)]
+#[non_exhaustive]
 pub enum Error {
     #[snafu(display("Generic {} error: {}", store, source))]
     Generic {
diff --git a/object_store/src/path/mod.rs b/object_store/src/path/mod.rs
index f914862bc..59e08e2ea 100644
--- a/object_store/src/path/mod.rs
+++ b/object_store/src/path/mod.rs
@@ -37,6 +37,7 @@ pub use parts::{InvalidPart, PathPart};
 /// Error returned by [`Path::parse`]
 #[derive(Debug, Snafu)]
 #[allow(missing_docs)]
+#[non_exhaustive]
 pub enum Error {
     #[snafu(display("Path \"{}\" contained empty path segment", path))]
     EmptySegment { path: String },
diff --git a/parquet/Cargo.toml b/parquet/Cargo.toml
index 7391d0964..bc59c2f3f 100644
--- a/parquet/Cargo.toml
+++ b/parquet/Cargo.toml
@@ -44,7 +44,7 @@ arrow-schema = { workspace = true, optional = true }
 arrow-select = { workspace = true, optional = true }
 arrow-ipc = { workspace = true, optional = true }
 # Intentionally not a path dependency as object_store is released separately
-object_store = { version = "0.10.0", default-features = false, optional = true }
+object_store = { version = "0.10.2", default-features = false, optional = true }
 
 bytes = { version = "1.1", default-features = false, features = ["std"] }
 thrift = { version = "0.17", default-features = false }
@@ -67,7 +67,7 @@ hashbrown = { version = "0.14", default-features = false }
 twox-hash = { version = "1.6", default-features = false }
 paste = { version = "1.0" }
 half = { version = "2.1", default-features = false, features = ["num-traits"] }
-sysinfo = { version = "0.30.12", optional = true, default-features = false }
+sysinfo = { version = "0.31.2", optional = true, default-features = false, features = ["system"] }
 
 [dev-dependencies]
 base64 = { version = "0.22", default-features = false, features = ["std"] }
@@ -82,7 +82,7 @@ serde_json = { version = "1.0", features = ["std"], default-features = false }
 arrow = { workspace = true, features = ["ipc", "test_utils", "prettyprint", "json"] }
 tokio = { version = "1.0", default-features = false, features = ["macros", "rt", "io-util", "fs"] }
 rand = { version = "0.8", default-features = false, features = ["std", "std_rng"] }
-object_store = { version = "0.10.0", default-features = false, features = ["azure"] }
+object_store = { version = "0.10.2", default-features = false, features = ["azure"] }
 
 # TODO: temporary to fix parquet wasm build
 # upstream issue: https://github.com/gyscos/zstd-rs/issues/269
diff --git a/parquet/examples/write_parquet.rs b/parquet/examples/write_parquet.rs
index d2ef550df..9ee88facc 100644
--- a/parquet/examples/write_parquet.rs
+++ b/parquet/examples/write_parquet.rs
@@ -28,7 +28,7 @@ use parquet::arrow::ArrowWriter as ParquetWriter;
 use parquet::basic::Encoding;
 use parquet::errors::Result;
 use parquet::file::properties::{BloomFilterPosition, WriterProperties};
-use sysinfo::{MemoryRefreshKind, Pid, ProcessRefreshKind, RefreshKind, System};
+use sysinfo::{MemoryRefreshKind, ProcessRefreshKind, ProcessesToUpdate, RefreshKind, System};
 
 #[derive(ValueEnum, Clone)]
 enum BloomFilterPositionArg {
@@ -61,8 +61,16 @@ fn now() -> String {
 }
 
 fn mem(system: &mut System) -> String {
-    let pid = Pid::from(std::process::id() as usize);
-    system.refresh_process_specifics(pid, ProcessRefreshKind::new().with_memory());
+    let pid = match sysinfo::get_current_pid() {
+        Ok(pid) => pid,
+        Err(e) => return format!("Can't get process PID: {e}"),
+    };
+
+    system.refresh_processes_specifics(
+        ProcessesToUpdate::Some(&[pid]),
+        ProcessRefreshKind::everything(),
+    );
+
     system
         .process(pid)
         .map(|proc| format!("{}MB", proc.memory() / 1_000_000))
diff --git a/parquet/src/arrow/arrow_reader/statistics.rs b/parquet/src/arrow/arrow_reader/statistics.rs
index 6a1434bce..c42f92838 100644
--- a/parquet/src/arrow/arrow_reader/statistics.rs
+++ b/parquet/src/arrow/arrow_reader/statistics.rs
@@ -26,7 +26,8 @@ use crate::file::page_index::index::{Index, PageIndex};
 use crate::file::statistics::Statistics as ParquetStatistics;
 use crate::schema::types::SchemaDescriptor;
 use arrow_array::builder::{
-    BooleanBuilder, FixedSizeBinaryBuilder, LargeStringBuilder, StringBuilder,
+    BinaryViewBuilder, BooleanBuilder, FixedSizeBinaryBuilder, LargeStringBuilder, StringBuilder,
+    StringViewBuilder,
 };
 use arrow_array::{
     new_empty_array, new_null_array, ArrayRef, BinaryArray, BooleanArray, Date32Array, Date64Array,
@@ -446,14 +447,43 @@ macro_rules! get_statistics {
             },
             DataType::Dictionary(_, value_type) => {
                 [<$stat_type_prefix:lower _ statistics>](value_type, $iterator)
+            },
+            DataType::Utf8View => {
+                let iterator = [<$stat_type_prefix ByteArrayStatsIterator>]::new($iterator);
+                let mut builder = StringViewBuilder::new();
+                for x in iterator {
+                    let Some(x) = x else {
+                        builder.append_null(); // no statistics value
+                        continue;
+                    };
+
+                    let Ok(x) = std::str::from_utf8(x) else {
+                        builder.append_null();
+                        continue;
+                    };
+
+                    builder.append_value(x);
+                }
+                Ok(Arc::new(builder.finish()))
+            },
+            DataType::BinaryView => {
+                let iterator = [<$stat_type_prefix ByteArrayStatsIterator>]::new($iterator);
+                let mut builder = BinaryViewBuilder::new();
+                for x in iterator {
+                    let Some(x) = x else {
+                        builder.append_null(); // no statistics value
+                        continue;
+                    };
+
+                    builder.append_value(x);
+                }
+                Ok(Arc::new(builder.finish()))
             }
 
             DataType::Map(_,_) |
             DataType::Duration(_) |
             DataType::Interval(_) |
             DataType::Null |
-            DataType::BinaryView |
-            DataType::Utf8View |
             DataType::List(_) |
             DataType::ListView(_) |
             DataType::FixedSizeList(_, _) |
@@ -919,7 +949,7 @@ macro_rules! get_data_page_statistics {
                         }
                     })
                 },
-               Some(DataType::FixedSizeBinary(size)) => {
+                Some(DataType::FixedSizeBinary(size)) => {
                     let mut builder = FixedSizeBinaryBuilder::new(*size);
                     let iterator = [<$stat_type_prefix FixedLenByteArrayDataPageStatsIterator>]::new($iterator);
                     for x in iterator {
@@ -943,7 +973,58 @@ macro_rules! get_data_page_statistics {
                     }
                     Ok(Arc::new(builder.finish()))
                 },
-                _ => unimplemented!()
+                Some(DataType::Utf8View) => {
+                    let mut builder = StringViewBuilder::new();
+                    let iterator = [<$stat_type_prefix ByteArrayDataPageStatsIterator>]::new($iterator);
+                    for x in iterator {
+                        for x in x.into_iter() {
+                            let Some(x) = x else {
+                                builder.append_null(); // no statistics value
+                                continue;
+                            };
+
+                            let Ok(x) = std::str::from_utf8(x.data()) else {
+                                builder.append_null();
+                                continue;
+                            };
+
+                            builder.append_value(x);
+                        }
+                    }
+                    Ok(Arc::new(builder.finish()))
+                },
+                Some(DataType::BinaryView) => {
+                    let mut builder = BinaryViewBuilder::new();
+                    let iterator = [<$stat_type_prefix ByteArrayDataPageStatsIterator>]::new($iterator);
+                    for x in iterator {
+                        for x in x.into_iter() {
+                            let Some(x) = x else {
+                                builder.append_null(); // no statistics value
+                                continue;
+                            };
+
+                            builder.append_value(x);
+                        }
+                    }
+                    Ok(Arc::new(builder.finish()))
+                },
+                Some(DataType::Null) |
+                Some(DataType::Duration(_)) |
+                Some(DataType::Interval(_)) |
+                Some(DataType::List(_)) |
+                Some(DataType::ListView(_)) |
+                Some(DataType::FixedSizeList(_, _)) |
+                Some(DataType::LargeList(_)) |
+                Some(DataType::LargeListView(_)) |
+                Some(DataType::Struct(_)) |
+                Some(DataType::Union(_, _)) |
+                Some(DataType::Map(_, _)) |
+                Some(DataType::RunEndEncoded(_, _)) => {
+                    let len = $iterator.count();
+                    // don't know how to extract statistics, so return a null array
+                    Ok(new_null_array($data_type.unwrap(), len))
+                },
+                None => unimplemented!()  // not sure how to handle this
             }
         }
     }
@@ -1499,10 +1580,10 @@ mod test {
     use arrow::datatypes::{i256, Date32Type, Date64Type};
     use arrow::util::test_util::parquet_test_data;
     use arrow_array::{
-        new_empty_array, new_null_array, Array, ArrayRef, BinaryArray, BooleanArray, Date32Array,
-        Date64Array, Decimal128Array, Decimal256Array, Float32Array, Float64Array, Int16Array,
-        Int32Array, Int64Array, Int8Array, LargeBinaryArray, RecordBatch, StringArray, StructArray,
-        TimestampNanosecondArray,
+        new_empty_array, new_null_array, Array, ArrayRef, BinaryArray, BinaryViewArray,
+        BooleanArray, Date32Array, Date64Array, Decimal128Array, Decimal256Array, Float32Array,
+        Float64Array, Int16Array, Int32Array, Int64Array, Int8Array, LargeBinaryArray, RecordBatch,
+        StringArray, StringViewArray, StructArray, TimestampNanosecondArray,
     };
     use arrow_schema::{DataType, Field, SchemaRef};
     use bytes::Bytes;
@@ -1916,6 +1997,65 @@ mod test {
         .run()
     }
 
+    #[test]
+    fn roundtrip_string_view() {
+        Test {
+            input: string_view_array([
+                // row group 1
+                Some("A"),
+                None,
+                Some("Q"),
+                // row group 2
+                Some("ZZ"),
+                Some("A_longerthan12"),
+                None,
+                // row group 3
+                Some("A_longerthan12"),
+                None,
+                None,
+            ]),
+            expected_min: string_view_array([
+                Some("A"),
+                Some("A_longerthan12"),
+                Some("A_longerthan12"),
+            ]),
+            expected_max: string_view_array([Some("Q"), Some("ZZ"), Some("A_longerthan12")]),
+        }
+        .run()
+    }
+
+    #[test]
+    fn roundtrip_binary_view() {
+        let input: Vec<Option<&[u8]>> = vec![
+            // row group 1
+            Some(b"A"),
+            None,
+            Some(b"Q"),
+            // row group 2
+            Some(b"ZZ"),
+            Some(b"A_longerthan12"),
+            None,
+            // row group 3
+            Some(b"A_longerthan12"),
+            None,
+            None,
+        ];
+
+        let expected_min: Vec<Option<&[u8]>> =
+            vec![Some(b"A"), Some(b"A_longerthan12"), Some(b"A_longerthan12")];
+        let expected_max: Vec<Option<&[u8]>> =
+            vec![Some(b"Q"), Some(b"ZZ"), Some(b"A_longerthan12")];
+
+        let array = binary_view_array(input);
+
+        Test {
+            input: array,
+            expected_min: binary_view_array(expected_min),
+            expected_max: binary_view_array(expected_max),
+        }
+        .run()
+    }
+
     #[test]
     fn roundtrip_struct() {
         let mut test = Test {
@@ -2539,4 +2679,19 @@ mod test {
 
         Arc::new(array)
     }
+
+    fn string_view_array<'a>(input: impl IntoIterator<Item = Option<&'a str>>) -> ArrayRef {
+        let array: StringViewArray = input
+            .into_iter()
+            .map(|s| s.map(|s| s.to_string()))
+            .collect();
+
+        Arc::new(array)
+    }
+
+    fn binary_view_array(input: Vec<Option<&[u8]>>) -> ArrayRef {
+        let array = BinaryViewArray::from(input.into_iter().collect::<Vec<Option<&[u8]>>>());
+
+        Arc::new(array)
+    }
 }
diff --git a/parquet/src/arrow/arrow_writer/mod.rs b/parquet/src/arrow/arrow_writer/mod.rs
index 8f7b514cc..db5d4502c 100644
--- a/parquet/src/arrow/arrow_writer/mod.rs
+++ b/parquet/src/arrow/arrow_writer/mod.rs
@@ -43,7 +43,7 @@ use crate::column::writer::{
 };
 use crate::data_type::{ByteArray, FixedLenByteArray};
 use crate::errors::{ParquetError, Result};
-use crate::file::metadata::{ColumnChunkMetaData, KeyValue, RowGroupMetaData};
+use crate::file::metadata::{KeyValue, RowGroupMetaData};
 use crate::file::properties::{WriterProperties, WriterPropertiesPtr};
 use crate::file::reader::{ChunkReader, Length};
 use crate::file::writer::{SerializedFileWriter, SerializedRowGroupWriter};
@@ -489,11 +489,6 @@ impl PageWriter for ArrowPageWriter {
         Ok(spec)
     }
 
-    fn write_metadata(&mut self, _metadata: &ColumnChunkMetaData) -> Result<()> {
-        // Skip writing metadata as won't be copied anyway
-        Ok(())
-    }
-
     fn close(&mut self) -> Result<()> {
         Ok(())
     }
diff --git a/parquet/src/arrow/async_writer/mod.rs b/parquet/src/arrow/async_writer/mod.rs
index 274d8fef8..50bb5c046 100644
--- a/parquet/src/arrow/async_writer/mod.rs
+++ b/parquet/src/arrow/async_writer/mod.rs
@@ -22,6 +22,7 @@
 //! read the documentation there before using this API.
 //!
 //! Here is an example for using [`AsyncArrowWriter`]:
+//!
 //! ```
 //! # #[tokio::main(flavor="current_thread")]
 //! # async fn main() {
@@ -49,6 +50,13 @@
 //! assert_eq!(to_write, read);
 //! # }
 //! ```
+//!
+//! [`object_store`] provides it's native implementation of [`AsyncFileWriter`] by [`ParquetObjectWriter`].
+
+#[cfg(feature = "object_store")]
+mod store;
+#[cfg(feature = "object_store")]
+pub use store::*;
 
 use crate::{
     arrow::arrow_writer::ArrowWriterOptions,
@@ -65,7 +73,7 @@ use futures::FutureExt;
 use std::mem;
 use tokio::io::{AsyncWrite, AsyncWriteExt};
 
-/// The asynchronous interface used by [`AsyncArrowWriter`] to write parquet files
+/// The asynchronous interface used by [`AsyncArrowWriter`] to write parquet files.
 pub trait AsyncFileWriter: Send {
     /// Write the provided bytes to the underlying writer
     ///
diff --git a/parquet/src/arrow/async_writer/store.rs b/parquet/src/arrow/async_writer/store.rs
new file mode 100644
index 000000000..ad09eae49
--- /dev/null
+++ b/parquet/src/arrow/async_writer/store.rs
@@ -0,0 +1,157 @@
+// 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 bytes::Bytes;
+use futures::future::BoxFuture;
+use std::sync::Arc;
+
+use crate::arrow::async_writer::AsyncFileWriter;
+use crate::errors::{ParquetError, Result};
+use object_store::buffered::BufWriter;
+use object_store::path::Path;
+use object_store::ObjectStore;
+use tokio::io::AsyncWriteExt;
+
+/// [`ParquetObjectWriter`] for writing to parquet to [`ObjectStore`]
+///
+/// ```
+/// # use arrow_array::{ArrayRef, Int64Array, RecordBatch};
+/// # use object_store::memory::InMemory;
+/// # use object_store::path::Path;
+/// # use object_store::ObjectStore;
+/// # use std::sync::Arc;
+///
+/// # use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder;
+/// # use parquet::arrow::async_writer::ParquetObjectWriter;
+/// # use parquet::arrow::AsyncArrowWriter;
+///
+/// # #[tokio::main(flavor="current_thread")]
+/// # async fn main() {
+///     let store = Arc::new(InMemory::new());
+///
+///     let col = Arc::new(Int64Array::from_iter_values([1, 2, 3])) as ArrayRef;
+///     let to_write = RecordBatch::try_from_iter([("col", col)]).unwrap();
+///
+///     let object_store_writer = ParquetObjectWriter::new(store.clone(), Path::from("test"));
+///     let mut writer =
+///         AsyncArrowWriter::try_new(object_store_writer, to_write.schema(), None).unwrap();
+///     writer.write(&to_write).await.unwrap();
+///     writer.close().await.unwrap();
+///
+///     let buffer = store
+///         .get(&Path::from("test"))
+///         .await
+///         .unwrap()
+///         .bytes()
+///         .await
+///         .unwrap();
+///     let mut reader = ParquetRecordBatchReaderBuilder::try_new(buffer)
+///         .unwrap()
+///         .build()
+///         .unwrap();
+///     let read = reader.next().unwrap().unwrap();
+///
+///     assert_eq!(to_write, read);
+/// # }
+/// ```
+#[derive(Debug)]
+pub struct ParquetObjectWriter {
+    w: BufWriter,
+}
+
+impl ParquetObjectWriter {
+    /// Create a new [`ParquetObjectWriter`] that writes to the specified path in the given store.
+    ///
+    /// To configure the writer behavior, please build [`BufWriter`] and then use [`Self::from_buf_writer`]
+    pub fn new(store: Arc<dyn ObjectStore>, path: Path) -> Self {
+        Self::from_buf_writer(BufWriter::new(store, path))
+    }
+
+    /// Construct a new ParquetObjectWriter via a existing BufWriter.
+    pub fn from_buf_writer(w: BufWriter) -> Self {
+        Self { w }
+    }
+
+    /// Consume the writer and return the underlying BufWriter.
+    pub fn into_inner(self) -> BufWriter {
+        self.w
+    }
+}
+
+impl AsyncFileWriter for ParquetObjectWriter {
+    fn write(&mut self, bs: Bytes) -> BoxFuture<'_, Result<()>> {
+        Box::pin(async {
+            self.w
+                .put(bs)
+                .await
+                .map_err(|err| ParquetError::External(Box::new(err)))
+        })
+    }
+
+    fn complete(&mut self) -> BoxFuture<'_, Result<()>> {
+        Box::pin(async {
+            self.w
+                .shutdown()
+                .await
+                .map_err(|err| ParquetError::External(Box::new(err)))
+        })
+    }
+}
+impl From<BufWriter> for ParquetObjectWriter {
+    fn from(w: BufWriter) -> Self {
+        Self::from_buf_writer(w)
+    }
+}
+#[cfg(test)]
+mod tests {
+    use arrow_array::{ArrayRef, Int64Array, RecordBatch};
+    use object_store::memory::InMemory;
+    use std::sync::Arc;
+
+    use super::*;
+    use crate::arrow::arrow_reader::ParquetRecordBatchReaderBuilder;
+    use crate::arrow::AsyncArrowWriter;
+
+    #[tokio::test]
+    async fn test_async_writer() {
+        let store = Arc::new(InMemory::new());
+
+        let col = Arc::new(Int64Array::from_iter_values([1, 2, 3])) as ArrayRef;
+        let to_write = RecordBatch::try_from_iter([("col", col)]).unwrap();
+
+        let object_store_writer = ParquetObjectWriter::new(store.clone(), Path::from("test"));
+        let mut writer =
+            AsyncArrowWriter::try_new(object_store_writer, to_write.schema(), None).unwrap();
+        writer.write(&to_write).await.unwrap();
+        writer.close().await.unwrap();
+
+        let buffer = store
+            .get(&Path::from("test"))
+            .await
+            .unwrap()
+            .bytes()
+            .await
+            .unwrap();
+        let mut reader = ParquetRecordBatchReaderBuilder::try_new(buffer)
+            .unwrap()
+            .build()
+            .unwrap();
+        let read = reader.next().unwrap().unwrap();
+
+        assert_eq!(to_write, read);
+    }
+}
diff --git a/parquet/src/column/page.rs b/parquet/src/column/page.rs
index 947a633f4..585d1951c 100644
--- a/parquet/src/column/page.rs
+++ b/parquet/src/column/page.rs
@@ -21,7 +21,7 @@ use bytes::Bytes;
 
 use crate::basic::{Encoding, PageType};
 use crate::errors::{ParquetError, Result};
-use crate::file::{metadata::ColumnChunkMetaData, statistics::Statistics};
+use crate::file::statistics::Statistics;
 use crate::format::PageHeader;
 
 /// Parquet Page definition.
@@ -350,12 +350,6 @@ pub trait PageWriter: Send {
     /// either data page or dictionary page.
     fn write_page(&mut self, page: CompressedPage) -> Result<PageWriteSpec>;
 
-    /// Writes column chunk metadata into the output stream/sink.
-    ///
-    /// This method is called once before page writer is closed, normally when writes are
-    /// finalised in column writer.
-    fn write_metadata(&mut self, metadata: &ColumnChunkMetaData) -> Result<()>;
-
     /// Closes resources and flushes underlying sink.
     /// Page writer should not be used after this method is called.
     fn close(&mut self) -> Result<()>;
diff --git a/parquet/src/column/writer/mod.rs b/parquet/src/column/writer/mod.rs
index 54d8fd3cc..519a219f9 100644
--- a/parquet/src/column/writer/mod.rs
+++ b/parquet/src/column/writer/mod.rs
@@ -579,7 +579,7 @@ impl<'a, E: ColumnValueEncoder> GenericColumnWriter<'a, E> {
             self.write_dictionary_page()?;
         }
         self.flush_data_pages()?;
-        let metadata = self.write_column_metadata()?;
+        let metadata = self.build_column_metadata()?;
         self.page_writer.close()?;
 
         let boundary_order = match (
@@ -1041,8 +1041,8 @@ impl<'a, E: ColumnValueEncoder> GenericColumnWriter<'a, E> {
         Ok(())
     }
 
-    /// Assembles and writes column chunk metadata.
-    fn write_column_metadata(&mut self) -> Result<ColumnChunkMetaData> {
+    /// Assembles column chunk metadata.
+    fn build_column_metadata(&mut self) -> Result<ColumnChunkMetaData> {
         let total_compressed_size = self.column_metrics.total_compressed_size as i64;
         let total_uncompressed_size = self.column_metrics.total_uncompressed_size as i64;
         let num_values = self.column_metrics.total_num_values as i64;
@@ -1050,15 +1050,9 @@ impl<'a, E: ColumnValueEncoder> GenericColumnWriter<'a, E> {
         // If data page offset is not set, then no pages have been written
         let data_page_offset = self.column_metrics.data_page_offset.unwrap_or(0) as i64;
 
-        let file_offset = match dict_page_offset {
-            Some(dict_offset) => dict_offset + total_compressed_size,
-            None => data_page_offset + total_compressed_size,
-        };
-
         let mut builder = ColumnChunkMetaData::builder(self.descr.clone())
             .set_compression(self.codec)
             .set_encodings(self.encodings.iter().cloned().collect())
-            .set_file_offset(file_offset)
             .set_total_compressed_size(total_compressed_size)
             .set_total_uncompressed_size(total_uncompressed_size)
             .set_num_values(num_values)
@@ -1138,8 +1132,6 @@ impl<'a, E: ColumnValueEncoder> GenericColumnWriter<'a, E> {
         }
 
         let metadata = builder.build()?;
-        self.page_writer.write_metadata(&metadata)?;
-
         Ok(metadata)
     }
 
@@ -3589,10 +3581,6 @@ mod tests {
             Ok(res)
         }
 
-        fn write_metadata(&mut self, _metadata: &ColumnChunkMetaData) -> Result<()> {
-            Ok(())
-        }
-
         fn close(&mut self) -> Result<()> {
             Ok(())
         }
diff --git a/parquet/src/file/metadata/mod.rs b/parquet/src/file/metadata/mod.rs
index d99cd9510..86c673bbd 100644
--- a/parquet/src/file/metadata/mod.rs
+++ b/parquet/src/file/metadata/mod.rs
@@ -682,7 +682,12 @@ impl ColumnChunkMetaData {
         self.file_path.as_deref()
     }
 
-    /// Byte offset in `file_path()`.
+    /// Byte offset of `ColumnMetaData` in `file_path()`.
+    ///
+    /// Note that the meaning of this field has been inconsistent between implementations
+    /// so its use has since been deprecated in the Parquet specification. Modern implementations
+    /// will set this to `0` to indicate that the `ColumnMetaData` is solely contained in the
+    /// `ColumnChunk` struct.
     pub fn file_offset(&self) -> i64 {
         self.file_offset
     }
@@ -1040,6 +1045,14 @@ impl ColumnChunkMetaDataBuilder {
     }
 
     /// Sets file offset in bytes.
+    ///
+    /// This field was meant to provide an alternate to storing `ColumnMetadata` directly in
+    /// the `ColumnChunkMetadata`. However, most Parquet readers assume the `ColumnMetadata`
+    /// is stored inline and ignore this field.
+    #[deprecated(
+        since = "53.0.0",
+        note = "The Parquet specification requires this field to be 0"
+    )]
     pub fn set_file_offset(mut self, value: i64) -> Self {
         self.0.file_offset = value;
         self
@@ -1453,7 +1466,6 @@ mod tests {
         let col_metadata = ColumnChunkMetaData::builder(column_descr.clone())
             .set_encodings(vec![Encoding::PLAIN, Encoding::RLE])
             .set_file_path("file_path".to_owned())
-            .set_file_offset(100)
             .set_num_values(1000)
             .set_compression(Compression::SNAPPY)
             .set_total_compressed_size(2000)
diff --git a/parquet/src/file/writer.rs b/parquet/src/file/writer.rs
index 89aaf028d..bcf8449b4 100644
--- a/parquet/src/file/writer.rs
+++ b/parquet/src/file/writer.rs
@@ -649,13 +649,10 @@ impl<'a, W: Write + Send> SerializedRowGroupWriter<'a, W> {
             ));
         }
 
-        let file_offset = self.buf.bytes_written() as i64;
-
         let map_offset = |x| x - src_offset + write_offset as i64;
         let mut builder = ColumnChunkMetaData::builder(metadata.column_descr_ptr())
             .set_compression(metadata.compression())
             .set_encodings(metadata.encodings().clone())
-            .set_file_offset(file_offset)
             .set_total_compressed_size(metadata.compressed_size())
             .set_total_uncompressed_size(metadata.uncompressed_size())
             .set_num_values(metadata.num_values())
@@ -680,7 +677,6 @@ impl<'a, W: Write + Send> SerializedRowGroupWriter<'a, W> {
             }
         }
 
-        SerializedPageWriter::new(self.buf).write_metadata(&metadata)?;
         let (_, on_close) = self.get_on_close();
         on_close(close)
     }
@@ -808,14 +804,6 @@ impl<'a, W: Write + Send> PageWriter for SerializedPageWriter<'a, W> {
         Ok(spec)
     }
 
-    fn write_metadata(&mut self, metadata: &ColumnChunkMetaData) -> Result<()> {
-        let mut protocol = TCompactOutputProtocol::new(&mut self.sink);
-        metadata
-            .to_column_metadata_thrift()
-            .write_to_out_protocol(&mut protocol)?;
-        Ok(())
-    }
-
     fn close(&mut self) -> Result<()> {
         self.sink.flush()?;
         Ok(())


Test patch:
diff --git a/parquet/tests/arrow_reader/mod.rs b/parquet/tests/arrow_reader/mod.rs
index 4f63a5054..7e979dcf3 100644
--- a/parquet/tests/arrow_reader/mod.rs
+++ b/parquet/tests/arrow_reader/mod.rs
@@ -17,13 +17,13 @@
 
 use arrow_array::types::{Int32Type, Int8Type};
 use arrow_array::{
-    Array, ArrayRef, BinaryArray, BooleanArray, Date32Array, Date64Array, Decimal128Array,
-    Decimal256Array, DictionaryArray, FixedSizeBinaryArray, Float16Array, Float32Array,
-    Float64Array, Int16Array, Int32Array, Int64Array, Int8Array, LargeBinaryArray,
-    LargeStringArray, RecordBatch, StringArray, StructArray, Time32MillisecondArray,
-    Time32SecondArray, Time64MicrosecondArray, Time64NanosecondArray, TimestampMicrosecondArray,
-    TimestampMillisecondArray, TimestampNanosecondArray, TimestampSecondArray, UInt16Array,
-    UInt32Array, UInt64Array, UInt8Array,
+    Array, ArrayRef, BinaryArray, BinaryViewArray, BooleanArray, Date32Array, Date64Array,
+    Decimal128Array, Decimal256Array, DictionaryArray, FixedSizeBinaryArray, Float16Array,
+    Float32Array, Float64Array, Int16Array, Int32Array, Int64Array, Int8Array, LargeBinaryArray,
+    LargeStringArray, RecordBatch, StringArray, StringViewArray, StructArray,
+    Time32MillisecondArray, Time32SecondArray, Time64MicrosecondArray, Time64NanosecondArray,
+    TimestampMicrosecondArray, TimestampMillisecondArray, TimestampNanosecondArray,
+    TimestampSecondArray, UInt16Array, UInt32Array, UInt64Array, UInt8Array,
 };
 use arrow_buffer::i256;
 use arrow_schema::{DataType, Field, Schema, TimeUnit};
@@ -88,6 +88,8 @@ enum Scenario {
     PeriodsInColumnNames,
     StructArray,
     UTF8,
+    UTF8View,
+    BinaryView,
 }
 
 fn make_boolean_batch(v: Vec<Option<bool>>) -> RecordBatch {
@@ -589,6 +591,16 @@ fn make_utf8_batch(value: Vec<Option<&str>>) -> RecordBatch {
     .unwrap()
 }
 
+fn make_utf8_view_batch(value: Vec<Option<&str>>) -> RecordBatch {
+    let utf8_view = StringViewArray::from(value);
+    RecordBatch::try_from_iter(vec![("utf8_view", Arc::new(utf8_view) as _)]).unwrap()
+}
+
+fn make_binary_view_batch(value: Vec<Option<&[u8]>>) -> RecordBatch {
+    let binary_view = BinaryViewArray::from(value);
+    RecordBatch::try_from_iter(vec![("binary_view", Arc::new(binary_view) as _)]).unwrap()
+}
+
 fn make_dict_batch() -> RecordBatch {
     let values = [
         Some("abc"),
@@ -972,6 +984,35 @@ fn create_data_batch(scenario: Scenario) -> Vec<RecordBatch> {
                 make_utf8_batch(vec![Some("e"), Some("f"), Some("g"), Some("h"), Some("i")]),
             ]
         }
+        Scenario::UTF8View => {
+            // Make utf8_view batch including string length <12 and >12 bytes
+            // as the internal representation of StringView is differed for strings
+            // shorter and longer than that length
+            vec![
+                make_utf8_view_batch(vec![Some("a"), Some("b"), Some("c"), Some("d"), None]),
+                make_utf8_view_batch(vec![Some("a"), Some("e_longerthan12"), None, None, None]),
+                make_utf8_view_batch(vec![
+                    Some("e_longerthan12"),
+                    Some("f_longerthan12"),
+                    Some("g_longerthan12"),
+                    Some("h_longerthan12"),
+                    Some("i_longerthan12"),
+                ]),
+            ]
+        }
+        Scenario::BinaryView => {
+            vec![
+                make_binary_view_batch(vec![Some(b"a"), Some(b"b"), Some(b"c"), Some(b"d"), None]),
+                make_binary_view_batch(vec![Some(b"a"), Some(b"e_longerthan12"), None, None, None]),
+                make_binary_view_batch(vec![
+                    Some(b"e_longerthan12"),
+                    Some(b"f_longerthan12"),
+                    Some(b"g_longerthan12"),
+                    Some(b"h_longerthan12"),
+                    Some(b"i_longerthan12"),
+                ]),
+            ]
+        }
     }
 }
 
diff --git a/parquet/tests/arrow_reader/statistics.rs b/parquet/tests/arrow_reader/statistics.rs
index 5702967ff..75a73ac13 100644
--- a/parquet/tests/arrow_reader/statistics.rs
+++ b/parquet/tests/arrow_reader/statistics.rs
@@ -29,11 +29,11 @@ use arrow::datatypes::{
     TimestampNanosecondType, TimestampSecondType,
 };
 use arrow_array::{
-    make_array, new_null_array, Array, ArrayRef, BinaryArray, BooleanArray, Date32Array,
-    Date64Array, Decimal128Array, Decimal256Array, FixedSizeBinaryArray, Float16Array,
+    make_array, new_null_array, Array, ArrayRef, BinaryArray, BinaryViewArray, BooleanArray,
+    Date32Array, Date64Array, Decimal128Array, Decimal256Array, FixedSizeBinaryArray, Float16Array,
     Float32Array, Float64Array, Int16Array, Int32Array, Int64Array, Int8Array, LargeBinaryArray,
-    LargeStringArray, RecordBatch, StringArray, Time32MillisecondArray, Time32SecondArray,
-    Time64MicrosecondArray, Time64NanosecondArray, TimestampMicrosecondArray,
+    LargeStringArray, RecordBatch, StringArray, StringViewArray, Time32MillisecondArray,
+    Time32SecondArray, Time64MicrosecondArray, Time64NanosecondArray, TimestampMicrosecondArray,
     TimestampMillisecondArray, TimestampNanosecondArray, TimestampSecondArray, UInt16Array,
     UInt32Array, UInt64Array, UInt8Array,
 };
@@ -2059,6 +2059,60 @@ async fn test_utf8() {
     .run();
 }
 
+// UTF8View
+#[tokio::test]
+async fn test_utf8_view() {
+    let reader = TestReader {
+        scenario: Scenario::UTF8View,
+        row_per_group: 5,
+    }
+    .build()
+    .await;
+
+    // test for utf8_view
+    Test {
+        reader: &reader,
+        expected_min: Arc::new(StringViewArray::from(vec!["a", "a", "e_longerthan12"])),
+        expected_max: Arc::new(StringViewArray::from(vec![
+            "d",
+            "e_longerthan12",
+            "i_longerthan12",
+        ])),
+        expected_null_counts: UInt64Array::from(vec![1, 3, 0]),
+        expected_row_counts: Some(UInt64Array::from(vec![5, 5, 5])),
+        column_name: "utf8_view",
+        check: Check::Both,
+    }
+    .run()
+}
+
+// BinaryView
+#[tokio::test]
+async fn test_binary_view() {
+    let reader = TestReader {
+        scenario: Scenario::BinaryView,
+        row_per_group: 5,
+    }
+    .build()
+    .await;
+
+    let expected_min: Vec<Option<&[u8]>> = vec![Some(b"a"), Some(b"a"), Some(b"e_longerthan12")];
+    let expected_max: Vec<Option<&[u8]>> =
+        vec![Some(b"d"), Some(b"e_longerthan12"), Some(b"i_longerthan12")];
+
+    // test for utf8_view
+    Test {
+        reader: &reader,
+        expected_min: Arc::new(BinaryViewArray::from(expected_min)),
+        expected_max: Arc::new(BinaryViewArray::from(expected_max)),
+        expected_null_counts: UInt64Array::from(vec![1, 3, 0]),
+        expected_row_counts: Some(UInt64Array::from(vec![5, 5, 5])),
+        column_name: "binary_view",
+        check: Check::Both,
+    }
+    .run()
+}
+
 ////// Files with missing statistics ///////
 
 #[tokio::test]