diff --git a/CHANGELOG.md b/CHANGELOG.md
index af56ae1c..b0cafded 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -22,6 +22,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   This has also added a new field, `api_version`, to the `AppInstanceSettings` and
   `DataSourceInstanceSettings` structs.
 - The `ArrayRefIntoField` trait is now correctly hidden behind the `arrow` feature flag.
+- Switch from `arrow2` to `arrow`.
+  The `arrow2` crate has been deprecated and is no longer maintained. The SDK now uses the
+  `arrow` crate instead, which is maintained by the Arrow project.
+  This is a breaking change because the `Field::values()` method now returns an
+  `Arc<dyn arrow::array::Array>` instead of the `arrow2` equivalent, and the
+  APIs for working with arrow arrays differs between the two crates.
+  Users who are just using the simple `Field::set_values` method or the various
+  `IntoField` / `IntoOptField` traits should not be affected.
 
 ## [0.5.0] - 2024-09-17
 
diff --git a/crates/grafana-plugin-sdk/Cargo.toml b/crates/grafana-plugin-sdk/Cargo.toml
index 89ba6f60..f7fff356 100644
--- a/crates/grafana-plugin-sdk/Cargo.toml
+++ b/crates/grafana-plugin-sdk/Cargo.toml
@@ -9,9 +9,7 @@ repository = "https://github.com/grafana/grafana-plugin-sdk-rust"
 description = "SDK for building Grafana backend plugins."
 
 [dependencies]
-arrow-array = { version = ">=40", optional = true } # synced with arrow2
-arrow-schema = { version = ">=40", optional = true } # synced with arrow2
-arrow2 = { version = "0.18.0", features = ["io_ipc"] }
+arrow = { version = "53.0.0", default-features = false, features = ["ipc"] }
 cfg-if = "1.0.0"
 chrono = "0.4.26"
 futures-core = "0.3.28"
@@ -63,7 +61,6 @@ all-features = true
 rustdoc-args = ["--cfg", "docsrs"]
 
 [features]
-arrow = ["dep:arrow-array", "dep:arrow-schema", "arrow2/arrow"]
 reqwest = ["reqwest_lib"]
 # Since prost 0.11 we no longer use prost-build at build time to generate code,
 # because it requires protoc. The generated code is instead checked in to source
diff --git a/crates/grafana-plugin-sdk/src/data/error.rs b/crates/grafana-plugin-sdk/src/data/error.rs
index 00e78dea..f69c4147 100644
--- a/crates/grafana-plugin-sdk/src/data/error.rs
+++ b/crates/grafana-plugin-sdk/src/data/error.rs
@@ -1,5 +1,5 @@
 //! Error types returned by the SDK.
-use arrow2::datatypes::DataType;
+use arrow::datatypes::DataType;
 use itertools::Itertools;
 use thiserror::Error;
 
diff --git a/crates/grafana-plugin-sdk/src/data/field.rs b/crates/grafana-plugin-sdk/src/data/field.rs
index be0c563c..5039d7c3 100644
--- a/crates/grafana-plugin-sdk/src/data/field.rs
+++ b/crates/grafana-plugin-sdk/src/data/field.rs
@@ -2,9 +2,10 @@
 use std::{
     collections::{BTreeMap, HashMap},
     iter::FromIterator,
+    sync::Arc,
 };
 
-use arrow2::{
+use arrow::{
     array::Array,
     datatypes::{DataType, Field as ArrowField, TimeUnit},
 };
@@ -40,7 +41,7 @@ pub struct Field {
     /// The types of values contained within the `Array` MUST match the
     /// type information in `type_info` at all times. The various `into_field`-like
     /// functions and the `set_values` methods should ensure this.
-    pub(crate) values: Box<dyn Array>,
+    pub(crate) values: Arc<dyn Array>,
     /// Type information for this field, as understood by Grafana.
     pub(crate) type_info: TypeInfo,
 }
@@ -62,12 +63,12 @@ impl Field {
             .into_iter()
             .collect(),
         };
-        Ok(ArrowField {
-            name: self.name.clone(),
-            data_type: self.type_info.frame.into(),
-            is_nullable: self.type_info.nullable.unwrap_or_default(),
-            metadata,
-        })
+        Ok(ArrowField::new(
+            self.name.clone(),
+            self.type_info.frame.into(),
+            self.type_info.nullable.unwrap_or_default(),
+        )
+        .with_metadata(metadata))
     }
 
     /// Return a new field with the given name.
@@ -131,11 +132,6 @@ impl Field {
 
     /// Get the values of this field as a [`&dyn Array`].
     pub fn values(&self) -> &dyn Array {
-        // This nightly clippy lint creates spurious suggestions for `&dyn Trait` return
-        // types. This can be uncommented when https://github.com/rust-lang/rust-clippy/pull/9126
-        // is released.
-        #[allow(unknown_lints)]
-        #[allow(clippy::explicit_auto_deref)]
         &*self.values
     }
 
@@ -147,7 +143,7 @@ impl Field {
     /// do not match the types of the existing data.
     ///
     /// ```rust
-    /// use arrow2::array::Utf8Array;
+    /// use arrow::array::StringArray;
     /// use grafana_plugin_sdk::prelude::*;
     ///
     /// let mut field = ["a", "b", "c"]
@@ -157,7 +153,7 @@ impl Field {
     ///     field
     ///         .values()
     ///         .as_any()
-    ///         .downcast_ref::<Utf8Array<i32>>()
+    ///         .downcast_ref::<StringArray>()
     ///         .unwrap()
     ///         .iter()
     ///         .collect::<Vec<_>>(),
@@ -171,7 +167,8 @@ impl Field {
         T: IntoIterator<Item = U>,
         U: IntoFieldType<ElementType = V>,
         V: FieldType,
-        V::Array: Array + FromIterator<Option<V>> + 'static,
+        V::InArray: Array + FromIterator<Option<V>> + 'static,
+        V::OutArray: Array + FromIterator<Option<V>> + 'static,
     {
         let new_data_type: DataType = U::TYPE_INFO_TYPE.into();
         if self.values.data_type() != &new_data_type {
@@ -181,12 +178,11 @@ impl Field {
                 field: self.name.clone(),
             });
         }
-        self.values = Box::new(V::convert_arrow_array(
+        self.values = Arc::new(V::convert_arrow_array(
             values
                 .into_iter()
                 .map(U::into_field_type)
-                .collect::<V::Array>(),
-            new_data_type,
+                .collect::<V::InArray>(),
         ));
         self.type_info.nullable = Some(false);
         Ok(())
@@ -200,7 +196,7 @@ impl Field {
     /// do not match the types of the existing data.
     ///
     /// ```rust
-    /// use arrow2::array::Utf8Array;
+    /// use arrow::array::StringArray;
     /// use grafana_plugin_sdk::prelude::*;
     ///
     /// let mut field = ["a", "b", "c"]
@@ -210,7 +206,7 @@ impl Field {
     ///     field
     ///         .values()
     ///         .as_any()
-    ///         .downcast_ref::<Utf8Array<i32>>()
+    ///         .downcast_ref::<StringArray>()
     ///         .unwrap()
     ///         .iter()
     ///         .collect::<Vec<_>>(),
@@ -224,7 +220,8 @@ impl Field {
         T: IntoIterator<Item = Option<U>>,
         U: IntoFieldType<ElementType = V>,
         V: FieldType,
-        V::Array: Array + FromIterator<Option<V>> + 'static,
+        V::InArray: Array + FromIterator<Option<V>> + 'static,
+        V::OutArray: Array + FromIterator<Option<V>> + 'static,
     {
         let new_data_type: DataType = U::TYPE_INFO_TYPE.into();
         if self.values.data_type() != &new_data_type {
@@ -234,12 +231,11 @@ impl Field {
                 field: self.name.clone(),
             });
         }
-        self.values = Box::new(V::convert_arrow_array(
+        self.values = Arc::new(V::convert_arrow_array(
             values
                 .into_iter()
                 .map(|x| x.and_then(U::into_field_type))
-                .collect::<V::Array>(),
-            new_data_type,
+                .collect::<V::InArray>(),
         ));
         self.type_info.nullable = Some(true);
         Ok(())
@@ -253,25 +249,25 @@ impl Field {
     /// do not match the types of the existing data.
     ///
     /// ```rust
-    /// use arrow2::array::{PrimitiveArray, Utf8Array};
+    /// use arrow::array::{PrimitiveArray, StringArray};
     /// use grafana_plugin_sdk::prelude::*;
     ///
     /// let mut field = ["a", "b", "c"]
     ///     .into_field("x");
-    /// let new_values = Utf8Array::<i32>::from(["d", "e", "f"].map(Some));
+    /// let new_values = StringArray::from(["d", "e", "f"].map(Some).to_vec());
     /// assert!(field.set_values_array(new_values).is_ok());
     /// assert_eq!(
     ///     field
     ///         .values()
     ///         .as_any()
-    ///         .downcast_ref::<Utf8Array<i32>>()
+    ///         .downcast_ref::<StringArray>()
     ///         .unwrap()
     ///         .iter()
     ///         .collect::<Vec<_>>(),
     ///     vec![Some("d"), Some("e"), Some("f")],
     /// );
     ///
-    /// let bad_values = PrimitiveArray::<u32>::from([1, 2, 3].map(Some));
+    /// let bad_values = PrimitiveArray::from([1u32, 2, 3].map(Some).to_vec());
     /// assert!(field.set_values_array(bad_values).is_err());
     /// ```
     pub fn set_values_array<T>(&mut self, values: T) -> Result<(), crate::data::error::Error>
@@ -285,7 +281,7 @@ impl Field {
                 field: self.name.clone(),
             });
         }
-        self.values = Box::new(values);
+        self.values = Arc::new(values);
         Ok(())
     }
 }
@@ -296,7 +292,7 @@ impl PartialEq for Field {
             && self.labels == other.labels
             && self.config == other.config
             && self.type_info == other.type_info
-            && arrow2::array::equal(&*self.values, &*other.values)
+            && other.values.eq(&self.values)
     }
 }
 
@@ -316,9 +312,9 @@ pub trait IntoField {
 impl<T, U, V> IntoField for T
 where
     T: IntoIterator<Item = U>,
-    U: IntoFieldType<ElementType = V>,
-    V: FieldType,
-    V::Array: Array + FromIterator<Option<V>> + 'static,
+    U: FieldType + IntoFieldType<ElementType = V>,
+    U::InArray: Array + FromIterator<Option<V>> + 'static,
+    U::OutArray: Array + FromIterator<Option<V>> + 'static,
 {
     fn into_field(self, name: impl Into<String>) -> Field {
         Field {
@@ -329,11 +325,10 @@ where
                 frame: U::TYPE_INFO_TYPE,
                 nullable: Some(false),
             },
-            values: Box::new(V::convert_arrow_array(
+            values: Arc::new(U::convert_arrow_array(
                 self.into_iter()
                     .map(U::into_field_type)
-                    .collect::<V::Array>(),
-                U::TYPE_INFO_TYPE.into(),
+                    .collect::<U::InArray>(),
             )),
         }
     }
@@ -348,9 +343,9 @@ pub trait IntoOptField {
 impl<T, U, V> IntoOptField for T
 where
     T: IntoIterator<Item = Option<U>>,
-    U: IntoFieldType<ElementType = V>,
-    V: FieldType,
-    V::Array: Array + FromIterator<Option<V>> + 'static,
+    U: FieldType + IntoFieldType<ElementType = V>,
+    U::InArray: Array + FromIterator<Option<V>> + 'static,
+    U::OutArray: Array + FromIterator<Option<V>> + 'static,
 {
     fn into_opt_field(self, name: impl Into<String>) -> Field {
         Field {
@@ -361,11 +356,10 @@ where
                 frame: U::TYPE_INFO_TYPE,
                 nullable: Some(true),
             },
-            values: Box::new(V::convert_arrow_array(
+            values: Arc::new(U::convert_arrow_array(
                 self.into_iter()
                     .map(|x| x.and_then(U::into_field_type))
-                    .collect::<V::Array>(),
-                U::TYPE_INFO_TYPE.into(),
+                    .collect::<U::InArray>(),
             )),
         }
     }
@@ -394,13 +388,12 @@ where
                 frame: self.data_type().try_into()?,
                 nullable: Some(true),
             },
-            values: Box::new(self),
+            values: Arc::new(self),
         })
     }
 }
 
-/// Helper trait for creating a [`Field`] from an [`Array`][arrow_array::Array].
-#[cfg(feature = "arrow")]
+/// Helper trait for creating a [`Field`] from an [`Array`].
 pub trait ArrayRefIntoField {
     /// Create a `Field` using `self` as the values.
     ///
@@ -410,8 +403,7 @@ pub trait ArrayRefIntoField {
     fn try_into_field(self, name: impl Into<String>) -> Result<Field, error::Error>;
 }
 
-#[cfg(feature = "arrow")]
-impl ArrayRefIntoField for &dyn ::arrow_array::Array {
+impl ArrayRefIntoField for Arc<dyn arrow::array::Array> {
     fn try_into_field(self, name: impl Into<String>) -> Result<Field, error::Error> {
         Ok(Field {
             name: name.into(),
@@ -421,7 +413,7 @@ impl ArrayRefIntoField for &dyn ::arrow_array::Array {
                 frame: self.data_type().try_into()?,
                 nullable: Some(true),
             },
-            values: self.into(),
+            values: self,
         })
     }
 }
@@ -523,51 +515,6 @@ impl From<TypeInfoType> for DataType {
     }
 }
 
-#[cfg(feature = "arrow")]
-impl TryFrom<&::arrow_schema::DataType> for TypeInfoType {
-    type Error = error::Error;
-    fn try_from(other: &::arrow_schema::DataType) -> Result<Self, Self::Error> {
-        Ok(match other {
-            ::arrow_schema::DataType::Int8 => Self::Int8,
-            ::arrow_schema::DataType::Int16 => Self::Int16,
-            ::arrow_schema::DataType::Int32 => Self::Int32,
-            ::arrow_schema::DataType::Int64 => Self::Int64,
-            ::arrow_schema::DataType::UInt8 => Self::UInt8,
-            ::arrow_schema::DataType::UInt16 => Self::UInt16,
-            ::arrow_schema::DataType::UInt32 => Self::UInt32,
-            ::arrow_schema::DataType::UInt64 => Self::UInt64,
-            ::arrow_schema::DataType::Float32 => Self::Float32,
-            ::arrow_schema::DataType::Float64 => Self::Float64,
-            ::arrow_schema::DataType::Utf8 => Self::String,
-            ::arrow_schema::DataType::Boolean => Self::Bool,
-            ::arrow_schema::DataType::Timestamp(..) => Self::Time,
-            // TODO - handle time correctly.
-            other => return Err(error::Error::UnsupportedArrowDataType(other.clone().into())),
-        })
-    }
-}
-
-#[cfg(feature = "arrow")]
-impl From<TypeInfoType> for ::arrow_schema::DataType {
-    fn from(other: TypeInfoType) -> Self {
-        match other {
-            TypeInfoType::Int8 => Self::Int8,
-            TypeInfoType::Int16 => Self::Int16,
-            TypeInfoType::Int32 => Self::Int32,
-            TypeInfoType::Int64 => Self::Int64,
-            TypeInfoType::UInt8 => Self::UInt8,
-            TypeInfoType::UInt16 => Self::UInt16,
-            TypeInfoType::UInt32 => Self::UInt32,
-            TypeInfoType::UInt64 => Self::UInt64,
-            TypeInfoType::Float32 => Self::Float32,
-            TypeInfoType::Float64 => Self::Float64,
-            TypeInfoType::String => Self::Utf8,
-            TypeInfoType::Bool => Self::Boolean,
-            TypeInfoType::Time => Self::Timestamp(::arrow_schema::TimeUnit::Nanosecond, None),
-        }
-    }
-}
-
 impl TypeInfoType {
     #[must_use]
     pub(crate) const fn simple_type(&self) -> SimpleType {
@@ -860,8 +807,8 @@ mod tests {
 
                 #[test]
                 #[allow(non_snake_case)]
-                fn [< create_field_from_array_ $t >]() {
-                    let array = <$t as FieldType>::Array::from_slice([<$t>::default()]);
+                fn [< create_field_from_vec_ $t >]() {
+                    let array = <$t as FieldType>::InArray::from(vec![<$t>::default()]);
                     let field = array.try_into_field("x".to_string()).unwrap();
                     assert_eq!(field.name, "x");
                     assert_eq!(field.values.len(), 1)
diff --git a/crates/grafana-plugin-sdk/src/data/field_type.rs b/crates/grafana-plugin-sdk/src/data/field_type.rs
index c81ac419..d4bd3940 100644
--- a/crates/grafana-plugin-sdk/src/data/field_type.rs
+++ b/crates/grafana-plugin-sdk/src/data/field_type.rs
@@ -1,9 +1,12 @@
 //! Types of field understood by the Grafana plugin SDK.
 use std::time::{SystemTime, UNIX_EPOCH};
 
-use arrow2::{
-    array::{BooleanArray, PrimitiveArray, Utf8Array},
-    datatypes::{DataType, TimeUnit},
+use arrow::{
+    array::{BooleanArray, PrimitiveArray, StringArray},
+    datatypes::{
+        DataType, Float32Type, Float64Type, Int16Type, Int32Type, Int64Type, Int8Type, TimeUnit,
+        TimestampNanosecondType, UInt16Type, UInt32Type, UInt64Type, UInt8Type,
+    },
 };
 use chrono::prelude::*;
 
@@ -12,17 +15,14 @@ use crate::data::TypeInfoType;
 /// Indicates that a type is can be stored in an Arrow array.
 pub trait FieldType {
     /// The type of arrow array this field type is stored in.
-    type Array;
+    type InArray;
+    /// The type of arrow array to convert to when storing values in a `Field`.
+    type OutArray;
     /// The logical arrow data type that an arrow array of this data should have.
     const ARROW_DATA_TYPE: DataType;
 
-    /// Convert the logical type of `Self::Array`, if needed.
-    ///
-    /// The default implementation is a no-op, but some field types may need to
-    /// implement this to ensure the underlying boxed Arrow array can be downcast correctly.
-    fn convert_arrow_array(array: Self::Array, _data_type: DataType) -> Self::Array {
-        array
-    }
+    /// Convert the logical type of `Self::InArray`.
+    fn convert_arrow_array(array: Self::InArray) -> Self::OutArray;
 }
 
 /// Indicates that a type can be converted to one that is [`FieldType`], and holds associated metadata.
@@ -56,12 +56,13 @@ where
 }
 
 macro_rules! impl_fieldtype_for_primitive {
-    ($ty: ty, $arrow_ty: expr, $type_info: expr) => {
+    ($ty: ty, $arrow_ty: ty, $arrow_data_ty: expr, $type_info: expr) => {
         impl FieldType for $ty {
-            type Array = PrimitiveArray<$ty>;
-            const ARROW_DATA_TYPE: DataType = $arrow_ty;
-            fn convert_arrow_array(array: Self::Array, data_type: DataType) -> Self::Array {
-                array.to(data_type)
+            type InArray = PrimitiveArray<$arrow_ty>;
+            type OutArray = PrimitiveArray<$arrow_ty>;
+            const ARROW_DATA_TYPE: DataType = $arrow_data_ty;
+            fn convert_arrow_array(array: Self::InArray) -> Self::OutArray {
+                array
             }
         }
 
@@ -75,24 +76,25 @@ macro_rules! impl_fieldtype_for_primitive {
     };
 }
 
-impl_fieldtype_for_primitive!(i8, DataType::Int8, TypeInfoType::Int8);
-impl_fieldtype_for_primitive!(i16, DataType::Int16, TypeInfoType::Int16);
-impl_fieldtype_for_primitive!(i32, DataType::Int32, TypeInfoType::Int32);
-impl_fieldtype_for_primitive!(i64, DataType::Int64, TypeInfoType::Int64);
-impl_fieldtype_for_primitive!(u8, DataType::UInt8, TypeInfoType::UInt8);
-impl_fieldtype_for_primitive!(u16, DataType::UInt16, TypeInfoType::UInt16);
-impl_fieldtype_for_primitive!(u32, DataType::UInt32, TypeInfoType::UInt32);
-impl_fieldtype_for_primitive!(u64, DataType::UInt64, TypeInfoType::UInt64);
-impl_fieldtype_for_primitive!(f32, DataType::Float32, TypeInfoType::Float32);
-impl_fieldtype_for_primitive!(f64, DataType::Float64, TypeInfoType::Float64);
+impl_fieldtype_for_primitive!(i8, Int8Type, DataType::Int8, TypeInfoType::Int8);
+impl_fieldtype_for_primitive!(i16, Int16Type, DataType::Int16, TypeInfoType::Int16);
+impl_fieldtype_for_primitive!(i32, Int32Type, DataType::Int32, TypeInfoType::Int32);
+impl_fieldtype_for_primitive!(i64, Int64Type, DataType::Int64, TypeInfoType::Int64);
+impl_fieldtype_for_primitive!(u8, UInt8Type, DataType::UInt8, TypeInfoType::UInt8);
+impl_fieldtype_for_primitive!(u16, UInt16Type, DataType::UInt16, TypeInfoType::UInt16);
+impl_fieldtype_for_primitive!(u32, UInt32Type, DataType::UInt32, TypeInfoType::UInt32);
+impl_fieldtype_for_primitive!(u64, UInt64Type, DataType::UInt64, TypeInfoType::UInt64);
+impl_fieldtype_for_primitive!(f32, Float32Type, DataType::Float32, TypeInfoType::Float32);
+impl_fieldtype_for_primitive!(f64, Float64Type, DataType::Float64, TypeInfoType::Float64);
 
 // Boolean impl.
 
 impl FieldType for bool {
-    type Array = BooleanArray;
+    type InArray = BooleanArray;
+    type OutArray = BooleanArray;
     const ARROW_DATA_TYPE: DataType = DataType::Boolean;
 
-    fn convert_arrow_array(array: Self::Array, _data_type: DataType) -> Self::Array {
+    fn convert_arrow_array(array: Self::InArray) -> Self::OutArray {
         array
     }
 }
@@ -109,12 +111,13 @@ impl IntoFieldType for bool {
 // DateTime impls.
 
 impl FieldType for SystemTime {
-    type Array = PrimitiveArray<i64>;
+    type InArray = PrimitiveArray<Int64Type>;
+    type OutArray = PrimitiveArray<TimestampNanosecondType>;
     const ARROW_DATA_TYPE: DataType = DataType::Timestamp(TimeUnit::Nanosecond, None);
 
     /// Convert the logical type of `Self::Array` to `DataType::Timestamp`.
-    fn convert_arrow_array(array: Self::Array, data_type: DataType) -> Self::Array {
-        array.to(data_type)
+    fn convert_arrow_array(array: Self::InArray) -> Self::OutArray {
+        array.reinterpret_cast()
     }
 }
 
@@ -133,12 +136,13 @@ impl<T> FieldType for DateTime<T>
 where
     T: Offset + TimeZone,
 {
-    type Array = PrimitiveArray<i64>;
+    type InArray = PrimitiveArray<Int64Type>;
+    type OutArray = PrimitiveArray<TimestampNanosecondType>;
     const ARROW_DATA_TYPE: DataType = DataType::Timestamp(TimeUnit::Nanosecond, None);
 
     /// Convert the logical type of `Self::Array` to `DataType::Timestamp`.
-    fn convert_arrow_array(array: Self::Array, data_type: DataType) -> Self::Array {
-        array.to(data_type)
+    fn convert_arrow_array(array: Self::InArray) -> Self::OutArray {
+        array.reinterpret_cast()
     }
 }
 
@@ -160,12 +164,13 @@ impl<T> FieldType for Date<T>
 where
     T: Offset + TimeZone,
 {
-    type Array = PrimitiveArray<i64>;
+    type InArray = PrimitiveArray<Int64Type>;
+    type OutArray = PrimitiveArray<TimestampNanosecondType>;
     const ARROW_DATA_TYPE: DataType = DataType::Timestamp(TimeUnit::Nanosecond, None);
 
     /// Convert the logical type of `Self::Array` to `DataType::Timestamp`.
-    fn convert_arrow_array(array: Self::Array, data_type: DataType) -> Self::Array {
-        array.to(data_type)
+    fn convert_arrow_array(array: Self::InArray) -> Self::OutArray {
+        array.reinterpret_cast()
     }
 }
 
@@ -186,12 +191,13 @@ where
 }
 
 impl FieldType for NaiveDate {
-    type Array = PrimitiveArray<i64>;
+    type InArray = PrimitiveArray<Int64Type>;
+    type OutArray = PrimitiveArray<TimestampNanosecondType>;
     const ARROW_DATA_TYPE: DataType = DataType::Timestamp(TimeUnit::Nanosecond, None);
 
     /// Convert the logical type of `Self::Array` to `DataType::Timestamp`.
-    fn convert_arrow_array(array: Self::Array, data_type: DataType) -> Self::Array {
-        array.to(data_type)
+    fn convert_arrow_array(array: Self::InArray) -> Self::OutArray {
+        array.reinterpret_cast()
     }
 }
 
@@ -207,12 +213,13 @@ impl IntoFieldType for NaiveDate {
 }
 
 impl FieldType for NaiveDateTime {
-    type Array = PrimitiveArray<i64>;
+    type InArray = PrimitiveArray<Int64Type>;
+    type OutArray = PrimitiveArray<TimestampNanosecondType>;
     const ARROW_DATA_TYPE: DataType = DataType::Timestamp(TimeUnit::Nanosecond, None);
 
     /// Convert the logical type of `Self::Array` to `DataType::Timestamp`.
-    fn convert_arrow_array(array: Self::Array, data_type: DataType) -> Self::Array {
-        array.to(data_type)
+    fn convert_arrow_array(array: Self::InArray) -> Self::OutArray {
+        array.reinterpret_cast()
     }
 }
 
@@ -227,8 +234,13 @@ impl IntoFieldType for NaiveDateTime {
 // String impls.
 
 impl FieldType for &str {
-    type Array = Utf8Array<i32>;
+    type InArray = StringArray;
+    type OutArray = StringArray;
     const ARROW_DATA_TYPE: DataType = DataType::Utf8;
+
+    fn convert_arrow_array(array: Self::InArray) -> Self::OutArray {
+        array
+    }
 }
 
 impl<'a> IntoFieldType for &'a str {
@@ -240,8 +252,13 @@ impl<'a> IntoFieldType for &'a str {
 }
 
 impl FieldType for String {
-    type Array = Utf8Array<i32>;
+    type InArray = StringArray;
+    type OutArray = StringArray;
     const ARROW_DATA_TYPE: DataType = DataType::Utf8;
+
+    fn convert_arrow_array(array: Self::InArray) -> Self::OutArray {
+        array
+    }
 }
 
 impl IntoFieldType for String {
diff --git a/crates/grafana-plugin-sdk/src/data/frame/de.rs b/crates/grafana-plugin-sdk/src/data/frame/de.rs
index a044a51d..a7d6dca1 100644
--- a/crates/grafana-plugin-sdk/src/data/frame/de.rs
+++ b/crates/grafana-plugin-sdk/src/data/frame/de.rs
@@ -1,19 +1,22 @@
 //! Deserialization of [`Frame`]s from the JSON format.
-use std::{collections::BTreeMap, fmt, marker::PhantomData};
+use std::{collections::BTreeMap, fmt, marker::PhantomData, sync::Arc};
 
-use arrow2::{
+use arrow::{
     array::{
-        Array, MutableArray, MutableBooleanArray, MutablePrimitiveArray, MutableUtf8Array, TryPush,
+        Array, BooleanArray, Float32Array, Float64Array, Int16Array, Int32Array, Int64Array,
+        Int8Array, StringArray, TimestampNanosecondArray, UInt16Array, UInt32Array, UInt64Array,
+        UInt8Array,
+    },
+    datatypes::{
+        ArrowPrimitiveType, Float32Type, Float64Type, Int16Type, Int32Type, Int64Type, Int8Type,
+        UInt16Type, UInt32Type, UInt64Type, UInt8Type,
     },
-    datatypes::{DataType, TimeUnit},
-    types::{NativeType, Offset},
 };
 use num_traits::Float;
 use serde::{
-    de::{Deserializer, Error, MapAccess, SeqAccess, Visitor},
+    de::{DeserializeOwned, DeserializeSeed, Deserializer, Error, MapAccess, SeqAccess, Visitor},
     Deserialize,
 };
-use serde_json::from_str;
 
 use crate::data::{
     field::{FieldConfig, SimpleType, TypeInfo, TypeInfoType},
@@ -136,7 +139,7 @@ struct RawData<'a> {
 
 #[derive(Debug)]
 struct Data {
-    values: Vec<Box<dyn Array>>,
+    values: Vec<Arc<dyn Array>>,
 }
 
 impl TryFrom<(&'_ Schema, RawData<'_>)> for Data {
@@ -161,50 +164,33 @@ impl TryFrom<(&'_ Schema, RawData<'_>)> for Data {
                 .zip(entities)
                 .map(|((f, v), e)| {
                     let s = v.get();
-                    let arr: Box<dyn Array> = match f.type_info.frame {
-                        TypeInfoType::Int8 => {
-                            parse_array::<MutablePrimitiveArray<i8>, i8, ()>(s)?.as_box()
-                        }
-                        TypeInfoType::Int16 => {
-                            parse_array::<MutablePrimitiveArray<i16>, i16, ()>(s)?.as_box()
-                        }
-                        TypeInfoType::Int32 => {
-                            parse_array::<MutablePrimitiveArray<i32>, i32, ()>(s)?.as_box()
-                        }
-                        TypeInfoType::Int64 => {
-                            parse_array::<MutablePrimitiveArray<i64>, i64, ()>(s)?.as_box()
-                        }
-                        TypeInfoType::UInt8 => {
-                            parse_array::<MutablePrimitiveArray<u8>, u8, ()>(s)?.as_box()
-                        }
+                    let arr: Arc<dyn Array> = match f.type_info.frame {
+                        TypeInfoType::Int8 => parse_primitive_array::<Int8Array, Int8Type>(s)?,
+                        TypeInfoType::Int16 => parse_primitive_array::<Int16Array, Int16Type>(s)?,
+                        TypeInfoType::Int32 => parse_primitive_array::<Int32Array, Int32Type>(s)?,
+                        TypeInfoType::Int64 => parse_primitive_array::<Int64Array, Int64Type>(s)?,
+                        TypeInfoType::UInt8 => parse_primitive_array::<UInt8Array, UInt8Type>(s)?,
                         TypeInfoType::UInt16 => {
-                            parse_array::<MutablePrimitiveArray<u16>, u16, ()>(s)?.as_box()
+                            parse_primitive_array::<UInt16Array, UInt16Type>(s)?
                         }
                         TypeInfoType::UInt32 => {
-                            parse_array::<MutablePrimitiveArray<u32>, u32, ()>(s)?.as_box()
+                            parse_primitive_array::<UInt32Array, UInt32Type>(s)?
                         }
                         TypeInfoType::UInt64 => {
-                            parse_array::<MutablePrimitiveArray<u64>, u64, ()>(s)?.as_box()
+                            parse_primitive_array::<UInt64Array, UInt64Type>(s)?
                         }
+                        // TypeInfoType::Float16 => {
+                        //     parse_array_with_entities::<Float16Array, Float16Type>(s, e)?
+                        // }
                         TypeInfoType::Float32 => {
-                            parse_array_with_entities::<MutablePrimitiveArray<f32>, f32>(s, e)?
-                                .as_box()
+                            parse_array_with_entities::<Float32Array, Float32Type>(s, e)?
                         }
                         TypeInfoType::Float64 => {
-                            parse_array_with_entities::<MutablePrimitiveArray<f64>, f64>(s, e)?
-                                .as_box()
-                        }
-                        TypeInfoType::String => {
-                            parse_array::<MutableUtf8Array<i32>, String, ()>(s)?.as_box()
-                        }
-                        TypeInfoType::Bool => {
-                            parse_array::<MutableBooleanArray, bool, ()>(s)?.as_box()
-                        }
-                        TypeInfoType::Time => {
-                            parse_array::<MutablePrimitiveArray<i64>, i64, TimestampProcessor>(s)?
-                                .to(DataType::Timestamp(TimeUnit::Nanosecond, None))
-                                .as_box()
+                            parse_array_with_entities::<Float64Array, Float64Type>(s, e)?
                         }
+                        TypeInfoType::String => parse_string_array(s)?,
+                        TypeInfoType::Bool => parse_bool_array(s)?,
+                        TypeInfoType::Time => parse_timestamp_array(s)?,
                     };
                     Ok(arr)
                 })
@@ -220,13 +206,32 @@ impl TryFrom<(&'_ Schema, RawData<'_>)> for Data {
 /// Returns an error if the string is invalid JSON, if the elements of
 /// the array are not of type `U`,  or if the Arrow buffer could not be
 /// created.
-fn parse_array<'de, T, U, V>(s: &'de str) -> Result<T, serde_json::Error>
+fn parse_primitive_array<T, U>(s: &str) -> Result<Arc<dyn Array>, serde_json::Error>
 where
-    T: Default + MutableArray + TryPush<Option<U>> + WithCapacity,
-    U: Deserialize<'de>,
-    V: ElementProcessor<U>,
+    T: From<Vec<Option<U::Native>>> + Array + 'static,
+    U: ArrowPrimitiveType,
+    U::Native: DeserializeOwned,
 {
-    Ok(from_str::<DeArray<T, U, V>>(s)?.array)
+    let v: Vec<Option<U::Native>> = serde_json::from_str(s)?;
+    Ok(Arc::new(T::from(v)))
+}
+
+fn parse_timestamp_array(s: &str) -> Result<Arc<dyn Array>, serde_json::Error> {
+    let v: Vec<Option<i64>> = serde_json::from_str::<Vec<Option<i64>>>(s)?
+        .into_iter()
+        .map(|opt| opt.map(|x| x * 1_000_000))
+        .collect();
+    Ok(Arc::new(TimestampNanosecondArray::from(v)))
+}
+
+fn parse_string_array(s: &str) -> Result<Arc<dyn Array>, serde_json::Error> {
+    let v: Vec<Option<String>> = serde_json::from_str(s)?;
+    Ok(Arc::new(StringArray::from(v)))
+}
+
+fn parse_bool_array(s: &str) -> Result<Arc<dyn Array>, serde_json::Error> {
+    let v: Vec<Option<bool>> = serde_json::from_str(s)?;
+    Ok(Arc::new(BooleanArray::from(v)))
 }
 
 /// Parse a JSON array containing elements of `U` into a mutable Arrow array `T`,
@@ -242,48 +247,38 @@ where
 ///
 /// Panics if any of the indexes in `entities` are greater than the length
 /// of the parsed array.
-fn parse_array_with_entities<'de, T, U>(
-    s: &'de str,
+fn parse_array_with_entities<T, U>(
+    s: &str,
     entities: Option<Entities>,
-) -> Result<T, serde_json::Error>
+) -> Result<Arc<dyn Array>, serde_json::Error>
 where
-    T: Default + MutableArray + SetArray<Option<U>> + TryPush<Option<U>> + WithCapacity,
-    U: Deserialize<'de> + Float,
+    T: From<Vec<Option<U::Native>>> + Array + 'static,
+    U: ArrowPrimitiveType,
+    U::Native: DeserializeOwned + Float,
 {
-    let mut arr = from_str::<DeArray<T, U>>(s)?.array;
-    if let Some(e) = entities {
-        e.nan.iter().for_each(|idx| arr.set(*idx, Some(U::nan())));
-        e.inf
-            .iter()
-            .for_each(|idx| arr.set(*idx, Some(U::infinity())));
-        e.neg_inf
-            .iter()
-            .for_each(|idx| arr.set(*idx, Some(U::neg_infinity())));
-    }
-    Ok(arr)
-}
-
-trait ElementProcessor<T> {
-    fn process_element(el: T) -> T {
-        el
-    }
-}
-
-impl<T> ElementProcessor<T> for () {}
-
-struct TimestampProcessor;
-impl ElementProcessor<i64> for TimestampProcessor {
-    fn process_element(el: i64) -> i64 {
-        el * 1_000_000
-    }
+    let de = DeArray::<T, U>::new(entities);
+    let mut deserializer = serde_json::Deserializer::from_str(s);
+    Ok(Arc::new(de.deserialize(&mut deserializer)?))
 }
 
 /// Helper struct used to deserialize a sequence into an Arrow `Array`.
 #[derive(Debug)]
-struct DeArray<T, U, V = ()> {
-    array: T,
+struct DeArray<T, U> {
+    entities: Option<Entities>,
+    // The type of the final array.
+    t: PhantomData<T>,
+    // The type of the elements in the array.
     u: PhantomData<U>,
-    v: PhantomData<V>,
+}
+
+impl<T, U> DeArray<T, U> {
+    fn new(entities: Option<Entities>) -> Self {
+        Self {
+            entities,
+            t: PhantomData,
+            u: PhantomData,
+        }
+    }
 }
 
 // Deserialization for mutable Arrow arrays.
@@ -292,25 +287,27 @@ struct DeArray<T, U, V = ()> {
 // All of the `Mutable` variants of Arrow arrays implement `TryPush<Option<U>>`
 // for some relevant `U`, and here we just impose that the `U` is `Deserialize`
 // and gradually build up the array.
-impl<'de, T, U, V> Deserialize<'de> for DeArray<T, U, V>
+impl<'de, T, U> DeserializeSeed<'de> for DeArray<T, U>
 where
-    T: Default + TryPush<Option<U>> + WithCapacity,
-    U: Deserialize<'de>,
-    V: ElementProcessor<U>,
+    T: From<Vec<Option<U::Native>>> + Array + 'static,
+    U: ArrowPrimitiveType,
+    U::Native: Deserialize<'de> + Float,
 {
-    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+    type Value = T;
+
+    fn deserialize<D>(self, deserializer: D) -> Result<T, D::Error>
     where
         D: Deserializer<'de>,
     {
-        struct ArrayVisitor<T, U, V>(PhantomData<T>, PhantomData<U>, PhantomData<V>);
+        struct ArrayVisitor<T, U>(Option<Entities>, PhantomData<T>, PhantomData<U>);
 
-        impl<'de, T, U, V> Visitor<'de> for ArrayVisitor<T, U, V>
+        impl<'de, T, U> Visitor<'de> for ArrayVisitor<T, U>
         where
-            T: Default + TryPush<Option<U>> + WithCapacity,
-            U: Deserialize<'de>,
-            V: ElementProcessor<U>,
+            T: From<Vec<Option<U::Native>>> + Array + 'static,
+            U: ArrowPrimitiveType,
+            U::Native: Deserialize<'de> + Float,
         {
-            type Value = DeArray<T, U, V>;
+            type Value = T;
 
             fn expecting(&self, formatter: &mut fmt::Formatter) -> fmt::Result {
                 formatter.write_str("a heterogeneous array of compatible values")
@@ -320,67 +317,37 @@ where
             where
                 S: SeqAccess<'de>,
             {
-                let mut array = seq.size_hint().map_or_else(T::default, T::with_capacity);
-                while let Some(x) = seq.next_element::<Option<U>>()? {
-                    array.try_push(x.map(V::process_element)).map_err(|e| {
-                        S::Error::custom(format!("could not push to Arrow array: {}", e))
-                    })?;
+                let mut array = Vec::with_capacity(seq.size_hint().unwrap_or(0));
+                while let Some(opt) = seq.next_element::<Option<U::Native>>()? {
+                    array.push(opt);
                 }
-                Ok(Self::Value {
-                    array,
-                    u: PhantomData,
-                    v: PhantomData,
-                })
+
+                if let Some(e) = self.0 {
+                    e.nan.iter().for_each(|idx| {
+                        if let Some(v) = array.get_mut(*idx) {
+                            *v = Some(U::Native::nan());
+                        }
+                    });
+                    e.inf.iter().for_each(|idx| {
+                        if let Some(v) = array.get_mut(*idx) {
+                            *v = Some(U::Native::infinity());
+                        }
+                    });
+                    e.neg_inf.iter().for_each(|idx| {
+                        if let Some(v) = array.get_mut(*idx) {
+                            *v = Some(U::Native::neg_infinity());
+                        }
+                    });
+                }
+                Ok(array.into())
             }
         }
 
-        deserializer.deserialize_seq(ArrayVisitor(PhantomData, PhantomData, PhantomData))
-    }
-}
-
-/// Indicates that an array can have its elements mutated.
-trait SetArray<T> {
-    /// Set the value at `index` to `value`.
-    ///
-    /// Note that if it is the first time a null appears in this array,
-    /// this initializes the validity bitmap (`O(N)`).
-    ///
-    /// # Panics
-    ///
-    /// Panics iff index is larger than `self.len()`.
-    fn set(&mut self, index: usize, value: T);
-}
-
-impl<T: NativeType> SetArray<Option<T>> for MutablePrimitiveArray<T> {
-    fn set(&mut self, index: usize, value: Option<T>) {
-        self.set(index, value);
-    }
-}
-
-/// Indicates that an array can be created with a specified capacity.
-trait WithCapacity {
-    /// Create `Self` with the given `capacity` preallocated.
-    ///
-    /// This generally just delegates to the `with_capacity` method on
-    /// individual arrays.
-    fn with_capacity(capacity: usize) -> Self;
-}
-
-impl<T: NativeType> WithCapacity for MutablePrimitiveArray<T> {
-    fn with_capacity(capacity: usize) -> Self {
-        Self::with_capacity(capacity)
-    }
-}
-
-impl WithCapacity for MutableBooleanArray {
-    fn with_capacity(capacity: usize) -> Self {
-        Self::with_capacity(capacity)
-    }
-}
-
-impl<T: Offset> WithCapacity for MutableUtf8Array<T> {
-    fn with_capacity(capacity: usize) -> Self {
-        Self::with_capacity(capacity)
+        deserializer.deserialize_seq(ArrayVisitor::<T, U>(
+            self.entities,
+            PhantomData,
+            PhantomData,
+        ))
     }
 }
 
diff --git a/crates/grafana-plugin-sdk/src/data/frame/mod.rs b/crates/grafana-plugin-sdk/src/data/frame/mod.rs
index 4a71f4ca..128b2bf9 100644
--- a/crates/grafana-plugin-sdk/src/data/frame/mod.rs
+++ b/crates/grafana-plugin-sdk/src/data/frame/mod.rs
@@ -180,7 +180,7 @@ impl Frame {
     /// # Example
     ///
     /// ```rust
-    /// use arrow2::array::{PrimitiveArray, Utf8Array};
+    /// use arrow::{array::{PrimitiveArray, StringArray}, datatypes::UInt32Type};
     /// use grafana_plugin_sdk::prelude::*;
     ///
     /// // Create an initial `Frame`.
@@ -199,18 +199,18 @@ impl Frame {
     ///         .fields()[0]
     ///         .values()
     ///         .as_any()
-    ///         .downcast_ref::<PrimitiveArray<u32>>()
+    ///         .downcast_ref::<PrimitiveArray<UInt32Type>>()
     ///         .unwrap()
     ///         .iter()
-    ///         .collect::<Vec<_>>(),
-    ///     vec![Some(&4), Some(&5), Some(&6)],
+    ///         .collect::<Vec<Option<u32>>>(),
+    ///     vec![Some(4), Some(5), Some(6)],
     /// );
     /// assert_eq!(
     ///     frame
     ///         .fields()[1]
     ///         .values()
     ///         .as_any()
-    ///         .downcast_ref::<Utf8Array<i32>>()
+    ///         .downcast_ref::<StringArray>()
     ///         .unwrap()
     ///         .iter()
     ///         .collect::<Vec<_>>(),
@@ -236,7 +236,6 @@ impl Frame {
     /// # Example
     ///
     /// ```rust
-    /// use arrow2::array::{PrimitiveArray, Utf8Array};
     /// use grafana_plugin_sdk::prelude::*;
     ///
     /// assert!(
diff --git a/crates/grafana-plugin-sdk/src/data/frame/ser.rs b/crates/grafana-plugin-sdk/src/data/frame/ser.rs
index f93e6a06..57164ad8 100644
--- a/crates/grafana-plugin-sdk/src/data/frame/ser.rs
+++ b/crates/grafana-plugin-sdk/src/data/frame/ser.rs
@@ -1,11 +1,15 @@
 //! Serialization of [`Frame`]s to the JSON format.
 use std::{cell::RefCell, collections::BTreeMap};
 
-use arrow2::{
-    array::{Array, BooleanArray, PrimitiveArray, Utf8Array},
-    datatypes::{DataType, TimeUnit},
+use arrow::{
+    array::{Array, BooleanArray, PrimitiveArray, StringArray},
+    datatypes::{
+        ArrowPrimitiveType, DataType, Date32Type, Date64Type, Float32Type, Float64Type, Int16Type,
+        Int32Type, Int64Type, Int8Type, TimeUnit, TimestampMicrosecondType,
+        TimestampMillisecondType, TimestampNanosecondType, TimestampSecondType, UInt16Type,
+        UInt32Type, UInt64Type, UInt8Type,
+    },
     temporal_conversions::MILLISECONDS_IN_DAY,
-    types::NativeType,
 };
 use num_traits::Float;
 use serde::{
@@ -153,62 +157,62 @@ impl<'a> Serialize for SerializableArray<'a> {
                     .unwrap()
                     .iter(),
             ),
-            DataType::Utf8 | DataType::LargeUtf8 => serializer.collect_seq(
-                array
-                    .as_any()
-                    .downcast_ref::<Utf8Array<i32>>()
-                    .unwrap()
-                    .iter(),
-            ),
-            DataType::Int8 => serializer.collect_seq(primitive_array_iter::<i8>(array)),
-            DataType::Int16 => serializer.collect_seq(primitive_array_iter::<i16>(array)),
-            DataType::Int32 => serializer.collect_seq(primitive_array_iter::<i32>(array)),
-            DataType::Int64 => serializer.collect_seq(primitive_array_iter::<i64>(array)),
+            DataType::Utf8 | DataType::LargeUtf8 => {
+                serializer.collect_seq(array.as_any().downcast_ref::<StringArray>().unwrap().iter())
+            }
+            DataType::Int8 => serializer.collect_seq(primitive_array_iter::<Int8Type>(array)),
+            DataType::Int16 => serializer.collect_seq(primitive_array_iter::<Int16Type>(array)),
+            DataType::Int32 => serializer.collect_seq(primitive_array_iter::<Int32Type>(array)),
+            DataType::Int64 => serializer.collect_seq(primitive_array_iter::<Int64Type>(array)),
             DataType::Date32 => serializer.collect_seq(
-                primitive_array_iter::<i32>(array)
-                    .map(|opt| opt.map(|&x| i64::from(x) * MILLISECONDS_IN_DAY)),
+                primitive_array_iter::<Date32Type>(array)
+                    .map(|opt| opt.map(|x| i64::from(x) * MILLISECONDS_IN_DAY)),
             ),
-            DataType::Date64 => serializer.collect_seq(primitive_array_iter::<i64>(array)),
+            DataType::Date64 => serializer.collect_seq(primitive_array_iter::<Date64Type>(array)),
             DataType::Timestamp(TimeUnit::Second, _) => {
                 // Timestamps should be serialized to JSON as milliseconds.
                 serializer.collect_seq(
-                    primitive_array_iter::<i64>(array).map(|opt| opt.map(|x| x * 1_000)),
+                    primitive_array_iter::<TimestampSecondType>(array)
+                        .map(|opt| opt.map(|x| x * 1_000)),
                 )
             }
             DataType::Timestamp(TimeUnit::Millisecond, _) => {
                 // Timestamps should be serialized to JSON as milliseconds.
-                serializer.collect_seq(primitive_array_iter::<i64>(array))
+                serializer.collect_seq(primitive_array_iter::<TimestampMillisecondType>(array))
             }
             DataType::Timestamp(TimeUnit::Microsecond, _) => {
                 // Timestamps should be serialized to JSON as milliseconds.
                 serializer.collect_seq(
-                    primitive_array_iter::<i64>(array).map(|opt| opt.map(|x| x / 1_000)),
+                    primitive_array_iter::<TimestampMicrosecondType>(array)
+                        .map(|opt| opt.map(|x| x / 1_000)),
                 )
             }
             DataType::Timestamp(TimeUnit::Nanosecond, _) => {
                 // Timestamps should be serialized to JSON as milliseconds.
                 serializer.collect_seq(
-                    primitive_array_iter::<i64>(array).map(|opt| opt.map(|x| x / 1_000_000)),
+                    primitive_array_iter::<TimestampNanosecondType>(array)
+                        .map(|opt| opt.map(|x| x / 1_000_000)),
                 )
             }
-            DataType::UInt8 => serializer.collect_seq(primitive_array_iter::<u8>(array)),
-            DataType::UInt16 => serializer.collect_seq(primitive_array_iter::<u16>(array)),
-            DataType::UInt32 => serializer.collect_seq(primitive_array_iter::<u32>(array)),
-            DataType::UInt64 => serializer.collect_seq(primitive_array_iter::<u64>(array)),
+            DataType::UInt8 => serializer.collect_seq(primitive_array_iter::<UInt8Type>(array)),
+            DataType::UInt16 => serializer.collect_seq(primitive_array_iter::<UInt16Type>(array)),
+            DataType::UInt32 => serializer.collect_seq(primitive_array_iter::<UInt32Type>(array)),
+            DataType::UInt64 => serializer.collect_seq(primitive_array_iter::<UInt64Type>(array)),
             DataType::Float32 => {
-                serialize_floats_and_collect_entities::<S, f32>(serializer, array, self.1)
+                serialize_floats_and_collect_entities::<S, Float32Type>(serializer, array, self.1)
             }
             DataType::Float64 => {
-                serialize_floats_and_collect_entities::<S, f64>(serializer, array, self.1)
+                serialize_floats_and_collect_entities::<S, Float64Type>(serializer, array, self.1)
             }
             _ => Err(S::Error::custom("unsupported arrow datatype")),
         }
     }
 }
 
-fn primitive_array_iter<T>(array: &dyn Array) -> impl Iterator<Item = Option<&T>>
+fn primitive_array_iter<T>(array: &dyn Array) -> impl Iterator<Item = Option<T::Native>> + '_
 where
-    T: NativeType + Clone,
+    T: ArrowPrimitiveType,
+    <T as ArrowPrimitiveType>::Native: Clone,
 {
     array
         .as_any()
@@ -224,7 +228,8 @@ fn serialize_floats_and_collect_entities<S, T>(
 ) -> Result<S::Ok, S::Error>
 where
     S: Serializer,
-    T: NativeType + Float + Serialize,
+    T: ArrowPrimitiveType,
+    <T as ArrowPrimitiveType>::Native: Float + Serialize,
 {
     let array = array.as_any().downcast_ref::<PrimitiveArray<T>>().unwrap();
     let mut seq = serializer.serialize_seq(Some(array.len()))?;
@@ -256,10 +261,8 @@ pub(crate) struct Entities {
 
 #[cfg(test)]
 mod test {
-    use arrow2::{
-        array::PrimitiveArray,
-        datatypes::{DataType, TimeUnit},
-    };
+
+    use chrono::{NaiveDate, TimeZone, Utc};
     use pretty_assertions::assert_eq;
     use serde_json::{from_str, json, to_string, to_string_pretty};
 
@@ -285,123 +288,20 @@ mod test {
                 ..Default::default()
             }),
             fields: vec![
-                Field {
-                    name: "int8_values".to_string(),
-                    labels: Default::default(),
-                    config: None,
-                    values: Box::new(PrimitiveArray::<i8>::from_slice([-128, -128, 0, 127, 127])),
-                    type_info: TypeInfo {
-                        frame: TypeInfoType::Int8,
-                        nullable: Some(false),
-                    },
-                },
-                Field {
-                    name: "date32_values".to_string(),
-                    labels: Default::default(),
-                    config: None,
-                    values: Box::new(
-                        PrimitiveArray::<i32>::from_slice([18895, 18896, 18897, 18898, 18899])
-                            .to(DataType::Date32),
-                    ),
-                    type_info: TypeInfo {
-                        frame: TypeInfoType::Time,
-                        nullable: Some(false),
-                    },
-                },
-                Field {
-                    name: "date64_values".to_string(),
-                    labels: Default::default(),
-                    config: None,
-                    values: Box::new(
-                        PrimitiveArray::<i64>::from_slice([
-                            1632528000000,
-                            1632614400000,
-                            1632700800000,
-                            1632787200000,
-                            1632873600000,
-                        ])
-                        .to(DataType::Date64),
-                    ),
-                    type_info: TypeInfo {
-                        frame: TypeInfoType::Time,
-                        nullable: Some(false),
-                    },
-                },
-                Field {
-                    name: "timestamp_s_values".to_string(),
-                    labels: Default::default(),
-                    config: None,
-                    values: Box::new(
-                        PrimitiveArray::<i64>::from_slice([
-                            1632855151, 1632855152, 1632855153, 1632855154, 1632855155,
-                        ])
-                        .to(DataType::Timestamp(TimeUnit::Second, None)),
-                    ),
-                    type_info: TypeInfo {
-                        frame: TypeInfoType::Time,
-                        nullable: Some(false),
-                    },
-                },
-                Field {
-                    name: "timestamp_ms_values".to_string(),
-                    labels: Default::default(),
-                    config: None,
-                    values: Box::new(
-                        PrimitiveArray::<i64>::from_slice([
-                            1632855151000,
-                            1632855152000,
-                            1632855153000,
-                            1632855154000,
-                            1632855155000,
-                        ])
-                        .to(DataType::Timestamp(TimeUnit::Millisecond, None)),
-                    ),
-                    type_info: TypeInfo {
-                        frame: TypeInfoType::Time,
-                        nullable: Some(false),
-                    },
-                },
-                Field {
-                    name: "timestamp_us_values".to_string(),
-                    labels: Default::default(),
-                    config: None,
-                    values: Box::new(
-                        PrimitiveArray::<i64>::from_slice([
-                            1632855151000000,
-                            1632855152000000,
-                            1632855153000000,
-                            1632855154000000,
-                            1632855155000000,
-                        ])
-                        .to(DataType::Timestamp(TimeUnit::Microsecond, None)),
-                    ),
-                    type_info: TypeInfo {
-                        frame: TypeInfoType::Time,
-                        nullable: Some(false),
-                    },
-                },
-                Field {
-                    name: "timestamp_ns_values".to_string(),
-                    labels: Default::default(),
-                    config: None,
-                    values: Box::new(
-                        PrimitiveArray::<i64>::from_slice([
-                            1632855151000000000,
-                            1632855152000000000,
-                            1632855153000000000,
-                            1632855154000000000,
-                            1632855155000000000,
-                        ])
-                        .to(DataType::Timestamp(
-                            TimeUnit::Nanosecond,
-                            Some("+12:00".to_string()),
-                        )),
-                    ),
-                    type_info: TypeInfo {
-                        frame: TypeInfoType::Time,
-                        nullable: Some(false),
-                    },
-                },
+                vec![-128i8, -128, 0, 127, 127].into_field("int8_values"),
+                vec!["foo", "bar", "baz", "qux", "quux"].into_field("string_values"),
+                vec![10000i32, 20000, 30000, 40000, 50000].into_field("int32_values"),
+                vec![10000u32, 20000, 30000, 40000, 50000].into_field("uint32_values"),
+                vec![1.0f32, 2.0, f32::NAN, f32::INFINITY, f32::NEG_INFINITY]
+                    .into_field("float32_values"),
+                vec![1.0f64, 2.0, f64::NAN, f64::INFINITY, f64::NEG_INFINITY]
+                    .into_field("float64_values"),
+                vec![Utc
+                    .with_ymd_and_hms(2024, 1, 1, 12, 13, 14)
+                    .single()
+                    .unwrap()]
+                .into_field("datetime_values"),
+                vec![NaiveDate::from_ymd_opt(2024, 1, 1).unwrap()].into_field("date_values"),
             ],
         };
         let jdoc = to_string_pretty(&f).unwrap();
diff --git a/crates/grafana-plugin-sdk/src/data/frame/to_arrow.rs b/crates/grafana-plugin-sdk/src/data/frame/to_arrow.rs
index dc3fb707..125a8c22 100644
--- a/crates/grafana-plugin-sdk/src/data/frame/to_arrow.rs
+++ b/crates/grafana-plugin-sdk/src/data/frame/to_arrow.rs
@@ -1,7 +1,7 @@
 //! Conversion of [`Frame`][crate::data::Frame]s to the Arrow IPC format.
-use std::collections::BTreeMap;
+use std::{collections::HashMap, sync::Arc};
 
-use arrow2::{chunk::Chunk, datatypes::Schema, io::ipc::write::FileWriter};
+use arrow::{datatypes::Schema, ipc::writer::FileWriter, record_batch::RecordBatch};
 use thiserror::Error;
 
 use crate::data::{field::Field, frame::CheckedFrame};
@@ -15,10 +15,10 @@ pub enum Error {
     Json(#[from] serde_json::Error),
     /// An error occurred creating the Arrow record batch.
     #[error("Error creating record batch: {0}")]
-    CreateRecordBatch(arrow2::error::Error),
+    CreateRecordBatch(arrow::error::ArrowError),
     /// An error occurred when attempting to create or write data to the output buffer.
     #[error("Error writing data to Arrow buffer")]
-    WriteBuffer(arrow2::error::Error),
+    WriteBuffer(arrow::error::ArrowError),
 }
 
 impl CheckedFrame<'_> {
@@ -33,7 +33,7 @@ impl CheckedFrame<'_> {
             .iter()
             .map(Field::to_arrow_field)
             .collect::<Result<_, _>>()?;
-        let mut metadata: BTreeMap<String, String> = [
+        let mut metadata: HashMap<String, String> = [
             ("name".to_string(), self.0.name.to_string()),
             (
                 "refId".to_string(),
@@ -51,7 +51,7 @@ impl CheckedFrame<'_> {
         if let Some(meta) = &self.0.meta {
             metadata.insert("meta".to_string(), serde_json::to_string(&meta)?);
         }
-        Ok(Schema::from(fields).with_metadata(metadata))
+        Ok(Schema::new_with_metadata(fields, metadata))
     }
 
     /// Convert this [`Frame`][crate::data::Frame] to Arrow using the IPC format.
@@ -59,23 +59,26 @@ impl CheckedFrame<'_> {
     /// If `ref_id` is provided, it is passed down to the various conversion
     /// function and takes precedence over the `ref_id` set on the frame.
     pub(crate) fn to_arrow(&self, ref_id: Option<String>) -> Result<Vec<u8>, Error> {
-        let schema = self.arrow_schema(ref_id)?;
+        let schema = Arc::new(self.arrow_schema(ref_id)?);
 
         let records = if self.0.fields.is_empty() {
             None
         } else {
             Some(
-                Chunk::try_new(self.0.fields.iter().map(|f| f.values.clone()).collect())
-                    .map_err(Error::CreateRecordBatch)?,
+                RecordBatch::try_new(
+                    Arc::clone(&schema),
+                    self.0.fields.iter().map(|f| f.values.clone()).collect(),
+                )
+                .map_err(Error::CreateRecordBatch)?,
             )
         };
 
         let mut buf = Vec::new();
         {
-            let mut writer = FileWriter::try_new(&mut buf, schema, None, Default::default())
-                .map_err(Error::WriteBuffer)?;
+            let mut writer =
+                FileWriter::try_new_buffered(&mut buf, &schema).map_err(Error::WriteBuffer)?;
             if let Some(records) = records {
-                writer.write(&records, None).map_err(Error::WriteBuffer)?;
+                writer.write(&records).map_err(Error::WriteBuffer)?;
             }
             writer.finish().map_err(Error::WriteBuffer)?;
         }
diff --git a/crates/grafana-plugin-sdk/src/lib.rs b/crates/grafana-plugin-sdk/src/lib.rs
index be712ead..c6dfc972 100644
--- a/crates/grafana-plugin-sdk/src/lib.rs
+++ b/crates/grafana-plugin-sdk/src/lib.rs
@@ -34,12 +34,12 @@ The following feature flags enable additional functionality for this crate:
 #![cfg_attr(docsrs, feature(doc_notable_trait))]
 #![deny(missing_docs)]
 
-/// Re-export of the arrow2 crate depended on by this crate.
+/// Re-export of the arrow crate depended on by this crate.
 ///
-/// We recommend that you use this re-export rather than depending on arrow2
+/// We recommend that you use this re-export rather than depending on arrow
 /// directly to ensure compatibility; otherwise, rustc/cargo may emit mysterious
 /// error messages.
-pub use arrow2;
+pub use arrow;
 
 #[doc(hidden)]
 pub use serde_json;