diff --git a/Cargo.lock b/Cargo.lock
index 18e1b8b7f723f..a5c23f212bb18 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1898,6 +1898,7 @@ dependencies = [
  "apache-avro",
  "arrow",
  "arrow-ipc",
+ "arrow-schema",
  "chrono",
  "criterion",
  "half",
@@ -1915,6 +1916,7 @@ dependencies = [
  "recursive",
  "sqlparser",
  "tokio",
+ "uuid",
  "web-time",
 ]
 
@@ -2151,6 +2153,7 @@ name = "datafusion-expr"
 version = "52.1.0"
 dependencies = [
  "arrow",
+ "arrow-schema",
  "async-trait",
  "chrono",
  "ctor",
diff --git a/datafusion-examples/README.md b/datafusion-examples/README.md
index 2cf0ec52409f8..8463f6b19e284 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                          |
+| ----------- | --------------------------------------------------------------------------- | ------------------------------------ |
+| temperature | [`extension_types/temperature.rs`](examples/extension_types/temperature.rs) | Extension type for temperature data. |
+
 ## External Dependency Examples
 
 ### Group: `external_dependency`
diff --git a/datafusion-examples/examples/extension_types/main.rs b/datafusion-examples/examples/extension_types/main.rs
new file mode 100644
index 0000000000000..97c00fdcb64f8
--- /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|temperature]
+//! ```
+//!
+//! Each subcommand runs a corresponding example:
+//! - `all` — run all examples included in this module
+//!
+//! - `temperature`
+//!   (file: temperature.rs, desc: Extension type for temperature data.)
+
+mod temperature;
+
+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,
+    Temperature,
+}
+
+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::Temperature => {
+                temperature::temperature_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-examples/examples/extension_types/temperature.rs b/datafusion-examples/examples/extension_types/temperature.rs
new file mode 100644
index 0000000000000..6031da1660dc0
--- /dev/null
+++ b/datafusion-examples/examples/extension_types/temperature.rs
@@ -0,0 +1,268 @@
+// 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, Float64Array, RecordBatch, StringArray};
+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
+/// semantic type [`TemperatureExtensionType`].
+pub async fn temperature_example() -> Result<()> {
+    let ctx = create_session_context()?;
+    register_temperature_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> {
+    let registry = MemoryExtensionTypeRegistry::new_empty();
+
+    // The registration creates a new instance of the extension type with the deserialized metadata.
+    let temp_registration = DefaultExtensionTypeRegistration::new_arc(|metadata| {
+        Ok(TemperatureExtensionType(metadata))
+    });
+    registry.add_extension_type_registration(temp_registration)?;
+
+    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_temperature_table(ctx: &SessionContext) -> Result<DataFrame> {
+    let schema = example_schema();
+
+    let city_names = Arc::new(StringArray::from(vec![
+        "Vienna", "Tokyo", "New York", "Sydney",
+    ]));
+
+    // We'll use the same raw float values across columns to show how the
+    // extension type changes the formatting behavior.
+    let celsius_temps = vec![15.1, 22.5, 18.98, 25.0];
+    let fahrenheit_temps = vec![59.18, 72.5, 66.164, 77.0];
+    let kelvin_temps = vec![288.25, 295.65, 292.13, 298.15];
+
+    let batch = RecordBatch::try_new(
+        schema,
+        vec![
+            city_names,
+            Arc::new(Float64Array::from(celsius_temps)),
+            Arc::new(Float64Array::from(fahrenheit_temps)),
+            Arc::new(Float64Array::from(kelvin_temps)),
+        ],
+    )?;
+
+    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("city", DataType::Utf8, false),
+        Field::new("celsius", DataType::Float64, false)
+            .with_extension_type(TemperatureExtensionType(TemperatureUnit::Celsius)),
+        Field::new("fahrenheit", DataType::Float64, false)
+            .with_extension_type(TemperatureExtensionType(TemperatureUnit::Fahrenheit)),
+        Field::new("kelvin", DataType::Float64, false)
+            .with_extension_type(TemperatureExtensionType(TemperatureUnit::Kelvin)),
+    ]))
+}
+
+/// Represents the unit of a temperature reading.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum TemperatureUnit {
+    Celsius,
+    Fahrenheit,
+    Kelvin,
+}
+
+/// Represents a float that semantically represents a temperature. The temperature can be one of
+/// the supported [`TemperatureUnit`]s.
+///
+/// The unit is realized as an additional extension type metadata and is stored alongside the
+/// extension type name in the Arrow field metadata. This metadata can also be stored within files,
+/// allowing DataFusion to read temperature data from, for example, Parquet files.
+///
+/// The field metadata for a Celsius temperature field will look like this (serialized as JSON):
+/// ```json
+/// {
+///     "ARROW:extension:name": "custom.temperature",
+///     "ARROW:extension:metadata": "celsius"
+/// }
+/// ```
+///
+/// See [the official Arrow documentation](https://arrow.apache.org/docs/format/Columnar.html#extension-types)
+/// for more details on the extension type mechanism.
+#[derive(Debug)]
+pub struct TemperatureExtensionType(TemperatureUnit);
+
+/// Implementation of [`ExtensionType`] for [`TemperatureExtensionType`].
+impl ExtensionType for TemperatureExtensionType {
+    /// Arrow extension type name that is stored in the `ARROW:extension:name` field.
+    const NAME: &'static str = "custom.temperature";
+    type Metadata = TemperatureUnit;
+
+    fn metadata(&self) -> &Self::Metadata {
+        &self.0
+    }
+
+    /// Arrow extension type metadata is encoded as a string and stored in the
+    /// `ARROW:extension:metadata` field. As we only store the name of the unit, a simple string
+    /// suffices. Extension types can store more complex metadata using serialization formats like
+    /// JSON.
+    fn serialize_metadata(&self) -> Option<String> {
+        let s = match self.0 {
+            TemperatureUnit::Celsius => "celsius",
+            TemperatureUnit::Fahrenheit => "fahrenheit",
+            TemperatureUnit::Kelvin => "kelvin",
+        };
+        Some(s.to_string())
+    }
+
+    fn deserialize_metadata(
+        metadata: Option<&str>,
+    ) -> std::result::Result<Self::Metadata, ArrowError> {
+        match metadata {
+            Some("celsius") => Ok(TemperatureUnit::Celsius),
+            Some("fahrenheit") => Ok(TemperatureUnit::Fahrenheit),
+            Some("kelvin") => Ok(TemperatureUnit::Kelvin),
+            Some(other) => Err(ArrowError::InvalidArgumentError(format!(
+                "Invalid metadata for temperature type: {other}"
+            ))),
+            None => Err(ArrowError::InvalidArgumentError(
+                "Temperature type requires metadata (unit)".to_owned(),
+            )),
+        }
+    }
+
+    fn supports_data_type(
+        &self,
+        data_type: &DataType,
+    ) -> std::result::Result<(), ArrowError> {
+        match data_type {
+            DataType::Float64 => Ok(()),
+            _ => Err(ArrowError::InvalidArgumentError(format!(
+                "Invalid data type: {data_type} for temperature type, expected Float64",
+            ))),
+        }
+    }
+
+    fn try_new(
+        data_type: &DataType,
+        metadata: Self::Metadata,
+    ) -> std::result::Result<Self, ArrowError> {
+        let instance = Self(metadata);
+        instance.supports_data_type(data_type)?;
+        Ok(instance)
+    }
+}
+
+/// Implementation of [`DFExtensionType`] for [`TemperatureExtensionType`].
+impl DFExtensionType for TemperatureExtensionType {
+    fn create_array_formatter<'fmt>(
+        &self,
+        array: &'fmt dyn Array,
+        options: &FormatOptions<'fmt>,
+    ) -> Result<Option<ArrayFormatter<'fmt>>> {
+        if array.data_type() != &DataType::Float64 {
+            return internal_err!("Wrong array type for Temperature");
+        }
+
+        let display_index = TemperatureDisplayIndex {
+            array: array.as_any().downcast_ref().unwrap(),
+            null_str: options.null(),
+            unit: self.0,
+        };
+        Ok(Some(ArrayFormatter::new(
+            Box::new(display_index),
+            options.safe(),
+        )))
+    }
+}
+
+/// Pretty printer for temperatures.
+#[derive(Debug)]
+struct TemperatureDisplayIndex<'a> {
+    array: &'a Float64Array,
+    null_str: &'a str,
+    unit: TemperatureUnit,
+}
+
+/// Implements the custom display logic.
+impl DisplayIndex for TemperatureDisplayIndex<'_> {
+    fn write(&self, idx: usize, f: &mut dyn Write) -> FormatResult {
+        if self.array.is_null(idx) {
+            write!(f, "{}", self.null_str)?;
+            return Ok(());
+        }
+
+        let value = self.array.value(idx);
+        let suffix = match self.unit {
+            TemperatureUnit::Celsius => "°C",
+            TemperatureUnit::Fahrenheit => "°F",
+            TemperatureUnit::Kelvin => "K",
+        };
+
+        write!(f, "{value:.2} {suffix}")?;
+
+        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_temperature_table(&ctx).await?;
+
+        assert_snapshot!(
+            table.to_string().await?,
+            @r"
+        +----------+----------+------------+----------+
+        | city     | celsius  | fahrenheit | kelvin   |
+        +----------+----------+------------+----------+
+        | Vienna   | 15.10 °C | 59.18 °F   | 288.25 K |
+        | Tokyo    | 22.50 °C | 72.50 °F   | 295.65 K |
+        | New York | 18.98 °C | 66.16 °F   | 292.13 K |
+        | Sydney   | 25.00 °C | 77.00 °F   | 298.15 K |
+        +----------+----------+------------+----------+
+        "
+        );
+
+        Ok(())
+    }
+}
diff --git a/datafusion/common/Cargo.toml b/datafusion/common/Cargo.toml
index 82e7aafcee2b1..ab67c5a3f2f6b 100644
--- a/datafusion/common/Cargo.toml
+++ b/datafusion/common/Cargo.toml
@@ -67,6 +67,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 }
 half = { workspace = true }
 hashbrown = { workspace = true }
@@ -81,6 +82,7 @@ paste = { workspace = 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..1eb3a09bf3fdc
--- /dev/null
+++ b/datafusion/common/src/types/canonical_extensions/uuid.rs
@@ -0,0 +1,95 @@
+// 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::_internal_err;
+use crate::types::extension::DFExtensionType;
+use arrow::array::{Array, FixedSizeBinaryArray};
+use arrow::datatypes::DataType;
+use arrow::util::display::{ArrayFormatter, DisplayIndex, FormatOptions, FormatResult};
+use std::fmt::Write;
+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(),
+        )))
+    }
+}
+
+/// 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(())
+    }
+}
+
+#[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..1cd19b595f649
--- /dev/null
+++ b/datafusion/common/src/types/extension.rs
@@ -0,0 +1,71 @@
+// 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;
+use arrow::util::display::{ArrayFormatter, FormatOptions};
+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)
+    }
+}
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..f83be57721876 100644
--- a/datafusion/core/src/execution/session_state.rs
+++ b/datafusion/core/src/execution/session_state.rs
@@ -56,7 +56,10 @@ 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::{
+    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 +161,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 +271,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 +995,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 +1036,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 +1092,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 +1138,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_empty()))
+            .extend(&SessionStateDefaults::default_extension_types())
+            .expect("MemoryExtensionTypeRegistry is not read-only.");
+
         self.table_functions
             .get_or_insert_with(HashMap::new)
             .extend(
@@ -1316,6 +1333,15 @@ 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
+    }
+
     /// Set the [`SerializerRegistry`]
     pub fn with_serializer_registry(
         mut self,
@@ -1454,6 +1480,7 @@ impl SessionStateBuilder {
             scalar_functions,
             aggregate_functions,
             window_functions,
+            extension_types,
             serializer_registry,
             file_formats,
             table_options,
@@ -1490,6 +1517,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 +1587,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,
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..aa08bd1b1f519
--- /dev/null
+++ b/datafusion/core/tests/extension_types/pretty_printing.rs
@@ -0,0 +1,78 @@
+// 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, Schema, SchemaRef};
+use datafusion::dataframe::DataFrame;
+use datafusion::error::Result;
+use datafusion::execution::SessionStateBuilder;
+use datafusion::prelude::SessionContext;
+use datafusion_expr::registry::MemoryExtensionTypeRegistry;
+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_extension_type_registry(Arc::new(
+            MemoryExtensionTypeRegistry::new_with_canonical_extension_types(),
+        ))
+        .build();
+    let ctx = SessionContext::new_with_state(state);
+
+    ctx.register_batch("test", batch)?;
+
+    ctx.table("test").await
+}
+
+#[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(())
+}
diff --git a/datafusion/datasource-arrow/src/file_format.rs b/datafusion/datasource-arrow/src/file_format.rs
index 9997d23d4c61f..de164e70d1cb8 100644
--- a/datafusion/datasource-arrow/src/file_format.rs
+++ b/datafusion/datasource-arrow/src/file_format.rs
@@ -554,6 +554,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};
@@ -609,6 +610,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 0c274806c09c3..c80f9f097d1db 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;
@@ -1220,6 +1221,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 75aa59595bed5..e9887f2d21a1f 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/registry.rs b/datafusion/expr/src/registry.rs
index 472e065211aac..0f13f9873f316 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,281 @@ 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 {
+        Self::new_empty()
+    }
+}
+
+impl MemoryExtensionTypeRegistry {
+    /// Creates an empty [MemoryExtensionTypeRegistry].
+    pub fn new_empty() -> Self {
+        Self {
+            extension_types: Arc::new(RwLock::new(HashMap::new())),
+        }
+    }
+
+    /// Pre-registers the [canonical extension types](https://arrow.apache.org/docs/format/CanonicalExtensions.html)
+    /// in the extension type registry.
+    pub fn new_with_canonical_extension_types() -> Self {
+        let mapping = [DefaultExtensionTypeRegistration::new_arc(|_| {
+            Ok(arrow_schema::extension::Uuid {})
+        })];
+
+        let mut extension_types = HashMap::new();
+        for registration in mapping.into_iter() {
+            extension_types.insert(registration.type_name().to_owned(), registration);
+        }
+
+        Self {
+            extension_types: Arc::new(RwLock::new(HashMap::from(extension_types))),
+        }
+    }
+
+    /// 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/ffi/src/session/mod.rs b/datafusion/ffi/src/session/mod.rs
index aa910abb9149a..69cc8d9c6954f 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.
@@ -356,6 +356,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,
@@ -424,6 +425,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(),
             })
@@ -515,6 +517,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/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>;