diff --git a/Cargo.lock b/Cargo.lock
index 8101e2fe22d92..71440ca24b3ea 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1918,6 +1918,7 @@ dependencies = [
  "apache-avro",
  "arrow",
  "arrow-ipc",
+ "arrow-schema",
  "chrono",
  "criterion",
  "foldhash 0.2.0",
@@ -1935,6 +1936,7 @@ dependencies = [
  "recursive",
  "sqlparser",
  "tokio",
+ "uuid",
  "web-time",
 ]
 
@@ -2173,6 +2175,7 @@ name = "datafusion-expr"
 version = "52.3.0"
 dependencies = [
  "arrow",
+ "arrow-schema",
  "async-trait",
  "chrono",
  "ctor",
diff --git a/datafusion-examples/README.md b/datafusion-examples/README.md
index 2cf0ec52409f8..45fb8159ec451 100644
--- a/datafusion-examples/README.md
+++ b/datafusion-examples/README.md
@@ -125,6 +125,16 @@ cargo run --example dataframe -- dataframe
 | mem_pool_tracking  | [`execution_monitoring/memory_pool_tracking.rs`](examples/execution_monitoring/memory_pool_tracking.rs)             | Demonstrates memory tracking             |
 | tracing            | [`execution_monitoring/tracing.rs`](examples/execution_monitoring/tracing.rs)                                       | Demonstrates tracing integration         |
 
+## Extension Types Examples
+
+### Group: `extension_types`
+
+#### Category: Single Process
+
+| Subcommand | File Path                                                             | Description                                               |
+| ---------- | --------------------------------------------------------------------- | --------------------------------------------------------- |
+| event_id   | [`extension_types/event_id.rs`](examples/extension_types/event_id.rs) | A custom wrapper around integers that represent event ids |
+
 ## External Dependency Examples
 
 ### Group: `external_dependency`
diff --git a/datafusion-examples/examples/extension_types/event_id.rs b/datafusion-examples/examples/extension_types/event_id.rs
new file mode 100644
index 0000000000000..11a27c0742f62
--- /dev/null
+++ b/datafusion-examples/examples/extension_types/event_id.rs
@@ -0,0 +1,297 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use arrow::array::{Array, RecordBatch, StringArray, UInt32Array};
+use arrow::util::display::{ArrayFormatter, DisplayIndex, FormatOptions, FormatResult};
+use arrow_schema::extension::ExtensionType;
+use arrow_schema::{ArrowError, DataType, Field, Schema, SchemaRef};
+use datafusion::dataframe::DataFrame;
+use datafusion::error::Result;
+use datafusion::execution::SessionStateBuilder;
+use datafusion::prelude::SessionContext;
+use datafusion_common::internal_err;
+use datafusion_common::types::DFExtensionType;
+use datafusion_expr::registry::{
+    DefaultExtensionTypeRegistration, ExtensionTypeRegistry, MemoryExtensionTypeRegistry,
+};
+use std::fmt::Write;
+use std::sync::Arc;
+
+/// This example demonstrates using DataFusion's extension type API to create a custom identifier
+/// type [`EventIdExtensionType`].
+///
+/// The following use cases are demonstrated:
+/// - Use a custom implementation for pretty-printing data frames.
+pub async fn event_id_example() -> Result<()> {
+    let ctx = create_session_context()?;
+    register_events_table(&ctx).await?;
+
+    // Print the example table with the custom pretty-printer.
+    ctx.table("example").await?.show().await
+}
+
+/// Creates the DataFusion session context with the custom extension type implementation.
+fn create_session_context() -> Result<SessionContext> {
+    // Create a registry with a reference to the custom extension type implementation.
+    let registry = MemoryExtensionTypeRegistry::new();
+    let event_id_registration = DefaultExtensionTypeRegistration::new_arc(|metadata| {
+        Ok(EventIdExtensionType(metadata))
+    });
+    registry.add_extension_type_registration(event_id_registration)?;
+
+    // Set the extension type registry in the session state so that DataFusion can use it.
+    let state = SessionStateBuilder::default()
+        .with_extension_type_registry(Arc::new(registry))
+        .build();
+    Ok(SessionContext::new_with_state(state))
+}
+
+/// Registers the example table and returns the data frame.
+async fn register_events_table(ctx: &SessionContext) -> Result<DataFrame> {
+    let schema = example_schema();
+    let batch = RecordBatch::try_new(
+        schema,
+        vec![
+            Arc::new(UInt32Array::from(vec![
+                2001000000, 2001000001, 2103000000, 2103000001, 2103000002,
+            ])),
+            Arc::new(UInt32Array::from(vec![
+                2020010000, 2020010001, 2021030000, 2021030001, 2021030002,
+            ])),
+            Arc::new(StringArray::from(vec![
+                "First Event Jan 2020",
+                "Second Event Jan 2020",
+                "First Event Mar 2021",
+                "Second Event Mar 2021",
+                "Third Event Mar 2021",
+            ])),
+        ],
+    )?;
+
+    // Register the table and return the data frame.
+    ctx.register_batch("example", batch)?;
+    ctx.table("example").await
+}
+
+/// The schema of the example table.
+fn example_schema() -> SchemaRef {
+    Arc::new(Schema::new(vec![
+        Field::new("event_id_short", DataType::UInt32, false)
+            .with_extension_type(EventIdExtensionType(IdYearMode::Short)),
+        Field::new("event_id_long", DataType::UInt32, false)
+            .with_extension_type(EventIdExtensionType(IdYearMode::Long)),
+        Field::new("name", DataType::Utf8, false),
+    ]))
+}
+
+/// Represents a 32-bit custom identifier that represents a single event. Using this format is
+/// probably not a good idea in practice, but it is useful for demonstrating the API usage.
+///
+/// An event is constructed of three parts:
+/// - The year
+/// - The month
+/// - An auto-incrementing counter within the month
+///
+/// For example, the event id `2024-01-0000` represents the first event in 2024.
+///
+/// # Year Mode
+///
+/// In addition, each event id can be represented in two modes. A short year mode `24-01-000000` and
+/// a long year mode `2024-01-0000`. This showcases how extension types can be parameterized using
+/// metadata.
+#[derive(Debug)]
+pub struct EventIdExtensionType(IdYearMode);
+
+/// Represents whether the id uses the short or long format.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub enum IdYearMode {
+    /// The short year format (e.g., `24-01-000000`). Allows for more events per month.
+    Short,
+    /// The long year format (e.g., `2024-01-0000`). Allows distinguishing between centuries.
+    Long,
+}
+
+/// Implementation of [`ExtensionType`] for [`EventIdExtensionType`].
+///
+/// This is for the arrow-rs side of the API usage. The [`ExtensionType::Metadata`] type provides
+/// static guarantees on the deserialized metadata for the extension type. We will use this
+/// implementation to read and write the type metadata to arrow [`Field`]s.
+///
+/// This trait does allow users to customize the behavior of DataFusion for this extension type.
+/// This is done in [`DFExtensionType`].
+impl ExtensionType for EventIdExtensionType {
+    const NAME: &'static str = "custom.event_id";
+    type Metadata = IdYearMode;
+
+    fn metadata(&self) -> &Self::Metadata {
+        &self.0
+    }
+
+    fn serialize_metadata(&self) -> Option<String> {
+        // Arrow extension type metadata is encoded as a string. We simply use the lowercase name.
+        // In a more involved scenario, more complex serialization formats such as JSON are
+        // appropriate.
+        Some(format!("{:?}", self.0).to_lowercase())
+    }
+
+    fn deserialize_metadata(
+        metadata: Option<&str>,
+    ) -> std::result::Result<Self::Metadata, ArrowError> {
+        match metadata {
+            None => Err(ArrowError::InvalidArgumentError(
+                "Event id type requires metadata".to_owned(),
+            )),
+            Some(metadata) => match metadata {
+                "short" => Ok(IdYearMode::Short),
+                "long" => Ok(IdYearMode::Long),
+                _ => Err(ArrowError::InvalidArgumentError(format!(
+                    "Invalid metadata for event id type: {metadata}"
+                ))),
+            },
+        }
+    }
+
+    fn supports_data_type(
+        &self,
+        data_type: &DataType,
+    ) -> std::result::Result<(), ArrowError> {
+        match data_type {
+            DataType::UInt32 => Ok(()),
+            _ => Err(ArrowError::InvalidArgumentError(format!(
+                "Invalid data type: {data_type} for event id type",
+            ))),
+        }
+    }
+
+    fn try_new(
+        data_type: &DataType,
+        metadata: Self::Metadata,
+    ) -> std::result::Result<Self, ArrowError> {
+        let instance = Self(metadata);
+        instance.supports_data_type(data_type)?; // Check that the data type is supported.
+        Ok(instance)
+    }
+}
+
+/// Implementation of [`ExtensionType`] for [`EventIdExtensionType`].
+///
+/// This is for the DataFusion side of the API usage. Here users can override the default behavior
+/// of DataFusion for supported extension points.
+impl DFExtensionType for EventIdExtensionType {
+    fn create_array_formatter<'fmt>(
+        &self,
+        array: &'fmt dyn Array,
+        options: &FormatOptions<'fmt>,
+    ) -> Result<Option<ArrayFormatter<'fmt>>> {
+        if array.data_type() != &DataType::UInt32 {
+            return internal_err!("Wrong array type for Event Id");
+        }
+
+        // Create the formatter and pass in the year formatting mode of the type
+        let display_index = EventIdDisplayIndex {
+            array: array.as_any().downcast_ref().unwrap(),
+            null_str: options.null(),
+            mode: self.0,
+        };
+        Ok(Some(ArrayFormatter::new(
+            Box::new(display_index),
+            options.safe(),
+        )))
+    }
+}
+
+/// Pretty printer for event ids.
+#[derive(Debug)]
+struct EventIdDisplayIndex<'a> {
+    array: &'a UInt32Array,
+    null_str: &'a str,
+    mode: IdYearMode,
+}
+
+/// This implements the arrow-rs API for printing individual values of a column. DataFusion will
+/// automatically pass in the reference to this implementation if a column is annotated with the
+/// extension type metadata.
+impl DisplayIndex for EventIdDisplayIndex<'_> {
+    fn write(&self, idx: usize, f: &mut dyn Write) -> FormatResult {
+        // Handle nulls first
+        if self.array.is_null(idx) {
+            write!(f, "{}", self.null_str)?;
+            return Ok(());
+        }
+
+        let value = self.array.value(idx);
+
+        match self.mode {
+            IdYearMode::Short => {
+                // Format: YY-MM-CCCCCC
+                // Logic:
+                //  - The last 6 digits are the counter.
+                //  - The next 2 digits are the month.
+                //  - The remaining digits are the year.
+                let counter = value % 1_000_000;
+                let rest = value / 1_000_000;
+                let month = rest % 100;
+                let year = rest / 100;
+
+                write!(f, "{year:02}-{month:02}-{counter:06}")?;
+            }
+            IdYearMode::Long => {
+                // Format: YYYY-MM-CCCC
+                // Logic:
+                //  - The last 4 digits are the counter.
+                //  - The next 2 digits are the month.
+                //  - The remaining digits are the year.
+                let counter = value % 10_000;
+                let rest = value / 10_000;
+                let month = rest % 100;
+                let year = rest / 100;
+
+                write!(f, "{year:04}-{month:02}-{counter:04}")?;
+            }
+        }
+        Ok(())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use insta::assert_snapshot;
+
+    #[tokio::test]
+    async fn test_print_example_table() -> Result<()> {
+        let ctx = create_session_context()?;
+        let table = register_events_table(&ctx).await?;
+
+        assert_snapshot!(
+            table.to_string().await?,
+            @r"
+        +----------------+---------------+-----------------------+
+        | event_id_short | event_id_long | name                  |
+        +----------------+---------------+-----------------------+
+        | 20-01-000000   | 2020-01-0000  | First Event Jan 2020  |
+        | 20-01-000001   | 2020-01-0001  | Second Event Jan 2020 |
+        | 21-03-000000   | 2021-03-0000  | First Event Mar 2021  |
+        | 21-03-000001   | 2021-03-0001  | Second Event Mar 2021 |
+        | 21-03-000002   | 2021-03-0002  | Third Event Mar 2021  |
+        +----------------+---------------+-----------------------+
+        "
+        );
+
+        Ok(())
+    }
+}
diff --git a/datafusion-examples/examples/extension_types/main.rs b/datafusion-examples/examples/extension_types/main.rs
new file mode 100644
index 0000000000000..672f339871f95
--- /dev/null
+++ b/datafusion-examples/examples/extension_types/main.rs
@@ -0,0 +1,84 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! # Extension type usage examples
+//!
+//! These examples demonstrate the API for creating and using custom extension types.
+//!
+//! ## Usage
+//! ```bash
+//! cargo run --example extension_types -- [all|event_id]
+//! ```
+//!
+//! Each subcommand runs a corresponding example:
+//! - `all` — run all examples included in this module
+//!
+//! - `event_id`
+//!   (file: event_id.rs, desc: A custom wrapper around integers that represent event ids)
+
+mod event_id;
+
+use datafusion::error::{DataFusionError, Result};
+use strum::{IntoEnumIterator, VariantNames};
+use strum_macros::{Display, EnumIter, EnumString, VariantNames};
+
+#[derive(EnumIter, EnumString, Display, VariantNames)]
+#[strum(serialize_all = "snake_case")]
+enum ExampleKind {
+    All,
+    EventId,
+}
+
+impl ExampleKind {
+    const EXAMPLE_NAME: &str = "extension_types";
+
+    fn runnable() -> impl Iterator<Item = ExampleKind> {
+        ExampleKind::iter().filter(|v| !matches!(v, ExampleKind::All))
+    }
+
+    async fn run(&self) -> Result<()> {
+        match self {
+            ExampleKind::All => {
+                for example in ExampleKind::runnable() {
+                    println!("Running example: {example}");
+                    Box::pin(example.run()).await?;
+                }
+            }
+            ExampleKind::EventId => {
+                event_id::event_id_example().await?;
+            }
+        }
+        Ok(())
+    }
+}
+
+#[tokio::main]
+async fn main() -> Result<()> {
+    let usage = format!(
+        "Usage: cargo run --example {} -- [{}]",
+        ExampleKind::EXAMPLE_NAME,
+        ExampleKind::VARIANTS.join("|")
+    );
+
+    let example: ExampleKind = std::env::args()
+        .nth(1)
+        .unwrap_or_else(|| ExampleKind::All.to_string())
+        .parse()
+        .map_err(|_| DataFusionError::Execution(format!("Unknown example. {usage}")))?;
+
+    example.run().await
+}
diff --git a/datafusion/common/Cargo.toml b/datafusion/common/Cargo.toml
index 92dd76aa97d47..033ecef834f68 100644
--- a/datafusion/common/Cargo.toml
+++ b/datafusion/common/Cargo.toml
@@ -74,6 +74,7 @@ apache-avro = { workspace = true, features = [
 ], optional = true }
 arrow = { workspace = true }
 arrow-ipc = { workspace = true }
+arrow-schema = { workspace = true, features = ["canonical_extension_types"] }
 chrono = { workspace = true }
 foldhash = "0.2"
 half = { workspace = true }
@@ -88,6 +89,7 @@ parquet = { workspace = true, optional = true, default-features = true }
 recursive = { workspace = true, optional = true }
 sqlparser = { workspace = true, optional = true }
 tokio = { workspace = true }
+uuid = { workspace = true, features = ["v4"] }
 
 [target.'cfg(target_family = "wasm")'.dependencies]
 web-time = "1.1.0"
diff --git a/datafusion/common/src/types/canonical_extensions/mod.rs b/datafusion/common/src/types/canonical_extensions/mod.rs
new file mode 100644
index 0000000000000..e61c415b44811
--- /dev/null
+++ b/datafusion/common/src/types/canonical_extensions/mod.rs
@@ -0,0 +1,18 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+mod uuid;
diff --git a/datafusion/common/src/types/canonical_extensions/uuid.rs b/datafusion/common/src/types/canonical_extensions/uuid.rs
new file mode 100644
index 0000000000000..0bd4f77dda7e9
--- /dev/null
+++ b/datafusion/common/src/types/canonical_extensions/uuid.rs
@@ -0,0 +1,236 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use crate::cast::{as_fixed_size_binary_array, as_string_array};
+use crate::error::_internal_err;
+use crate::types::CastExtension;
+use crate::types::extension::DFExtensionType;
+use arrow::array::{
+    Array, ArrayRef, FixedSizeBinaryArray, StringBuilder, builder::FixedSizeBinaryBuilder,
+};
+use arrow::compute::{CastOptions, cast};
+use arrow::datatypes::DataType;
+use arrow::util::display::{ArrayFormatter, DisplayIndex, FormatOptions, FormatResult};
+use arrow_schema::Field;
+use std::fmt::Write;
+use std::sync::Arc;
+use uuid::{Bytes, Uuid};
+
+/// Defines the extension type logic for the canonical `arrow.uuid` extension type.
+///
+/// See [`DFExtensionType`] for information on DataFusion's extension type mechanism.
+impl DFExtensionType for arrow_schema::extension::Uuid {
+    fn create_array_formatter<'fmt>(
+        &self,
+        array: &'fmt dyn Array,
+        options: &FormatOptions<'fmt>,
+    ) -> crate::Result<Option<ArrayFormatter<'fmt>>> {
+        if array.data_type() != &DataType::FixedSizeBinary(16) {
+            return _internal_err!("Wrong array type for Uuid");
+        }
+
+        let display_index = UuidValueDisplayIndex {
+            array: array.as_any().downcast_ref().unwrap(),
+            null_str: options.null(),
+        };
+        Ok(Some(ArrayFormatter::new(
+            Box::new(display_index),
+            options.safe(),
+        )))
+    }
+
+    fn cast_from(&self) -> crate::Result<Arc<dyn CastExtension>> {
+        Ok(Arc::new(CastToUuid {}))
+    }
+
+    fn cast_to(&self) -> crate::Result<Arc<dyn CastExtension>> {
+        Ok(Arc::new(CastFromUuid {}))
+    }
+}
+
+/// Pretty printer for binary UUID values.
+#[derive(Debug, Clone, Copy)]
+struct UuidValueDisplayIndex<'a> {
+    array: &'a FixedSizeBinaryArray,
+    null_str: &'a str,
+}
+
+impl DisplayIndex for UuidValueDisplayIndex<'_> {
+    fn write(&self, idx: usize, f: &mut dyn Write) -> FormatResult {
+        if self.array.is_null(idx) {
+            write!(f, "{}", self.null_str)?;
+            return Ok(());
+        }
+
+        let bytes = Bytes::try_from(self.array.value(idx))
+            .expect("FixedSizeBinaryArray length checked in create_array_formatter");
+        let uuid = Uuid::from_bytes(bytes);
+        write!(f, "{uuid}")?;
+        Ok(())
+    }
+}
+
+#[derive(Debug)]
+struct CastFromUuid {}
+
+impl CastExtension for CastFromUuid {
+    fn can_cast(
+        &self,
+        _from: &Field,
+        to: &Field,
+        options: &CastOptions,
+    ) -> crate::Result<bool> {
+        if to.extension_type_name().is_some() {
+            return Ok(false);
+        }
+
+        match to.data_type() {
+            // Only explicit casts to string
+            DataType::Utf8 | DataType::Utf8View | DataType::LargeUtf8 => {
+                if options.safe {
+                    Ok(false)
+                } else {
+                    Ok(true)
+                }
+            }
+            // Can implicitly cast to storage
+            DataType::FixedSizeBinary(16) => Ok(true),
+            _ => Ok(false),
+        }
+    }
+
+    fn cast(
+        &self,
+        value: ArrayRef,
+        from: &Field,
+        to: &Field,
+        options: &CastOptions,
+    ) -> crate::Result<ArrayRef> {
+        if !self.can_cast(from, to, options)? {
+            return _internal_err!("Unhandled cast");
+        }
+
+        let storage = as_fixed_size_binary_array(&value)?;
+        match to.data_type() {
+            DataType::Utf8 | DataType::Utf8View | DataType::LargeUtf8 => {
+                let mut builder =
+                    StringBuilder::with_capacity(storage.len(), storage.len() * 36);
+                for bytes_opt in storage {
+                    match bytes_opt {
+                        Some(bytes) => {
+                            let bytes16 = Bytes::try_from(bytes).map_err(|e| {
+                                crate::DataFusionError::Execution(e.to_string())
+                            })?;
+                            let uuid = Uuid::from_bytes(bytes16);
+                            write!(builder, "{uuid}")?;
+                            builder.append_value("");
+                        }
+                        None => builder.append_null(),
+                    }
+                }
+
+                let string_array = Arc::new(builder.finish()) as ArrayRef;
+                return Ok(cast(&string_array, to.data_type())?);
+            }
+            DataType::FixedSizeBinary(16) => return Ok(value),
+            _ => {}
+        }
+
+        _internal_err!("Unexpected difference between can_cast()")
+    }
+}
+
+#[derive(Debug)]
+struct CastToUuid {}
+
+impl CastExtension for CastToUuid {
+    fn can_cast(
+        &self,
+        from: &Field,
+        to: &Field,
+        options: &CastOptions,
+    ) -> crate::Result<bool> {
+        CastFromUuid {}.can_cast(to, from, options)
+    }
+
+    fn cast(
+        &self,
+        value: ArrayRef,
+        from: &Field,
+        to: &Field,
+        options: &CastOptions,
+    ) -> crate::Result<ArrayRef> {
+        if !self.can_cast(from, to, options)? {
+            return _internal_err!("Unhandled cast");
+        }
+
+        match from.data_type() {
+            DataType::Utf8 | DataType::Utf8View | DataType::LargeUtf8 => {
+                let string_array_ref = cast(&value, &DataType::Utf8)?;
+                let string_array = as_string_array(&string_array_ref)?;
+                let mut builder = FixedSizeBinaryBuilder::new(16);
+                for string_opt in string_array {
+                    match string_opt {
+                        Some(string) => {
+                            let uuid = Uuid::try_parse(string).map_err(|_| {
+                                crate::DataFusionError::Execution(format!(
+                                    "Failed to parsed string '{string}' as UUID"
+                                ))
+                            })?;
+                            builder.append_value(uuid.as_bytes())?;
+                        }
+                        None => {
+                            builder.append_null();
+                        }
+                    }
+                }
+
+                return Ok(Arc::new(builder.finish()));
+            }
+            // Can implicitly cast from storage
+            DataType::FixedSizeBinary(16) => return Ok(value),
+            _ => {}
+        }
+
+        _internal_err!("Unexpected difference between can_cast_from()")
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::ScalarValue;
+
+    #[test]
+    pub fn test_pretty_print_uuid() {
+        let my_uuid = Uuid::nil();
+        let uuid = ScalarValue::FixedSizeBinary(16, Some(my_uuid.as_bytes().to_vec()))
+            .to_array_of_size(1)
+            .unwrap();
+
+        let extension_type = arrow_schema::extension::Uuid {};
+        let formatter = extension_type
+            .create_array_formatter(uuid.as_ref(), &FormatOptions::default())
+            .unwrap()
+            .unwrap();
+
+        assert_eq!(
+            formatter.value(0).to_string(),
+            "00000000-0000-0000-0000-000000000000"
+        );
+    }
+}
diff --git a/datafusion/common/src/types/extension.rs b/datafusion/common/src/types/extension.rs
new file mode 100644
index 0000000000000..44bbdc65ccf29
--- /dev/null
+++ b/datafusion/common/src/types/extension.rs
@@ -0,0 +1,113 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use crate::error::Result;
+use arrow::array::{Array, ArrayRef};
+use arrow::compute::CastOptions;
+use arrow::util::display::{ArrayFormatter, FormatOptions};
+use arrow_schema::Field;
+use std::fmt::Debug;
+use std::sync::Arc;
+
+/// A cheaply cloneable pointer to a [`DFExtensionType`].
+pub type DFExtensionTypeRef = Arc<dyn DFExtensionType>;
+
+/// Represents an implementation of a DataFusion extension type.
+///
+/// This allows users to customize the behavior of DataFusion for certain types. Having this ability
+/// is necessary because extension types affect how columns should be treated by the query engine.
+/// This effect includes which operations are possible on a column and what are the expected results
+/// from these operations. The extension type mechanism allows users to define how these operations
+/// apply to a particular extension type.
+///
+/// For example, adding two values of [`Int64`](arrow::datatypes::DataType::Int64) is a sensible
+/// thing to do. However, if the same column is annotated with an extension type like `custom.id`,
+/// the correct interpretation of a column changes. Adding together two `custom.id` values, even
+/// though they are stored as integers, may no longer make sense.
+///
+/// Note that DataFusion's extension type support is still young and therefore might not cover all
+/// relevant use cases. Currently, the following operations can be customized:
+/// - Pretty-printing values in record batches
+///
+/// # Relation to Arrow's `ExtensionType`
+///
+/// The purpose of Arrow's `ExtensionType` trait, for the time being, is to allow reading and
+/// writing extension type metadata in a type-safe manner. The trait does not provide any
+/// customization options. Therefore, downstream users (such as DataFusion) have the flexibility to
+/// implement the extension type mechanism according to their needs. [`DFExtensionType`] is
+/// DataFusion's implementation of this extension type mechanism.
+///
+/// Furthermore, the current trait in arrow-rs is not dyn-compatible, which we need for implementing
+/// extension type registries. In the future, the two implementations may increasingly converge.
+///
+/// # Examples
+///
+/// Examples for using the extension type machinery can be found in the DataFusion examples
+/// directory.
+pub trait DFExtensionType: Debug + Send + Sync {
+    /// Returns an [`ArrayFormatter`] that can format values of this type.
+    ///
+    /// If `Ok(None)` is returned, the default implementation will be used.
+    /// If an error is returned, there was an error creating the formatter.
+    fn create_array_formatter<'fmt>(
+        &self,
+        _array: &'fmt dyn Array,
+        _options: &FormatOptions<'fmt>,
+    ) -> Result<Option<ArrayFormatter<'fmt>>> {
+        Ok(None)
+    }
+
+    fn cast_from(&self) -> Result<Arc<dyn CastExtension>> {
+        Ok(Arc::new(DefaultExtensionCast {}))
+    }
+
+    fn cast_to(&self) -> Result<Arc<dyn CastExtension>> {
+        Ok(Arc::new(DefaultExtensionCast {}))
+    }
+}
+
+pub trait CastExtension: Debug + Send + Sync {
+    fn can_cast(&self, from: &Field, to: &Field, options: &CastOptions) -> Result<bool>;
+
+    // None for fallback
+    fn cast(
+        &self,
+        value: ArrayRef,
+        from: &Field,
+        to: &Field,
+        options: &CastOptions,
+    ) -> Result<ArrayRef>;
+}
+
+#[derive(Debug)]
+struct DefaultExtensionCast {}
+
+impl CastExtension for DefaultExtensionCast {
+    fn can_cast(&self, from: &Field, to: &Field, _options: &CastOptions) -> Result<bool> {
+        Ok(from.data_type() == to.data_type())
+    }
+
+    fn cast(
+        &self,
+        value: ArrayRef,
+        _from: &Field,
+        _to: &Field,
+        _options: &CastOptions,
+    ) -> Result<ArrayRef> {
+        Ok(value)
+    }
+}
diff --git a/datafusion/common/src/types/mod.rs b/datafusion/common/src/types/mod.rs
index 2f9ce4ce02827..82455063bc6ce 100644
--- a/datafusion/common/src/types/mod.rs
+++ b/datafusion/common/src/types/mod.rs
@@ -16,11 +16,14 @@
 // under the License.
 
 mod builtin;
+mod canonical_extensions;
+mod extension;
 mod field;
 mod logical;
 mod native;
 
 pub use builtin::*;
+pub use extension::*;
 pub use field::*;
 pub use logical::*;
 pub use native::*;
diff --git a/datafusion/core/src/dataframe/mod.rs b/datafusion/core/src/dataframe/mod.rs
index 2292f5855bfde..6591c943264da 100644
--- a/datafusion/core/src/dataframe/mod.rs
+++ b/datafusion/core/src/dataframe/mod.rs
@@ -69,6 +69,7 @@ use datafusion_functions_aggregate::expr_fn::{
     avg, count, max, median, min, stddev, sum,
 };
 
+use crate::extension_types::DFArrayFormatterFactory;
 use async_trait::async_trait;
 use datafusion_catalog::Session;
 
@@ -1516,6 +1517,11 @@ impl DataFrame {
         let options = self.session_state.config().options().format.clone();
         let arrow_options: arrow::util::display::FormatOptions = (&options).try_into()?;
 
+        let registry = self.session_state.extension_type_registry();
+        let formatter_factory = DFArrayFormatterFactory::new(Arc::clone(registry));
+        let arrow_options =
+            arrow_options.with_formatter_factory(Some(&formatter_factory));
+
         let results = self.collect().await?;
         Ok(
             pretty::pretty_format_batches_with_options(&results, &arrow_options)?
diff --git a/datafusion/core/src/datasource/listing_table_factory.rs b/datafusion/core/src/datasource/listing_table_factory.rs
index f85f15a6d8c63..196bc35671bc0 100644
--- a/datafusion/core/src/datasource/listing_table_factory.rs
+++ b/datafusion/core/src/datasource/listing_table_factory.rs
@@ -241,6 +241,7 @@ mod tests {
 
     use datafusion_common::parsers::CompressionTypeVariant;
     use datafusion_common::{DFSchema, TableReference};
+    use datafusion_expr::registry::ExtensionTypeRegistryRef;
 
     #[tokio::test]
     async fn test_create_using_non_std_file_ext() {
@@ -609,6 +610,11 @@ mod tests {
             ) -> &HashMap<String, Arc<datafusion_expr::WindowUDF>> {
                 unimplemented!()
             }
+
+            fn extension_type_registry(&self) -> &ExtensionTypeRegistryRef {
+                unreachable!()
+            }
+
             fn runtime_env(&self) -> &Arc<datafusion_execution::runtime_env::RuntimeEnv> {
                 unimplemented!()
             }
diff --git a/datafusion/core/src/execution/session_state.rs b/datafusion/core/src/execution/session_state.rs
index 9560616c1b6da..7ecf53d9cfd61 100644
--- a/datafusion/core/src/execution/session_state.rs
+++ b/datafusion/core/src/execution/session_state.rs
@@ -56,7 +56,11 @@ use datafusion_expr::expr_rewriter::FunctionRewrite;
 use datafusion_expr::planner::ExprPlanner;
 #[cfg(feature = "sql")]
 use datafusion_expr::planner::{RelationPlanner, TypePlanner};
-use datafusion_expr::registry::{FunctionRegistry, SerializerRegistry};
+use datafusion_expr::registry::{
+    DefaultExtensionTypeRegistration, ExtensionTypeRegistration,
+    ExtensionTypeRegistrationRef, ExtensionTypeRegistry, ExtensionTypeRegistryRef,
+    FunctionRegistry, MemoryExtensionTypeRegistry, SerializerRegistry,
+};
 use datafusion_expr::simplify::SimplifyContext;
 use datafusion_expr::{AggregateUDF, Explain, Expr, LogicalPlan, ScalarUDF, WindowUDF};
 use datafusion_optimizer::simplify_expressions::ExprSimplifier;
@@ -158,6 +162,8 @@ pub struct SessionState {
     aggregate_functions: HashMap<String, Arc<AggregateUDF>>,
     /// Window functions registered in the context
     window_functions: HashMap<String, Arc<WindowUDF>>,
+    /// Extension types registry for extensions.
+    extension_types: ExtensionTypeRegistryRef,
     /// Deserializer registry for extensions.
     serializer_registry: Arc<dyn SerializerRegistry>,
     /// Holds registered external FileFormat implementations
@@ -266,6 +272,10 @@ impl Session for SessionState {
         &self.window_functions
     }
 
+    fn extension_type_registry(&self) -> &ExtensionTypeRegistryRef {
+        &self.extension_types
+    }
+
     fn runtime_env(&self) -> &Arc<RuntimeEnv> {
         self.runtime_env()
     }
@@ -986,6 +996,7 @@ pub struct SessionStateBuilder {
     scalar_functions: Option<Vec<Arc<ScalarUDF>>>,
     aggregate_functions: Option<Vec<Arc<AggregateUDF>>>,
     window_functions: Option<Vec<Arc<WindowUDF>>>,
+    extension_types: Option<ExtensionTypeRegistryRef>,
     serializer_registry: Option<Arc<dyn SerializerRegistry>>,
     file_formats: Option<Vec<Arc<dyn FileFormatFactory>>>,
     config: Option<SessionConfig>,
@@ -1026,6 +1037,7 @@ impl SessionStateBuilder {
             scalar_functions: None,
             aggregate_functions: None,
             window_functions: None,
+            extension_types: None,
             serializer_registry: None,
             file_formats: None,
             table_options: None,
@@ -1081,6 +1093,7 @@ impl SessionStateBuilder {
                 existing.aggregate_functions.into_values().collect_vec(),
             ),
             window_functions: Some(existing.window_functions.into_values().collect_vec()),
+            extension_types: Some(existing.extension_types),
             serializer_registry: Some(existing.serializer_registry),
             file_formats: Some(existing.file_formats.into_values().collect_vec()),
             config: Some(new_config),
@@ -1126,6 +1139,11 @@ impl SessionStateBuilder {
             .get_or_insert_with(Vec::new)
             .extend(SessionStateDefaults::default_window_functions());
 
+        self.extension_types
+            .get_or_insert_with(|| Arc::new(MemoryExtensionTypeRegistry::new()))
+            .extend(&SessionStateDefaults::default_extension_types())
+            .expect("MemoryExtensionTypeRegistry is not read-only.");
+
         self.table_functions
             .get_or_insert_with(HashMap::new)
             .extend(
@@ -1316,6 +1334,44 @@ impl SessionStateBuilder {
         self
     }
 
+    /// Set the map of [`ExtensionTypeRegistration`]s
+    pub fn with_extension_type_registry(
+        mut self,
+        registry: ExtensionTypeRegistryRef,
+    ) -> Self {
+        self.extension_types = Some(registry);
+        self
+    }
+
+    /// Registers [canonical extension types](https://arrow.apache.org/docs/format/CanonicalExtensions.html)
+    /// in DataFusion's extension type registry. For more information see [`ExtensionTypeRegistry`].
+    ///
+    /// # Errors
+    ///
+    /// May fail if an already registered [`ExtensionTypeRegistry`] raises an error while
+    /// registering the canonical extension types.
+    pub fn with_canonical_extension_types(mut self) -> datafusion_common::Result<Self> {
+        let uuid = DefaultExtensionTypeRegistration::new_arc(|_| {
+            Ok(arrow_schema::extension::Uuid {})
+        });
+        let canonical_extension_types = vec![uuid];
+
+        match &self.extension_types {
+            None => {
+                let registry = Arc::new(MemoryExtensionTypeRegistry::new());
+                registry
+                    .extend(&canonical_extension_types)
+                    .expect("Adding valid extension types to MemoryExtensionTypeRegistry always succeeds.");
+                self.extension_types = Some(registry);
+            }
+            Some(registry) => {
+                registry.extend(&canonical_extension_types)?;
+            }
+        }
+
+        Ok(self)
+    }
+
     /// Set the [`SerializerRegistry`]
     pub fn with_serializer_registry(
         mut self,
@@ -1454,6 +1510,7 @@ impl SessionStateBuilder {
             scalar_functions,
             aggregate_functions,
             window_functions,
+            extension_types,
             serializer_registry,
             file_formats,
             table_options,
@@ -1490,6 +1547,7 @@ impl SessionStateBuilder {
             scalar_functions: HashMap::new(),
             aggregate_functions: HashMap::new(),
             window_functions: HashMap::new(),
+            extension_types: Arc::new(MemoryExtensionTypeRegistry::default()),
             serializer_registry: serializer_registry
                 .unwrap_or_else(|| Arc::new(EmptySerializerRegistry)),
             file_formats: HashMap::new(),
@@ -1559,6 +1617,10 @@ impl SessionStateBuilder {
             });
         }
 
+        if let Some(extension_types) = extension_types {
+            state.extension_types = extension_types;
+        }
+
         if state.config.create_default_catalog_and_schema() {
             let default_catalog = SessionStateDefaults::default_catalog(
                 &state.config,
@@ -1597,6 +1659,10 @@ impl SessionStateBuilder {
             }
         }
 
+        // Temporary hack while we figure out how to get the extension types where they
+        // need to go
+        state.execution_props.extension_types = Some(Arc::clone(&state.extension_types));
+
         state
     }
 
@@ -2071,6 +2137,35 @@ impl datafusion_execution::TaskContextProvider for SessionState {
     }
 }
 
+impl ExtensionTypeRegistry for SessionState {
+    fn extension_type_registration(
+        &self,
+        name: &str,
+    ) -> datafusion_common::Result<ExtensionTypeRegistrationRef> {
+        self.extension_types.extension_type_registration(name)
+    }
+
+    fn extension_type_registrations(&self) -> Vec<Arc<dyn ExtensionTypeRegistration>> {
+        self.extension_types.extension_type_registrations()
+    }
+
+    fn add_extension_type_registration(
+        &self,
+        extension_type: ExtensionTypeRegistrationRef,
+    ) -> datafusion_common::Result<Option<ExtensionTypeRegistrationRef>> {
+        self.extension_types
+            .add_extension_type_registration(extension_type)
+    }
+
+    fn remove_extension_type_registration(
+        &self,
+        name: &str,
+    ) -> datafusion_common::Result<Option<ExtensionTypeRegistrationRef>> {
+        self.extension_types
+            .remove_extension_type_registration(name)
+    }
+}
+
 impl OptimizerConfig for SessionState {
     fn query_execution_start_time(&self) -> Option<DateTime<Utc>> {
         self.execution_props.query_execution_start_time
@@ -2087,6 +2182,10 @@ impl OptimizerConfig for SessionState {
     fn function_registry(&self) -> Option<&dyn FunctionRegistry> {
         Some(self)
     }
+
+    fn extension_types(&self) -> Option<Arc<dyn ExtensionTypeRegistry>> {
+        Some(Arc::clone(&self.extension_types))
+    }
 }
 
 /// Create a new task context instance from SessionState
diff --git a/datafusion/core/src/execution/session_state_defaults.rs b/datafusion/core/src/execution/session_state_defaults.rs
index 721710d4e057e..8ef041e5bf640 100644
--- a/datafusion/core/src/execution/session_state_defaults.rs
+++ b/datafusion/core/src/execution/session_state_defaults.rs
@@ -36,6 +36,7 @@ use datafusion_execution::config::SessionConfig;
 use datafusion_execution::object_store::ObjectStoreUrl;
 use datafusion_execution::runtime_env::RuntimeEnv;
 use datafusion_expr::planner::ExprPlanner;
+use datafusion_expr::registry::ExtensionTypeRegistrationRef;
 use datafusion_expr::{AggregateUDF, ScalarUDF, WindowUDF};
 use std::collections::HashMap;
 use std::sync::Arc;
@@ -122,6 +123,13 @@ impl SessionStateDefaults {
         functions_window::all_default_window_functions()
     }
 
+    /// Returns the list of default extension types.
+    ///
+    /// For now, we do not register any extension types by default.
+    pub fn default_extension_types() -> Vec<ExtensionTypeRegistrationRef> {
+        vec![]
+    }
+
     /// returns the list of default [`TableFunction`]s
     pub fn default_table_functions() -> Vec<Arc<TableFunction>> {
         functions_table::all_default_table_functions()
diff --git a/datafusion/core/src/extension_types/array_formatter_factory.rs b/datafusion/core/src/extension_types/array_formatter_factory.rs
new file mode 100644
index 0000000000000..f10576e816bda
--- /dev/null
+++ b/datafusion/core/src/extension_types/array_formatter_factory.rs
@@ -0,0 +1,67 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use arrow::array::Array;
+use arrow::util::display::{ArrayFormatter, ArrayFormatterFactory, FormatOptions};
+use arrow_schema::{ArrowError, Field};
+use datafusion_expr::registry::ExtensionTypeRegistryRef;
+
+/// A factory for creating [`ArrayFormatter`]s that checks whether a registered extension type can
+/// format a given array based on its metadata.
+#[derive(Debug)]
+pub struct DFArrayFormatterFactory {
+    /// The extension type registry
+    registry: ExtensionTypeRegistryRef,
+}
+
+impl DFArrayFormatterFactory {
+    /// Creates a new [`DFArrayFormatterFactory`].
+    pub fn new(registry: ExtensionTypeRegistryRef) -> Self {
+        Self { registry }
+    }
+}
+
+impl ArrayFormatterFactory for DFArrayFormatterFactory {
+    fn create_array_formatter<'formatter>(
+        &self,
+        array: &'formatter dyn Array,
+        options: &FormatOptions<'formatter>,
+        field: Option<&'formatter Field>,
+    ) -> Result<Option<ArrayFormatter<'formatter>>, ArrowError> {
+        let Some(field) = field else {
+            return Ok(None);
+        };
+
+        let Some(extension_type_name) = field.extension_type_name() else {
+            return Ok(None);
+        };
+
+        let Some(registration) = self
+            .registry
+            .extension_type_registration(extension_type_name)
+            .ok()
+        else {
+            // If the extension type is not registered, we fall back to the default formatter
+            return Ok(None);
+        };
+
+        registration
+            .create_df_extension_type(field.extension_type_metadata())?
+            .create_array_formatter(array, options)
+            .map_err(ArrowError::from)
+    }
+}
diff --git a/datafusion/core/src/extension_types/mod.rs b/datafusion/core/src/extension_types/mod.rs
new file mode 100644
index 0000000000000..55ec1ad95b5a1
--- /dev/null
+++ b/datafusion/core/src/extension_types/mod.rs
@@ -0,0 +1,22 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! This module contains code that enables DataFusion's extension type capabilities.
+
+mod array_formatter_factory;
+
+pub use array_formatter_factory::*;
diff --git a/datafusion/core/src/lib.rs b/datafusion/core/src/lib.rs
index 349eee5592abe..40705d79f645a 100644
--- a/datafusion/core/src/lib.rs
+++ b/datafusion/core/src/lib.rs
@@ -762,13 +762,11 @@
 //! [`RecordBatchReader`]: arrow::record_batch::RecordBatchReader
 //! [`Array`]: arrow::array::Array
 
-/// DataFusion crate version
-pub const DATAFUSION_VERSION: &str = env!("CARGO_PKG_VERSION");
-
 extern crate core;
-
 #[cfg(feature = "sql")]
 extern crate sqlparser;
+/// DataFusion crate version
+pub const DATAFUSION_VERSION: &str = env!("CARGO_PKG_VERSION");
 
 pub mod dataframe;
 pub mod datasource;
@@ -896,6 +894,7 @@ pub mod variable {
 #[cfg(not(target_arch = "wasm32"))]
 pub mod test;
 
+mod extension_types;
 mod schema_equivalence;
 pub mod test_util;
 
diff --git a/datafusion/core/tests/core_integration.rs b/datafusion/core/tests/core_integration.rs
index bdbe72245323d..99783427f022e 100644
--- a/datafusion/core/tests/core_integration.rs
+++ b/datafusion/core/tests/core_integration.rs
@@ -60,6 +60,9 @@ mod catalog_listing;
 /// Run all tests that are found in the `tracing` directory
 mod tracing;
 
+/// Run all tests that are found in the `extension_types` directory
+mod extension_types;
+
 #[cfg(test)]
 #[ctor::ctor]
 fn init() {
diff --git a/datafusion/core/tests/extension_types/mod.rs b/datafusion/core/tests/extension_types/mod.rs
new file mode 100644
index 0000000000000..bfe0c2e34927e
--- /dev/null
+++ b/datafusion/core/tests/extension_types/mod.rs
@@ -0,0 +1,18 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+mod pretty_printing;
diff --git a/datafusion/core/tests/extension_types/pretty_printing.rs b/datafusion/core/tests/extension_types/pretty_printing.rs
new file mode 100644
index 0000000000000..1a281dca59c37
--- /dev/null
+++ b/datafusion/core/tests/extension_types/pretty_printing.rs
@@ -0,0 +1,164 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use arrow::array::{FixedSizeBinaryArray, RecordBatch};
+use arrow_schema::extension::Uuid;
+use arrow_schema::{DataType, Field, FieldRef, Schema, SchemaRef};
+use datafusion::assert_batches_eq;
+use datafusion::dataframe::DataFrame;
+use datafusion::error::Result;
+use datafusion::execution::SessionStateBuilder;
+use datafusion::prelude::SessionContext;
+use datafusion_expr::planner::TypePlanner;
+use insta::assert_snapshot;
+use std::sync::Arc;
+
+fn test_schema() -> SchemaRef {
+    Arc::new(Schema::new(vec![uuid_field()]))
+}
+
+fn uuid_field() -> Field {
+    Field::new("my_uuids", DataType::FixedSizeBinary(16), false).with_extension_type(Uuid)
+}
+
+async fn create_test_table() -> Result<DataFrame> {
+    let schema = test_schema();
+
+    // define data.
+    let batch = RecordBatch::try_new(
+        schema,
+        vec![Arc::new(FixedSizeBinaryArray::from(vec![
+            &[0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
+            &[0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 5, 6],
+        ]))],
+    )?;
+
+    let state = SessionStateBuilder::default()
+        .with_canonical_extension_types()?
+        .build();
+    let ctx = SessionContext::new_with_state(state);
+
+    ctx.register_batch("test", batch)?;
+
+    ctx.table("test").await
+}
+
+// Test here
+
+#[tokio::test]
+async fn test_pretty_print_logical_plan() -> Result<()> {
+    let result = create_test_table().await?.to_string().await?;
+
+    assert_snapshot!(
+        result,
+        @r"
+    +--------------------------------------+
+    | my_uuids                             |
+    +--------------------------------------+
+    | 00000000-0000-0000-0000-000000000000 |
+    | 00010203-0405-0607-0809-000102030506 |
+    +--------------------------------------+
+    "
+    );
+
+    Ok(())
+}
+
+#[tokio::test]
+async fn create_cast_uuid_to_char() -> Result<()> {
+    let schema = test_schema();
+
+    // define data.
+    let batch = RecordBatch::try_new(
+        schema,
+        vec![Arc::new(FixedSizeBinaryArray::from(vec![
+            &[0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
+            &[0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 5, 6],
+        ]))],
+    )?;
+
+    let state = SessionStateBuilder::default()
+        .with_canonical_extension_types()?
+        .with_type_planner(Arc::new(CustomTypePlanner {}))
+        .build();
+    let ctx = SessionContext::new_with_state(state);
+
+    ctx.register_batch("test", batch)?;
+
+    let df = ctx.sql("SELECT my_uuids::VARCHAR FROM test").await?;
+    let batches = df.collect().await?;
+
+    assert_batches_eq!(
+        [
+            "+--------------------------------------+",
+            "| test.my_uuids                        |",
+            "+--------------------------------------+",
+            "| 00000000-0000-0000-0000-000000000000 |",
+            "| 00010203-0405-0607-0809-000102030506 |",
+            "+--------------------------------------+",
+        ],
+        &batches
+    );
+
+    Ok(())
+}
+
+#[tokio::test]
+async fn create_cast_char_to_uuid() -> Result<()> {
+    let state = SessionStateBuilder::default()
+        .with_canonical_extension_types()?
+        .with_type_planner(Arc::new(CustomTypePlanner {}))
+        .build();
+    let ctx = SessionContext::new_with_state(state);
+
+    let df = ctx
+        .sql("SELECT '00010203-0405-0607-0809-000102030506'::UUID AS uuid")
+        .await?;
+    let batches = df.collect().await?;
+    assert_batches_eq!(
+        [
+            "+----------------------------------+",
+            "| uuid                             |",
+            "+----------------------------------+",
+            "| 00010203040506070809000102030506 |",
+            "+----------------------------------+",
+        ],
+        &batches
+    );
+
+    Ok(())
+}
+
+#[derive(Debug)]
+pub struct CustomTypePlanner {}
+
+impl TypePlanner for CustomTypePlanner {
+    fn plan_type_field(
+        &self,
+        sql_type: &sqlparser::ast::DataType,
+    ) -> Result<Option<FieldRef>> {
+        match sql_type {
+            sqlparser::ast::DataType::Uuid => Ok(Some(Arc::new(
+                Field::new("", DataType::FixedSizeBinary(16), true).with_metadata(
+                    [("ARROW:extension:name".to_string(), "arrow.uuid".to_string())]
+                        .into(),
+                ),
+            ))),
+            _ => Ok(None),
+        }
+    }
+}
diff --git a/datafusion/datasource-arrow/src/file_format.rs b/datafusion/datasource-arrow/src/file_format.rs
index f60bce3249935..05088a885da0f 100644
--- a/datafusion/datasource-arrow/src/file_format.rs
+++ b/datafusion/datasource-arrow/src/file_format.rs
@@ -555,6 +555,7 @@ mod tests {
     use datafusion_execution::config::SessionConfig;
     use datafusion_execution::runtime_env::RuntimeEnv;
     use datafusion_expr::execution_props::ExecutionProps;
+    use datafusion_expr::registry::ExtensionTypeRegistryRef;
     use datafusion_expr::{AggregateUDF, Expr, LogicalPlan, ScalarUDF, WindowUDF};
     use datafusion_physical_expr_common::physical_expr::PhysicalExpr;
     use object_store::{chunked::ChunkedStore, memory::InMemory, path::Path};
@@ -610,6 +611,10 @@ mod tests {
             unimplemented!()
         }
 
+        fn extension_type_registry(&self) -> &ExtensionTypeRegistryRef {
+            unimplemented!()
+        }
+
         fn runtime_env(&self) -> &Arc<RuntimeEnv> {
             &self.runtime_env
         }
diff --git a/datafusion/datasource/src/url.rs b/datafusion/datasource/src/url.rs
index 39d1047984ff6..ef81354007c15 100644
--- a/datafusion/datasource/src/url.rs
+++ b/datafusion/datasource/src/url.rs
@@ -517,6 +517,7 @@ mod tests {
     use datafusion_execution::config::SessionConfig;
     use datafusion_execution::runtime_env::RuntimeEnv;
     use datafusion_expr::execution_props::ExecutionProps;
+    use datafusion_expr::registry::ExtensionTypeRegistryRef;
     use datafusion_expr::{AggregateUDF, Expr, LogicalPlan, ScalarUDF, WindowUDF};
     use datafusion_physical_expr_common::physical_expr::PhysicalExpr;
     use datafusion_physical_plan::ExecutionPlan;
@@ -1216,6 +1217,10 @@ mod tests {
             unimplemented!()
         }
 
+        fn extension_type_registry(&self) -> &ExtensionTypeRegistryRef {
+            unimplemented!()
+        }
+
         fn runtime_env(&self) -> &Arc<RuntimeEnv> {
             &self.runtime_env
         }
diff --git a/datafusion/expr/Cargo.toml b/datafusion/expr/Cargo.toml
index 6990714585001..265651cfe0eb0 100644
--- a/datafusion/expr/Cargo.toml
+++ b/datafusion/expr/Cargo.toml
@@ -46,7 +46,8 @@ recursive_protection = ["dep:recursive"]
 sql = ["sqlparser"]
 
 [dependencies]
-arrow = { workspace = true }
+arrow = { workspace = true, features = ["canonical_extension_types"] }
+arrow-schema = { workspace = true, features = ["canonical_extension_types"] }
 async-trait = { workspace = true }
 chrono = { workspace = true }
 datafusion-common = { workspace = true, default-features = false }
diff --git a/datafusion/expr/src/execution_props.rs b/datafusion/expr/src/execution_props.rs
index 3bf6978eb60ee..2808a7ae5935a 100644
--- a/datafusion/expr/src/execution_props.rs
+++ b/datafusion/expr/src/execution_props.rs
@@ -15,6 +15,7 @@
 // specific language governing permissions and limitations
 // under the License.
 
+use crate::registry::ExtensionTypeRegistry;
 use crate::var_provider::{VarProvider, VarType};
 use chrono::{DateTime, Utc};
 use datafusion_common::HashMap;
@@ -42,6 +43,7 @@ pub struct ExecutionProps {
     pub config_options: Option<Arc<ConfigOptions>>,
     /// Providers for scalar variables
     pub var_providers: Option<HashMap<VarType, Arc<dyn VarProvider + Send + Sync>>>,
+    pub extension_types: Option<Arc<dyn ExtensionTypeRegistry>>,
 }
 
 impl Default for ExecutionProps {
@@ -58,6 +60,7 @@ impl ExecutionProps {
             alias_generator: Arc::new(AliasGenerator::new()),
             config_options: None,
             var_providers: None,
+            extension_types: None,
         }
     }
 
@@ -126,7 +129,7 @@ mod test {
     fn debug() {
         let props = ExecutionProps::new();
         assert_eq!(
-            "ExecutionProps { query_execution_start_time: None, alias_generator: AliasGenerator { next_id: 1 }, config_options: None, var_providers: None }",
+            "ExecutionProps { query_execution_start_time: None, alias_generator: AliasGenerator { next_id: 1 }, config_options: None, var_providers: None, extension_types: None }",
             format!("{props:?}")
         );
     }
diff --git a/datafusion/expr/src/registry.rs b/datafusion/expr/src/registry.rs
index 472e065211aac..53d67126b04ae 100644
--- a/datafusion/expr/src/registry.rs
+++ b/datafusion/expr/src/registry.rs
@@ -20,10 +20,13 @@
 use crate::expr_rewriter::FunctionRewrite;
 use crate::planner::ExprPlanner;
 use crate::{AggregateUDF, ScalarUDF, UserDefinedLogicalNode, WindowUDF};
+use arrow::datatypes::Field;
+use arrow_schema::extension::ExtensionType;
+use datafusion_common::types::{DFExtensionType, DFExtensionTypeRef};
 use datafusion_common::{HashMap, Result, not_impl_err, plan_datafusion_err};
 use std::collections::HashSet;
-use std::fmt::Debug;
-use std::sync::Arc;
+use std::fmt::{Debug, Formatter};
+use std::sync::{Arc, RwLock};
 
 /// A registry knows how to build logical expressions out of user-defined function' names
 pub trait FunctionRegistry {
@@ -215,3 +218,266 @@ impl FunctionRegistry for MemoryFunctionRegistry {
         self.udwfs.keys().cloned().collect()
     }
 }
+
+/// A cheaply cloneable pointer to an [ExtensionTypeRegistration].
+pub type ExtensionTypeRegistrationRef = Arc<dyn ExtensionTypeRegistration>;
+
+/// The registration of an extension type. Implementations of this trait are responsible for
+/// *creating* instances of [`DFExtensionType`] that represent the entire semantics of an extension
+/// type.
+///
+/// # Why do we need a Registration?
+///
+/// A good question is why this trait is even necessary. Why not directly register the
+/// [`DFExtensionType`] in a registration?
+///
+/// While this works for extension types without parameters (e.g., `arrow.uuid`), it does not work
+/// for more complex extension types that may have another extension type as a parameter. For
+/// example, consider an extension type `custom.shortened(n)` that aims to short the pretty-printing
+/// string to `n` characters. Here, `n` is a parameter of the extension type and should be a field
+/// in the concrete struct that implements the [`DFExtensionType`]. The job of the registration is
+/// to read the metadata from the field and create the corresponding [`DFExtensionType`] instance
+/// with the correct `n` set.
+///
+/// The [`DefaultExtensionTypeRegistration`] provides a convenient way of creating registrations.
+pub trait ExtensionTypeRegistration: Debug + Send + Sync {
+    /// The name of the extension type.
+    ///
+    /// This name will be used to find the correct [ExtensionTypeRegistration] when an extension
+    /// type is encountered.
+    fn type_name(&self) -> &str;
+
+    /// Creates an extension type instance from the optional metadata. The name of the extension
+    /// type is not a parameter as it's already defined by the registration itself.
+    fn create_df_extension_type(
+        &self,
+        metadata: Option<&str>,
+    ) -> Result<DFExtensionTypeRef>;
+}
+
+/// A cheaply cloneable pointer to an [ExtensionTypeRegistry].
+pub type ExtensionTypeRegistryRef = Arc<dyn ExtensionTypeRegistry>;
+
+/// Manages [`ExtensionTypeRegistration`]s, which allow users to register custom behavior for
+/// extension types.
+///
+/// Each registration is connected to the extension type name, which can also be looked up to get
+/// the registration.
+pub trait ExtensionTypeRegistry: Debug + Send + Sync {
+    /// Returns a reference to registration of an extension type named `name`.
+    ///
+    /// Returns an error if there is no extension type with that name.
+    fn extension_type_registration(
+        &self,
+        name: &str,
+    ) -> Result<ExtensionTypeRegistrationRef>;
+
+    /// Creates a [`DFExtensionTypeRef`] from the type information in the `field`.
+    ///
+    /// The result `Ok(None)` indicates that there is no extension type metadata. Returns an error
+    /// if the extension type in the metadata is not found.
+    fn create_extension_type_for_field(
+        &self,
+        field: &Field,
+    ) -> Result<Option<DFExtensionTypeRef>> {
+        let Some(extension_type_name) = field.extension_type_name() else {
+            return Ok(None);
+        };
+
+        let registration = self.extension_type_registration(extension_type_name)?;
+        registration
+            .create_df_extension_type(field.extension_type_metadata())
+            .map(Some)
+    }
+
+    /// Returns all registered [ExtensionTypeRegistration].
+    fn extension_type_registrations(&self) -> Vec<Arc<dyn ExtensionTypeRegistration>>;
+
+    /// Registers a new [ExtensionTypeRegistrationRef], returning any previously registered
+    /// implementation.
+    ///
+    /// Returns an error if the type cannot be registered, for example, if the registry is
+    /// read-only.
+    fn add_extension_type_registration(
+        &self,
+        extension_type: ExtensionTypeRegistrationRef,
+    ) -> Result<Option<ExtensionTypeRegistrationRef>>;
+
+    /// Extends the registry with the provided extension types.
+    ///
+    /// Returns an error if the type cannot be registered, for example, if the registry is
+    /// read-only.
+    fn extend(&self, extension_types: &[ExtensionTypeRegistrationRef]) -> Result<()> {
+        for extension_type in extension_types.iter().cloned() {
+            self.add_extension_type_registration(extension_type)?;
+        }
+        Ok(())
+    }
+
+    /// Deregisters an extension type registration with the name `name`, returning the
+    /// implementation that was deregistered.
+    ///
+    /// Returns an error if the type cannot be deregistered, for example, if the registry is
+    /// read-only.
+    fn remove_extension_type_registration(
+        &self,
+        name: &str,
+    ) -> Result<Option<ExtensionTypeRegistrationRef>>;
+}
+
+/// A default implementation of [ExtensionTypeRegistration] that parses the metadata from the
+/// given extension type and passes it to a constructor function.
+pub struct DefaultExtensionTypeRegistration<
+    TExtensionType: ExtensionType + DFExtensionType + 'static,
+> {
+    /// A function that creates an instance of [`DFExtensionTypeRef`] from the metadata.
+    factory:
+        Box<dyn Fn(TExtensionType::Metadata) -> Result<TExtensionType> + Send + Sync>,
+}
+
+impl<TExtensionType: ExtensionType + DFExtensionType + 'static>
+    DefaultExtensionTypeRegistration<TExtensionType>
+{
+    /// Creates a new registration for the given `name` and `logical_type`.
+    pub fn new_arc(
+        factory: impl Fn(TExtensionType::Metadata) -> Result<TExtensionType>
+        + Send
+        + Sync
+        + 'static,
+    ) -> ExtensionTypeRegistrationRef {
+        Arc::new(Self {
+            factory: Box::new(factory),
+        })
+    }
+}
+
+impl<TExtensionType: ExtensionType + DFExtensionType> ExtensionTypeRegistration
+    for DefaultExtensionTypeRegistration<TExtensionType>
+{
+    fn type_name(&self) -> &str {
+        TExtensionType::NAME
+    }
+
+    fn create_df_extension_type(
+        &self,
+        metadata: Option<&str>,
+    ) -> Result<DFExtensionTypeRef> {
+        let metadata = TExtensionType::deserialize_metadata(metadata)?;
+        self.factory.as_ref()(metadata)
+            .map(|extension_type| Arc::new(extension_type) as DFExtensionTypeRef)
+    }
+}
+
+impl<TExtensionType: ExtensionType + DFExtensionType> Debug
+    for DefaultExtensionTypeRegistration<TExtensionType>
+{
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("DefaultExtensionTypeRegistration")
+            .field("type_name", &TExtensionType::NAME)
+            .finish()
+    }
+}
+
+/// An [`ExtensionTypeRegistry`] that uses in memory [`HashMap`]s.
+#[derive(Clone, Debug)]
+pub struct MemoryExtensionTypeRegistry {
+    /// Holds a mapping between the name of an extension type and its logical type.
+    extension_types: Arc<RwLock<HashMap<String, ExtensionTypeRegistrationRef>>>,
+}
+
+impl Default for MemoryExtensionTypeRegistry {
+    fn default() -> Self {
+        MemoryExtensionTypeRegistry {
+            extension_types: Arc::new(RwLock::new(HashMap::new())),
+        }
+    }
+}
+
+impl MemoryExtensionTypeRegistry {
+    /// Creates an empty [MemoryExtensionTypeRegistry].
+    pub fn new() -> Self {
+        Self {
+            extension_types: Arc::new(RwLock::new(HashMap::new())),
+        }
+    }
+
+    /// Creates a new [MemoryExtensionTypeRegistry] with the provided `types`.
+    ///
+    /// # Errors
+    ///
+    /// Returns an error if one of the `types` is a native type.
+    pub fn new_with_types(
+        types: impl IntoIterator<Item = ExtensionTypeRegistrationRef>,
+    ) -> Result<Self> {
+        let extension_types = types
+            .into_iter()
+            .map(|t| (t.type_name().to_owned(), t))
+            .collect::<HashMap<_, _>>();
+        Ok(Self {
+            extension_types: Arc::new(RwLock::new(extension_types)),
+        })
+    }
+
+    /// Returns a list of all registered types.
+    pub fn all_extension_types(&self) -> Vec<ExtensionTypeRegistrationRef> {
+        self.extension_types
+            .read()
+            .expect("Extension type registry lock poisoned")
+            .values()
+            .cloned()
+            .collect()
+    }
+}
+
+impl ExtensionTypeRegistry for MemoryExtensionTypeRegistry {
+    fn extension_type_registration(
+        &self,
+        name: &str,
+    ) -> Result<ExtensionTypeRegistrationRef> {
+        self.extension_types
+            .write()
+            .expect("Extension type registry lock poisoned")
+            .get(name)
+            .ok_or_else(|| plan_datafusion_err!("Logical type not found."))
+            .cloned()
+    }
+
+    fn extension_type_registrations(&self) -> Vec<Arc<dyn ExtensionTypeRegistration>> {
+        self.extension_types
+            .read()
+            .expect("Extension type registry lock poisoned")
+            .values()
+            .cloned()
+            .collect()
+    }
+
+    fn add_extension_type_registration(
+        &self,
+        extension_type: ExtensionTypeRegistrationRef,
+    ) -> Result<Option<ExtensionTypeRegistrationRef>> {
+        Ok(self
+            .extension_types
+            .write()
+            .expect("Extension type registry lock poisoned")
+            .insert(extension_type.type_name().to_owned(), extension_type))
+    }
+
+    fn remove_extension_type_registration(
+        &self,
+        name: &str,
+    ) -> Result<Option<ExtensionTypeRegistrationRef>> {
+        Ok(self
+            .extension_types
+            .write()
+            .expect("Extension type registry lock poisoned")
+            .remove(name))
+    }
+}
+
+impl From<HashMap<String, ExtensionTypeRegistrationRef>> for MemoryExtensionTypeRegistry {
+    fn from(value: HashMap<String, ExtensionTypeRegistrationRef>) -> Self {
+        Self {
+            extension_types: Arc::new(RwLock::new(value)),
+        }
+    }
+}
diff --git a/datafusion/expr/src/simplify.rs b/datafusion/expr/src/simplify.rs
index 8c68067a55a37..ae26af834731d 100644
--- a/datafusion/expr/src/simplify.rs
+++ b/datafusion/expr/src/simplify.rs
@@ -24,6 +24,7 @@ use chrono::{DateTime, Utc};
 use datafusion_common::config::ConfigOptions;
 use datafusion_common::{DFSchema, DFSchemaRef, Result};
 
+use crate::registry::ExtensionTypeRegistry;
 use crate::{Expr, ExprSchemable};
 
 /// Provides simplification information based on schema, query execution time,
@@ -38,6 +39,7 @@ pub struct SimplifyContext {
     schema: DFSchemaRef,
     query_execution_start_time: Option<DateTime<Utc>>,
     config_options: Arc<ConfigOptions>,
+    extension_types: Option<Arc<dyn ExtensionTypeRegistry>>,
 }
 
 impl Default for SimplifyContext {
@@ -46,6 +48,7 @@ impl Default for SimplifyContext {
             schema: Arc::new(DFSchema::empty()),
             query_execution_start_time: None,
             config_options: Arc::new(ConfigOptions::default()),
+            extension_types: None,
         }
     }
 }
@@ -78,6 +81,14 @@ impl SimplifyContext {
         self
     }
 
+    pub fn with_extension_types(
+        mut self,
+        extension_types: Option<Arc<dyn ExtensionTypeRegistry>>,
+    ) -> Self {
+        self.extension_types = extension_types;
+        self
+    }
+
     /// Returns the schema
     pub fn schema(&self) -> &DFSchemaRef {
         &self.schema
@@ -108,6 +119,10 @@ impl SimplifyContext {
     pub fn config_options(&self) -> &Arc<ConfigOptions> {
         &self.config_options
     }
+
+    pub fn extension_types(&self) -> Option<&Arc<dyn ExtensionTypeRegistry>> {
+        self.extension_types.as_ref()
+    }
 }
 
 /// Was the expression simplified?
diff --git a/datafusion/ffi/src/session/mod.rs b/datafusion/ffi/src/session/mod.rs
index 007181356e1b8..f7bbc088aac67 100644
--- a/datafusion/ffi/src/session/mod.rs
+++ b/datafusion/ffi/src/session/mod.rs
@@ -20,6 +20,17 @@ use std::collections::HashMap;
 use std::ffi::c_void;
 use std::sync::Arc;
 
+use crate::arrow_wrappers::WrappedSchema;
+use crate::execution::FFI_TaskContext;
+use crate::execution_plan::FFI_ExecutionPlan;
+use crate::physical_expr::FFI_PhysicalExpr;
+use crate::proto::logical_extension_codec::FFI_LogicalExtensionCodec;
+use crate::session::config::FFI_SessionConfig;
+use crate::udaf::FFI_AggregateUDF;
+use crate::udf::FFI_ScalarUDF;
+use crate::udwf::FFI_WindowUDF;
+use crate::util::FFIResult;
+use crate::{df_result, rresult, rresult_return};
 use abi_stable::StableAbi;
 use abi_stable::std_types::{RHashMap, RResult, RStr, RString, RVec};
 use arrow_schema::SchemaRef;
@@ -32,6 +43,7 @@ use datafusion_execution::TaskContext;
 use datafusion_execution::config::SessionConfig;
 use datafusion_execution::runtime_env::RuntimeEnv;
 use datafusion_expr::execution_props::ExecutionProps;
+use datafusion_expr::registry::{ExtensionTypeRegistryRef, MemoryExtensionTypeRegistry};
 use datafusion_expr::{
     AggregateUDF, AggregateUDFImpl, Expr, LogicalPlan, ScalarUDF, ScalarUDFImpl,
     WindowUDF, WindowUDFImpl,
@@ -47,18 +59,6 @@ use datafusion_session::Session;
 use prost::Message;
 use tokio::runtime::Handle;
 
-use crate::arrow_wrappers::WrappedSchema;
-use crate::execution::FFI_TaskContext;
-use crate::execution_plan::FFI_ExecutionPlan;
-use crate::physical_expr::FFI_PhysicalExpr;
-use crate::proto::logical_extension_codec::FFI_LogicalExtensionCodec;
-use crate::session::config::FFI_SessionConfig;
-use crate::udaf::FFI_AggregateUDF;
-use crate::udf::FFI_ScalarUDF;
-use crate::udwf::FFI_WindowUDF;
-use crate::util::FFIResult;
-use crate::{df_result, rresult, rresult_return};
-
 pub mod config;
 
 /// A stable struct for sharing [`Session`] across FFI boundaries.
@@ -378,6 +378,7 @@ pub struct ForeignSession {
     scalar_functions: HashMap<String, Arc<ScalarUDF>>,
     aggregate_functions: HashMap<String, Arc<AggregateUDF>>,
     window_functions: HashMap<String, Arc<WindowUDF>>,
+    extension_types: ExtensionTypeRegistryRef,
     table_options: TableOptions,
     runtime_env: Arc<RuntimeEnv>,
     props: ExecutionProps,
@@ -446,6 +447,7 @@ impl TryFrom<&FFI_SessionRef> for ForeignSession {
                 scalar_functions,
                 aggregate_functions,
                 window_functions,
+                extension_types: Arc::new(MemoryExtensionTypeRegistry::default()),
                 runtime_env: Default::default(),
                 props: Default::default(),
             })
@@ -592,6 +594,10 @@ impl Session for ForeignSession {
         &self.window_functions
     }
 
+    fn extension_type_registry(&self) -> &ExtensionTypeRegistryRef {
+        &self.extension_types
+    }
+
     fn runtime_env(&self) -> &Arc<RuntimeEnv> {
         &self.runtime_env
     }
diff --git a/datafusion/optimizer/src/optimizer.rs b/datafusion/optimizer/src/optimizer.rs
index bdea6a83072cd..b57f31e3eb585 100644
--- a/datafusion/optimizer/src/optimizer.rs
+++ b/datafusion/optimizer/src/optimizer.rs
@@ -21,7 +21,7 @@ use std::fmt::Debug;
 use std::sync::Arc;
 
 use chrono::{DateTime, Utc};
-use datafusion_expr::registry::FunctionRegistry;
+use datafusion_expr::registry::{ExtensionTypeRegistry, FunctionRegistry};
 use datafusion_expr::{InvariantLevel, assert_expected_schema};
 use log::{debug, warn};
 
@@ -146,6 +146,10 @@ pub trait OptimizerConfig {
     fn function_registry(&self) -> Option<&dyn FunctionRegistry> {
         None
     }
+
+    fn extension_types(&self) -> Option<Arc<dyn ExtensionTypeRegistry>> {
+        None
+    }
 }
 
 /// A standalone [`OptimizerConfig`] that can be used independently
diff --git a/datafusion/optimizer/src/simplify_expressions/expr_simplifier.rs b/datafusion/optimizer/src/simplify_expressions/expr_simplifier.rs
index 28fcdf1dede0b..5ed19822a881c 100644
--- a/datafusion/optimizer/src/simplify_expressions/expr_simplifier.rs
+++ b/datafusion/optimizer/src/simplify_expressions/expr_simplifier.rs
@@ -41,6 +41,7 @@ use datafusion_common::{
 use datafusion_expr::{
     BinaryExpr, Case, ColumnarValue, Expr, ExprSchemable, Like, Operator, Volatility,
     and, binary::BinaryTypeCoercer, lit, or, preimage::PreimageResult,
+    registry::ExtensionTypeRegistry,
 };
 use datafusion_expr::{Cast, TryCast, simplify::ExprSimplifyResult};
 use datafusion_expr::{expr::ScalarFunction, interval_arithmetic::NullableInterval};
@@ -209,7 +210,10 @@ impl ExprSimplifier {
     ) -> Result<(Transformed<Expr>, u32)> {
         let mut simplifier = Simplifier::new(&self.info);
         let config_options = Some(Arc::clone(self.info.config_options()));
-        let mut const_evaluator = ConstEvaluator::try_new(config_options)?;
+        let mut const_evaluator = ConstEvaluator::try_new(
+            config_options,
+            self.info.extension_types().cloned(),
+        )?;
         let mut shorten_in_list_simplifier = ShortenInListSimplifier::new();
         let guarantees_map: HashMap<&Expr, &NullableInterval> =
             self.guarantees.iter().map(|(k, v)| (k, v)).collect();
@@ -585,7 +589,10 @@ impl ConstEvaluator {
     ///
     /// The `config_options` parameter is used to pass session configuration
     /// (like timezone) to scalar functions during constant evaluation.
-    pub fn try_new(config_options: Option<Arc<ConfigOptions>>) -> Result<Self> {
+    pub fn try_new(
+        config_options: Option<Arc<ConfigOptions>>,
+        extension_types: Option<Arc<dyn ExtensionTypeRegistry>>,
+    ) -> Result<Self> {
         // The dummy column name is unused and doesn't matter as only
         // expressions without column references can be evaluated
         static DUMMY_COL_NAME: &str = ".";
@@ -601,6 +608,7 @@ impl ConstEvaluator {
 
         let mut execution_props = ExecutionProps::new();
         execution_props.config_options = config_options;
+        execution_props.extension_types = extension_types;
 
         Ok(Self {
             can_evaluate: vec![],
diff --git a/datafusion/optimizer/src/simplify_expressions/simplify_exprs.rs b/datafusion/optimizer/src/simplify_expressions/simplify_exprs.rs
index 29ee59342273e..8a6f87e2adf99 100644
--- a/datafusion/optimizer/src/simplify_expressions/simplify_exprs.rs
+++ b/datafusion/optimizer/src/simplify_expressions/simplify_exprs.rs
@@ -101,7 +101,8 @@ impl SimplifyExpressions {
         let info = SimplifyContext::default()
             .with_schema(schema)
             .with_config_options(config.options())
-            .with_query_execution_start_time(config.query_execution_start_time());
+            .with_query_execution_start_time(config.query_execution_start_time())
+            .with_extension_types(config.extension_types().clone());
 
         // Inputs have already been rewritten (due to bottom-up traversal handled by Optimizer)
         // Just need to rewrite our own expressions
diff --git a/datafusion/physical-expr/src/expressions/cast.rs b/datafusion/physical-expr/src/expressions/cast.rs
index 5a80daf663a36..2884cf03f40d6 100644
--- a/datafusion/physical-expr/src/expressions/cast.rs
+++ b/datafusion/physical-expr/src/expressions/cast.rs
@@ -28,7 +28,8 @@ use arrow::record_batch::RecordBatch;
 use datafusion_common::datatype::DataTypeExt;
 use datafusion_common::format::DEFAULT_FORMAT_OPTIONS;
 use datafusion_common::nested_struct::validate_struct_compatibility;
-use datafusion_common::{Result, not_impl_err};
+use datafusion_common::types::CastExtension;
+use datafusion_common::{Result, ScalarValue, not_impl_err};
 use datafusion_expr_common::columnar_value::ColumnarValue;
 use datafusion_expr_common::interval_arithmetic::Interval;
 use datafusion_expr_common::sort_properties::ExprProperties;
@@ -60,7 +61,7 @@ fn can_cast_struct_types(source: &DataType, target: &DataType) -> bool {
 }
 
 /// CAST expression casts an expression to a specific data type and returns a runtime error on invalid cast
-#[derive(Debug, Clone, Eq)]
+#[derive(Debug, Clone)]
 pub struct CastExpr {
     /// The expression to cast
     pub expr: Arc<dyn PhysicalExpr>,
@@ -68,6 +69,8 @@ pub struct CastExpr {
     target_field: FieldRef,
     /// Cast options
     cast_options: CastOptions<'static>,
+    // CastExtension
+    cast_extension: Option<Arc<dyn CastExtension>>,
 }
 
 // Manually derive PartialEq and Hash to work around https://github.com/rust-lang/rust/issues/78808
@@ -87,6 +90,8 @@ impl Hash for CastExpr {
     }
 }
 
+impl Eq for CastExpr {}
+
 impl CastExpr {
     /// Create a new `CastExpr` using only a `DataType`.
     ///
@@ -132,6 +137,17 @@ impl CastExpr {
             expr,
             target_field,
             cast_options: cast_options.unwrap_or(DEFAULT_CAST_OPTIONS),
+            cast_extension: None,
+        }
+    }
+
+    pub fn with_cast_extension(
+        self,
+        cast_extension: Option<Arc<dyn CastExtension>>,
+    ) -> Self {
+        Self {
+            cast_extension,
+            ..self
         }
     }
 
@@ -159,6 +175,7 @@ impl CastExpr {
         self.target_field.name().is_empty()
             && self.target_field.is_nullable()
             && self.target_field.metadata().is_empty()
+            && self.cast_extension.is_none()
     }
 
     fn resolved_target_field(&self, input_schema: &Schema) -> Result<FieldRef> {
@@ -255,7 +272,35 @@ impl PhysicalExpr for CastExpr {
 
     fn evaluate(&self, batch: &RecordBatch) -> Result<ColumnarValue> {
         let value = self.expr.evaluate(batch)?;
-        value.cast_to(self.cast_type(), Some(&self.cast_options))
+        if let Some(cast_extension) = &self.cast_extension {
+            let from_field = self.expr.return_field(&batch.schema())?;
+            let to_field = self.return_field(&batch.schema())?;
+            match value {
+                ColumnarValue::Array(array) => {
+                    Ok(ColumnarValue::Array(cast_extension.cast(
+                        array,
+                        &from_field,
+                        &to_field,
+                        &self.cast_options,
+                    )?))
+                }
+                ColumnarValue::Scalar(scalar_value) => {
+                    let array = scalar_value.to_array()?;
+                    let array_result = cast_extension.cast(
+                        array,
+                        &from_field,
+                        &to_field,
+                        &self.cast_options,
+                    )?;
+                    Ok(ColumnarValue::Scalar(ScalarValue::try_from_array(
+                        &array_result,
+                        0,
+                    )?))
+                }
+            }
+        } else {
+            value.cast_to(self.cast_type(), Some(&self.cast_options))
+        }
     }
 
     fn return_field(&self, input_schema: &Schema) -> Result<FieldRef> {
@@ -352,6 +397,18 @@ pub fn cast(
     cast_with_options(expr, input_schema, cast_type, None)
 }
 
+pub fn cast_with_extension(
+    expr: Arc<dyn PhysicalExpr>,
+    _input_schema: &Schema,
+    cast_type: DataType,
+    cast_extension: Arc<dyn CastExtension>,
+) -> Result<Arc<dyn PhysicalExpr>> {
+    Ok(Arc::new(
+        CastExpr::new(expr, cast_type, Some(DEFAULT_CAST_OPTIONS))
+            .with_cast_extension(Some(cast_extension)),
+    ))
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
diff --git a/datafusion/physical-expr/src/expressions/mod.rs b/datafusion/physical-expr/src/expressions/mod.rs
index c9e02708d6c28..f256ca73c7d25 100644
--- a/datafusion/physical-expr/src/expressions/mod.rs
+++ b/datafusion/physical-expr/src/expressions/mod.rs
@@ -41,7 +41,7 @@ pub use crate::aggregate::stats::StatsType;
 
 pub use binary::{BinaryExpr, binary, similar_to};
 pub use case::{CaseExpr, case};
-pub use cast::{CastExpr, cast};
+pub use cast::{CastExpr, cast, cast_with_extension};
 pub use cast_column::CastColumnExpr;
 pub use column::{Column, col, with_new_schema};
 pub use datafusion_expr::utils::format_state_name;
diff --git a/datafusion/physical-expr/src/planner.rs b/datafusion/physical-expr/src/planner.rs
index 5c170700d9833..86dab578c4c7e 100644
--- a/datafusion/physical-expr/src/planner.rs
+++ b/datafusion/physical-expr/src/planner.rs
@@ -23,8 +23,10 @@ use crate::{
     expressions::{self, Column, Literal, binary, like, similar_to},
 };
 
+use arrow::compute::CastOptions;
 use arrow::datatypes::Schema;
 use datafusion_common::config::ConfigOptions;
+use datafusion_common::format::DEFAULT_FORMAT_OPTIONS;
 use datafusion_common::metadata::{FieldMetadata, format_type_and_metadata};
 use datafusion_common::{
     DFSchema, Result, ScalarValue, ToDFSchema, exec_err, not_impl_err, plan_err,
@@ -289,8 +291,32 @@ pub fn create_physical_expr(
             Ok(expressions::case(expr, when_then_expr, else_expr)?)
         }
         Expr::Cast(Cast { expr, field }) => {
+            let (_, src_field) = expr.to_field(input_dfschema)?;
+            const DEFAULT_CAST_OPTIONS: CastOptions<'static> = CastOptions {
+                safe: false,
+                format_options: DEFAULT_FORMAT_OPTIONS,
+            };
+
             if !field.metadata().is_empty() {
-                let (_, src_field) = expr.to_field(input_dfschema)?;
+                if let Some(registry) = &execution_props.extension_types
+                    && let Some(extension_type) =
+                        registry.create_extension_type_for_field(field)?
+                {
+                    let cast_extension = extension_type.cast_from()?;
+                    if cast_extension.can_cast(
+                        &src_field,
+                        field,
+                        &DEFAULT_CAST_OPTIONS,
+                    )? {
+                        return expressions::cast_with_extension(
+                            create_physical_expr(expr, input_dfschema, execution_props)?,
+                            input_schema,
+                            field.data_type().clone(),
+                            cast_extension,
+                        );
+                    }
+                }
+
                 return plan_err!(
                     "Cast from {} to {} is not supported",
                     format_type_and_metadata(
@@ -299,6 +325,31 @@ pub fn create_physical_expr(
                     ),
                     format_type_and_metadata(field.data_type(), Some(field.metadata()))
                 );
+            } else if let Some(registry) = &execution_props.extension_types
+                && let Some(extension_type) =
+                    registry.create_extension_type_for_field(&src_field)?
+            {
+                let cast_extension = extension_type.cast_to()?;
+                if cast_extension.can_cast(&src_field, field, &DEFAULT_CAST_OPTIONS)? {
+                    return expressions::cast_with_extension(
+                        create_physical_expr(expr, input_dfschema, execution_props)?,
+                        input_schema,
+                        field.data_type().clone(),
+                        cast_extension,
+                    );
+                } else {
+                    return plan_err!(
+                        "Cast from {} to {} is not supported",
+                        format_type_and_metadata(
+                            src_field.data_type(),
+                            Some(src_field.metadata()),
+                        ),
+                        format_type_and_metadata(
+                            field.data_type(),
+                            Some(field.metadata())
+                        )
+                    );
+                }
             }
 
             expressions::cast(
diff --git a/datafusion/session/src/session.rs b/datafusion/session/src/session.rs
index 2593e8cd71f4c..34506a0a1718d 100644
--- a/datafusion/session/src/session.rs
+++ b/datafusion/session/src/session.rs
@@ -22,6 +22,7 @@ use datafusion_execution::TaskContext;
 use datafusion_execution::config::SessionConfig;
 use datafusion_execution::runtime_env::RuntimeEnv;
 use datafusion_expr::execution_props::ExecutionProps;
+use datafusion_expr::registry::ExtensionTypeRegistryRef;
 use datafusion_expr::{AggregateUDF, Expr, LogicalPlan, ScalarUDF, WindowUDF};
 use datafusion_physical_plan::{ExecutionPlan, PhysicalExpr};
 use parking_lot::{Mutex, RwLock};
@@ -116,6 +117,9 @@ pub trait Session: Send + Sync {
     /// Return reference to window functions
     fn window_functions(&self) -> &HashMap<String, Arc<WindowUDF>>;
 
+    /// Return a reference to the extension type registry
+    fn extension_type_registry(&self) -> &ExtensionTypeRegistryRef;
+
     /// Return the runtime env
     fn runtime_env(&self) -> &Arc<RuntimeEnv>;