Repository: apache/arrow-rs
Pull request #6026 (2024-07-09 09:31:21)

Add begin/end_transaction methods in FlightSqlServiceClient
# 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 #.

# Rationale for this change
 
<!--
Why are you proposing this change? If this is already explained clearly in the issue then this section is not needed.
Explaining clearly why changes are proposed helps reviewers understand your changes and offer better suggestions for fixes.
-->
`FlightSqlService` suppots transaction feature, so add some helper methods for client.

# What changes are included in this PR?

<!--
There is no need to duplicate the description in the issue here but it is sometimes worth providing a summary of the individual changes in this PR.
-->

# Are there any user-facing changes?


<!--
If there are user-facing changes then we may require documentation to be updated before approving the PR.
-->

<!---
If there are any breaking changes to public APIs, please add the `breaking change` label.
-->



Patch:
diff --git a/.github/workflows/object_store.yml b/.github/workflows/object_store.yml
index 422f676cb..bdbfc0bec 100644
--- a/.github/workflows/object_store.yml
+++ b/.github/workflows/object_store.yml
@@ -198,3 +198,16 @@ jobs:
         run: cargo build --target wasm32-unknown-unknown
       - name: Build wasm32-wasi
         run: cargo build --target wasm32-wasi
+
+  windows:
+    name: cargo test LocalFileSystem (win64)
+    runs-on: windows-latest
+    defaults:
+      run:
+        working-directory: object_store
+    steps:
+      - uses: actions/checkout@v4
+        with:
+          submodules: true
+      - name: Run LocalFileSystem tests
+        run: cargo test local::tests
diff --git a/arrow-array/Cargo.toml b/arrow-array/Cargo.toml
index bf6e27f6b..7c989fe4d 100644
--- a/arrow-array/Cargo.toml
+++ b/arrow-array/Cargo.toml
@@ -48,7 +48,7 @@ chrono = { workspace = true }
 chrono-tz = { version = "0.9", optional = true }
 num = { version = "0.4.1", default-features = false, features = ["std"] }
 half = { version = "2.1", default-features = false, features = ["num-traits"] }
-hashbrown = { version = "0.14", default-features = false }
+hashbrown = { version = "0.14.2", default-features = false }
 
 [features]
 ffi = ["arrow-schema/ffi", "arrow-data/ffi"]
diff --git a/arrow-array/src/array/byte_view_array.rs b/arrow-array/src/array/byte_view_array.rs
index a00bf7271..7017add49 100644
--- a/arrow-array/src/array/byte_view_array.rs
+++ b/arrow-array/src/array/byte_view_array.rs
@@ -46,7 +46,7 @@ use super::ByteArrayType;
 /// `N` elements is stored as `N` fixed length "views" and a variable number
 /// of variable length "buffers".
 ///
-/// Each view is a `u128` value  layout is different depending on the
+/// Each view is a `u128` value whose layout is different depending on the
 /// length of the string stored at that location:
 ///
 /// ```text
@@ -324,9 +324,11 @@ impl<T: ByteViewType + ?Sized> GenericByteViewArray<T> {
     /// Note that it will copy the array regardless of whether the original array is compact.
     /// Use with caution as this can be an expensive operation, only use it when you are sure that the view
     /// array is significantly smaller than when it is originally created, e.g., after filtering or slicing.
+    ///
+    /// Note: this function does not attempt to canonicalize / deduplicate values. For this
+    /// feature see  [`GenericByteViewBuilder::with_deduplicate_strings`].
     pub fn gc(&self) -> Self {
-        let mut builder =
-            GenericByteViewBuilder::<T>::with_capacity(self.len()).with_deduplicate_strings();
+        let mut builder = GenericByteViewBuilder::<T>::with_capacity(self.len());
 
         for v in self.iter() {
             builder.append_option(v);
diff --git a/arrow-array/src/array/dictionary_array.rs b/arrow-array/src/array/dictionary_array.rs
index 045917a1b..d6c5dd4c3 100644
--- a/arrow-array/src/array/dictionary_array.rs
+++ b/arrow-array/src/array/dictionary_array.rs
@@ -1025,13 +1025,13 @@ mod tests {
         let value_data = ArrayData::builder(DataType::Int8)
             .len(8)
             .add_buffer(Buffer::from(
-                &[10_i8, 11, 12, 13, 14, 15, 16, 17].to_byte_slice(),
+                [10_i8, 11, 12, 13, 14, 15, 16, 17].to_byte_slice(),
             ))
             .build()
             .unwrap();
 
         // Construct a buffer for value offsets, for the nested array:
-        let keys = Buffer::from(&[2_i16, 3, 4].to_byte_slice());
+        let keys = Buffer::from([2_i16, 3, 4].to_byte_slice());
 
         // Construct a dictionary array from the above two
         let key_type = DataType::Int16;
diff --git a/arrow-array/src/array/fixed_size_list_array.rs b/arrow-array/src/array/fixed_size_list_array.rs
index 6f3a76908..0d57d9a69 100644
--- a/arrow-array/src/array/fixed_size_list_array.rs
+++ b/arrow-array/src/array/fixed_size_list_array.rs
@@ -674,7 +674,7 @@ mod tests {
         assert_eq!(err.to_string(), "Invalid argument error: Found unmasked nulls for non-nullable FixedSizeListArray field \"item\"");
 
         // Valid as nulls in child masked by parent
-        let nulls = NullBuffer::new(BooleanBuffer::new(vec![0b0000101].into(), 0, 3));
+        let nulls = NullBuffer::new(BooleanBuffer::new(Buffer::from([0b0000101]), 0, 3));
         FixedSizeListArray::new(field, 2, values.clone(), Some(nulls));
 
         let field = Arc::new(Field::new("item", DataType::Int64, true));
diff --git a/arrow-array/src/array/map_array.rs b/arrow-array/src/array/map_array.rs
index ed22ba1d8..bddf202bd 100644
--- a/arrow-array/src/array/map_array.rs
+++ b/arrow-array/src/array/map_array.rs
@@ -448,20 +448,20 @@ mod tests {
         // Construct key and values
         let keys_data = ArrayData::builder(DataType::Int32)
             .len(8)
-            .add_buffer(Buffer::from(&[0, 1, 2, 3, 4, 5, 6, 7].to_byte_slice()))
+            .add_buffer(Buffer::from([0, 1, 2, 3, 4, 5, 6, 7].to_byte_slice()))
             .build()
             .unwrap();
         let values_data = ArrayData::builder(DataType::UInt32)
             .len(8)
             .add_buffer(Buffer::from(
-                &[0u32, 10, 20, 30, 40, 50, 60, 70].to_byte_slice(),
+                [0u32, 10, 20, 30, 40, 50, 60, 70].to_byte_slice(),
             ))
             .build()
             .unwrap();
 
         // Construct a buffer for value offsets, for the nested array:
         //  [[0, 1, 2], [3, 4, 5], [6, 7]]
-        let entry_offsets = Buffer::from(&[0, 3, 6, 8].to_byte_slice());
+        let entry_offsets = Buffer::from([0, 3, 6, 8].to_byte_slice());
 
         let keys = Arc::new(Field::new("keys", DataType::Int32, false));
         let values = Arc::new(Field::new("values", DataType::UInt32, false));
@@ -493,13 +493,13 @@ mod tests {
         // Construct key and values
         let key_data = ArrayData::builder(DataType::Int32)
             .len(8)
-            .add_buffer(Buffer::from(&[0, 1, 2, 3, 4, 5, 6, 7].to_byte_slice()))
+            .add_buffer(Buffer::from([0, 1, 2, 3, 4, 5, 6, 7].to_byte_slice()))
             .build()
             .unwrap();
         let value_data = ArrayData::builder(DataType::UInt32)
             .len(8)
             .add_buffer(Buffer::from(
-                &[0u32, 10, 20, 0, 40, 0, 60, 70].to_byte_slice(),
+                [0u32, 10, 20, 0, 40, 0, 60, 70].to_byte_slice(),
             ))
             .null_bit_buffer(Some(Buffer::from(&[0b11010110])))
             .build()
@@ -507,7 +507,7 @@ mod tests {
 
         // Construct a buffer for value offsets, for the nested array:
         //  [[0, 1, 2], [3, 4, 5], [6, 7]]
-        let entry_offsets = Buffer::from(&[0, 3, 6, 8].to_byte_slice());
+        let entry_offsets = Buffer::from([0, 3, 6, 8].to_byte_slice());
 
         let keys_field = Arc::new(Field::new("keys", DataType::Int32, false));
         let values_field = Arc::new(Field::new("values", DataType::UInt32, true));
@@ -617,18 +617,18 @@ mod tests {
         // Construct key and values
         let keys_data = ArrayData::builder(DataType::Int32)
             .len(5)
-            .add_buffer(Buffer::from(&[3, 4, 5, 6, 7].to_byte_slice()))
+            .add_buffer(Buffer::from([3, 4, 5, 6, 7].to_byte_slice()))
             .build()
             .unwrap();
         let values_data = ArrayData::builder(DataType::UInt32)
             .len(5)
-            .add_buffer(Buffer::from(&[30u32, 40, 50, 60, 70].to_byte_slice()))
+            .add_buffer(Buffer::from([30u32, 40, 50, 60, 70].to_byte_slice()))
             .build()
             .unwrap();
 
         // Construct a buffer for value offsets, for the nested array:
         //  [[3, 4, 5], [6, 7]]
-        let entry_offsets = Buffer::from(&[0, 3, 5].to_byte_slice());
+        let entry_offsets = Buffer::from([0, 3, 5].to_byte_slice());
 
         let keys = Arc::new(Field::new("keys", DataType::Int32, false));
         let values = Arc::new(Field::new("values", DataType::UInt32, false));
diff --git a/arrow-array/src/array/struct_array.rs b/arrow-array/src/array/struct_array.rs
index ae292944e..44a7f38c3 100644
--- a/arrow-array/src/array/struct_array.rs
+++ b/arrow-array/src/array/struct_array.rs
@@ -549,7 +549,7 @@ mod tests {
         let expected_string_data = ArrayData::builder(DataType::Utf8)
             .len(4)
             .null_bit_buffer(Some(Buffer::from(&[9_u8])))
-            .add_buffer(Buffer::from(&[0, 3, 3, 3, 7].to_byte_slice()))
+            .add_buffer(Buffer::from([0, 3, 3, 3, 7].to_byte_slice()))
             .add_buffer(Buffer::from(b"joemark"))
             .build()
             .unwrap();
@@ -557,7 +557,7 @@ mod tests {
         let expected_int_data = ArrayData::builder(DataType::Int32)
             .len(4)
             .null_bit_buffer(Some(Buffer::from(&[11_u8])))
-            .add_buffer(Buffer::from(&[1, 2, 0, 4].to_byte_slice()))
+            .add_buffer(Buffer::from([1, 2, 0, 4].to_byte_slice()))
             .build()
             .unwrap();
 
diff --git a/arrow-array/src/builder/generic_bytes_view_builder.rs b/arrow-array/src/builder/generic_bytes_view_builder.rs
index dda553545..587255cc6 100644
--- a/arrow-array/src/builder/generic_bytes_view_builder.rs
+++ b/arrow-array/src/builder/generic_bytes_view_builder.rs
@@ -201,7 +201,8 @@ impl<T: ByteViewType + ?Sized> GenericByteViewBuilder<T> {
 
     /// Returns the value at the given index
     /// Useful if we want to know what value has been inserted to the builder
-    fn get_value(&self, index: usize) -> &[u8] {
+    /// The index has to be smaller than `self.len()`, otherwise it will panic
+    pub fn get_value(&self, index: usize) -> &[u8] {
         let view = self.views_builder.as_slice().get(index).unwrap();
         let len = *view as u32;
         if len <= 12 {
@@ -337,6 +338,19 @@ impl<T: ByteViewType + ?Sized> GenericByteViewBuilder<T> {
     pub fn validity_slice(&self) -> Option<&[u8]> {
         self.null_buffer_builder.as_slice()
     }
+
+    /// Return the allocated size of this builder in bytes, useful for memory accounting.
+    pub fn allocated_size(&self) -> usize {
+        let views = self.views_builder.capacity() * std::mem::size_of::<u128>();
+        let null = self.null_buffer_builder.allocated_size();
+        let buffer_size = self.completed.iter().map(|b| b.capacity()).sum::<usize>();
+        let in_progress = self.in_progress.capacity();
+        let tracker = match &self.string_tracker {
+            Some((ht, _)) => ht.capacity() * std::mem::size_of::<usize>(),
+            None => 0,
+        };
+        buffer_size + in_progress + tracker + views + null
+    }
 }
 
 impl<T: ByteViewType + ?Sized> Default for GenericByteViewBuilder<T> {
diff --git a/arrow-buffer/src/buffer/immutable.rs b/arrow-buffer/src/buffer/immutable.rs
index 2c743842f..52e201ca1 100644
--- a/arrow-buffer/src/buffer/immutable.rs
+++ b/arrow-buffer/src/buffer/immutable.rs
@@ -543,7 +543,7 @@ mod tests {
 
     #[test]
     fn test_access_concurrently() {
-        let buffer = Buffer::from(vec![1, 2, 3, 4, 5]);
+        let buffer = Buffer::from([1, 2, 3, 4, 5]);
         let buffer2 = buffer.clone();
         assert_eq!([1, 2, 3, 4, 5], buffer.as_slice());
 
diff --git a/arrow-buffer/src/builder/null.rs b/arrow-buffer/src/builder/null.rs
index 55b3303c9..a1cea6ef2 100644
--- a/arrow-buffer/src/builder/null.rs
+++ b/arrow-buffer/src/builder/null.rs
@@ -161,6 +161,14 @@ impl NullBufferBuilder {
     pub fn as_slice_mut(&mut self) -> Option<&mut [u8]> {
         self.bitmap_builder.as_mut().map(|b| b.as_slice_mut())
     }
+
+    /// Return the allocated size of this builder, in bytes, useful for memory accounting.
+    pub fn allocated_size(&self) -> usize {
+        self.bitmap_builder
+            .as_ref()
+            .map(|b| b.capacity())
+            .unwrap_or(0)
+    }
 }
 
 impl NullBufferBuilder {
diff --git a/arrow-buffer/src/util/bit_chunk_iterator.rs b/arrow-buffer/src/util/bit_chunk_iterator.rs
index 9e4fb8268..440450908 100644
--- a/arrow-buffer/src/util/bit_chunk_iterator.rs
+++ b/arrow-buffer/src/util/bit_chunk_iterator.rs
@@ -456,7 +456,7 @@ mod tests {
         const ALLOC_SIZE: usize = 4 * 1024;
         let input = vec![0xFF_u8; ALLOC_SIZE];
 
-        let buffer: Buffer = Buffer::from(input);
+        let buffer: Buffer = Buffer::from_vec(input);
 
         let bitchunks = buffer.bit_chunks(57, ALLOC_SIZE * 8 - 57);
 
diff --git a/arrow-cast/src/base64.rs b/arrow-cast/src/base64.rs
index 50c742362..534b21878 100644
--- a/arrow-cast/src/base64.rs
+++ b/arrow-cast/src/base64.rs
@@ -20,7 +20,7 @@
 //! [`StringArray`]: arrow_array::StringArray
 
 use arrow_array::{Array, GenericBinaryArray, GenericStringArray, OffsetSizeTrait};
-use arrow_buffer::OffsetBuffer;
+use arrow_buffer::{Buffer, OffsetBuffer};
 use arrow_schema::ArrowError;
 use base64::encoded_len;
 use base64::engine::Config;
@@ -50,7 +50,9 @@ pub fn b64_encode<E: Engine, O: OffsetSizeTrait>(
     assert_eq!(offset, buffer_len);
 
     // Safety: Base64 is valid UTF-8
-    unsafe { GenericStringArray::new_unchecked(offsets, buffer.into(), array.nulls().cloned()) }
+    unsafe {
+        GenericStringArray::new_unchecked(offsets, Buffer::from_vec(buffer), array.nulls().cloned())
+    }
 }
 
 /// Base64 decode each element of `array` with the provided [`Engine`]
@@ -79,7 +81,7 @@ pub fn b64_decode<E: Engine, O: OffsetSizeTrait>(
 
     Ok(GenericBinaryArray::new(
         offsets,
-        buffer.into(),
+        Buffer::from_vec(buffer),
         array.nulls().cloned(),
     ))
 }
diff --git a/arrow-data/src/data.rs b/arrow-data/src/data.rs
index b6ff4518e..d6d6b90b6 100644
--- a/arrow-data/src/data.rs
+++ b/arrow-data/src/data.rs
@@ -1959,7 +1959,7 @@ mod tests {
             .len(20)
             .offset(5)
             .add_buffer(b1)
-            .null_bit_buffer(Some(Buffer::from(vec![
+            .null_bit_buffer(Some(Buffer::from([
                 0b01011111, 0b10110101, 0b01100011, 0b00011110,
             ])))
             .build()
@@ -2164,7 +2164,7 @@ mod tests {
 
     #[test]
     fn test_count_nulls() {
-        let buffer = Buffer::from(vec![0b00010110, 0b10011111]);
+        let buffer = Buffer::from([0b00010110, 0b10011111]);
         let buffer = NullBuffer::new(BooleanBuffer::new(buffer, 0, 16));
         let count = count_nulls(Some(&buffer), 0, 16);
         assert_eq!(count, 7);
diff --git a/arrow-flight/Cargo.toml b/arrow-flight/Cargo.toml
index 60ba3ae82..111bf94d8 100644
--- a/arrow-flight/Cargo.toml
+++ b/arrow-flight/Cargo.toml
@@ -76,6 +76,7 @@ pin-project-lite = "0.2"
 tempfile = "3.3"
 tokio-stream = { version = "0.1", features = ["net"] }
 tower = "0.4.13"
+uuid = { version = "1.10.0", features = ["v4"] }
 
 [[example]]
 name = "flight_sql_server"
diff --git a/arrow-flight/src/sql/client.rs b/arrow-flight/src/sql/client.rs
index df5c17676..91790898b 100644
--- a/arrow-flight/src/sql/client.rs
+++ b/arrow-flight/src/sql/client.rs
@@ -28,15 +28,19 @@ use crate::decode::FlightRecordBatchStream;
 use crate::encode::FlightDataEncoderBuilder;
 use crate::error::FlightError;
 use crate::flight_service_client::FlightServiceClient;
-use crate::sql::server::{CLOSE_PREPARED_STATEMENT, CREATE_PREPARED_STATEMENT};
+use crate::sql::gen::action_end_transaction_request::EndTransaction;
+use crate::sql::server::{
+    BEGIN_TRANSACTION, CLOSE_PREPARED_STATEMENT, CREATE_PREPARED_STATEMENT, END_TRANSACTION,
+};
 use crate::sql::{
+    ActionBeginTransactionRequest, ActionBeginTransactionResult,
     ActionClosePreparedStatementRequest, ActionCreatePreparedStatementRequest,
-    ActionCreatePreparedStatementResult, Any, CommandGetCatalogs, CommandGetCrossReference,
-    CommandGetDbSchemas, CommandGetExportedKeys, CommandGetImportedKeys, CommandGetPrimaryKeys,
-    CommandGetSqlInfo, CommandGetTableTypes, CommandGetTables, CommandGetXdbcTypeInfo,
-    CommandPreparedStatementQuery, CommandPreparedStatementUpdate, CommandStatementQuery,
-    CommandStatementUpdate, DoPutPreparedStatementResult, DoPutUpdateResult, ProstMessageExt,
-    SqlInfo,
+    ActionCreatePreparedStatementResult, ActionEndTransactionRequest, Any, CommandGetCatalogs,
+    CommandGetCrossReference, CommandGetDbSchemas, CommandGetExportedKeys, CommandGetImportedKeys,
+    CommandGetPrimaryKeys, CommandGetSqlInfo, CommandGetTableTypes, CommandGetTables,
+    CommandGetXdbcTypeInfo, CommandPreparedStatementQuery, CommandPreparedStatementUpdate,
+    CommandStatementQuery, CommandStatementUpdate, DoPutPreparedStatementResult, DoPutUpdateResult,
+    ProstMessageExt, SqlInfo,
 };
 use crate::trailers::extract_lazy_trailers;
 use crate::{
@@ -399,6 +403,54 @@ impl FlightSqlServiceClient<Channel> {
         ))
     }
 
+    /// Request to begin a transaction.
+    pub async fn begin_transaction(&mut self) -> Result<Bytes, ArrowError> {
+        let cmd = ActionBeginTransactionRequest {};
+        let action = Action {
+            r#type: BEGIN_TRANSACTION.to_string(),
+            body: cmd.as_any().encode_to_vec().into(),
+        };
+        let req = self.set_request_headers(action.into_request())?;
+        let mut result = self
+            .flight_client
+            .do_action(req)
+            .await
+            .map_err(status_to_arrow_error)?
+            .into_inner();
+        let result = result
+            .message()
+            .await
+            .map_err(status_to_arrow_error)?
+            .unwrap();
+        let any = Any::decode(&*result.body).map_err(decode_error_to_arrow_error)?;
+        let begin_result: ActionBeginTransactionResult = any.unpack()?.unwrap();
+        Ok(begin_result.transaction_id)
+    }
+
+    /// Request to commit/rollback a transaction.
+    pub async fn end_transaction(
+        &mut self,
+        transaction_id: Bytes,
+        action: EndTransaction,
+    ) -> Result<(), ArrowError> {
+        let cmd = ActionEndTransactionRequest {
+            transaction_id,
+            action: action as i32,
+        };
+        let action = Action {
+            r#type: END_TRANSACTION.to_string(),
+            body: cmd.as_any().encode_to_vec().into(),
+        };
+        let req = self.set_request_headers(action.into_request())?;
+        let _ = self
+            .flight_client
+            .do_action(req)
+            .await
+            .map_err(status_to_arrow_error)?
+            .into_inner();
+        Ok(())
+    }
+
     /// Explicitly shut down and clean up the client.
     pub async fn close(&mut self) -> Result<(), ArrowError> {
         // TODO: consume self instead of &mut self to explicitly prevent reuse?
diff --git a/arrow-flight/src/sql/mod.rs b/arrow-flight/src/sql/mod.rs
index b3b9882ee..61eb67b69 100644
--- a/arrow-flight/src/sql/mod.rs
+++ b/arrow-flight/src/sql/mod.rs
@@ -49,6 +49,7 @@ mod gen {
     include!("arrow.flight.protocol.sql.rs");
 }
 
+pub use gen::action_end_transaction_request::EndTransaction;
 pub use gen::ActionBeginSavepointRequest;
 pub use gen::ActionBeginSavepointResult;
 pub use gen::ActionBeginTransactionRequest;
diff --git a/arrow-flight/src/utils.rs b/arrow-flight/src/utils.rs
index c1e2d61fc..37d7ff9e7 100644
--- a/arrow-flight/src/utils.rs
+++ b/arrow-flight/src/utils.rs
@@ -94,7 +94,7 @@ pub fn flight_data_to_arrow_batch(
         })
         .map(|batch| {
             reader::read_record_batch(
-                &Buffer::from(&data.data_body),
+                &Buffer::from(data.data_body.as_ref()),
                 batch,
                 schema,
                 dictionaries_by_id,
diff --git a/arrow-ipc/src/compression.rs b/arrow-ipc/src/compression.rs
index 0d8b7b4c1..47ea7785c 100644
--- a/arrow-ipc/src/compression.rs
+++ b/arrow-ipc/src/compression.rs
@@ -103,8 +103,8 @@ impl CompressionCodec {
         } else if let Ok(decompressed_length) = usize::try_from(decompressed_length) {
             // decompress data using the codec
             let input_data = &input[(LENGTH_OF_PREFIX_DATA as usize)..];
-            self.decompress(input_data, decompressed_length as _)?
-                .into()
+            let v = self.decompress(input_data, decompressed_length as _)?;
+            Buffer::from_vec(v)
         } else {
             return Err(ArrowError::IpcError(format!(
                 "Invalid uncompressed length: {decompressed_length}"
diff --git a/arrow-json/src/reader/mod.rs b/arrow-json/src/reader/mod.rs
index 3e1c5d2fc..97d9c8962 100644
--- a/arrow-json/src/reader/mod.rs
+++ b/arrow-json/src/reader/mod.rs
@@ -1850,7 +1850,7 @@ mod tests {
         let c = ArrayDataBuilder::new(c_field.data_type().clone())
             .len(7)
             .add_child_data(d.to_data())
-            .null_bit_buffer(Some(Buffer::from(vec![0b00111011])))
+            .null_bit_buffer(Some(Buffer::from([0b00111011])))
             .build()
             .unwrap();
         let b = BooleanArray::from(vec![
@@ -1866,14 +1866,14 @@ mod tests {
             .len(7)
             .add_child_data(b.to_data())
             .add_child_data(c.clone())
-            .null_bit_buffer(Some(Buffer::from(vec![0b00111111])))
+            .null_bit_buffer(Some(Buffer::from([0b00111111])))
             .build()
             .unwrap();
         let a_list = ArrayDataBuilder::new(a_field.data_type().clone())
             .len(6)
             .add_buffer(Buffer::from_slice_ref([0i32, 2, 3, 6, 6, 6, 7]))
             .add_child_data(a)
-            .null_bit_buffer(Some(Buffer::from(vec![0b00110111])))
+            .null_bit_buffer(Some(Buffer::from([0b00110111])))
             .build()
             .unwrap();
         let expected = make_array(a_list);
diff --git a/arrow-json/src/writer.rs b/arrow-json/src/writer.rs
index ef4141d7a..86d2e88d9 100644
--- a/arrow-json/src/writer.rs
+++ b/arrow-json/src/writer.rs
@@ -927,12 +927,12 @@ mod tests {
 
         let a_values = StringArray::from(vec!["a", "a1", "b", "c", "d", "e"]);
         // list column rows: ["a", "a1"], ["b"], ["c"], ["d"], ["e"]
-        let a_value_offsets = Buffer::from(&[0, 2, 3, 4, 5, 6].to_byte_slice());
+        let a_value_offsets = Buffer::from([0, 2, 3, 4, 5, 6].to_byte_slice());
         let a_list_data = ArrayData::builder(field_c1.data_type().clone())
             .len(5)
             .add_buffer(a_value_offsets)
             .add_child_data(a_values.into_data())
-            .null_bit_buffer(Some(Buffer::from(vec![0b00011111])))
+            .null_bit_buffer(Some(Buffer::from([0b00011111])))
             .build()
             .unwrap();
         let a = ListArray::from(a_list_data);
@@ -976,17 +976,17 @@ mod tests {
         // list column rows: [[1, 2], [3]], [], [[4, 5, 6]]
         let a_values = Int32Array::from(vec![1, 2, 3, 4, 5, 6]);
 
-        let a_value_offsets = Buffer::from(&[0, 2, 3, 6].to_byte_slice());
+        let a_value_offsets = Buffer::from([0, 2, 3, 6].to_byte_slice());
         // Construct a list array from the above two
         let a_list_data = ArrayData::builder(list_inner_type.data_type().clone())
             .len(3)
             .add_buffer(a_value_offsets)
-            .null_bit_buffer(Some(Buffer::from(vec![0b00000111])))
+            .null_bit_buffer(Some(Buffer::from([0b00000111])))
             .add_child_data(a_values.into_data())
             .build()
             .unwrap();
 
-        let c1_value_offsets = Buffer::from(&[0, 2, 2, 3].to_byte_slice());
+        let c1_value_offsets = Buffer::from([0, 2, 2, 3].to_byte_slice());
         let c1_list_data = ArrayData::builder(field_c1.data_type().clone())
             .len(3)
             .add_buffer(c1_value_offsets)
@@ -1058,12 +1058,12 @@ mod tests {
         // [{"c11": 1, "c12": {"c121": "e"}}, {"c12": {"c121": "f"}}],
         // null,
         // [{"c11": 5, "c12": {"c121": "g"}}]
-        let c1_value_offsets = Buffer::from(&[0, 2, 2, 3].to_byte_slice());
+        let c1_value_offsets = Buffer::from([0, 2, 2, 3].to_byte_slice());
         let c1_list_data = ArrayData::builder(field_c1.data_type().clone())
             .len(3)
             .add_buffer(c1_value_offsets)
             .add_child_data(struct_values.into_data())
-            .null_bit_buffer(Some(Buffer::from(vec![0b00000101])))
+            .null_bit_buffer(Some(Buffer::from([0b00000101])))
             .build()
             .unwrap();
         let c1 = ListArray::from(c1_list_data);
@@ -1225,7 +1225,7 @@ mod tests {
         );
 
         // [{"foo": 10}, null, {}, {"bar": 20, "baz": 30, "qux": 40}, {"quux": 50}, {}]
-        let entry_offsets = Buffer::from(&[0, 1, 1, 1, 4, 5, 5].to_byte_slice());
+        let entry_offsets = Buffer::from([0, 1, 1, 1, 4, 5, 5].to_byte_slice());
         let valid_buffer = Buffer::from([0b00111101]);
 
         let map_data = ArrayData::builder(map_data_type.clone())
@@ -1408,7 +1408,7 @@ mod tests {
             );
 
             // [{"list":[{"int32":1,"utf8":"a"},{"int32":null,"utf8":"b"}]},{"list":null},{"list":[{int32":5,"utf8":null}]},{"list":null}]
-            let entry_offsets = Buffer::from(&[0, 2, 2, 3, 3].to_byte_slice());
+            let entry_offsets = Buffer::from([0, 2, 2, 3, 3].to_byte_slice());
             let data = ArrayData::builder(field.data_type().clone())
                 .len(4)
                 .add_buffer(entry_offsets)
diff --git a/arrow-row/Cargo.toml b/arrow-row/Cargo.toml
index dcf624f8b..3754afb4d 100644
--- a/arrow-row/Cargo.toml
+++ b/arrow-row/Cargo.toml
@@ -46,7 +46,6 @@ arrow-data = { workspace = true }
 arrow-schema = { workspace = true }
 
 half = { version = "2.1", default-features = false }
-hashbrown = { version = "0.14", default-features = false }
 
 [dev-dependencies]
 arrow-cast = { workspace = true }
diff --git a/arrow-row/src/variable.rs b/arrow-row/src/variable.rs
index c5aa7d8ac..4d4bcddc0 100644
--- a/arrow-row/src/variable.rs
+++ b/arrow-row/src/variable.rs
@@ -22,6 +22,7 @@ use arrow_buffer::bit_util::ceil;
 use arrow_buffer::MutableBuffer;
 use arrow_data::ArrayDataBuilder;
 use arrow_schema::{DataType, SortOptions};
+use builder::make_view;
 
 /// The block size of the variable length encoding
 pub const BLOCK_SIZE: usize = 32;
@@ -152,6 +153,8 @@ fn encode_blocks<const SIZE: usize>(out: &mut [u8], val: &[u8]) -> usize {
     end_offset
 }
 
+/// Decodes a single block of data
+/// The `f` function accepts a slice of the decoded data, it may be called multiple times
 pub fn decode_blocks(row: &[u8], options: SortOptions, mut f: impl FnMut(&[u8])) -> usize {
     let (non_empty_sentinel, continuation) = match options.descending {
         true => (!NON_EMPTY_SENTINEL, !BLOCK_CONTINUATION),
@@ -243,6 +246,69 @@ pub fn decode_binary<I: OffsetSizeTrait>(
     unsafe { GenericBinaryArray::from(builder.build_unchecked()) }
 }
 
+fn decode_binary_view_inner(
+    rows: &mut [&[u8]],
+    options: SortOptions,
+    check_utf8: bool,
+) -> BinaryViewArray {
+    let len = rows.len();
+
+    let mut null_count = 0;
+
+    let nulls = MutableBuffer::collect_bool(len, |x| {
+        let valid = rows[x][0] != null_sentinel(options);
+        null_count += !valid as usize;
+        valid
+    });
+
+    let values_capacity: usize = rows.iter().map(|row| decoded_len(row, options)).sum();
+    let mut values = MutableBuffer::new(values_capacity);
+    let mut views = BufferBuilder::<u128>::new(len);
+
+    for row in rows {
+        let start_offset = values.len();
+        let offset = decode_blocks(row, options, |b| values.extend_from_slice(b));
+        if row[0] == null_sentinel(options) {
+            debug_assert_eq!(offset, 1);
+            debug_assert_eq!(start_offset, values.len());
+            views.append(0);
+        } else {
+            // Safety: we just appended the data to the end of the buffer
+            let val = unsafe { values.get_unchecked_mut(start_offset..) };
+
+            if options.descending {
+                val.iter_mut().for_each(|o| *o = !*o);
+            }
+
+            let view = make_view(val, 0, start_offset as u32);
+            views.append(view);
+        }
+        *row = &row[offset..];
+    }
+
+    if check_utf8 {
+        // the values contains all data, no matter if it is short or long
+        // we can validate utf8 in one go.
+        std::str::from_utf8(values.as_slice()).unwrap();
+    }
+
+    let builder = ArrayDataBuilder::new(DataType::BinaryView)
+        .len(len)
+        .null_count(null_count)
+        .null_bit_buffer(Some(nulls.into()))
+        .add_buffer(views.finish())
+        .add_buffer(values.into());
+
+    // SAFETY:
+    // Valid by construction above
+    unsafe { BinaryViewArray::from(builder.build_unchecked()) }
+}
+
+/// Decodes a binary view array from `rows` with the provided `options`
+pub fn decode_binary_view(rows: &mut [&[u8]], options: SortOptions) -> BinaryViewArray {
+    decode_binary_view_inner(rows, options, false)
+}
+
 /// Decodes a string array from `rows` with the provided `options`
 ///
 /// # Safety
@@ -269,16 +335,6 @@ pub unsafe fn decode_string<I: OffsetSizeTrait>(
     GenericStringArray::from(builder.build_unchecked())
 }
 
-/// Decodes a binary view array from `rows` with the provided `options`
-pub fn decode_binary_view(rows: &mut [&[u8]], options: SortOptions) -> BinaryViewArray {
-    let decoded: GenericBinaryArray<i64> = decode_binary(rows, options);
-
-    // Better performance might be to directly build the binary view instead of building to BinaryArray and then casting
-    // I suspect that the overhead is not a big deal.
-    // If it is, we can reimplement the `decode_binary_view` function to directly build the StringViewArray
-    BinaryViewArray::from(&decoded)
-}
-
 /// Decodes a string view array from `rows` with the provided `options`
 ///
 /// # Safety
@@ -289,9 +345,6 @@ pub unsafe fn decode_string_view(
     options: SortOptions,
     validate_utf8: bool,
 ) -> StringViewArray {
-    let decoded: GenericStringArray<i64> = decode_string(rows, options, validate_utf8);
-    // Better performance might be to directly build the string view instead of building to StringArray and then casting
-    // I suspect that the overhead is not a big deal.
-    // If it is, we can reimplement the `decode_string_view` function to directly build the StringViewArray
-    StringViewArray::from(&decoded)
+    let view = decode_binary_view_inner(rows, options, validate_utf8);
+    view.to_string_view_unchecked()
 }
diff --git a/arrow-select/src/dictionary.rs b/arrow-select/src/dictionary.rs
index d0b6fcfc3..2a532600b 100644
--- a/arrow-select/src/dictionary.rs
+++ b/arrow-select/src/dictionary.rs
@@ -297,7 +297,7 @@ mod tests {
 
     #[test]
     fn test_merge_nulls() {
-        let buffer = Buffer::from("helloworldbingohelloworld");
+        let buffer = Buffer::from(b"helloworldbingohelloworld");
         let offsets = OffsetBuffer::from_lengths([5, 5, 5, 5, 5]);
         let nulls = NullBuffer::from(vec![true, false, true, true, true]);
         let values = StringArray::new(offsets, buffer, Some(nulls));
diff --git a/arrow-string/src/substring.rs b/arrow-string/src/substring.rs
index f5fe81103..fc2f6c8ad 100644
--- a/arrow-string/src/substring.rs
+++ b/arrow-string/src/substring.rs
@@ -732,7 +732,7 @@ mod tests {
     }
 
     fn generic_string_with_non_zero_offset<O: OffsetSizeTrait>() {
-        let values = "hellotherearrow";
+        let values = b"hellotherearrow";
         let offsets = &[
             O::zero(),
             O::from_usize(5).unwrap(),
@@ -867,7 +867,7 @@ mod tests {
         let data = ArrayData::builder(GenericStringArray::<O>::DATA_TYPE)
             .len(2)
             .add_buffer(Buffer::from_slice_ref(offsets))
-            .add_buffer(Buffer::from(values))
+            .add_buffer(Buffer::from(values.as_bytes()))
             .null_bit_buffer(Some(Buffer::from(bitmap)))
             .offset(1)
             .build()
diff --git a/arrow/benches/row_format.rs b/arrow/benches/row_format.rs
index b5298cbe3..0fb63b5b3 100644
--- a/arrow/benches/row_format.rs
+++ b/arrow/benches/row_format.rs
@@ -24,7 +24,7 @@ use arrow::datatypes::{Int64Type, UInt64Type};
 use arrow::row::{RowConverter, SortField};
 use arrow::util::bench_util::{
     create_boolean_array, create_dict_from_values, create_primitive_array,
-    create_string_array_with_len, create_string_dict_array,
+    create_string_array_with_len, create_string_dict_array, create_string_view_array_with_len,
 };
 use arrow_array::types::Int32Type;
 use arrow_array::Array;
@@ -87,6 +87,18 @@ fn row_bench(c: &mut Criterion) {
     let cols = vec![Arc::new(create_string_array_with_len::<i32>(4096, 0.5, 100)) as ArrayRef];
     do_bench(c, "4096 string(100, 0.5)", cols);
 
+    let cols = vec![Arc::new(create_string_view_array_with_len(4096, 0., 10, false)) as ArrayRef];
+    do_bench(c, "4096 string view(10, 0)", cols);
+
+    let cols = vec![Arc::new(create_string_view_array_with_len(4096, 0., 30, false)) as ArrayRef];
+    do_bench(c, "4096 string view(30, 0)", cols);
+
+    let cols = vec![Arc::new(create_string_view_array_with_len(40960, 0., 100, false)) as ArrayRef];
+    do_bench(c, "40960 string view(100, 0)", cols);
+
+    let cols = vec![Arc::new(create_string_view_array_with_len(4096, 0.5, 100, false)) as ArrayRef];
+    do_bench(c, "4096 string view(100, 0.5)", cols);
+
     let cols = vec![Arc::new(create_string_dict_array::<Int32Type>(4096, 0., 10)) as ArrayRef];
     do_bench(c, "4096 string_dictionary(10, 0)", cols);
 
diff --git a/arrow/examples/builders.rs b/arrow/examples/builders.rs
index ad6b87964..5c8cd51c5 100644
--- a/arrow/examples/builders.rs
+++ b/arrow/examples/builders.rs
@@ -88,13 +88,13 @@ fn main() {
     // buffer.
     let value_data = ArrayData::builder(DataType::Int32)
         .len(8)
-        .add_buffer(Buffer::from(&[0, 1, 2, 3, 4, 5, 6, 7].to_byte_slice()))
+        .add_buffer(Buffer::from([0, 1, 2, 3, 4, 5, 6, 7].to_byte_slice()))
         .build()
         .unwrap();
 
     // Construct a buffer for value offsets, for the nested array:
     //  [[0, 1, 2], [3, 4, 5], [6, 7]]
-    let value_offsets = Buffer::from(&[0, 3, 6, 8].to_byte_slice());
+    let value_offsets = Buffer::from([0, 3, 6, 8].to_byte_slice());
 
     // Construct a list array from the above two
     let list_data_type = DataType::List(Arc::new(Field::new("item", DataType::Int32, false)));
diff --git a/arrow/examples/tensor_builder.rs b/arrow/examples/tensor_builder.rs
index 90ad1b486..45440533d 100644
--- a/arrow/examples/tensor_builder.rs
+++ b/arrow/examples/tensor_builder.rs
@@ -57,7 +57,7 @@ fn main() -> Result<()> {
 
     // In order to build a tensor from an array the function to_byte_slice add the
     // required padding to the elements in the array.
-    let buf = Buffer::from(&[0, 1, 2, 3, 4, 5, 6, 7, 9, 10].to_byte_slice());
+    let buf = Buffer::from([0, 1, 2, 3, 4, 5, 6, 7, 9, 10].to_byte_slice());
     let tensor = Int32Tensor::try_new(buf, Some(vec![2, 5]), None, None)?;
     println!("\nInt32 Tensor");
     println!("{tensor:?}");
diff --git a/object_store/Cargo.toml b/object_store/Cargo.toml
index 5a0df233c..3f7b2c0b2 100644
--- a/object_store/Cargo.toml
+++ b/object_store/Cargo.toml
@@ -46,7 +46,7 @@ walkdir = "2"
 # Cloud storage support
 base64 = { version = "0.22", default-features = false, features = ["std"], optional = true }
 hyper = { version = "1.2", default-features = false, optional = true }
-quick-xml = { version = "0.35.0", features = ["serialize", "overlapped-lists"], optional = true }
+quick-xml = { version = "0.36.0", features = ["serialize", "overlapped-lists"], optional = true }
 serde = { version = "1.0", default-features = false, features = ["derive"], optional = true }
 serde_json = { version = "1.0", default-features = false, optional = true }
 rand = { version = "0.8", default-features = false, features = ["std", "std_rng"], optional = true }
diff --git a/object_store/src/local.rs b/object_store/src/local.rs
index 484738929..db4b4b050 100644
--- a/object_store/src/local.rs
+++ b/object_store/src/local.rs
@@ -297,7 +297,22 @@ impl LocalFileSystem {
                 path: location.as_ref()
             }
         );
-        self.config.prefix_to_filesystem(location)
+        let path = self.config.prefix_to_filesystem(location)?;
+
+        #[cfg(target_os = "windows")]
+        let path = {
+            let path = path.to_string_lossy();
+
+            // Assume the first char is the drive letter and the next is a colon.
+            let mut out = String::new();
+            let drive = &path[..2]; // The drive letter and colon (e.g., "C:")
+            let filepath = &path[2..].replace(':', "%3A"); // Replace subsequent colons
+            out.push_str(drive);
+            out.push_str(filepath);
+            PathBuf::from(out)
+        };
+
+        Ok(path)
     }
 
     /// Enable automatic cleanup of empty directories when deleting files
@@ -1053,6 +1068,7 @@ mod tests {
     use super::*;
 
     #[tokio::test]
+    #[cfg(target_family = "unix")]
     async fn file_test() {
         let root = TempDir::new().unwrap();
         let integration = LocalFileSystem::new_with_prefix(root.path()).unwrap();
@@ -1069,6 +1085,7 @@ mod tests {
     }
 
     #[test]
+    #[cfg(target_family = "unix")]
     fn test_non_tokio() {
         let root = TempDir::new().unwrap();
         let integration = LocalFileSystem::new_with_prefix(root.path()).unwrap();
@@ -1481,6 +1498,28 @@ mod tests {
         assert_eq!(list, vec![c, a]);
     }
 
+    #[tokio::test]
+    #[cfg(target_os = "windows")]
+    async fn filesystem_filename_with_colon() {
+        let root = TempDir::new().unwrap();
+        let integration = LocalFileSystem::new_with_prefix(root.path()).unwrap();
+        let path = Path::parse("file%3Aname.parquet").unwrap();
+        let location = Path::parse("file:name.parquet").unwrap();
+
+        integration.put(&location, "test".into()).await.unwrap();
+        let list = flatten_list_stream(&integration, None).await.unwrap();
+        assert_eq!(list, vec![path.clone()]);
+
+        let result = integration
+            .get(&location)
+            .await
+            .unwrap()
+            .bytes()
+            .await
+            .unwrap();
+        assert_eq!(result, Bytes::from("test"));
+    }
+
     #[tokio::test]
     async fn delete_dirs_automatically() {
         let root = TempDir::new().unwrap();
diff --git a/parquet/src/arrow/array_reader/byte_array.rs b/parquet/src/arrow/array_reader/byte_array.rs
index 80c448edc..ed5961586 100644
--- a/parquet/src/arrow/array_reader/byte_array.rs
+++ b/parquet/src/arrow/array_reader/byte_array.rs
@@ -599,7 +599,7 @@ mod tests {
     use super::*;
     use crate::arrow::array_reader::test_util::{byte_array_all_encodings, utf8_column};
     use crate::arrow::record_reader::buffer::ValuesBuffer;
-    use arrow_array::{Array, StringArray, StringViewArray};
+    use arrow_array::{Array, StringArray};
     use arrow_buffer::Buffer;
 
     #[test]
@@ -657,64 +657,6 @@ mod tests {
         }
     }
 
-    #[test]
-    fn test_byte_array_string_view_decoder() {
-        let (pages, encoded_dictionary) =
-            byte_array_all_encodings(vec!["hello", "world", "large payload over 12 bytes", "b"]);
-
-        let column_desc = utf8_column();
-        let mut decoder = ByteArrayColumnValueDecoder::new(&column_desc);
-
-        decoder
-            .set_dict(encoded_dictionary, 4, Encoding::RLE_DICTIONARY, false)
-            .unwrap();
-
-        for (encoding, page) in pages {
-            let mut output = OffsetBuffer::<i32>::default();
-            decoder.set_data(encoding, page, 4, Some(4)).unwrap();
-
-            assert_eq!(decoder.read(&mut output, 1).unwrap(), 1);
-
-            assert_eq!(output.values.as_slice(), "hello".as_bytes());
-            assert_eq!(output.offsets.as_slice(), &[0, 5]);
-
-            assert_eq!(decoder.read(&mut output, 1).unwrap(), 1);
-            assert_eq!(output.values.as_slice(), "helloworld".as_bytes());
-            assert_eq!(output.offsets.as_slice(), &[0, 5, 10]);
-
-            assert_eq!(decoder.read(&mut output, 2).unwrap(), 2);
-            assert_eq!(
-                output.values.as_slice(),
-                "helloworldlarge payload over 12 bytesb".as_bytes()
-            );
-            assert_eq!(output.offsets.as_slice(), &[0, 5, 10, 37, 38]);
-
-            assert_eq!(decoder.read(&mut output, 4).unwrap(), 0);
-
-            let valid = [false, false, true, true, false, true, true, false, false];
-            let valid_buffer = Buffer::from_iter(valid.iter().cloned());
-
-            output.pad_nulls(0, 4, valid.len(), valid_buffer.as_slice());
-            let array = output.into_array(Some(valid_buffer), ArrowType::Utf8View);
-            let strings = array.as_any().downcast_ref::<StringViewArray>().unwrap();
-
-            assert_eq!(
-                strings.iter().collect::<Vec<_>>(),
-                vec![
-                    None,
-                    None,
-                    Some("hello"),
-                    Some("world"),
-                    None,
-                    Some("large payload over 12 bytes"),
-                    Some("b"),
-                    None,
-                    None,
-                ]
-            );
-        }
-    }
-
     #[test]
     fn test_byte_array_decoder_skip() {
         let (pages, encoded_dictionary) =
@@ -759,60 +701,6 @@ mod tests {
         }
     }
 
-    #[test]
-    fn test_byte_array_string_view_decoder_skip() {
-        let (pages, encoded_dictionary) =
-            byte_array_all_encodings(vec!["hello", "world", "a", "large payload over 12 bytes"]);
-
-        let column_desc = utf8_column();
-        let mut decoder = ByteArrayColumnValueDecoder::new(&column_desc);
-
-        decoder
-            .set_dict(encoded_dictionary, 4, Encoding::RLE_DICTIONARY, false)
-            .unwrap();
-
-        for (encoding, page) in pages {
-            let mut output = OffsetBuffer::<i32>::default();
-            decoder.set_data(encoding, page, 4, Some(4)).unwrap();
-
-            assert_eq!(decoder.read(&mut output, 1).unwrap(), 1);
-
-            assert_eq!(output.values.as_slice(), "hello".as_bytes());
-            assert_eq!(output.offsets.as_slice(), &[0, 5]);
-
-            assert_eq!(decoder.skip_values(1).unwrap(), 1);
-            assert_eq!(decoder.skip_values(1).unwrap(), 1);
-
-            assert_eq!(decoder.read(&mut output, 1).unwrap(), 1);
-            assert_eq!(
-                output.values.as_slice(),
-                "hellolarge payload over 12 bytes".as_bytes()
-            );
-            assert_eq!(output.offsets.as_slice(), &[0, 5, 32]);
-
-            assert_eq!(decoder.read(&mut output, 4).unwrap(), 0);
-
-            let valid = [false, false, true, true, false, false];
-            let valid_buffer = Buffer::from_iter(valid.iter().cloned());
-
-            output.pad_nulls(0, 2, valid.len(), valid_buffer.as_slice());
-            let array = output.into_array(Some(valid_buffer), ArrowType::Utf8View);
-            let strings = array.as_any().downcast_ref::<StringViewArray>().unwrap();
-
-            assert_eq!(
-                strings.iter().collect::<Vec<_>>(),
-                vec![
-                    None,
-                    None,
-                    Some("hello"),
-                    Some("large payload over 12 bytes"),
-                    None,
-                    None,
-                ]
-            );
-        }
-    }
-
     #[test]
     fn test_byte_array_decoder_nulls() {
         let (pages, encoded_dictionary) = byte_array_all_encodings(Vec::<&str>::new());
diff --git a/parquet/src/arrow/array_reader/byte_view_array.rs b/parquet/src/arrow/array_reader/byte_view_array.rs
index dc4ce3f9c..5845e2c08 100644
--- a/parquet/src/arrow/array_reader/byte_view_array.rs
+++ b/parquet/src/arrow/array_reader/byte_view_array.rs
@@ -28,6 +28,7 @@ use crate::encodings::decoding::{Decoder, DeltaBitPackDecoder};
 use crate::errors::{ParquetError, Result};
 use crate::schema::types::ColumnDescPtr;
 use arrow_array::{builder::make_view, ArrayRef};
+use arrow_buffer::Buffer;
 use arrow_data::ByteView;
 use arrow_schema::DataType as ArrowType;
 use bytes::Bytes;
@@ -71,7 +72,6 @@ struct ByteViewArrayReader {
 }
 
 impl ByteViewArrayReader {
-    #[allow(unused)]
     fn new(
         pages: Box<dyn PageIterator>,
         data_type: ArrowType,
@@ -316,7 +316,10 @@ impl ByteViewArrayDecoderPlain {
     }
 
     pub fn read(&mut self, output: &mut ViewBuffer, len: usize) -> Result<usize> {
-        let block_id = output.append_block(self.buf.clone().into());
+        // Here we convert `bytes::Bytes` into `arrow_buffer::Bytes`, which is zero copy
+        // Then we convert `arrow_buffer::Bytes` into `arrow_buffer:Buffer`, which is also zero copy
+        let buf = arrow_buffer::Buffer::from_bytes(self.buf.clone().into());
+        let block_id = output.append_block(buf);
 
         let to_read = len.min(self.max_remaining_values);
 
@@ -546,7 +549,10 @@ impl ByteViewArrayDecoderDeltaLength {
 
         let src_lengths = &self.lengths[self.length_offset..self.length_offset + to_read];
 
-        let block_id = output.append_block(self.data.clone().into());
+        // Here we convert `bytes::Bytes` into `arrow_buffer::Bytes`, which is zero copy
+        // Then we convert `arrow_buffer::Bytes` into `arrow_buffer:Buffer`, which is also zero copy
+        let bytes = arrow_buffer::Buffer::from_bytes(self.data.clone().into());
+        let block_id = output.append_block(bytes);
 
         let mut current_offset = self.data_offset;
         let initial_offset = current_offset;
@@ -661,7 +667,7 @@ impl ByteViewArrayDecoderDelta {
             v
         };
 
-        let actual_block_id = output.append_block(array_buffer.into());
+        let actual_block_id = output.append_block(Buffer::from_vec(array_buffer));
         assert_eq!(actual_block_id, buffer_id);
         Ok(read)
     }
diff --git a/parquet/src/arrow/array_reader/list_array.rs b/parquet/src/arrow/array_reader/list_array.rs
index 7c66c5c23..e1752f30c 100644
--- a/parquet/src/arrow/array_reader/list_array.rs
+++ b/parquet/src/arrow/array_reader/list_array.rs
@@ -213,7 +213,7 @@ impl<OffsetSize: OffsetSizeTrait> ArrayReader for ListArrayReader<OffsetSize> {
             return Err(general_err!("Failed to reconstruct list from level data"));
         }
 
-        let value_offsets = Buffer::from(&list_offsets.to_byte_slice());
+        let value_offsets = Buffer::from(list_offsets.to_byte_slice());
 
         let mut data_builder = ArrayData::builder(self.get_data_type().clone())
             .len(list_offsets.len() - 1)
diff --git a/parquet/src/arrow/arrow_reader/mod.rs b/parquet/src/arrow/arrow_reader/mod.rs
index cc369cec0..71bb0b5e2 100644
--- a/parquet/src/arrow/arrow_reader/mod.rs
+++ b/parquet/src/arrow/arrow_reader/mod.rs
@@ -2425,89 +2425,179 @@ mod tests {
 
     fn test_invalid_utf8_string_array_inner<O: OffsetSizeTrait>() {
         let cases = [
-            (
-                invalid_utf8_first_char::<O>(),
-                "Parquet argument error: Parquet error: encountered non UTF-8 data",
-            ),
-            (
-                invalid_utf8_later_char::<O>(),
-                "Parquet argument error: Parquet error: encountered non UTF-8 data: invalid utf-8 sequence of 1 bytes from index 6",
-            ),
+            invalid_utf8_first_char::<O>(),
+            invalid_utf8_first_char_long_strings::<O>(),
+            invalid_utf8_later_char::<O>(),
+            invalid_utf8_later_char_long_strings::<O>(),
+            invalid_utf8_later_char_really_long_strings::<O>(),
+            invalid_utf8_later_char_really_long_strings2::<O>(),
         ];
-        for (array, expected_error) in cases {
-            // data is not valid utf8 we can not construct a correct StringArray
-            // safely, so purposely create an invalid StringArray
-            let array = unsafe {
-                GenericStringArray::<O>::new_unchecked(
-                    array.offsets().clone(),
-                    array.values().clone(),
-                    array.nulls().cloned(),
-                )
-            };
-            let data_type = array.data_type().clone();
-            let data = write_to_parquet(Arc::new(array));
-            let err = read_from_parquet(data).unwrap_err();
-            assert_eq!(err.to_string(), expected_error, "data type: {data_type:?}")
+        for array in &cases {
+            for encoding in STRING_ENCODINGS {
+                // data is not valid utf8 we can not construct a correct StringArray
+                // safely, so purposely create an invalid StringArray
+                let array = unsafe {
+                    GenericStringArray::<O>::new_unchecked(
+                        array.offsets().clone(),
+                        array.values().clone(),
+                        array.nulls().cloned(),
+                    )
+                };
+                let data_type = array.data_type().clone();
+                let data = write_to_parquet_with_encoding(Arc::new(array), *encoding);
+                let err = read_from_parquet(data).unwrap_err();
+                let expected_err =
+                    "Parquet argument error: Parquet error: encountered non UTF-8 data";
+                assert!(
+                    err.to_string().contains(expected_err),
+                    "data type: {data_type:?}, expected: {expected_err}, got: {err}"
+                );
+            }
         }
     }
 
     #[test]
     fn test_invalid_utf8_string_view_array() {
         let cases = [
-            (
-                invalid_utf8_first_char::<i32>(),
-                "Parquet argument error: Parquet error: encountered non UTF-8 data: invalid utf-8 sequence of 1 bytes from index 11",
-            ),
-            (
-                invalid_utf8_later_char::<i32>(),
-                "Parquet argument error: Parquet error: encountered non UTF-8 data: invalid utf-8 sequence of 1 bytes from index 14",
-            ),
+            invalid_utf8_first_char::<i32>(),
+            invalid_utf8_first_char_long_strings::<i32>(),
+            invalid_utf8_later_char::<i32>(),
+            invalid_utf8_later_char_long_strings::<i32>(),
+            invalid_utf8_later_char_really_long_strings::<i32>(),
+            invalid_utf8_later_char_really_long_strings2::<i32>(),
         ];
-        for (array, expected_error) in cases {
-            let array = arrow_cast::cast(&array, &ArrowDataType::BinaryView).unwrap();
-            let array = array.as_binary_view();
-
-            // data is not valid utf8 we can not construct a correct StringArray
-            // safely, so purposely create an invalid StringArray
-            let array = unsafe {
-                StringViewArray::new_unchecked(
-                    array.views().clone(),
-                    array.data_buffers().to_vec(),
-                    array.nulls().cloned(),
-                )
-            };
-            let data_type = array.data_type().clone();
-            let data = write_to_parquet(Arc::new(array));
-            let err = read_from_parquet(data).unwrap_err();
-            assert_eq!(err.to_string(), expected_error, "data type: {data_type:?}")
+
+        for encoding in STRING_ENCODINGS {
+            for array in &cases {
+                let array = arrow_cast::cast(&array, &ArrowDataType::BinaryView).unwrap();
+                let array = array.as_binary_view();
+
+                // data is not valid utf8 we can not construct a correct StringArray
+                // safely, so purposely create an invalid StringViewArray
+                let array = unsafe {
+                    StringViewArray::new_unchecked(
+                        array.views().clone(),
+                        array.data_buffers().to_vec(),
+                        array.nulls().cloned(),
+                    )
+                };
+
+                let data_type = array.data_type().clone();
+                let data = write_to_parquet_with_encoding(Arc::new(array), *encoding);
+                let err = read_from_parquet(data).unwrap_err();
+                let expected_err =
+                    "Parquet argument error: Parquet error: encountered non UTF-8 data";
+                assert!(
+                    err.to_string().contains(expected_err),
+                    "data type: {data_type:?}, expected: {expected_err}, got: {err}"
+                );
+            }
         }
     }
 
+    /// Encodings suitable for string data
+    const STRING_ENCODINGS: &[Option<Encoding>] = &[
+        None,
+        Some(Encoding::PLAIN),
+        Some(Encoding::DELTA_LENGTH_BYTE_ARRAY),
+        Some(Encoding::DELTA_BYTE_ARRAY),
+    ];
+
+    /// Invalid Utf-8 sequence in the first character
+    /// <https://stackoverflow.com/questions/1301402/example-invalid-utf8-string>
+    const INVALID_UTF8_FIRST_CHAR: &[u8] = &[0xa0, 0xa1, 0x20, 0x20];
+
+    /// Invalid Utf=8 sequence in NOT the first character
+    /// <https://stackoverflow.com/questions/1301402/example-invalid-utf8-string>
+    const INVALID_UTF8_LATER_CHAR: &[u8] = &[0x20, 0x20, 0x20, 0xa0, 0xa1, 0x20, 0x20];
+
     /// returns a BinaryArray with invalid UTF8 data in the first character
     fn invalid_utf8_first_char<O: OffsetSizeTrait>() -> GenericBinaryArray<O> {
-        // invalid sequence in the first character
-        // https://stackoverflow.com/questions/1301402/example-invalid-utf8-string
         let valid: &[u8] = b"   ";
-        let invalid: &[u8] = &[0xa0, 0xa1, 0x20, 0x20];
+        let invalid = INVALID_UTF8_FIRST_CHAR;
         GenericBinaryArray::<O>::from_iter(vec![None, Some(valid), None, Some(invalid)])
     }
 
+    /// Returns a BinaryArray with invalid UTF8 data in the first character of a
+    /// string larger than 12 bytes which is handled specially when reading
+    /// `ByteViewArray`s
+    fn invalid_utf8_first_char_long_strings<O: OffsetSizeTrait>() -> GenericBinaryArray<O> {
+        let valid: &[u8] = b"   ";
+        let mut invalid = vec![];
+        invalid.extend_from_slice(b"ThisStringIsCertainlyLongerThan12Bytes");
+        invalid.extend_from_slice(INVALID_UTF8_FIRST_CHAR);
+        GenericBinaryArray::<O>::from_iter(vec![None, Some(valid), None, Some(&invalid)])
+    }
+
     /// returns a BinaryArray with invalid UTF8 data in a character other than
     /// the first (this is checked in a special codepath)
     fn invalid_utf8_later_char<O: OffsetSizeTrait>() -> GenericBinaryArray<O> {
-        // invalid sequence in NOT the first character
-        // https://stackoverflow.com/questions/1301402/example-invalid-utf8-string
         let valid: &[u8] = b"   ";
-        let invalid: &[u8] = &[0x20, 0x20, 0x20, 0xa0, 0xa1, 0x20, 0x20];
+        let invalid: &[u8] = INVALID_UTF8_LATER_CHAR;
         GenericBinaryArray::<O>::from_iter(vec![None, Some(valid), None, Some(invalid)])
     }
 
-    // writes the array into a single column parquet file
-    fn write_to_parquet(array: ArrayRef) -> Vec<u8> {
+    /// returns a BinaryArray with invalid UTF8 data in a character other than
+    /// the first in a string larger than 12 bytes which is handled specially
+    /// when reading `ByteViewArray`s (this is checked in a special codepath)
+    fn invalid_utf8_later_char_long_strings<O: OffsetSizeTrait>() -> GenericBinaryArray<O> {
+        let valid: &[u8] = b"   ";
+        let mut invalid = vec![];
+        invalid.extend_from_slice(b"ThisStringIsCertainlyLongerThan12Bytes");
+        invalid.extend_from_slice(INVALID_UTF8_LATER_CHAR);
+        GenericBinaryArray::<O>::from_iter(vec![None, Some(valid), None, Some(&invalid)])
+    }
+
+    /// returns a BinaryArray with invalid UTF8 data in a character other than
+    /// the first in a string larger than 128 bytes which is handled specially
+    /// when reading `ByteViewArray`s (this is checked in a special codepath)
+    fn invalid_utf8_later_char_really_long_strings<O: OffsetSizeTrait>() -> GenericBinaryArray<O> {
+        let valid: &[u8] = b"   ";
+        let mut invalid = vec![];
+        for _ in 0..10 {
+            // each instance is 38 bytes
+            invalid.extend_from_slice(b"ThisStringIsCertainlyLongerThan12Bytes");
+        }
+        invalid.extend_from_slice(INVALID_UTF8_LATER_CHAR);
+        GenericBinaryArray::<O>::from_iter(vec![None, Some(valid), None, Some(&invalid)])
+    }
+
+    /// returns a BinaryArray with small invalid UTF8 data followed by a large
+    /// invalid UTF8 data in a character other than the first in a string larger
+    fn invalid_utf8_later_char_really_long_strings2<O: OffsetSizeTrait>() -> GenericBinaryArray<O> {
+        let valid: &[u8] = b"   ";
+        let mut valid_long = vec![];
+        for _ in 0..10 {
+            // each instance is 38 bytes
+            valid_long.extend_from_slice(b"ThisStringIsCertainlyLongerThan12Bytes");
+        }
+        let invalid = INVALID_UTF8_LATER_CHAR;
+        GenericBinaryArray::<O>::from_iter(vec![
+            None,
+            Some(valid),
+            Some(invalid),
+            None,
+            Some(&valid_long),
+            Some(valid),
+        ])
+    }
+
+    /// writes the array into a single column parquet file with the specified
+    /// encoding.
+    ///
+    /// If no encoding is specified, use default (dictionary) encoding
+    fn write_to_parquet_with_encoding(array: ArrayRef, encoding: Option<Encoding>) -> Vec<u8> {
         let batch = RecordBatch::try_from_iter(vec![("c", array)]).unwrap();
         let mut data = vec![];
         let schema = batch.schema();
-        let props = None;
+        let props = encoding.map(|encoding| {
+            WriterProperties::builder()
+                // must disable dictionary encoding to actually use encoding
+                .set_dictionary_enabled(false)
+                .set_encoding(encoding)
+                .build()
+        });
+
         {
             let mut writer = ArrowWriter::try_new(&mut data, schema, props).unwrap();
             writer.write(&batch).unwrap();
diff --git a/parquet/src/arrow/arrow_writer/levels.rs b/parquet/src/arrow/arrow_writer/levels.rs
index c50e612b2..3e828bbdd 100644
--- a/parquet/src/arrow/arrow_writer/levels.rs
+++ b/parquet/src/arrow/arrow_writer/levels.rs
@@ -1050,7 +1050,7 @@ mod tests {
         let a_list_data = ArrayData::builder(a_list_type.clone())
             .len(5)
             .add_buffer(a_value_offsets)
-            .null_bit_buffer(Some(Buffer::from(vec![0b00011011])))
+            .null_bit_buffer(Some(Buffer::from([0b00011011])))
             .add_child_data(a_values.to_data())
             .build()
             .unwrap();
@@ -1116,7 +1116,7 @@ mod tests {
 
         // Construct a buffer for value offsets, for the nested array:
         //  [[1], [2, 3], null, [4, 5, 6], [7, 8, 9, 10]]
-        let g_value_offsets = arrow::buffer::Buffer::from(&[0, 1, 3, 3, 6, 10].to_byte_slice());
+        let g_value_offsets = arrow::buffer::Buffer::from([0, 1, 3, 3, 6, 10].to_byte_slice());
 
         // Construct a list array from the above two
         let g_list_data = ArrayData::builder(struct_field_g.data_type().clone())
diff --git a/parquet/src/arrow/arrow_writer/mod.rs b/parquet/src/arrow/arrow_writer/mod.rs
index f83e56c7a..cf46f3b64 100644
--- a/parquet/src/arrow/arrow_writer/mod.rs
+++ b/parquet/src/arrow/arrow_writer/mod.rs
@@ -1206,7 +1206,7 @@ mod tests {
 
         // Construct a buffer for value offsets, for the nested array:
         //  [[1], [2, 3], null, [4, 5, 6], [7, 8, 9, 10]]
-        let a_value_offsets = arrow::buffer::Buffer::from(&[0, 1, 3, 3, 6, 10].to_byte_slice());
+        let a_value_offsets = arrow::buffer::Buffer::from([0, 1, 3, 3, 6, 10].to_byte_slice());
 
         // Construct a list array from the above two
         let a_list_data = ArrayData::builder(DataType::List(Arc::new(Field::new(
@@ -1217,7 +1217,7 @@ mod tests {
         .len(5)
         .add_buffer(a_value_offsets)
         .add_child_data(a_values.into_data())
-        .null_bit_buffer(Some(Buffer::from(vec![0b00011011])))
+        .null_bit_buffer(Some(Buffer::from([0b00011011])))
         .build()
         .unwrap();
         let a = ListArray::from(a_list_data);
@@ -1246,7 +1246,7 @@ mod tests {
 
         // Construct a buffer for value offsets, for the nested array:
         //  [[1], [2, 3], [], [4, 5, 6], [7, 8, 9, 10]]
-        let a_value_offsets = arrow::buffer::Buffer::from(&[0, 1, 3, 3, 6, 10].to_byte_slice());
+        let a_value_offsets = arrow::buffer::Buffer::from([0, 1, 3, 3, 6, 10].to_byte_slice());
 
         // Construct a list array from the above two
         let a_list_data = ArrayData::builder(DataType::List(Arc::new(Field::new(
@@ -1405,7 +1405,7 @@ mod tests {
 
         // Construct a buffer for value offsets, for the nested array:
         //  [[1], [2, 3], [], [4, 5, 6], [7, 8, 9, 10]]
-        let g_value_offsets = arrow::buffer::Buffer::from(&[0, 1, 3, 3, 6, 10].to_byte_slice());
+        let g_value_offsets = arrow::buffer::Buffer::from([0, 1, 3, 3, 6, 10].to_byte_slice());
 
         // Construct a list array from the above two
         let g_list_data = ArrayData::builder(struct_field_g.data_type().clone())
@@ -1420,7 +1420,7 @@ mod tests {
             .len(5)
             .add_buffer(g_value_offsets)
             .add_child_data(g_value.to_data())
-            .null_bit_buffer(Some(Buffer::from(vec![0b00011011])))
+            .null_bit_buffer(Some(Buffer::from([0b00011011])))
             .build()
             .unwrap();
         let h = ListArray::from(h_list_data);
@@ -1525,14 +1525,14 @@ mod tests {
         let c = Int32Array::from(vec![Some(1), None, Some(3), None, None, Some(6)]);
         let b_data = ArrayDataBuilder::new(field_b.data_type().clone())
             .len(6)
-            .null_bit_buffer(Some(Buffer::from(vec![0b00100111])))
+            .null_bit_buffer(Some(Buffer::from([0b00100111])))
             .add_child_data(c.into_data())
             .build()
             .unwrap();
         let b = StructArray::from(b_data);
         let a_data = ArrayDataBuilder::new(field_a.data_type().clone())
             .len(6)
-            .null_bit_buffer(Some(Buffer::from(vec![0b00101111])))
+            .null_bit_buffer(Some(Buffer::from([0b00101111])))
             .add_child_data(b.into_data())
             .build()
             .unwrap();
@@ -1595,7 +1595,7 @@ mod tests {
         let c = Int32Array::from(vec![1, 2, 3, 4, 5, 6]);
         let b_data = ArrayDataBuilder::new(type_b)
             .len(6)
-            .null_bit_buffer(Some(Buffer::from(vec![0b00100111])))
+            .null_bit_buffer(Some(Buffer::from([0b00100111])))
             .add_child_data(c.into_data())
             .build()
             .unwrap();
@@ -2280,7 +2280,7 @@ mod tests {
 
         // Build [[], null, [null, null]]
         let a_values = NullArray::new(2);
-        let a_value_offsets = arrow::buffer::Buffer::from(&[0, 0, 0, 2].to_byte_slice());
+        let a_value_offsets = arrow::buffer::Buffer::from([0, 0, 0, 2].to_byte_slice());
         let a_list_data = ArrayData::builder(DataType::List(Arc::new(Field::new(
             "item",
             DataType::Null,
@@ -2288,7 +2288,7 @@ mod tests {
         ))))
         .len(3)
         .add_buffer(a_value_offsets)
-        .null_bit_buffer(Some(Buffer::from(vec![0b00000101])))
+        .null_bit_buffer(Some(Buffer::from([0b00000101])))
         .add_child_data(a_values.into_data())
         .build()
         .unwrap();
@@ -2310,7 +2310,7 @@ mod tests {
     #[test]
     fn list_single_column() {
         let a_values = Int32Array::from(vec![1, 2, 3, 4, 5, 6, 7, 8, 9, 10]);
-        let a_value_offsets = arrow::buffer::Buffer::from(&[0, 1, 3, 3, 6, 10].to_byte_slice());
+        let a_value_offsets = arrow::buffer::Buffer::from([0, 1, 3, 3, 6, 10].to_byte_slice());
         let a_list_data = ArrayData::builder(DataType::List(Arc::new(Field::new(
             "item",
             DataType::Int32,
@@ -2318,7 +2318,7 @@ mod tests {
         ))))
         .len(5)
         .add_buffer(a_value_offsets)
-        .null_bit_buffer(Some(Buffer::from(vec![0b00011011])))
+        .null_bit_buffer(Some(Buffer::from([0b00011011])))
         .add_child_data(a_values.into_data())
         .build()
         .unwrap();
@@ -2334,7 +2334,7 @@ mod tests {
     #[test]
     fn large_list_single_column() {
         let a_values = Int32Array::from(vec![1, 2, 3, 4, 5, 6, 7, 8, 9, 10]);
-        let a_value_offsets = arrow::buffer::Buffer::from(&[0i64, 1, 3, 3, 6, 10].to_byte_slice());
+        let a_value_offsets = arrow::buffer::Buffer::from([0i64, 1, 3, 3, 6, 10].to_byte_slice());
         let a_list_data = ArrayData::builder(DataType::LargeList(Arc::new(Field::new(
             "large_item",
             DataType::Int32,
@@ -2343,7 +2343,7 @@ mod tests {
         .len(5)
         .add_buffer(a_value_offsets)
         .add_child_data(a_values.into_data())
-        .null_bit_buffer(Some(Buffer::from(vec![0b00011011])))
+        .null_bit_buffer(Some(Buffer::from([0b00011011])))
         .build()
         .unwrap();
 
diff --git a/parquet/src/arrow/buffer/offset_buffer.rs b/parquet/src/arrow/buffer/offset_buffer.rs
index 806f144d9..ce9eb1142 100644
--- a/parquet/src/arrow/buffer/offset_buffer.rs
+++ b/parquet/src/arrow/buffer/offset_buffer.rs
@@ -18,13 +18,10 @@
 use crate::arrow::buffer::bit_util::iter_set_bits_rev;
 use crate::arrow::record_reader::buffer::ValuesBuffer;
 use crate::errors::{ParquetError, Result};
-use arrow_array::builder::GenericByteViewBuilder;
-use arrow_array::types::BinaryViewType;
 use arrow_array::{make_array, ArrayRef, OffsetSizeTrait};
 use arrow_buffer::{ArrowNativeType, Buffer};
 use arrow_data::ArrayDataBuilder;
 use arrow_schema::DataType as ArrowType;
-use std::sync::Arc;
 
 /// A buffer of variable-sized byte arrays that can be converted into
 /// a corresponding [`ArrayRef`]
@@ -128,51 +125,18 @@ impl<I: OffsetSizeTrait> OffsetBuffer<I> {
 
     /// Converts this into an [`ArrayRef`] with the provided `data_type` and `null_buffer`
     pub fn into_array(self, null_buffer: Option<Buffer>, data_type: ArrowType) -> ArrayRef {
-        match data_type {
-            ArrowType::Utf8View => {
-                let mut builder = self.build_generic_byte_view();
-                Arc::new(builder.finish().to_string_view().unwrap())
-            }
-            ArrowType::BinaryView => {
-                let mut builder = self.build_generic_byte_view();
-                Arc::new(builder.finish())
-            }
-            _ => {
-                let array_data_builder = ArrayDataBuilder::new(data_type)
-                    .len(self.len())
-                    .add_buffer(Buffer::from_vec(self.offsets))
-                    .add_buffer(Buffer::from_vec(self.values))
-                    .null_bit_buffer(null_buffer);
-
-                let data = match cfg!(debug_assertions) {
-                    true => array_data_builder.build().unwrap(),
-                    false => unsafe { array_data_builder.build_unchecked() },
-                };
-
-                make_array(data)
-            }
-        }
-    }
-
-    fn build_generic_byte_view(self) -> GenericByteViewBuilder<BinaryViewType> {
-        let mut builder = GenericByteViewBuilder::<BinaryViewType>::with_capacity(self.len());
-        let buffer = self.values.into();
-        let block = builder.append_block(buffer);
-        for window in self.offsets.windows(2) {
-            let start = window[0];
-            let end = window[1];
-            let len = (end - start).to_usize().unwrap();
-
-            if len != 0 {
-                // Safety: (1) the buffer is valid (2) the offsets are valid (3) the values in between are of ByteViewType
-                unsafe {
-                    builder.append_view_unchecked(block, start.as_usize() as u32, len as u32);
-                }
-            } else {
-                builder.append_null();
-            }
-        }
-        builder
+        let array_data_builder = ArrayDataBuilder::new(data_type)
+            .len(self.len())
+            .add_buffer(Buffer::from_vec(self.offsets))
+            .add_buffer(Buffer::from_vec(self.values))
+            .null_bit_buffer(null_buffer);
+
+        let data = match cfg!(debug_assertions) {
+            true => array_data_builder.build().unwrap(),
+            false => unsafe { array_data_builder.build_unchecked() },
+        };
+
+        make_array(data)
     }
 }
 
@@ -229,7 +193,7 @@ impl<I: OffsetSizeTrait> ValuesBuffer for OffsetBuffer<I> {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use arrow_array::{Array, LargeStringArray, StringArray, StringViewArray};
+    use arrow_array::{Array, LargeStringArray, StringArray};
 
     #[test]
     fn test_offset_buffer_empty() {
@@ -280,44 +244,6 @@ mod tests {
         );
     }
 
-    #[test]
-    fn test_string_view() {
-        let mut buffer = OffsetBuffer::<i32>::default();
-        for v in [
-            "hello",
-            "world",
-            "large payload over 12 bytes",
-            "a",
-            "b",
-            "c",
-        ] {
-            buffer.try_push(v.as_bytes(), false).unwrap()
-        }
-        let split = std::mem::take(&mut buffer);
-
-        let array = split.into_array(None, ArrowType::Utf8View);
-        let strings = array.as_any().downcast_ref::<StringViewArray>().unwrap();
-        assert_eq!(
-            strings.iter().map(|x| x.unwrap()).collect::<Vec<_>>(),
-            vec![
-                "hello",
-                "world",
-                "large payload over 12 bytes",
-                "a",
-                "b",
-                "c"
-            ]
-        );
-
-        buffer.try_push("test".as_bytes(), false).unwrap();
-        let array = buffer.into_array(None, ArrowType::Utf8View);
-        let strings = array.as_any().downcast_ref::<StringViewArray>().unwrap();
-        assert_eq!(
-            strings.iter().map(|x| x.unwrap()).collect::<Vec<_>>(),
-            vec!["test"]
-        );
-    }
-
     #[test]
     fn test_offset_buffer_pad_nulls() {
         let mut buffer = OffsetBuffer::<i32>::default();
diff --git a/parquet/src/arrow/buffer/view_buffer.rs b/parquet/src/arrow/buffer/view_buffer.rs
index ae83ac317..2256f4877 100644
--- a/parquet/src/arrow/buffer/view_buffer.rs
+++ b/parquet/src/arrow/buffer/view_buffer.rs
@@ -68,7 +68,6 @@ impl ViewBuffer {
     }
 
     /// Converts this into an [`ArrayRef`] with the provided `data_type` and `null_buffer`
-    #[allow(unused)]
     pub fn into_array(self, null_buffer: Option<Buffer>, data_type: &ArrowType) -> ArrayRef {
         let len = self.views.len();
         let views = Buffer::from_vec(self.views);
diff --git a/parquet/src/bin/parquet-fromcsv.rs b/parquet/src/bin/parquet-fromcsv.rs
index 140a36767..6f8c47c62 100644
--- a/parquet/src/bin/parquet-fromcsv.rs
+++ b/parquet/src/bin/parquet-fromcsv.rs
@@ -557,9 +557,10 @@ mod tests {
     fn test_parse_arg_compression_format_fail() {
         match parse_args(vec!["--parquet-compression", "zip"]) {
             Ok(_) => panic!("unexpected success"),
-            Err(e) => assert_eq!(
-                format!("{e}"),
-                "error: invalid value 'zip' for '--parquet-compression <PARQUET_COMPRESSION>': Unknown compression ZIP : possible values UNCOMPRESSED, SNAPPY, GZIP, LZO, BROTLI, LZ4, ZSTD \n\nFor more information try --help\n"),
+            Err(e) => {
+                let err = e.to_string();
+                assert!(err.contains("error: invalid value 'zip' for '--parquet-compression <PARQUET_COMPRESSION>': Unknown compression ZIP : possible values UNCOMPRESSED, SNAPPY, GZIP, LZO, BROTLI, LZ4, ZSTD \n\nFor more information try --help"), "{err}")
+            }
         }
     }
 
diff --git a/parquet/src/bloom_filter/mod.rs b/parquet/src/bloom_filter/mod.rs
index d2acdcd0b..a8d68d4b6 100644
--- a/parquet/src/bloom_filter/mod.rs
+++ b/parquet/src/bloom_filter/mod.rs
@@ -134,7 +134,8 @@ impl Block {
 
     #[inline]
     fn to_ne_bytes(self) -> [u8; 32] {
-        unsafe { std::mem::transmute(self) }
+        // SAFETY: [u32; 8] and [u8; 32] have the same size and neither has invalid bit patterns.
+        unsafe { std::mem::transmute(self.0) }
     }
 
     #[inline]
diff --git a/parquet/src/data_type.rs b/parquet/src/data_type.rs
index 5bcd2062c..b85a75cfd 100644
--- a/parquet/src/data_type.rs
+++ b/parquet/src/data_type.rs
@@ -468,6 +468,8 @@ macro_rules! gen_as_bytes {
         impl AsBytes for $source_ty {
             #[allow(clippy::size_of_in_element_count)]
             fn as_bytes(&self) -> &[u8] {
+                // SAFETY: macro is only used with primitive types that have no padding, so the
+                // resulting slice always refers to initialized memory.
                 unsafe {
                     std::slice::from_raw_parts(
                         self as *const $source_ty as *const u8,
@@ -481,6 +483,8 @@ macro_rules! gen_as_bytes {
             #[inline]
             #[allow(clippy::size_of_in_element_count)]
             fn slice_as_bytes(self_: &[Self]) -> &[u8] {
+                // SAFETY: macro is only used with primitive types that have no padding, so the
+                // resulting slice always refers to initialized memory.
                 unsafe {
                     std::slice::from_raw_parts(
                         self_.as_ptr() as *const u8,
@@ -492,10 +496,15 @@ macro_rules! gen_as_bytes {
             #[inline]
             #[allow(clippy::size_of_in_element_count)]
             unsafe fn slice_as_bytes_mut(self_: &mut [Self]) -> &mut [u8] {
-                std::slice::from_raw_parts_mut(
-                    self_.as_mut_ptr() as *mut u8,
-                    std::mem::size_of_val(self_),
-                )
+                // SAFETY: macro is only used with primitive types that have no padding, so the
+                // resulting slice always refers to initialized memory. Moreover, self has no
+                // invalid bit patterns, so all writes to the resulting slice will be valid.
+                unsafe {
+                    std::slice::from_raw_parts_mut(
+                        self_.as_mut_ptr() as *mut u8,
+                        std::mem::size_of_val(self_),
+                    )
+                }
             }
         }
     };
@@ -534,12 +543,15 @@ unimplemented_slice_as_bytes!(FixedLenByteArray);
 
 impl AsBytes for bool {
     fn as_bytes(&self) -> &[u8] {
+        // SAFETY: a bool is guaranteed to be either 0x00 or 0x01 in memory, so the memory is
+        // valid.
         unsafe { std::slice::from_raw_parts(self as *const bool as *const u8, 1) }
     }
 }
 
 impl AsBytes for Int96 {
     fn as_bytes(&self) -> &[u8] {
+        // SAFETY: Int96::data is a &[u32; 3].
         unsafe { std::slice::from_raw_parts(self.data() as *const [u32] as *const u8, 12) }
     }
 }
@@ -718,6 +730,7 @@ pub(crate) mod private {
 
                 #[inline]
                 fn encode<W: std::io::Write>(values: &[Self], writer: &mut W, _: &mut BitWriter) -> Result<()> {
+                    // SAFETY: Self is one of i32, i64, f32, f64, which have no padding.
                     let raw = unsafe {
                         std::slice::from_raw_parts(
                             values.as_ptr() as *const u8,
@@ -747,9 +760,10 @@ pub(crate) mod private {
                         return Err(eof_err!("Not enough bytes to decode"));
                     }
 
-                    // SAFETY: Raw types should be as per the standard rust bit-vectors
-                    unsafe {
-                        let raw_buffer = &mut Self::slice_as_bytes_mut(buffer)[..bytes_to_decode];
+                    {
+                        // SAFETY: Self has no invalid bit patterns, so writing to the slice
+                        // obtained with slice_as_bytes_mut is always safe.
+                        let raw_buffer = &mut unsafe { Self::slice_as_bytes_mut(buffer) }[..bytes_to_decode];
                         raw_buffer.copy_from_slice(data.slice(
                             decoder.start..decoder.start + bytes_to_decode
                         ).as_ref());
@@ -810,9 +824,7 @@ pub(crate) mod private {
             _: &mut BitWriter,
         ) -> Result<()> {
             for value in values {
-                let raw = unsafe {
-                    std::slice::from_raw_parts(value.data() as *const [u32] as *const u8, 12)
-                };
+                let raw = SliceAsBytes::slice_as_bytes(value.data());
                 writer.write_all(raw)?;
             }
             Ok(())
diff --git a/parquet/src/util/bit_util.rs b/parquet/src/util/bit_util.rs
index eaaf3ee10..adbf45014 100644
--- a/parquet/src/util/bit_util.rs
+++ b/parquet/src/util/bit_util.rs
@@ -42,7 +42,11 @@ fn array_from_slice<const N: usize>(bs: &[u8]) -> Result<[u8; N]> {
     }
 }
 
-pub trait FromBytes: Sized {
+/// # Safety
+/// All bit patterns 00000xxxx, where there are `BIT_CAPACITY` `x`s,
+/// must be valid, unless BIT_CAPACITY is 0.
+pub unsafe trait FromBytes: Sized {
+    const BIT_CAPACITY: usize;
     type Buffer: AsMut<[u8]> + Default;
     fn try_from_le_slice(b: &[u8]) -> Result<Self>;
     fn from_le_bytes(bs: Self::Buffer) -> Self;
@@ -51,7 +55,9 @@ pub trait FromBytes: Sized {
 macro_rules! from_le_bytes {
     ($($ty: ty),*) => {
         $(
-        impl FromBytes for $ty {
+        // SAFETY: this macro is used for types for which all bit patterns are valid.
+        unsafe impl FromBytes for $ty {
+            const BIT_CAPACITY: usize = std::mem::size_of::<$ty>() * 8;
             type Buffer = [u8; size_of::<Self>()];
             fn try_from_le_slice(b: &[u8]) -> Result<Self> {
                 Ok(Self::from_le_bytes(array_from_slice(b)?))
@@ -66,7 +72,9 @@ macro_rules! from_le_bytes {
 
 from_le_bytes! { u8, u16, u32, u64, i8, i16, i32, i64, f32, f64 }
 
-impl FromBytes for bool {
+// SAFETY: the 0000000x bit pattern is always valid for `bool`.
+unsafe impl FromBytes for bool {
+    const BIT_CAPACITY: usize = 1;
     type Buffer = [u8; 1];
 
     fn try_from_le_slice(b: &[u8]) -> Result<Self> {
@@ -77,7 +85,9 @@ impl FromBytes for bool {
     }
 }
 
-impl FromBytes for Int96 {
+// SAFETY: BIT_CAPACITY is 0.
+unsafe impl FromBytes for Int96 {
+    const BIT_CAPACITY: usize = 0;
     type Buffer = [u8; 12];
 
     fn try_from_le_slice(b: &[u8]) -> Result<Self> {
@@ -95,7 +105,9 @@ impl FromBytes for Int96 {
     }
 }
 
-impl FromBytes for ByteArray {
+// SAFETY: BIT_CAPACITY is 0.
+unsafe impl FromBytes for ByteArray {
+    const BIT_CAPACITY: usize = 0;
     type Buffer = Vec<u8>;
 
     fn try_from_le_slice(b: &[u8]) -> Result<Self> {
@@ -106,7 +118,9 @@ impl FromBytes for ByteArray {
     }
 }
 
-impl FromBytes for FixedLenByteArray {
+// SAFETY: BIT_CAPACITY is 0.
+unsafe impl FromBytes for FixedLenByteArray {
+    const BIT_CAPACITY: usize = 0;
     type Buffer = Vec<u8>;
 
     fn try_from_le_slice(b: &[u8]) -> Result<Self> {
@@ -457,10 +471,17 @@ impl BitReader {
             }
         }
 
+        assert_ne!(T::BIT_CAPACITY, 0);
+        assert!(num_bits <= T::BIT_CAPACITY);
+
         // Read directly into output buffer
         match size_of::<T>() {
             1 => {
                 let ptr = batch.as_mut_ptr() as *mut u8;
+                // SAFETY: batch is properly aligned and sized. Caller guarantees that all bit patterns
+                // in which only the lowest T::BIT_CAPACITY bits of T are set are valid,
+                // unpack{8,16,32,64} only set to non0 the lowest num_bits bits, and we
+                // checked that num_bits <= T::BIT_CAPACITY.
                 let out = unsafe { std::slice::from_raw_parts_mut(ptr, batch.len()) };
                 while values_to_read - i >= 8 {
                     let out_slice = (&mut out[i..i + 8]).try_into().unwrap();
@@ -471,6 +492,10 @@ impl BitReader {
             }
             2 => {
                 let ptr = batch.as_mut_ptr() as *mut u16;
+                // SAFETY: batch is properly aligned and sized. Caller guarantees that all bit patterns
+                // in which only the lowest T::BIT_CAPACITY bits of T are set are valid,
+                // unpack{8,16,32,64} only set to non0 the lowest num_bits bits, and we
+                // checked that num_bits <= T::BIT_CAPACITY.
                 let out = unsafe { std::slice::from_raw_parts_mut(ptr, batch.len()) };
                 while values_to_read - i >= 16 {
                     let out_slice = (&mut out[i..i + 16]).try_into().unwrap();
@@ -481,6 +506,10 @@ impl BitReader {
             }
             4 => {
                 let ptr = batch.as_mut_ptr() as *mut u32;
+                // SAFETY: batch is properly aligned and sized. Caller guarantees that all bit patterns
+                // in which only the lowest T::BIT_CAPACITY bits of T are set are valid,
+                // unpack{8,16,32,64} only set to non0 the lowest num_bits bits, and we
+                // checked that num_bits <= T::BIT_CAPACITY.
                 let out = unsafe { std::slice::from_raw_parts_mut(ptr, batch.len()) };
                 while values_to_read - i >= 32 {
                     let out_slice = (&mut out[i..i + 32]).try_into().unwrap();
@@ -491,6 +520,10 @@ impl BitReader {
             }
             8 => {
                 let ptr = batch.as_mut_ptr() as *mut u64;
+                // SAFETY: batch is properly aligned and sized. Caller guarantees that all bit patterns
+                // in which only the lowest T::BIT_CAPACITY bits of T are set are valid,
+                // unpack{8,16,32,64} only set to non0 the lowest num_bits bits, and we
+                // checked that num_bits <= T::BIT_CAPACITY.
                 let out = unsafe { std::slice::from_raw_parts_mut(ptr, batch.len()) };
                 while values_to_read - i >= 64 {
                     let out_slice = (&mut out[i..i + 64]).try_into().unwrap();


Test patch:
diff --git a/arrow-flight/tests/flight_sql_client_cli.rs b/arrow-flight/tests/flight_sql_client_cli.rs
index 317eb3900..631f5cd31 100644
--- a/arrow-flight/tests/flight_sql_client_cli.rs
+++ b/arrow-flight/tests/flight_sql_client_cli.rs
@@ -15,25 +15,21 @@
 // specific language governing permissions and limitations
 // under the License.
 
+use std::collections::HashMap;
 use std::{net::SocketAddr, pin::Pin, sync::Arc, time::Duration};
 
 use arrow_array::{ArrayRef, Int64Array, RecordBatch, StringArray};
+use arrow_flight::sql::client::FlightSqlServiceClient;
+use arrow_flight::sql::EndTransaction;
 use arrow_flight::{
     decode::FlightRecordBatchStream,
     flight_service_server::{FlightService, FlightServiceServer},
     sql::{
         server::{FlightSqlService, PeekableFlightDataStream},
-        ActionBeginSavepointRequest, ActionBeginSavepointResult, ActionBeginTransactionRequest,
-        ActionBeginTransactionResult, ActionCancelQueryRequest, ActionCancelQueryResult,
-        ActionClosePreparedStatementRequest, ActionCreatePreparedStatementRequest,
-        ActionCreatePreparedStatementResult, ActionCreatePreparedSubstraitPlanRequest,
-        ActionEndSavepointRequest, ActionEndTransactionRequest, Any, CommandGetCatalogs,
-        CommandGetCrossReference, CommandGetDbSchemas, CommandGetExportedKeys,
-        CommandGetImportedKeys, CommandGetPrimaryKeys, CommandGetSqlInfo, CommandGetTableTypes,
-        CommandGetTables, CommandGetXdbcTypeInfo, CommandPreparedStatementQuery,
-        CommandPreparedStatementUpdate, CommandStatementQuery, CommandStatementSubstraitPlan,
-        CommandStatementUpdate, DoPutPreparedStatementResult, ProstMessageExt, SqlInfo,
-        TicketStatementQuery,
+        ActionBeginTransactionRequest, ActionBeginTransactionResult,
+        ActionCreatePreparedStatementRequest, ActionCreatePreparedStatementResult,
+        ActionEndTransactionRequest, Any, CommandPreparedStatementQuery, CommandStatementQuery,
+        DoPutPreparedStatementResult, ProstMessageExt, SqlInfo,
     },
     utils::batches_to_flight_data,
     Action, FlightData, FlightDescriptor, FlightEndpoint, FlightInfo, HandshakeRequest,
@@ -45,8 +41,11 @@ use assert_cmd::Command;
 use bytes::Bytes;
 use futures::{Stream, TryStreamExt};
 use prost::Message;
+use tokio::sync::Mutex;
 use tokio::{net::TcpListener, task::JoinHandle};
+use tonic::transport::Endpoint;
 use tonic::{Request, Response, Status, Streaming};
+use uuid::Uuid;
 
 const QUERY: &str = "SELECT * FROM table;";
 
@@ -140,6 +139,7 @@ async fn test_do_put_prepared_statement(test_server: FlightSqlServiceImpl) {
 pub async fn test_do_put_prepared_statement_stateless() {
     test_do_put_prepared_statement(FlightSqlServiceImpl {
         stateless_prepared_statements: true,
+        transactions: Arc::new(Mutex::new(HashMap::new())),
     })
     .await
 }
@@ -148,10 +148,48 @@ pub async fn test_do_put_prepared_statement_stateless() {
 pub async fn test_do_put_prepared_statement_stateful() {
     test_do_put_prepared_statement(FlightSqlServiceImpl {
         stateless_prepared_statements: false,
+        transactions: Arc::new(Mutex::new(HashMap::new())),
     })
     .await
 }
 
+#[tokio::test]
+pub async fn test_begin_end_transaction() {
+    let test_server = FlightSqlServiceImpl {
+        stateless_prepared_statements: true,
+        transactions: Arc::new(Mutex::new(HashMap::new())),
+    };
+    let fixture = TestFixture::new(&test_server).await;
+    let addr = fixture.addr;
+    let channel = Endpoint::from_shared(format!("http://{}:{}", addr.ip(), addr.port()))
+        .unwrap()
+        .connect()
+        .await
+        .expect("error connecting");
+    let mut flight_sql_client = FlightSqlServiceClient::new(channel);
+
+    // begin commit
+    let transaction_id = flight_sql_client.begin_transaction().await.unwrap();
+    flight_sql_client
+        .end_transaction(transaction_id, EndTransaction::Commit)
+        .await
+        .unwrap();
+
+    // begin rollback
+    let transaction_id = flight_sql_client.begin_transaction().await.unwrap();
+    flight_sql_client
+        .end_transaction(transaction_id, EndTransaction::Rollback)
+        .await
+        .unwrap();
+
+    // unknown transaction id
+    let transaction_id = "UnknownTransactionId".to_string().into();
+    assert!(flight_sql_client
+        .end_transaction(transaction_id, EndTransaction::Commit)
+        .await
+        .is_err());
+}
+
 /// All tests must complete within this many seconds or else the test server is shutdown
 const DEFAULT_TIMEOUT_SECONDS: u64 = 30;
 
@@ -161,12 +199,14 @@ pub struct FlightSqlServiceImpl {
     /// prepared statements. stateful servers will not return an updated
     /// handle after executing `DoPut(CommandPreparedStatementQuery)`
     stateless_prepared_statements: bool,
+    transactions: Arc<Mutex<HashMap<String, ()>>>,
 }
 
 impl Default for FlightSqlServiceImpl {
     fn default() -> Self {
         Self {
             stateless_prepared_statements: true,
+            transactions: Arc::new(Mutex::new(HashMap::new())),
         }
     }
 }
@@ -318,244 +358,6 @@ impl FlightSqlService for FlightSqlServiceImpl {
         Ok(resp)
     }
 
-    async fn get_flight_info_substrait_plan(
-        &self,
-        _query: CommandStatementSubstraitPlan,
-        _request: Request<FlightDescriptor>,
-    ) -> Result<Response<FlightInfo>, Status> {
-        Err(Status::unimplemented(
-            "get_flight_info_substrait_plan not implemented",
-        ))
-    }
-
-    async fn get_flight_info_catalogs(
-        &self,
-        _query: CommandGetCatalogs,
-        _request: Request<FlightDescriptor>,
-    ) -> Result<Response<FlightInfo>, Status> {
-        Err(Status::unimplemented(
-            "get_flight_info_catalogs not implemented",
-        ))
-    }
-
-    async fn get_flight_info_schemas(
-        &self,
-        _query: CommandGetDbSchemas,
-        _request: Request<FlightDescriptor>,
-    ) -> Result<Response<FlightInfo>, Status> {
-        Err(Status::unimplemented(
-            "get_flight_info_schemas not implemented",
-        ))
-    }
-
-    async fn get_flight_info_tables(
-        &self,
-        _query: CommandGetTables,
-        _request: Request<FlightDescriptor>,
-    ) -> Result<Response<FlightInfo>, Status> {
-        Err(Status::unimplemented(
-            "get_flight_info_tables not implemented",
-        ))
-    }
-
-    async fn get_flight_info_table_types(
-        &self,
-        _query: CommandGetTableTypes,
-        _request: Request<FlightDescriptor>,
-    ) -> Result<Response<FlightInfo>, Status> {
-        Err(Status::unimplemented(
-            "get_flight_info_table_types not implemented",
-        ))
-    }
-
-    async fn get_flight_info_sql_info(
-        &self,
-        _query: CommandGetSqlInfo,
-        _request: Request<FlightDescriptor>,
-    ) -> Result<Response<FlightInfo>, Status> {
-        Err(Status::unimplemented(
-            "get_flight_info_sql_info not implemented",
-        ))
-    }
-
-    async fn get_flight_info_primary_keys(
-        &self,
-        _query: CommandGetPrimaryKeys,
-        _request: Request<FlightDescriptor>,
-    ) -> Result<Response<FlightInfo>, Status> {
-        Err(Status::unimplemented(
-            "get_flight_info_primary_keys not implemented",
-        ))
-    }
-
-    async fn get_flight_info_exported_keys(
-        &self,
-        _query: CommandGetExportedKeys,
-        _request: Request<FlightDescriptor>,
-    ) -> Result<Response<FlightInfo>, Status> {
-        Err(Status::unimplemented(
-            "get_flight_info_exported_keys not implemented",
-        ))
-    }
-
-    async fn get_flight_info_imported_keys(
-        &self,
-        _query: CommandGetImportedKeys,
-        _request: Request<FlightDescriptor>,
-    ) -> Result<Response<FlightInfo>, Status> {
-        Err(Status::unimplemented(
-            "get_flight_info_imported_keys not implemented",
-        ))
-    }
-
-    async fn get_flight_info_cross_reference(
-        &self,
-        _query: CommandGetCrossReference,
-        _request: Request<FlightDescriptor>,
-    ) -> Result<Response<FlightInfo>, Status> {
-        Err(Status::unimplemented(
-            "get_flight_info_imported_keys not implemented",
-        ))
-    }
-
-    async fn get_flight_info_xdbc_type_info(
-        &self,
-        _query: CommandGetXdbcTypeInfo,
-        _request: Request<FlightDescriptor>,
-    ) -> Result<Response<FlightInfo>, Status> {
-        Err(Status::unimplemented(
-            "get_flight_info_xdbc_type_info not implemented",
-        ))
-    }
-
-    // do_get
-    async fn do_get_statement(
-        &self,
-        _ticket: TicketStatementQuery,
-        _request: Request<Ticket>,
-    ) -> Result<Response<<Self as FlightService>::DoGetStream>, Status> {
-        Err(Status::unimplemented("do_get_statement not implemented"))
-    }
-
-    async fn do_get_prepared_statement(
-        &self,
-        _query: CommandPreparedStatementQuery,
-        _request: Request<Ticket>,
-    ) -> Result<Response<<Self as FlightService>::DoGetStream>, Status> {
-        Err(Status::unimplemented(
-            "do_get_prepared_statement not implemented",
-        ))
-    }
-
-    async fn do_get_catalogs(
-        &self,
-        _query: CommandGetCatalogs,
-        _request: Request<Ticket>,
-    ) -> Result<Response<<Self as FlightService>::DoGetStream>, Status> {
-        Err(Status::unimplemented("do_get_catalogs not implemented"))
-    }
-
-    async fn do_get_schemas(
-        &self,
-        _query: CommandGetDbSchemas,
-        _request: Request<Ticket>,
-    ) -> Result<Response<<Self as FlightService>::DoGetStream>, Status> {
-        Err(Status::unimplemented("do_get_schemas not implemented"))
-    }
-
-    async fn do_get_tables(
-        &self,
-        _query: CommandGetTables,
-        _request: Request<Ticket>,
-    ) -> Result<Response<<Self as FlightService>::DoGetStream>, Status> {
-        Err(Status::unimplemented("do_get_tables not implemented"))
-    }
-
-    async fn do_get_table_types(
-        &self,
-        _query: CommandGetTableTypes,
-        _request: Request<Ticket>,
-    ) -> Result<Response<<Self as FlightService>::DoGetStream>, Status> {
-        Err(Status::unimplemented("do_get_table_types not implemented"))
-    }
-
-    async fn do_get_sql_info(
-        &self,
-        _query: CommandGetSqlInfo,
-        _request: Request<Ticket>,
-    ) -> Result<Response<<Self as FlightService>::DoGetStream>, Status> {
-        Err(Status::unimplemented("do_get_sql_info not implemented"))
-    }
-
-    async fn do_get_primary_keys(
-        &self,
-        _query: CommandGetPrimaryKeys,
-        _request: Request<Ticket>,
-    ) -> Result<Response<<Self as FlightService>::DoGetStream>, Status> {
-        Err(Status::unimplemented("do_get_primary_keys not implemented"))
-    }
-
-    async fn do_get_exported_keys(
-        &self,
-        _query: CommandGetExportedKeys,
-        _request: Request<Ticket>,
-    ) -> Result<Response<<Self as FlightService>::DoGetStream>, Status> {
-        Err(Status::unimplemented(
-            "do_get_exported_keys not implemented",
-        ))
-    }
-
-    async fn do_get_imported_keys(
-        &self,
-        _query: CommandGetImportedKeys,
-        _request: Request<Ticket>,
-    ) -> Result<Response<<Self as FlightService>::DoGetStream>, Status> {
-        Err(Status::unimplemented(
-            "do_get_imported_keys not implemented",
-        ))
-    }
-
-    async fn do_get_cross_reference(
-        &self,
-        _query: CommandGetCrossReference,
-        _request: Request<Ticket>,
-    ) -> Result<Response<<Self as FlightService>::DoGetStream>, Status> {
-        Err(Status::unimplemented(
-            "do_get_cross_reference not implemented",
-        ))
-    }
-
-    async fn do_get_xdbc_type_info(
-        &self,
-        _query: CommandGetXdbcTypeInfo,
-        _request: Request<Ticket>,
-    ) -> Result<Response<<Self as FlightService>::DoGetStream>, Status> {
-        Err(Status::unimplemented(
-            "do_get_xdbc_type_info not implemented",
-        ))
-    }
-
-    // do_put
-    async fn do_put_statement_update(
-        &self,
-        _ticket: CommandStatementUpdate,
-        _request: Request<PeekableFlightDataStream>,
-    ) -> Result<i64, Status> {
-        Err(Status::unimplemented(
-            "do_put_statement_update not implemented",
-        ))
-    }
-
-    async fn do_put_substrait_plan(
-        &self,
-        _ticket: CommandStatementSubstraitPlan,
-        _request: Request<PeekableFlightDataStream>,
-    ) -> Result<i64, Status> {
-        Err(Status::unimplemented(
-            "do_put_substrait_plan not implemented",
-        ))
-    }
-
     async fn do_put_prepared_statement_query(
         &self,
         _query: CommandPreparedStatementQuery,
@@ -590,16 +392,6 @@ impl FlightSqlService for FlightSqlServiceImpl {
         Ok(result)
     }
 
-    async fn do_put_prepared_statement_update(
-        &self,
-        _query: CommandPreparedStatementUpdate,
-        _request: Request<PeekableFlightDataStream>,
-    ) -> Result<i64, Status> {
-        Err(Status::unimplemented(
-            "do_put_prepared_statement_update not implemented",
-        ))
-    }
-
     async fn do_action_create_prepared_statement(
         &self,
         _query: ActionCreatePreparedStatementRequest,
@@ -609,60 +401,38 @@ impl FlightSqlService for FlightSqlServiceImpl {
             .map_err(|e| Status::internal(format!("Unable to serialize schema: {e}")))
     }
 
-    async fn do_action_close_prepared_statement(
-        &self,
-        _query: ActionClosePreparedStatementRequest,
-        _request: Request<Action>,
-    ) -> Result<(), Status> {
-        unimplemented!("Implement do_action_close_prepared_statement")
-    }
-
-    async fn do_action_create_prepared_substrait_plan(
-        &self,
-        _query: ActionCreatePreparedSubstraitPlanRequest,
-        _request: Request<Action>,
-    ) -> Result<ActionCreatePreparedStatementResult, Status> {
-        unimplemented!("Implement do_action_create_prepared_substrait_plan")
-    }
-
     async fn do_action_begin_transaction(
         &self,
         _query: ActionBeginTransactionRequest,
         _request: Request<Action>,
     ) -> Result<ActionBeginTransactionResult, Status> {
-        unimplemented!("Implement do_action_begin_transaction")
+        let transaction_id = Uuid::new_v4().to_string();
+        self.transactions
+            .lock()
+            .await
+            .insert(transaction_id.clone(), ());
+        Ok(ActionBeginTransactionResult {
+            transaction_id: transaction_id.as_bytes().to_vec().into(),
+        })
     }
 
     async fn do_action_end_transaction(
         &self,
-        _query: ActionEndTransactionRequest,
+        query: ActionEndTransactionRequest,
         _request: Request<Action>,
     ) -> Result<(), Status> {
-        unimplemented!("Implement do_action_end_transaction")
-    }
-
-    async fn do_action_begin_savepoint(
-        &self,
-        _query: ActionBeginSavepointRequest,
-        _request: Request<Action>,
-    ) -> Result<ActionBeginSavepointResult, Status> {
-        unimplemented!("Implement do_action_begin_savepoint")
-    }
-
-    async fn do_action_end_savepoint(
-        &self,
-        _query: ActionEndSavepointRequest,
-        _request: Request<Action>,
-    ) -> Result<(), Status> {
-        unimplemented!("Implement do_action_end_savepoint")
-    }
-
-    async fn do_action_cancel_query(
-        &self,
-        _query: ActionCancelQueryRequest,
-        _request: Request<Action>,
-    ) -> Result<ActionCancelQueryResult, Status> {
-        unimplemented!("Implement do_action_cancel_query")
+        let transaction_id = String::from_utf8(query.transaction_id.to_vec())
+            .map_err(|_| Status::invalid_argument("Invalid transaction id"))?;
+        if self
+            .transactions
+            .lock()
+            .await
+            .remove(&transaction_id)
+            .is_none()
+        {
+            return Err(Status::invalid_argument("Transaction id not found"));
+        }
+        Ok(())
     }
 
     async fn register_sql_info(&self, _id: i32, _result: &SqlInfo) {}
diff --git a/arrow-integration-test/src/lib.rs b/arrow-integration-test/src/lib.rs
index 66fa9f332..d1486fd5a 100644
--- a/arrow-integration-test/src/lib.rs
+++ b/arrow-integration-test/src/lib.rs
@@ -696,7 +696,7 @@ pub fn array_from_json(
             let list_data = ArrayData::builder(field.data_type().clone())
                 .len(json_col.count)
                 .offset(0)
-                .add_buffer(Buffer::from(&offsets.to_byte_slice()))
+                .add_buffer(Buffer::from(offsets.to_byte_slice()))
                 .add_child_data(child_array.into_data())
                 .null_bit_buffer(Some(null_buf))
                 .build()
@@ -720,7 +720,7 @@ pub fn array_from_json(
             let list_data = ArrayData::builder(field.data_type().clone())
                 .len(json_col.count)
                 .offset(0)
-                .add_buffer(Buffer::from(&offsets.to_byte_slice()))
+                .add_buffer(Buffer::from(offsets.to_byte_slice()))
                 .add_child_data(child_array.into_data())
                 .null_bit_buffer(Some(null_buf))
                 .build()
@@ -839,7 +839,7 @@ pub fn array_from_json(
                 .collect();
             let array_data = ArrayData::builder(field.data_type().clone())
                 .len(json_col.count)
-                .add_buffer(Buffer::from(&offsets.to_byte_slice()))
+                .add_buffer(Buffer::from(offsets.to_byte_slice()))
                 .add_child_data(child_array.into_data())
                 .null_bit_buffer(Some(null_buf))
                 .build()
diff --git a/arrow-integration-testing/src/flight_client_scenarios/integration_test.rs b/arrow-integration-testing/src/flight_client_scenarios/integration_test.rs
index ec88ce36a..1a6c4e28a 100644
--- a/arrow-integration-testing/src/flight_client_scenarios/integration_test.rs
+++ b/arrow-integration-testing/src/flight_client_scenarios/integration_test.rs
@@ -262,7 +262,7 @@ async fn receive_batch_flight_data(
 
     while message.header_type() == ipc::MessageHeader::DictionaryBatch {
         reader::read_dictionary(
-            &Buffer::from(&data.data_body),
+            &Buffer::from(data.data_body.as_ref()),
             message
                 .header_as_dictionary_batch()
                 .expect("Error parsing dictionary"),
diff --git a/arrow-integration-testing/src/flight_server_scenarios/integration_test.rs b/arrow-integration-testing/src/flight_server_scenarios/integration_test.rs
index a03c1cd1a..76eb9d880 100644
--- a/arrow-integration-testing/src/flight_server_scenarios/integration_test.rs
+++ b/arrow-integration-testing/src/flight_server_scenarios/integration_test.rs
@@ -364,7 +364,7 @@ async fn save_uploaded_chunks(
 
                 let batch = record_batch_from_message(
                     message,
-                    &Buffer::from(data.data_body),
+                    &Buffer::from(data.data_body.as_ref()),
                     schema_ref.clone(),
                     &dictionaries_by_id,
                 )
@@ -375,7 +375,7 @@ async fn save_uploaded_chunks(
             ipc::MessageHeader::DictionaryBatch => {
                 dictionary_from_message(
                     message,
-                    &Buffer::from(data.data_body),
+                    &Buffer::from(data.data_body.as_ref()),
                     schema_ref.clone(),
                     &mut dictionaries_by_id,
                 )
diff --git a/arrow/tests/array_equal.rs b/arrow/tests/array_equal.rs
index 15011c547..7ed4dae1e 100644
--- a/arrow/tests/array_equal.rs
+++ b/arrow/tests/array_equal.rs
@@ -445,7 +445,7 @@ fn test_empty_offsets_list_equal() {
     .len(0)
     .add_buffer(Buffer::from([0i32, 2, 3, 4, 6, 7, 8].to_byte_slice()))
     .add_child_data(Int32Array::from(vec![1, 2, -1, -2, 3, 4, -3, -4]).into_data())
-    .null_bit_buffer(Some(Buffer::from(vec![0b00001001])))
+    .null_bit_buffer(Some(Buffer::from([0b00001001])))
     .build()
     .unwrap()
     .into();
@@ -483,7 +483,7 @@ fn test_list_null() {
     .len(6)
     .add_buffer(Buffer::from([0i32, 2, 3, 4, 6, 7, 8].to_byte_slice()))
     .add_child_data(c_values.into_data())
-    .null_bit_buffer(Some(Buffer::from(vec![0b00001001])))
+    .null_bit_buffer(Some(Buffer::from([0b00001001])))
     .build()
     .unwrap()
     .into();
@@ -506,7 +506,7 @@ fn test_list_null() {
     .len(6)
     .add_buffer(Buffer::from([0i32, 2, 3, 4, 6, 7, 8].to_byte_slice()))
     .add_child_data(d_values.into_data())
-    .null_bit_buffer(Some(Buffer::from(vec![0b00001001])))
+    .null_bit_buffer(Some(Buffer::from([0b00001001])))
     .build()
     .unwrap()
     .into();
@@ -807,7 +807,7 @@ fn test_struct_equal_null() {
         Field::new("f1", DataType::Utf8, true),
         Field::new("f2", DataType::Int32, true),
     ])))
-    .null_bit_buffer(Some(Buffer::from(vec![0b00001011])))
+    .null_bit_buffer(Some(Buffer::from([0b00001011])))
     .len(5)
     .add_child_data(strings.to_data())
     .add_child_data(ints.to_data())
@@ -819,7 +819,7 @@ fn test_struct_equal_null() {
         Field::new("f1", DataType::Utf8, true),
         Field::new("f2", DataType::Int32, true),
     ])))
-    .null_bit_buffer(Some(Buffer::from(vec![0b00001011])))
+    .null_bit_buffer(Some(Buffer::from([0b00001011])))
     .len(5)
     .add_child_data(strings.to_data())
     .add_child_data(ints_non_null.to_data())
@@ -835,7 +835,7 @@ fn test_struct_equal_null() {
         Field::new("f1", DataType::Utf8, true),
         Field::new("f2", DataType::Int32, true),
     ])))
-    .null_bit_buffer(Some(Buffer::from(vec![0b00001011])))
+    .null_bit_buffer(Some(Buffer::from([0b00001011])))
     .len(5)
     .add_child_data(strings.to_data())
     .add_child_data(c_ints_non_null.to_data())
@@ -849,7 +849,7 @@ fn test_struct_equal_null() {
     let a = ArrayData::builder(DataType::Struct(
         vec![Field::new("f3", a.data_type().clone(), true)].into(),
     ))
-    .null_bit_buffer(Some(Buffer::from(vec![0b00011110])))
+    .null_bit_buffer(Some(Buffer::from([0b00011110])))
     .len(5)
     .add_child_data(a.to_data())
     .build()
@@ -868,7 +868,7 @@ fn test_struct_equal_null() {
         Field::new("f1", DataType::Utf8, true),
         Field::new("f2", DataType::Int32, true),
     ])))
-    .null_bit_buffer(Some(Buffer::from(vec![0b00001011])))
+    .null_bit_buffer(Some(Buffer::from([0b00001011])))
     .len(5)
     .add_child_data(strings.to_data())
     .add_child_data(ints_non_null.to_data())
@@ -878,7 +878,7 @@ fn test_struct_equal_null() {
     let b = ArrayData::builder(DataType::Struct(
         vec![Field::new("f3", b.data_type().clone(), true)].into(),
     ))
-    .null_bit_buffer(Some(Buffer::from(vec![0b00011110])))
+    .null_bit_buffer(Some(Buffer::from([0b00011110])))
     .len(5)
     .add_child_data(b)
     .build()
@@ -909,7 +909,7 @@ fn test_struct_equal_null_variable_size() {
     let a = ArrayData::builder(DataType::Struct(
         vec![Field::new("f1", DataType::Utf8, true)].into(),
     ))
-    .null_bit_buffer(Some(Buffer::from(vec![0b00001010])))
+    .null_bit_buffer(Some(Buffer::from([0b00001010])))
     .len(5)
     .add_child_data(strings1.to_data())
     .build()
@@ -919,7 +919,7 @@ fn test_struct_equal_null_variable_size() {
     let b = ArrayData::builder(DataType::Struct(
         vec![Field::new("f1", DataType::Utf8, true)].into(),
     ))
-    .null_bit_buffer(Some(Buffer::from(vec![0b00001010])))
+    .null_bit_buffer(Some(Buffer::from([0b00001010])))
     .len(5)
     .add_child_data(strings2.to_data())
     .build()
@@ -939,7 +939,7 @@ fn test_struct_equal_null_variable_size() {
     let c = ArrayData::builder(DataType::Struct(
         vec![Field::new("f1", DataType::Utf8, true)].into(),
     ))
-    .null_bit_buffer(Some(Buffer::from(vec![0b00001011])))
+    .null_bit_buffer(Some(Buffer::from([0b00001011])))
     .len(5)
     .add_child_data(strings3.to_data())
     .build()
diff --git a/arrow/tests/array_validation.rs b/arrow/tests/array_validation.rs
index 41def9051..1321f10f9 100644
--- a/arrow/tests/array_validation.rs
+++ b/arrow/tests/array_validation.rs
@@ -63,7 +63,7 @@ fn test_fixed_width_overflow() {
 #[should_panic(expected = "null_bit_buffer size too small. got 1 needed 2")]
 fn test_bitmap_too_small() {
     let buffer = make_i32_buffer(9);
-    let null_bit_buffer = Buffer::from(vec![0b11111111]);
+    let null_bit_buffer = Buffer::from([0b11111111]);
 
     ArrayData::try_new(
         DataType::Int32,