diff --git a/README.md b/README.md
index afac007..6289e6e 100644
--- a/README.md
+++ b/README.md
@@ -246,6 +246,39 @@ struct Submodel {
 }
 ```
 
+### Flatten support
+
+Atmosphere supports flattening nested structs into the parent table using `#[sql(flatten)]`.
+This generates columns for the nested type at compile time and binds them on INSERT/UPDATE.
+
+```rust
+use atmosphere::prelude::*;
+
+#[derive(FlattenFields)]
+struct ContactInfo {
+    email: String,
+    phone: Option<String>,
+}
+
+#[table(schema = "public", name = "user")]
+struct User {
+    #[sql(pk)]
+    id: i32,
+    name: String,
+    #[sql(flatten, prefix)]
+    contact: ContactInfo,
+}
+```
+
+Prefix modes:
+- `#[sql(flatten)]` no prefix (columns must be unique)
+- `#[sql(flatten, prefix)]` or `#[sql(flatten, prefix = auto)]` auto prefix with field name
+- `#[sql(flatten, prefix = "c_")]` custom prefix
+
+Notes:
+- Changing prefix requires a database migration (column names change).
+- Recommended max nesting depth: 2 levels (Postgres identifier limit is 63 chars).
+
 ## Contribution
 
 We welcome contributions! Please see [our contribution guidelines](CONTRIBUTING.md) for more details.
diff --git a/atmosphere-core/src/flatten.rs b/atmosphere-core/src/flatten.rs
new file mode 100644
index 0000000..c001240
--- /dev/null
+++ b/atmosphere-core/src/flatten.rs
@@ -0,0 +1,524 @@
+//! Flatten Support for Atmosphere SQL Framework
+//!
+//! This module provides traits and utilities for flattening nested structs within
+//! Atmosphere tables. The `#[sql(flatten)]` attribute allows nested struct fields
+//! to be expanded into the parent table's column set, enabling code reuse and
+//! composition without requiring JSON serialization.
+//!
+//! # Overview
+//!
+//! The `FlattenFields` trait must be implemented by any type that will be flattened
+//! into a parent table. This can be done automatically using `#[derive(FlattenFields)]`
+//! or manually for external types.
+//!
+//! # Migration from #[sql(json)]
+//!
+//! If you were previously using `#[sql(json)]` to store nested structs as JSON,
+//! you can migrate to `#[sql(flatten)]` for better query performance and type safety:
+//!
+//! **Before (JSON serialization):**
+//! ```ignore
+//! #[table(schema = "public", name = "user")]
+//! struct User {
+//!     #[sql(pk)]
+//!     id: i32,
+//!     #[sql(json)]
+//!     contact: ContactInfo,  // Stored as JSON column
+//! }
+//! ```
+//!
+//! **After (flattened columns):**
+//! ```ignore
+//! #[derive(FlattenFields)]
+//! struct ContactInfo {
+//!     email: String,
+//!     phone: Option<String>,
+//! }
+//!
+//! #[table(schema = "public", name = "user")]
+//! struct User {
+//!     #[sql(pk)]
+//!     id: i32,
+//!     #[sql(flatten, prefix)]
+//!     contact: ContactInfo,  // Stored as contact_email, contact_phone columns
+//! }
+//! ```
+//!
+//! **Note:** This migration requires a database schema change. The JSON column must be
+//! replaced with individual columns for each flattened field.
+//!
+//! # Prefix Options
+//!
+//! - `#[sql(flatten)]` - No prefix, columns must be unique across parent and nested types
+//! - `#[sql(flatten, prefix)]` - Auto-prefix using field name (e.g., `contact_email`)
+//! - `#[sql(flatten, prefix = auto)]` - Same as above (explicit form)
+//! - `#[sql(flatten, prefix = "c_")]` - Custom prefix (e.g., `c_email`)
+//!
+//! # Limitations
+//!
+//! - Maximum recommended nesting depth: 2 levels
+//! - PostgreSQL has a 63-character limit for identifiers; deeply nested prefixes may exceed this
+//! - Changing prefix configuration requires a database migration
+//!
+//! # Example
+//!
+//! ```ignore
+//! use atmosphere::prelude::*;
+//!
+//! #[derive(FlattenFields)]
+//! struct ContactInfo {
+//!     email: String,
+//!     phone: Option<String>,
+//! }
+//!
+//! #[table(schema = "public", name = "user")]
+//! struct User {
+//!     #[sql(pk)]
+//!     id: i32,
+//!     name: String,
+//!     #[sql(flatten)]
+//!     contact: ContactInfo,
+//! }
+//! ```
+
+use crate::{Bindable, Result, Table};
+
+/// A trait for types that can be flattened into a parent table's columns.
+///
+/// This trait enables nested struct composition in Atmosphere tables. When a field
+/// is marked with `#[sql(flatten)]`, Atmosphere will expand the nested type's columns
+/// into the parent table's column metadata.
+///
+/// # Column Names
+///
+/// The `COLUMN_NAMES` constant provides the SQL column names for all fields in the
+/// flattened type. These names may be prefixed based on the flatten attribute configuration:
+///
+/// - `#[sql(flatten)]` - No prefix, columns must be unique
+/// - `#[sql(flatten, prefix)]` - Auto-prefix using field name (e.g., `contact_email`)
+/// - `#[sql(flatten, prefix = "x_")]` - Custom prefix (e.g., `x_email`)
+///
+/// # Bind Support
+///
+/// The `bind_field` method enables INSERT and UPDATE operations by delegating
+/// the binding of individual columns to the nested type.
+///
+/// # Implementing
+///
+/// For user-defined types, use `#[derive(FlattenFields)]`:
+///
+/// ```ignore
+/// #[derive(FlattenFields)]
+/// struct Address {
+///     street: String,
+///     city: String,
+///     zip: String,
+/// }
+/// ```
+///
+/// For external types, implement the trait manually:
+///
+/// ```ignore
+/// impl FlattenFields for ExternalType {
+///     const COLUMN_NAMES: &'static [&'static str] = &["field1", "field2"];
+///
+///     fn bind_field<'q, Q: Bindable<'q>>(
+///         &'q self,
+///         name: &str,
+///         query: Q,
+///     ) -> Result<Q> {
+///         match name {
+///             "field1" => Ok(query.dyn_bind(&self.field1)),
+///             "field2" => Ok(query.dyn_bind(&self.field2)),
+///             _ => Err(atmosphere::Error::Bind(
+///                 atmosphere::bind::BindError::Unknown(name)
+///             )),
+///         }
+///     }
+/// }
+/// ```
+pub trait FlattenFields {
+    /// The SQL column names for all fields in this type.
+    ///
+    /// These are the base column names without any prefix. The prefix is applied
+    /// at the parent table level based on the flatten attribute configuration.
+    const COLUMN_NAMES: &'static [&'static str];
+
+    /// Binds a single field to the query by its column name.
+    ///
+    /// # Arguments
+    ///
+    /// * `name` - The SQL column name (without prefix) to bind
+    /// * `query` - The query builder to bind the value to
+    ///
+    /// # Returns
+    ///
+    /// Returns the query with the value bound, or an error if the column name
+    /// is not recognized.
+    fn bind_field<'q, Q: Bindable<'q>>(&'q self, name: &str, query: Q) -> Result<Q>;
+
+    /// Binds a single field to the query by its column name, allowing for optional values.
+    ///
+    /// This helper is used to support `Option<T>` flatten fields by binding NULL when the
+    /// nested value is not present.
+    fn bind_field_optional<'q, Q: Bindable<'q>>(
+        value: Option<&'q Self>,
+        name: &str,
+        query: Q,
+    ) -> Result<Q> {
+        match value {
+            Some(value) => value.bind_field(name, query),
+            None => Err(crate::Error::Bind(crate::bind::BindError::Unknown(
+                "missing flattened field",
+            ))),
+        }
+    }
+}
+
+impl<T: FlattenFields> FlattenFields for Option<T> {
+    const COLUMN_NAMES: &'static [&'static str] = T::COLUMN_NAMES;
+
+    fn bind_field<'q, Q: Bindable<'q>>(&'q self, name: &str, query: Q) -> Result<Q> {
+        T::bind_field_optional(self.as_ref(), name, query)
+    }
+
+    fn bind_field_optional<'q, Q: Bindable<'q>>(
+        value: Option<&'q Self>,
+        name: &str,
+        query: Q,
+    ) -> Result<Q> {
+        match value {
+            Some(value) => T::bind_field_optional(value.as_ref(), name, query),
+            None => T::bind_field_optional(None, name, query),
+        }
+    }
+}
+
+/// Compile-time helper to assert all column names in a slice are unique.
+///
+/// This is used by generated code to detect column name collisions at compile time.
+/// If two columns have the same SQL name (considering prefixes), compilation will fail.
+///
+/// # Complexity
+///
+/// This function uses O(n²) comparison for simplicity and compile-time compatibility.
+/// This is acceptable because:
+/// - Typical structs have 2-20 columns, making n² negligible
+/// - The check runs at compile time, not runtime
+/// - Const-compatible O(n) algorithms (requiring const sorting or hashing) would add
+///   significant complexity for minimal practical benefit
+///
+/// For tables with more than ~50 columns, consider splitting into multiple flattened types.
+///
+/// # Panics
+///
+/// Panics at compile time if duplicate column names are detected.
+#[doc(hidden)]
+pub const fn assert_unique_columns(columns: &[&str]) {
+    let len = columns.len();
+    let mut i = 0;
+    while i < len {
+        let mut j = i + 1;
+        while j < len {
+            if const_str_eq(columns[i], columns[j]) {
+                panic!("duplicate column name detected");
+            }
+            j += 1;
+        }
+        i += 1;
+    }
+}
+
+/// Compile-time helper to assert all table column names (including flatten columns) are unique.
+#[doc(hidden)]
+pub const fn assert_unique_table_columns<T: Table>() {
+    let pk = T::PRIMARY_KEY.sql;
+
+    // Primary key vs others
+    let mut i = 0;
+    while i < T::FOREIGN_KEYS.len() {
+        let fk = T::FOREIGN_KEYS[i].sql;
+        if const_str_eq(pk, fk) {
+            panic!("duplicate column name detected");
+        }
+        i += 1;
+    }
+
+    i = 0;
+    while i < T::DATA_COLUMNS.len() {
+        let data = T::DATA_COLUMNS[i].sql;
+        if const_str_eq(pk, data) {
+            panic!("duplicate column name detected");
+        }
+        i += 1;
+    }
+
+    i = 0;
+    while i < T::TIMESTAMP_COLUMNS.len() {
+        let ts = T::TIMESTAMP_COLUMNS[i].sql;
+        if const_str_eq(pk, ts) {
+            panic!("duplicate column name detected");
+        }
+        i += 1;
+    }
+
+    // Foreign keys vs each other and other columns
+    let mut a = 0;
+    while a < T::FOREIGN_KEYS.len() {
+        let fk_a = T::FOREIGN_KEYS[a].sql;
+        let mut b = a + 1;
+        while b < T::FOREIGN_KEYS.len() {
+            let fk_b = T::FOREIGN_KEYS[b].sql;
+            if const_str_eq(fk_a, fk_b) {
+                panic!("duplicate column name detected");
+            }
+            b += 1;
+        }
+
+        b = 0;
+        while b < T::DATA_COLUMNS.len() {
+            let data = T::DATA_COLUMNS[b].sql;
+            if const_str_eq(fk_a, data) {
+                panic!("duplicate column name detected");
+            }
+            b += 1;
+        }
+
+        b = 0;
+        while b < T::TIMESTAMP_COLUMNS.len() {
+            let ts = T::TIMESTAMP_COLUMNS[b].sql;
+            if const_str_eq(fk_a, ts) {
+                panic!("duplicate column name detected");
+            }
+            b += 1;
+        }
+
+        a += 1;
+    }
+
+    // Data columns vs each other and timestamps
+    a = 0;
+    while a < T::DATA_COLUMNS.len() {
+        let data_a = T::DATA_COLUMNS[a].sql;
+        let mut b = a + 1;
+        while b < T::DATA_COLUMNS.len() {
+            let data_b = T::DATA_COLUMNS[b].sql;
+            if const_str_eq(data_a, data_b) {
+                panic!("duplicate column name detected");
+            }
+            b += 1;
+        }
+
+        b = 0;
+        while b < T::TIMESTAMP_COLUMNS.len() {
+            let ts = T::TIMESTAMP_COLUMNS[b].sql;
+            if const_str_eq(data_a, ts) {
+                panic!("duplicate column name detected");
+            }
+            b += 1;
+        }
+
+        a += 1;
+    }
+
+    // Timestamp columns vs each other
+    a = 0;
+    while a < T::TIMESTAMP_COLUMNS.len() {
+        let ts_a = T::TIMESTAMP_COLUMNS[a].sql;
+        let mut b = a + 1;
+        while b < T::TIMESTAMP_COLUMNS.len() {
+            let ts_b = T::TIMESTAMP_COLUMNS[b].sql;
+            if const_str_eq(ts_a, ts_b) {
+                panic!("duplicate column name detected");
+            }
+            b += 1;
+        }
+        a += 1;
+    }
+
+    // Flatten columns vs regular columns
+    let mut f = 0;
+    while f < T::FLATTEN_COLUMNS.len() {
+        let flatten = &T::FLATTEN_COLUMNS[f];
+        let mut n = 0;
+        while n < flatten.nested_columns.len() {
+            let nested = flatten.nested_columns[n];
+
+            if prefixed_eq(flatten.sql_prefix, nested, pk) {
+                panic!("duplicate column name detected");
+            }
+
+            i = 0;
+            while i < T::FOREIGN_KEYS.len() {
+                if prefixed_eq(flatten.sql_prefix, nested, T::FOREIGN_KEYS[i].sql) {
+                    panic!("duplicate column name detected");
+                }
+                i += 1;
+            }
+
+            i = 0;
+            while i < T::DATA_COLUMNS.len() {
+                if prefixed_eq(flatten.sql_prefix, nested, T::DATA_COLUMNS[i].sql) {
+                    panic!("duplicate column name detected");
+                }
+                i += 1;
+            }
+
+            i = 0;
+            while i < T::TIMESTAMP_COLUMNS.len() {
+                if prefixed_eq(flatten.sql_prefix, nested, T::TIMESTAMP_COLUMNS[i].sql) {
+                    panic!("duplicate column name detected");
+                }
+                i += 1;
+            }
+
+            n += 1;
+        }
+        f += 1;
+    }
+
+    // Flatten columns vs flatten columns
+    let mut fa = 0;
+    while fa < T::FLATTEN_COLUMNS.len() {
+        let flatten_a = &T::FLATTEN_COLUMNS[fa];
+        let mut na = 0;
+        while na < flatten_a.nested_columns.len() {
+            let nested_a = flatten_a.nested_columns[na];
+            let mut fb = fa;
+            while fb < T::FLATTEN_COLUMNS.len() {
+                let flatten_b = &T::FLATTEN_COLUMNS[fb];
+                let mut nb = if fa == fb { na + 1 } else { 0 };
+                while nb < flatten_b.nested_columns.len() {
+                    let nested_b = flatten_b.nested_columns[nb];
+                    if prefixed_eq_with_prefix(
+                        flatten_a.sql_prefix,
+                        nested_a,
+                        flatten_b.sql_prefix,
+                        nested_b,
+                    ) {
+                        panic!("duplicate column name detected");
+                    }
+                    nb += 1;
+                }
+                fb += 1;
+            }
+            na += 1;
+        }
+        fa += 1;
+    }
+}
+
+/// Const function to compare two string slices for equality.
+#[doc(hidden)]
+pub const fn const_str_eq(a: &str, b: &str) -> bool {
+    let a = a.as_bytes();
+    let b = b.as_bytes();
+    if a.len() != b.len() {
+        return false;
+    }
+    let mut i = 0;
+    while i < a.len() {
+        if a[i] != b[i] {
+            return false;
+        }
+        i += 1;
+    }
+    true
+}
+
+const fn prefixed_eq(prefix: Option<&str>, name: &str, other: &str) -> bool {
+    match prefix {
+        Some(prefix) => {
+            let p = prefix.as_bytes();
+            let n = name.as_bytes();
+            let o = other.as_bytes();
+            if p.len() + n.len() != o.len() {
+                return false;
+            }
+            let mut i = 0;
+            while i < p.len() {
+                if p[i] != o[i] {
+                    return false;
+                }
+                i += 1;
+            }
+            let mut j = 0;
+            while j < n.len() {
+                if n[j] != o[p.len() + j] {
+                    return false;
+                }
+                j += 1;
+            }
+            true
+        }
+        None => const_str_eq(name, other),
+    }
+}
+
+const fn prefixed_eq_with_prefix(
+    prefix_a: Option<&str>,
+    name_a: &str,
+    prefix_b: Option<&str>,
+    name_b: &str,
+) -> bool {
+    let len_a = prefix_len(prefix_a) + name_a.len();
+    let len_b = prefix_len(prefix_b) + name_b.len();
+    if len_a != len_b {
+        return false;
+    }
+    let mut i = 0;
+    while i < len_a {
+        if prefixed_byte(prefix_a, name_a, i) != prefixed_byte(prefix_b, name_b, i) {
+            return false;
+        }
+        i += 1;
+    }
+    true
+}
+
+const fn prefix_len(prefix: Option<&str>) -> usize {
+    match prefix {
+        Some(prefix) => prefix.len(),
+        None => 0,
+    }
+}
+
+const fn prefixed_byte(prefix: Option<&str>, name: &str, idx: usize) -> u8 {
+    let prefix_len = prefix_len(prefix);
+    if idx < prefix_len {
+        match prefix {
+            Some(prefix) => prefix.as_bytes()[idx],
+            None => 0,
+        }
+    } else {
+        name.as_bytes()[idx - prefix_len]
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_const_str_eq() {
+        assert!(const_str_eq("hello", "hello"));
+        assert!(!const_str_eq("hello", "world"));
+        assert!(!const_str_eq("hello", "hell"));
+        assert!(!const_str_eq("a", "ab"));
+    }
+
+    #[test]
+    fn test_assert_unique_columns_ok() {
+        // Should not panic
+        assert_unique_columns(&["a", "b", "c"]);
+        assert_unique_columns(&["email", "phone", "name"]);
+        assert_unique_columns(&[]);
+        assert_unique_columns(&["single"]);
+    }
+
+    #[test]
+    #[should_panic(expected = "duplicate column name detected")]
+    fn test_assert_unique_columns_duplicate() {
+        assert_unique_columns(&["a", "b", "a"]);
+    }
+}
diff --git a/atmosphere-core/src/lib.rs b/atmosphere-core/src/lib.rs
index a40ac2d..fb069c0 100644
--- a/atmosphere-core/src/lib.rs
+++ b/atmosphere-core/src/lib.rs
@@ -91,8 +91,12 @@ pub mod driver {
     pub type Pool = sqlx::SqlitePool;
 }
 
+/// Provides traits and utilities for flattening nested structs in SQL operations.
+pub mod flatten;
+
 pub use bind::*;
 pub use error::*;
+pub use flatten::*;
 pub use schema::*;
 
 #[doc(hidden)]
diff --git a/atmosphere-core/src/runtime/sql.rs b/atmosphere-core/src/runtime/sql.rs
index dc2989a..c4c5c91 100644
--- a/atmosphere-core/src/runtime/sql.rs
+++ b/atmosphere-core/src/runtime/sql.rs
@@ -15,12 +15,12 @@
 //!   table columns and the SQL queries they are bound to. This ensures that queries are executed with the correct
 //!   parameters and their values.
 
-use std::fmt;
+use std::{borrow::Cow, fmt};
 
 use sqlx::QueryBuilder;
 
 use crate::{
-    Bind, Column,
+    Bind, Column, FlattenedColumn,
     query::{self, Query},
 };
 
@@ -91,7 +91,7 @@ fn table<T: Bind>() -> String {
 ///
 /// SQL: `SELECT * FROM .. WHERE .. = $1`
 pub fn select<T: Bind>() -> Query<T> {
-    select_by(Column::PrimaryKey(&T::PRIMARY_KEY))
+    select_by(Column::PrimaryKey(T::PRIMARY_KEY))
 }
 
 /// Creates a `SELECT` query to retrieve rows from the table based on a specific column.
@@ -116,6 +116,16 @@ pub fn select_by<T: Bind>(c: Column<T>) -> Query<T> {
         separated.push(meta.sql);
     }
 
+    for flatten in T::FLATTEN_COLUMNS {
+        for nested in flatten.nested_columns {
+            let sql = match flatten.sql_prefix {
+                Some(prefix) => format!("{}{} AS {}", prefix, nested, nested),
+                None => nested.to_string(),
+            };
+            separated.push(sql);
+        }
+    }
+
     query.push(format!("\nFROM\n  {}\n", table::<T>()));
     query.push(format!("WHERE {} = $1", c.sql()));
 
@@ -149,6 +159,16 @@ pub fn select_all<T: Bind>() -> Query<T> {
         separated.push(meta.sql);
     }
 
+    for flatten in T::FLATTEN_COLUMNS {
+        for nested in flatten.nested_columns {
+            let sql = match flatten.sql_prefix {
+                Some(prefix) => format!("{}{} AS {}", prefix, nested, nested),
+                None => nested.to_string(),
+            };
+            separated.push(sql);
+        }
+    }
+
     query.push(format!("\nFROM\n  {}\n", table::<T>()));
 
     Query::new(
@@ -170,28 +190,60 @@ pub fn insert<T: Bind>() -> Query<T> {
     let mut separated = builder.separated(", ");
 
     separated.push(T::PRIMARY_KEY.sql.to_string());
-    bindings.push(Column::PrimaryKey(&T::PRIMARY_KEY));
+    bindings.push(Column::PrimaryKey(T::PRIMARY_KEY));
 
     for fk in T::FOREIGN_KEYS {
         separated.push(fk.sql.to_string());
-        bindings.push(Column::ForeignKey(fk));
+        bindings.push(Column::ForeignKey(fk.clone()));
     }
 
     for data in T::DATA_COLUMNS {
         separated.push(data.sql.to_string());
-        bindings.push(Column::Data(data));
+        bindings.push(Column::Data(data.clone()));
     }
 
     for meta in T::TIMESTAMP_COLUMNS {
         separated.push(meta.sql.to_string());
-        bindings.push(Column::Timestamp(meta));
+        bindings.push(Column::Timestamp(meta.clone()));
+    }
+
+    for flatten in T::FLATTEN_COLUMNS {
+        for nested in flatten.nested_columns {
+            match flatten.sql_prefix {
+                Some(prefix) => {
+                    let sql = format!("{}{}", prefix, nested);
+                    separated.push(sql.clone());
+                    bindings.push(Column::Flattened(FlattenedColumn::new(
+                        flatten.field,
+                        nested,
+                        Cow::Owned(sql),
+                    )));
+                }
+                None => {
+                    separated.push(nested.to_string());
+                    bindings.push(Column::Flattened(FlattenedColumn::new(
+                        flatten.field,
+                        nested,
+                        Cow::Borrowed(nested),
+                    )));
+                }
+            }
+        }
     }
 
     separated.push_unseparated(")\nVALUES\n  (");
 
     separated.push_unseparated("$1");
 
-    let columns = 1 + T::FOREIGN_KEYS.len() + T::DATA_COLUMNS.len() + T::TIMESTAMP_COLUMNS.len();
+    let flatten_count = T::FLATTEN_COLUMNS
+        .iter()
+        .map(|f| f.nested_columns.len())
+        .sum::<usize>();
+    let columns = 1
+        + T::FOREIGN_KEYS.len()
+        + T::DATA_COLUMNS.len()
+        + T::TIMESTAMP_COLUMNS.len()
+        + flatten_count;
 
     for c in 2..=columns {
         separated.push(format!("${c}"));
@@ -217,28 +269,53 @@ pub fn update<T: Bind>() -> Query<T> {
     let mut separated = builder.separated(",\n  ");
 
     separated.push(format!("{} = $1", T::PRIMARY_KEY.sql));
-    bindings.push(Column::PrimaryKey(&T::PRIMARY_KEY));
+    bindings.push(Column::PrimaryKey(T::PRIMARY_KEY));
 
     let mut col = 2;
 
     for fk in T::FOREIGN_KEYS {
         separated.push(format!("{} = ${col}", fk.sql));
-        bindings.push(Column::ForeignKey(fk));
+        bindings.push(Column::ForeignKey(fk.clone()));
         col += 1;
     }
 
     for data in T::DATA_COLUMNS {
         separated.push(format!("{} = ${col}", data.sql));
-        bindings.push(Column::Data(data));
+        bindings.push(Column::Data(data.clone()));
         col += 1;
     }
 
     for meta in T::TIMESTAMP_COLUMNS {
         separated.push(format!("{} = ${col}", meta.sql));
-        bindings.push(Column::Timestamp(meta));
+        bindings.push(Column::Timestamp(meta.clone()));
         col += 1;
     }
 
+    for flatten in T::FLATTEN_COLUMNS {
+        for nested in flatten.nested_columns {
+            match flatten.sql_prefix {
+                Some(prefix) => {
+                    let sql = format!("{}{}", prefix, nested);
+                    separated.push(format!("{} = ${col}", sql));
+                    bindings.push(Column::Flattened(FlattenedColumn::new(
+                        flatten.field,
+                        nested,
+                        Cow::Owned(sql),
+                    )));
+                }
+                None => {
+                    separated.push(format!("{} = ${col}", nested));
+                    bindings.push(Column::Flattened(FlattenedColumn::new(
+                        flatten.field,
+                        nested,
+                        Cow::Borrowed(nested),
+                    )));
+                }
+            }
+            col += 1;
+        }
+    }
+
     builder.push(format!("\nWHERE\n  {} = $1", T::PRIMARY_KEY.sql));
 
     Query::new(
@@ -277,6 +354,16 @@ pub fn upsert<T: Bind>() -> Query<T> {
         separated.push(format!("{} = EXCLUDED.{}", meta.sql, meta.sql));
     }
 
+    for flatten in T::FLATTEN_COLUMNS {
+        for nested in flatten.nested_columns {
+            let sql = match flatten.sql_prefix {
+                Some(prefix) => format!("{}{}", prefix, nested),
+                None => nested.to_string(),
+            };
+            separated.push(format!("{} = EXCLUDED.{}", sql, sql));
+        }
+    }
+
     Query::new(
         query::Operation::Upsert,
         query::Cardinality::One,
@@ -305,14 +392,14 @@ pub fn delete_by<T: Bind>(c: Column<T>) -> Query<T> {
         query::Operation::Delete,
         query::Cardinality::One,
         builder,
-        Bindings(vec![Column::PrimaryKey(&T::PRIMARY_KEY)]),
+        Bindings(vec![Column::PrimaryKey(T::PRIMARY_KEY)]),
     )
 }
 
 #[cfg(test)]
 mod tests {
     use crate::{
-        Bind, Bindable, Column, DataColumn, ForeignKey, PrimaryKey, Table, TimestampColumn,
+        Bind, Bindable, Column, DataColumn, FlattenColumn, ForeignKey, PrimaryKey, Table, TimestampColumn,
         runtime::sql::{self, Bindings},
     };
 
@@ -335,6 +422,7 @@ mod tests {
         const DATA_COLUMNS: &'static [DataColumn<Self>] =
             &[DataColumn::new("data", "data_sql_col")];
         const TIMESTAMP_COLUMNS: &'static [TimestampColumn<Self>] = &[];
+        const FLATTEN_COLUMNS: &'static [FlattenColumn<Self>] = &[];
 
         fn pk(&self) -> &Self::PrimaryKey {
             &self.id
@@ -365,7 +453,7 @@ mod tests {
 
         assert_eq!(
             bindings,
-            Bindings(vec![Column::PrimaryKey(&TestTable::PRIMARY_KEY),])
+            Bindings(vec![Column::PrimaryKey(TestTable::PRIMARY_KEY),])
         );
     }
 
@@ -383,9 +471,9 @@ mod tests {
         assert_eq!(
             bindings,
             Bindings(vec![
-                Column::PrimaryKey(&TestTable::PRIMARY_KEY),
-                Column::ForeignKey(&TestTable::FOREIGN_KEYS[0]),
-                Column::Data(&TestTable::DATA_COLUMNS[0]),
+                Column::PrimaryKey(TestTable::PRIMARY_KEY),
+                Column::ForeignKey(TestTable::FOREIGN_KEYS[0]),
+                Column::Data(TestTable::DATA_COLUMNS[0]),
             ])
         );
     }
@@ -404,9 +492,9 @@ mod tests {
         assert_eq!(
             bindings,
             Bindings(vec![
-                Column::PrimaryKey(&TestTable::PRIMARY_KEY),
-                Column::ForeignKey(&TestTable::FOREIGN_KEYS[0]),
-                Column::Data(&TestTable::DATA_COLUMNS[0]),
+                Column::PrimaryKey(TestTable::PRIMARY_KEY),
+                Column::ForeignKey(TestTable::FOREIGN_KEYS[0]),
+                Column::Data(TestTable::DATA_COLUMNS[0]),
             ])
         );
     }
@@ -425,9 +513,9 @@ mod tests {
         assert_eq!(
             bindings,
             Bindings(vec![
-                Column::PrimaryKey(&TestTable::PRIMARY_KEY),
-                Column::ForeignKey(&TestTable::FOREIGN_KEYS[0]),
-                Column::Data(&TestTable::DATA_COLUMNS[0]),
+                Column::PrimaryKey(TestTable::PRIMARY_KEY),
+                Column::ForeignKey(TestTable::FOREIGN_KEYS[0]),
+                Column::Data(TestTable::DATA_COLUMNS[0]),
             ])
         );
     }
@@ -444,7 +532,7 @@ mod tests {
         );
         assert_eq!(
             bindings,
-            Bindings(vec![Column::PrimaryKey(&TestTable::PRIMARY_KEY),])
+            Bindings(vec![Column::PrimaryKey(TestTable::PRIMARY_KEY),])
         );
     }
 }
diff --git a/atmosphere-core/src/schema/mod.rs b/atmosphere-core/src/schema/mod.rs
index 1feb132..6c4f2c7 100644
--- a/atmosphere-core/src/schema/mod.rs
+++ b/atmosphere-core/src/schema/mod.rs
@@ -17,7 +17,9 @@ pub use delete::Delete;
 pub use read::Read;
 pub use update::Update;
 
-pub use self::column::{Column, DataColumn, ForeignKey, PrimaryKey, TimestampColumn};
+pub use self::column::{
+    Column, DataColumn, FlattenColumn, FlattenedColumn, ForeignKey, PrimaryKey, TimestampColumn,
+};
 
 /// SQL Table Definition
 ///
@@ -46,6 +48,8 @@ where
     const DATA_COLUMNS: &'static [DataColumn<Self>];
     /// An array of timestamp columns.
     const TIMESTAMP_COLUMNS: &'static [TimestampColumn<Self>];
+    /// An array of flattened nested struct columns.
+    const FLATTEN_COLUMNS: &'static [FlattenColumn<Self>];
 
     /// Returns a reference to the primary key of the table instance.
     fn pk(&self) -> &Self::PrimaryKey;
@@ -68,30 +72,22 @@ impl<E: Create + Read + Update + Delete> Entity for E {}
 /// and execution within the framework.
 pub mod column {
     use crate::Table;
+    use std::borrow::Cow;
     use std::marker::PhantomData;
 
     /// An enum that encapsulates different column types of a table.
-    #[derive(Copy, Debug, PartialEq, Eq)]
+    #[derive(Clone, Debug, PartialEq, Eq)]
     pub enum Column<T: Table> {
         /// A primary key
-        PrimaryKey(&'static PrimaryKey<T>),
+        PrimaryKey(PrimaryKey<T>),
         /// A foreign key
-        ForeignKey(&'static ForeignKey<T>),
+        ForeignKey(ForeignKey<T>),
         /// A data column
-        Data(&'static DataColumn<T>),
+        Data(DataColumn<T>),
         /// A timestamp column
-        Timestamp(&'static TimestampColumn<T>),
-    }
-
-    impl<T: Table> Clone for Column<T> {
-        fn clone(&self) -> Self {
-            match self {
-                Self::PrimaryKey(pk) => Self::PrimaryKey(*pk),
-                Self::ForeignKey(fk) => Self::ForeignKey(*fk),
-                Self::Data(data) => Self::Data(*data),
-                Self::Timestamp(ts) => Self::Timestamp(*ts),
-            }
-        }
+        Timestamp(TimestampColumn<T>),
+        /// A flattened nested struct column
+        Flattened(FlattenedColumn<T>),
     }
 
     impl<T: Table> Column<T> {
@@ -101,15 +97,24 @@ pub mod column {
                 Self::ForeignKey(fk) => fk.field,
                 Self::Data(data) => data.field,
                 Self::Timestamp(ts) => ts.field,
+                Self::Flattened(fl) => fl.nested_field,
             }
         }
 
-        pub const fn sql(&self) -> &'static str {
+        pub fn sql(&self) -> &str {
             match self {
                 Self::PrimaryKey(pk) => pk.sql,
                 Self::ForeignKey(fk) => fk.sql,
                 Self::Data(data) => data.sql,
                 Self::Timestamp(ts) => ts.sql,
+                Self::Flattened(fl) => fl.sql(),
+            }
+        }
+
+        pub const fn flattened(&self) -> Option<&FlattenedColumn<T>> {
+            match self {
+                Self::Flattened(fl) => Some(fl),
+                _ => None,
             }
         }
     }
@@ -131,8 +136,8 @@ pub mod column {
             }
         }
 
-        pub const fn as_col(&'static self) -> Column<T> {
-            Column::PrimaryKey(self)
+        pub fn as_col(&self) -> Column<T> {
+            Column::PrimaryKey(self.clone())
         }
     }
 
@@ -165,8 +170,8 @@ pub mod column {
             }
         }
 
-        pub const fn as_col(&'static self) -> Column<T> {
-            Column::ForeignKey(self)
+        pub fn as_col(&self) -> Column<T> {
+            Column::ForeignKey(self.clone())
         }
 
         /// # Safety
@@ -208,8 +213,8 @@ pub mod column {
             }
         }
 
-        pub const fn as_col(&'static self) -> Column<T> {
-            Column::Data(self)
+        pub fn as_col(&self) -> Column<T> {
+            Column::Data(self.clone())
         }
     }
 
@@ -264,4 +269,84 @@ pub mod column {
             }
         }
     }
+
+    /// Represents a flattened nested struct column in the table.
+    ///
+    /// A flatten column expands a nested struct's fields into the parent table's
+    /// column set. The nested type must implement `FlattenFields`.
+    #[derive(Copy, Debug, PartialEq, Eq)]
+    pub struct FlattenColumn<T: Table> {
+        /// The rust field name of the nested struct
+        pub field: &'static str,
+        /// The SQL column prefix (None = no prefix, Some = prefix string)
+        pub sql_prefix: Option<&'static str>,
+        /// The column names from the nested type (without prefix)
+        pub nested_columns: &'static [&'static str],
+        table: PhantomData<T>,
+    }
+
+    impl<T: Table> FlattenColumn<T> {
+        /// Creates a new flatten column definition.
+        pub const fn new(
+            field: &'static str,
+            sql_prefix: Option<&'static str>,
+            nested_columns: &'static [&'static str],
+        ) -> Self {
+            Self {
+                field,
+                sql_prefix,
+                nested_columns,
+                table: PhantomData,
+            }
+        }
+
+        /// Returns the prefixed SQL column name for a nested column.
+        ///
+        /// If the flatten column has a prefix, it is prepended to the nested column name.
+        /// Otherwise, the nested column name is returned as-is.
+        pub fn prefixed_sql(&self, nested_column: &str) -> String {
+            match self.sql_prefix {
+                Some(prefix) => format!("{}{}", prefix, nested_column),
+                None => nested_column.to_string(),
+            }
+        }
+    }
+
+    impl<T: Table> Clone for FlattenColumn<T> {
+        fn clone(&self) -> Self {
+            Self {
+                field: self.field,
+                sql_prefix: self.sql_prefix,
+                nested_columns: self.nested_columns,
+                table: PhantomData,
+            }
+        }
+    }
+
+    /// Represents a flattened nested column as a concrete SQL column.
+    #[derive(Clone, Debug, PartialEq, Eq)]
+    pub struct FlattenedColumn<T: Table> {
+        /// The rust field name of the nested struct
+        pub field: &'static str,
+        /// The rust field name of the nested column
+        pub nested_field: &'static str,
+        /// The SQL column name (including any prefix)
+        pub sql: Cow<'static, str>,
+        table: PhantomData<T>,
+    }
+
+    impl<T: Table> FlattenedColumn<T> {
+        pub fn new(field: &'static str, nested_field: &'static str, sql: Cow<'static, str>) -> Self {
+            Self {
+                field,
+                nested_field,
+                sql,
+                table: PhantomData,
+            }
+        }
+
+        pub fn sql(&self) -> &str {
+            self.sql.as_ref()
+        }
+    }
 }
diff --git a/atmosphere-macros/src/derive/bindings.rs b/atmosphere-macros/src/derive/bindings.rs
index 343d9f5..5da9086 100644
--- a/atmosphere-macros/src/derive/bindings.rs
+++ b/atmosphere-macros/src/derive/bindings.rs
@@ -1,14 +1,66 @@
 use proc_macro2::TokenStream;
 use quote::quote;
-use syn::Ident;
+use syn::{Ident, Type};
 
 use crate::schema::table::Table;
 
+fn option_inner(ty: &Type) -> Option<&Type> {
+    let Type::Path(path) = ty else {
+        return None;
+    };
+
+    if path.qself.is_some() {
+        return None;
+    }
+
+    let segment = path.path.segments.last()?;
+    if segment.ident != "Option" {
+        return None;
+    }
+
+    let syn::PathArguments::AngleBracketed(args) = &segment.arguments else {
+        return None;
+    };
+
+    let syn::GenericArgument::Type(inner) = args.args.first()? else {
+        return None;
+    };
+
+    Some(inner)
+}
+
 pub fn bindings(table: &Table) -> TokenStream {
     let col = Ident::new("col", proc_macro2::Span::call_site());
     let query = Ident::new("query", proc_macro2::Span::call_site());
 
     let mut binds = TokenStream::new();
+    let mut flatten_binds = TokenStream::new();
+
+    // Handle flatten columns before other fields to avoid name collisions.
+    for flatten in &table.flatten_columns {
+        let field = flatten.name.field();
+        let ty = &flatten.ty;
+
+        let (flatten_ty, value_expr) = if let Some(inner) = option_inner(ty) {
+            (inner, quote!(self.#field.as_ref()))
+        } else {
+            (ty, quote!(Some(&self.#field)))
+        };
+
+        flatten_binds.extend(quote!(
+            if let ::atmosphere::Column::Flattened(flattened) = #col {
+                if flattened.field == stringify!(#field) {
+                    return <#flatten_ty as ::atmosphere::FlattenFields>::bind_field_optional(
+                        #value_expr,
+                        flattened.nested_field,
+                        #query
+                    );
+                }
+            }
+        ));
+    }
+
+    binds.extend(flatten_binds);
 
     {
         let field = &table.primary_key.name.field();
diff --git a/atmosphere-macros/src/derive/flatten.rs b/atmosphere-macros/src/derive/flatten.rs
new file mode 100644
index 0000000..4b7cbf3
--- /dev/null
+++ b/atmosphere-macros/src/derive/flatten.rs
@@ -0,0 +1,272 @@
+//! Derive macro implementation for FlattenFields trait
+
+use proc_macro2::TokenStream;
+use quote::quote;
+use syn::{DeriveInput, Fields, Ident, Type};
+
+/// Information about a field in a struct being derived for FlattenFields
+struct FieldInfo {
+    ident: Ident,
+    #[allow(dead_code)]
+    ty: Type, // Reserved for future use (e.g., nested flatten support)
+    sql_name: String,
+    null_ty: Type,
+}
+
+fn option_inner(ty: &Type) -> Option<&Type> {
+    let Type::Path(path) = ty else {
+        return None;
+    };
+
+    if path.qself.is_some() {
+        return None;
+    }
+
+    let segment = path.path.segments.last()?;
+    if segment.ident != "Option" {
+        return None;
+    }
+
+    let syn::PathArguments::AngleBracketed(args) = &segment.arguments else {
+        return None;
+    };
+
+    let syn::GenericArgument::Type(inner) = args.args.first()? else {
+        return None;
+    };
+
+    Some(inner)
+}
+
+fn nullable_type(ty: &Type) -> Type {
+    option_inner(ty).cloned().unwrap_or_else(|| ty.clone())
+}
+
+/// Generate the FlattenFields trait implementation for a struct
+pub fn derive_flatten_fields(input: DeriveInput) -> syn::Result<TokenStream> {
+    let ident = &input.ident;
+
+    let fields = match &input.data {
+        syn::Data::Struct(data) => match &data.fields {
+            Fields::Named(named) => &named.named,
+            Fields::Unnamed(_) => {
+                return Err(syn::Error::new_spanned(
+                    ident,
+                    "FlattenFields can only be derived for structs with named fields",
+                ));
+            }
+            Fields::Unit => {
+                return Err(syn::Error::new_spanned(
+                    ident,
+                    "FlattenFields cannot be derived for unit structs",
+                ));
+            }
+        },
+        syn::Data::Enum(_) => {
+            return Err(syn::Error::new_spanned(
+                ident,
+                "FlattenFields cannot be derived for enums",
+            ));
+        }
+        syn::Data::Union(_) => {
+            return Err(syn::Error::new_spanned(
+                ident,
+                "FlattenFields cannot be derived for unions",
+            ));
+        }
+    };
+
+    // Extract field information
+    let field_infos: Vec<FieldInfo> = fields
+        .iter()
+        .filter_map(|f| {
+            let ident = f.ident.clone()?;
+            let ty = f.ty.clone();
+
+            // Check for #[sql(rename = "...")] attribute
+            let sql_name = f
+                .attrs
+                .iter()
+                .find(|attr| attr.path().is_ident("sql"))
+                .and_then(|attr| {
+                    // Try to parse rename from sql attribute
+                    let parsed: Result<SqlAttr, _> = attr.parse_args();
+                    parsed.ok().and_then(|a| a.rename)
+                })
+                .unwrap_or_else(|| ident.to_string());
+
+            let null_ty = nullable_type(&ty);
+
+            Some(FieldInfo {
+                ident,
+                ty,
+                sql_name,
+                null_ty,
+            })
+        })
+        .collect();
+
+    // Generate COLUMN_NAMES array
+    let column_names: Vec<&str> = field_infos.iter().map(|f| f.sql_name.as_str()).collect();
+
+    // Generate bind_field match arms
+    let bind_arms = field_infos.iter().map(|f| {
+        let field_ident = &f.ident;
+        let sql_name = &f.sql_name;
+
+        quote!(
+            #sql_name => {
+                use ::atmosphere::Bindable;
+                Ok(query.dyn_bind(&self.#field_ident))
+            }
+        )
+    });
+
+    let bind_optional_arms = field_infos.iter().map(|f| {
+        let field_ident = &f.ident;
+        let sql_name = &f.sql_name;
+        let null_ty = &f.null_ty;
+
+        quote!(
+            #sql_name => {
+                use ::atmosphere::Bindable;
+                match value {
+                    Some(value) => Ok(query.dyn_bind(&value.#field_ident)),
+                    None => Ok(query.dyn_bind(::core::option::Option::<#null_ty>::None)),
+                }
+            }
+        )
+    });
+
+    // Generate a static error message at compile time to avoid runtime allocation
+    let error_msg = format!("unknown column in {}", ident);
+
+    Ok(quote!(
+        #[automatically_derived]
+        impl ::atmosphere::FlattenFields for #ident {
+            const COLUMN_NAMES: &'static [&'static str] = &[#(#column_names),*];
+
+            /// Binds a field value to a query by column name.
+            ///
+            /// # Lifetimes
+            ///
+            /// - `'q`: The query lifetime. The borrowed `self` and returned query share this lifetime,
+            ///   ensuring the bound values remain valid for the duration of query execution.
+            fn bind_field<'q, Q: ::atmosphere::Bindable<'q>>(
+                &'q self,
+                name: &str,
+                query: Q,
+            ) -> ::atmosphere::Result<Q> {
+                match name {
+                    #(#bind_arms)*
+                    _ => {
+                        // Use a compile-time generated static error message
+                        // to avoid memory leaks from runtime string allocation
+                        Err(::atmosphere::Error::Bind(
+                            ::atmosphere::bind::BindError::Unknown(#error_msg)
+                        ))
+                    }
+                }
+            }
+
+            fn bind_field_optional<'q, Q: ::atmosphere::Bindable<'q>>(
+                value: Option<&'q Self>,
+                name: &str,
+                query: Q,
+            ) -> ::atmosphere::Result<Q> {
+                match name {
+                    #(#bind_optional_arms)*
+                    _ => {
+                        Err(::atmosphere::Error::Bind(
+                            ::atmosphere::bind::BindError::Unknown(#error_msg)
+                        ))
+                    }
+                }
+            }
+        }
+    ))
+}
+
+/// Helper struct for parsing #[sql(...)] attributes for FlattenFields derive.
+///
+/// This parser extracts `rename` values and gracefully handles other sql attributes
+/// (like `pk`, `fk`, `flatten`, etc.) that may be present but are not relevant here.
+struct SqlAttr {
+    rename: Option<String>,
+}
+
+impl syn::parse::Parse for SqlAttr {
+    fn parse(input: syn::parse::ParseStream) -> syn::Result<Self> {
+        let mut rename = None;
+
+        while !input.is_empty() {
+            let ident: syn::Ident = input.parse()?;
+
+            match ident.to_string().as_str() {
+                "rename" => {
+                    input.parse::<syn::Token![=]>()?;
+                    // Require a string literal for rename value
+                    if input.peek(syn::LitStr) {
+                        let lit: syn::LitStr = input.parse()?;
+                        rename = Some(lit.value());
+                    } else {
+                        return Err(syn::Error::new(
+                            input.span(),
+                            "expected a string literal for `rename` value, e.g., rename = \"column_name\"",
+                        ));
+                    }
+                }
+                // Known attributes that take `= value` - skip them gracefully
+                "timestamp" => {
+                    if input.peek(syn::Token![=]) {
+                        input.parse::<syn::Token![=]>()?;
+                        let _: syn::Ident = input.parse()?;
+                    }
+                }
+                "prefix" => {
+                    if input.peek(syn::Token![=]) {
+                        input.parse::<syn::Token![=]>()?;
+                        // prefix can be identifier (auto) or string literal
+                        if input.peek(syn::Ident) {
+                            let _: syn::Ident = input.parse()?;
+                        } else if input.peek(syn::LitStr) {
+                            let _: syn::LitStr = input.parse()?;
+                        }
+                    }
+                }
+                // Known attributes that take `-> Type` syntax
+                "fk" => {
+                    if input.peek(syn::Token![-]) {
+                        input.parse::<syn::Token![-]>()?;
+                        input.parse::<syn::Token![>]>()?;
+                        let _: syn::Ident = input.parse()?;
+                    }
+                }
+                // Known flag attributes (pk, unique, json, flatten) - no value needed
+                "pk" | "unique" | "json" | "flatten" => {}
+                // Unknown attribute - skip it gracefully but warn in generated code
+                _ => {
+                    if input.peek(syn::Token![=]) {
+                        input.parse::<syn::Token![=]>()?;
+                        // Try to skip any literal value
+                        if input.peek(syn::LitStr) {
+                            let _: syn::LitStr = input.parse()?;
+                        } else if input.peek(syn::Ident) {
+                            let _: syn::Ident = input.parse()?;
+                        } else if input.peek(syn::LitInt) {
+                            let _: syn::LitInt = input.parse()?;
+                        } else if input.peek(syn::LitBool) {
+                            let _: syn::LitBool = input.parse()?;
+                        }
+                    }
+                }
+            }
+
+            if input.peek(syn::Token![,]) {
+                input.parse::<syn::Token![,]>()?;
+            }
+        }
+
+        Ok(Self { rename })
+    }
+}
diff --git a/atmosphere-macros/src/derive/mod.rs b/atmosphere-macros/src/derive/mod.rs
index b885b71..be41339 100644
--- a/atmosphere-macros/src/derive/mod.rs
+++ b/atmosphere-macros/src/derive/mod.rs
@@ -4,20 +4,48 @@ use quote::quote;
 use crate::schema::table::Table;
 
 mod bindings;
+pub mod flatten;
 mod hooks;
 mod queries;
 mod relationships;
 mod table;
 
+/// Generate const assertions for flatten fields to verify trait bounds at compile time
+fn flatten_assertions(table: &Table) -> TokenStream {
+    if table.flatten_columns.is_empty() {
+        return TokenStream::new();
+    }
+
+    let flatten_types: Vec<_> = table.flatten_columns.iter().map(|f| &f.ty).collect();
+    let ident = &table.ident;
+
+    quote!(
+        #[doc(hidden)]
+        const _: () = {
+            // Assert that all flatten field types implement FlattenFields
+            fn _assert_flatten_fields<T: ::atmosphere::FlattenFields>() {}
+
+            fn _check_flatten_bounds() {
+                #(
+                    _assert_flatten_fields::<#flatten_types>();
+                )*
+            }
+
+            ::atmosphere::assert_unique_table_columns::<#ident>();
+        };
+    )
+}
+
 pub fn all(table: &Table) -> TokenStream {
     let bindings = bindings::bindings(table);
     let queries = queries::queries(table);
     let relationships = relationships::relationships(table);
     let hooks = hooks::hooks(table);
-    let table = table::table(table);
+    let table_impl = table::table(table);
+    let flatten_asserts = flatten_assertions(table);
 
     quote!(
-        #table
+        #table_impl
 
         #bindings
 
@@ -26,5 +54,7 @@ pub fn all(table: &Table) -> TokenStream {
         #relationships
 
         #hooks
+
+        #flatten_asserts
     )
 }
diff --git a/atmosphere-macros/src/derive/queries/unique.rs b/atmosphere-macros/src/derive/queries/unique.rs
index 8fd8ccf..894be7a 100644
--- a/atmosphere-macros/src/derive/queries/unique.rs
+++ b/atmosphere-macros/src/derive/queries/unique.rs
@@ -51,9 +51,8 @@ pub fn queries(table: &Table) -> TokenStream {
                         Error
                     };
 
-                    const COLUMN: ::atmosphere::Column<#ident> = #column.as_col();
-
-                    let query = sql::select_by::<#ident>(COLUMN.clone());
+                    let column = #column.as_col();
+                    let query = sql::select_by::<#ident>(column);
 
                     ::atmosphere::sqlx::query_as(query.sql())
                         .bind(value)
@@ -79,9 +78,8 @@ pub fn queries(table: &Table) -> TokenStream {
                         Error
                     };
 
-                    const COLUMN: ::atmosphere::Column<#ident> = #column.as_col();
-
-                    let query = sql::delete_by::<#ident>(COLUMN.clone());
+                    let column = #column.as_col();
+                    let query = sql::delete_by::<#ident>(column);
 
                     ::atmosphere::sqlx::query(query.sql())
                         .bind(value)
diff --git a/atmosphere-macros/src/derive/table.rs b/atmosphere-macros/src/derive/table.rs
index 4bf4fe8..e23b5b9 100644
--- a/atmosphere-macros/src/derive/table.rs
+++ b/atmosphere-macros/src/derive/table.rs
@@ -11,6 +11,7 @@ pub fn table(table: &Table) -> TokenStream {
         foreign_keys,
         data_columns,
         timestamp_columns,
+        flatten_columns,
         ..
     } = table;
 
@@ -24,6 +25,7 @@ pub fn table(table: &Table) -> TokenStream {
     let foreign_keys = foreign_keys.iter().map(|r| r.quote());
     let data = data_columns.iter().map(|d| d.quote());
     let timestamps = timestamp_columns.iter().map(|d| d.quote());
+    let flatten = flatten_columns.iter().map(|f| f.quote());
 
     quote!(
         #[automatically_derived]
@@ -37,6 +39,7 @@ pub fn table(table: &Table) -> TokenStream {
             const FOREIGN_KEYS: &'static [::atmosphere::ForeignKey<#ident>] = &[#(#foreign_keys),*];
             const DATA_COLUMNS: &'static [::atmosphere::DataColumn<#ident>] = &[#(#data),*];
             const TIMESTAMP_COLUMNS: &'static [::atmosphere::TimestampColumn<#ident>] = &[#(#timestamps),*];
+            const FLATTEN_COLUMNS: &'static [::atmosphere::FlattenColumn<#ident>] = &[#(#flatten),*];
 
             fn pk(&self) -> &Self::PrimaryKey {
                 &self.#pk_field
diff --git a/atmosphere-macros/src/lib.rs b/atmosphere-macros/src/lib.rs
index 6309a07..24c36fc 100644
--- a/atmosphere-macros/src/lib.rs
+++ b/atmosphere-macros/src/lib.rs
@@ -23,22 +23,67 @@ use schema::table::Table;
 
 /// An attribute macro that stores metadata about the sql table and derives needed traits.
 ///
-/// Keys:
+/// # Table Attributes
 ///
 /// - `schema` - sets schema name.
 /// - `name` - sets table name.
 ///
-/// Usage:
+/// # Field Attributes
+///
+/// - `#[sql(pk)]` - marks field as primary key
+/// - `#[sql(fk->Table)]` - marks field as foreign key referencing Table
+/// - `#[sql(unique)]` - marks field as unique
+/// - `#[sql(json)]` - serializes field as JSON
+/// - `#[sql(timestamp = created|updated|deleted)]` - marks as timestamp column
+/// - `#[sql(rename = "column_name")]` - renames SQL column
+/// - `#[sql(flatten)]` - flattens nested struct into parent table's columns
+/// - `#[sql(flatten, prefix)]` - flattens with auto-prefix using field name
+/// - `#[sql(flatten, prefix = "custom_")]` - flattens with custom prefix
+///
+/// # Flatten Support
+///
+/// The `#[sql(flatten)]` attribute allows nested structs to be expanded into the parent
+/// table's column set. The nested type must implement `FlattenFields` (use `#[derive(FlattenFields)]`).
+///
+/// ## Prefix Options
+///
+/// - `#[sql(flatten)]` - No prefix, columns must be unique
+/// - `#[sql(flatten, prefix)]` - Auto-prefix with field name (e.g., `contact_email`)
+/// - `#[sql(flatten, prefix = auto)]` - Same as above (explicit)
+/// - `#[sql(flatten, prefix = "c_")]` - Custom prefix (e.g., `c_email`)
+///
+/// ## Example
+///
+/// ```ignore
+/// use atmosphere::prelude::*;
+///
+/// #[derive(FlattenFields)]
+/// struct ContactInfo {
+///     email: String,
+///     phone: Option<String>,
+/// }
+///
+/// #[table(schema = "public", name = "scientist")]
+/// struct Scientist {
+///     #[sql(pk)]
+///     id: i32,
+///     name: String,
+///     #[sql(flatten, prefix)]  // Creates contact_email, contact_phone columns
+///     contact: ContactInfo,
+/// }
+/// ```
+///
+/// # Basic Usage
 ///
 /// ```ignore
 /// # use atmosphere::prelude::*;
 /// #[table(schema = "public", name = "user")]
-/// # struct User {
-/// #     #[sql(pk)]
-/// #     id: i32,
-/// #     #[sql(unique)]
-/// #     username: String,
-/// # }
+/// struct User {
+///     #[sql(pk)]
+///     id: i32,
+///     #[sql(unique)]
+///     username: String,
+/// }
 /// ```
 #[proc_macro_attribute]
 pub fn table(table_args: TokenStream, input: TokenStream) -> TokenStream {
@@ -78,6 +123,29 @@ pub fn table(table_args: TokenStream, input: TokenStream) -> TokenStream {
 
             field.attrs.push(rename);
         }
+
+        // Synthesize #[sqlx(flatten)] for flatten fields
+        if let schema::column::attribute::ColumnKind::Flatten { .. } = attribute.kind {
+            struct ExtractFlatten {
+                flatten: syn::Attribute,
+            }
+
+            impl syn::parse::Parse for ExtractFlatten {
+                fn parse(input: syn::parse::ParseStream) -> syn::Result<Self> {
+                    Ok(Self {
+                        flatten: input
+                            .call(syn::Attribute::parse_outer)?
+                            .into_iter()
+                            .next()
+                            .unwrap(),
+                    })
+                }
+            }
+
+            let ExtractFlatten { flatten } = syn::parse_str("#[sqlx(flatten)]").unwrap();
+
+            field.attrs.push(flatten);
+        }
     }
 
     let table = match Table::parse_struct(&model, table_args) {
@@ -133,3 +201,42 @@ pub fn hooks(attr: TokenStream, input: TokenStream) -> TokenStream {
     let _ = parse_macro_input!(attr as hooks::Hooks);
     quote! { #model }.into()
 }
+
+/// Derive macro for the `FlattenFields` trait.
+///
+/// This macro generates an implementation of `FlattenFields` for a struct,
+/// allowing it to be used with `#[sql(flatten)]` in a parent table.
+///
+/// # Example
+///
+/// ```ignore
+/// use atmosphere::prelude::*;
+///
+/// #[derive(FlattenFields)]
+/// struct ContactInfo {
+///     email: String,
+///     phone: Option<String>,
+/// }
+///
+/// #[table(schema = "public", name = "user")]
+/// struct User {
+///     #[sql(pk)]
+///     id: i32,
+///     name: String,
+///     #[sql(flatten)]
+///     contact: ContactInfo,
+/// }
+/// ```
+///
+/// # Supported Attributes
+///
+/// - `#[sql(rename = "column_name")]` - Rename a field's SQL column name
+#[proc_macro_derive(FlattenFields, attributes(sql))]
+pub fn derive_flatten_fields(input: TokenStream) -> TokenStream {
+    let input = parse_macro_input!(input as syn::DeriveInput);
+
+    match derive::flatten::derive_flatten_fields(input) {
+        Ok(tokens) => tokens.into(),
+        Err(err) => err.into_compile_error().into(),
+    }
+}
diff --git a/atmosphere-macros/src/schema/column.rs b/atmosphere-macros/src/schema/column.rs
index e3e88b1..b3e402a 100644
--- a/atmosphere-macros/src/schema/column.rs
+++ b/atmosphere-macros/src/schema/column.rs
@@ -92,16 +92,62 @@ impl DataColumn {
     }
 }
 
+/// Represents the prefix mode for a flattened column
+#[derive(Clone, Debug, PartialEq, Eq, Hash)]
+pub enum FlattenPrefix {
+    /// No prefix - columns must be unique across parent and nested types
+    None,
+    /// Auto-prefix using the field name (e.g., `scientist` → `scientist_name`)
+    Auto,
+    /// Custom prefix string (e.g., `"sci_"` → `sci_name`)
+    Custom(String),
+}
+
+/// Represents a flattened nested struct column
+#[derive(Clone, Debug, PartialEq, Eq, Hash)]
+pub struct FlattenColumn {
+    pub modifiers: ColumnModifiers,
+    pub name: NameSet,
+    pub ty: Type,
+    pub prefix: FlattenPrefix,
+}
+
+impl FlattenColumn {
+    pub fn quote(&self) -> TokenStream {
+        let field = self.name.field();
+        let prefix = match &self.prefix {
+            FlattenPrefix::None => quote!(None),
+            FlattenPrefix::Auto => {
+                let prefix_str = format!("{}_", field);
+                quote!(Some(#prefix_str))
+            }
+            FlattenPrefix::Custom(s) => quote!(Some(#s)),
+        };
+        let ty = &self.ty;
+
+        // We need to reference the nested type's COLUMN_NAMES at runtime
+        // The FlattenColumn in atmosphere-core takes nested_columns as a param
+        quote!(::atmosphere::FlattenColumn::new(
+            stringify!(#field),
+            #prefix,
+            <#ty as ::atmosphere::FlattenFields>::COLUMN_NAMES
+        ))
+    }
+}
+
 #[derive(Clone, Debug, PartialEq, Eq)]
 pub enum Column {
     PrimaryKey(PrimaryKey),
     ForeignKey(ForeignKey),
     Data(DataColumn),
     Timestamp(TimestampColumn),
+    Flatten(FlattenColumn),
 }
 
 impl Hash for Column {
     fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
+        // Include a discriminant to avoid hash collisions between different column types
+        core::mem::discriminant(self).hash(state);
         self.name().field().to_string().hash(state);
     }
 }
@@ -113,6 +159,7 @@ impl Column {
             Self::ForeignKey(fk) => fk.quote(),
             Self::Data(data) => data.quote(),
             Self::Timestamp(time) => time.quote(),
+            Self::Flatten(flatten) => flatten.quote(),
         }
     }
 
@@ -122,6 +169,7 @@ impl Column {
             Self::ForeignKey(fk) => &fk.ty,
             Self::Data(data) => &data.ty,
             Self::Timestamp(ts) => &ts.ty,
+            Self::Flatten(fl) => &fl.ty,
         }
     }
 }
@@ -129,7 +177,7 @@ impl Column {
 pub mod attribute {
     use syn::{Error, Ident, LitStr, Token, parse::Parse};
 
-    use super::{ColumnModifiers, TimestampKind};
+    use super::{ColumnModifiers, FlattenPrefix, TimestampKind};
 
     pub const PATH: &str = "sql";
 
@@ -138,6 +186,9 @@ pub mod attribute {
     const UNIQUE: &str = "unique";
     const JSON: &str = "json";
     const TIMESTAMP: &str = "timestamp";
+    const FLATTEN: &str = "flatten";
+    const PREFIX: &str = "prefix";
+    const AUTO: &str = "auto";
 
     const TIMESTAMP_CREATED: &str = "created";
     const TIMESTAMP_UPDATED: &str = "updated";
@@ -149,6 +200,7 @@ pub mod attribute {
         ForeignKey { on: Ident },
         Data,
         Timestamp { kind: TimestampKind },
+        Flatten { prefix: FlattenPrefix },
     }
 
     impl Parse for ColumnKind {
@@ -193,6 +245,62 @@ pub mod attribute {
 
                         kind = ColumnKind::Timestamp { kind: ty }
                     }
+                    FLATTEN => {
+                        let _: Ident = input.parse()?;
+
+                        // Check for optional prefix modifier
+                        let prefix = if input.peek(Token![,]) {
+                            input.parse::<Token![,]>()?;
+
+                            // Look for 'prefix' keyword
+                            if let Some((id, _)) = input.cursor().ident() {
+                                if id.to_string().as_str() == PREFIX {
+                                    let _: Ident = input.parse()?;
+
+                                    // Check what follows: nothing, '= auto', or '= "custom"'
+                                    if input.peek(Token![=]) {
+                                        input.parse::<Token![=]>()?;
+
+                                        // Check if it's 'auto' identifier or a string literal
+                                        if let Some((id, _)) = input.cursor().ident() {
+                                            if id.to_string().as_str() == AUTO {
+                                                let _: Ident = input.parse()?;
+                                                FlattenPrefix::Auto
+                                            } else {
+                                                return Err(syn::Error::new_spanned(
+                                                    id,
+                                                    "expected `auto` or a string literal for prefix value",
+                                                ));
+                                            }
+                                        } else {
+                                            // Must be a string literal for custom prefix
+                                            let lit: LitStr = input.parse()?;
+                                            FlattenPrefix::Custom(lit.value())
+                                        }
+                                    } else {
+                                        // Bare 'prefix' without '=' means auto
+                                        FlattenPrefix::Auto
+                                    }
+                                } else {
+                                    // Unknown identifier after flatten comma - error with helpful message
+                                    let unknown: Ident = input.parse()?;
+                                    return Err(syn::Error::new_spanned(
+                                        &unknown,
+                                        format!(
+                                            "unknown flatten option `{}`; expected `prefix`, `prefix = auto`, or `prefix = \"custom_\"`",
+                                            unknown
+                                        ),
+                                    ));
+                                }
+                            } else {
+                                FlattenPrefix::None
+                            }
+                        } else {
+                            FlattenPrefix::None
+                        };
+
+                        kind = ColumnKind::Flatten { prefix }
+                    }
                     _ => {}
                 };
 
@@ -344,6 +452,12 @@ impl TryFrom<Field> for Column {
                 name,
                 ty,
             })),
+            attribute::ColumnKind::Flatten { prefix } => Ok(Self::Flatten(FlattenColumn {
+                modifiers,
+                name,
+                ty,
+                prefix,
+            })),
         }
     }
 }
@@ -355,6 +469,7 @@ impl Column {
             Self::ForeignKey(fk) => &fk.name,
             Self::Data(data) => &data.name,
             Self::Timestamp(ts) => &ts.name,
+            Self::Flatten(fl) => &fl.name,
         }
     }
 }
@@ -388,4 +503,46 @@ impl Column {
             _ => None,
         }
     }
+
+    pub const fn as_flatten_column(&self) -> Option<&FlattenColumn> {
+        match self {
+            Self::Flatten(c) => Some(c),
+            _ => None,
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::attribute::ColumnKind;
+    use super::FlattenPrefix;
+
+    #[test]
+    fn parse_flatten_no_prefix() {
+        let kind: ColumnKind = syn::parse_str("flatten").unwrap();
+        assert_eq!(kind, ColumnKind::Flatten { prefix: FlattenPrefix::None });
+    }
+
+    #[test]
+    fn parse_flatten_prefix_auto_shorthand() {
+        let kind: ColumnKind = syn::parse_str("flatten, prefix").unwrap();
+        assert_eq!(kind, ColumnKind::Flatten { prefix: FlattenPrefix::Auto });
+    }
+
+    #[test]
+    fn parse_flatten_prefix_auto_explicit() {
+        let kind: ColumnKind = syn::parse_str("flatten, prefix = auto").unwrap();
+        assert_eq!(kind, ColumnKind::Flatten { prefix: FlattenPrefix::Auto });
+    }
+
+    #[test]
+    fn parse_flatten_prefix_custom() {
+        let kind: ColumnKind = syn::parse_str("flatten, prefix = \"sci_\"").unwrap();
+        assert_eq!(
+            kind,
+            ColumnKind::Flatten {
+                prefix: FlattenPrefix::Custom("sci_".to_string())
+            }
+        );
+    }
 }
diff --git a/atmosphere-macros/src/schema/table.rs b/atmosphere-macros/src/schema/table.rs
index 6112a5e..20c283c 100644
--- a/atmosphere-macros/src/schema/table.rs
+++ b/atmosphere-macros/src/schema/table.rs
@@ -2,10 +2,10 @@ use std::collections::HashSet;
 
 use syn::parse::{Parse, ParseStream};
 use syn::spanned::Spanned as _;
-use syn::{Error, Fields, Ident, LitStr, Token};
+use syn::{Error, Fields, Ident, LitStr, Token, Type};
 
 use crate::hooks::Hooks;
-use crate::schema::column::{Column, DataColumn, TimestampColumn};
+use crate::schema::column::{Column, DataColumn, FlattenColumn, TimestampColumn};
 use crate::schema::keys::{ForeignKey, PrimaryKey};
 
 #[derive(Clone, Debug)]
@@ -64,10 +64,45 @@ pub struct Table {
     pub foreign_keys: HashSet<ForeignKey>,
     pub data_columns: HashSet<DataColumn>,
     pub timestamp_columns: HashSet<TimestampColumn>,
+    pub flatten_columns: HashSet<FlattenColumn>,
 
     pub hooks: Hooks,
 }
 
+fn is_self_type(ty: &Type, ident: &Ident) -> bool {
+    let Type::Path(path) = ty else {
+        return false;
+    };
+
+    if path.qself.is_some() {
+        return false;
+    }
+
+    let segment = match path.path.segments.last() {
+        Some(segment) => segment,
+        None => return false,
+    };
+
+    if segment.ident == *ident {
+        return true;
+    }
+
+    if segment.ident != "Option" {
+        return false;
+    }
+
+    let syn::PathArguments::AngleBracketed(args) = &segment.arguments else {
+        return false;
+    };
+
+    let inner = match args.args.first() {
+        Some(syn::GenericArgument::Type(inner)) => inner,
+        _ => return false,
+    };
+
+    is_self_type(inner, ident)
+}
+
 impl Table {
     pub fn parse_struct(
         item: &syn::ItemStruct,
@@ -144,6 +179,22 @@ impl Table {
             .cloned()
             .collect();
 
+        let flatten_columns: HashSet<FlattenColumn> = columns
+            .iter()
+            .filter_map(|c| c.as_flatten_column())
+            .cloned()
+            .collect();
+
+        for flatten in &flatten_columns {
+            let ty: &Type = &flatten.ty;
+            if is_self_type(ty, ident) {
+                return Err(Error::new(
+                    flatten.ty.span(),
+                    "recursive flatten detected: type contains itself",
+                ));
+            }
+        }
+
         Ok(Self {
             ident: ident.clone(),
             id,
@@ -151,6 +202,7 @@ impl Table {
             foreign_keys,
             data_columns,
             timestamp_columns,
+            flatten_columns,
             hooks,
         })
     }
diff --git a/docs/src/getting-started/schema.md b/docs/src/getting-started/schema.md
index 9960213..d7c0977 100644
--- a/docs/src/getting-started/schema.md
+++ b/docs/src/getting-started/schema.md
@@ -59,4 +59,40 @@ struct User {
 Every struct member corresponds to one row of your backing table. Here you can
 use the `#[sql]` annotation to add metadata.
 
+## Flatten nested structs
+
+Use `#[sql(flatten)]` to expand a nested struct into the parent table columns.
+The nested type must implement `FlattenFields` (derive it on the nested struct).
+
+```rust
+use atmosphere::prelude::*;
+
+#[derive(FlattenFields)]
+struct ContactInfo {
+    email: String,
+    phone: Option<String>,
+}
+
+#[table(schema = "public", name = "user")]
+struct User {
+    #[sql(pk)]
+    id: i32,
+    name: String,
+    #[sql(flatten, prefix)]
+    contact: ContactInfo,
+}
+```
+
+### Prefix options
+
+- `#[sql(flatten)]` no prefix (columns must be unique)
+- `#[sql(flatten, prefix)]` auto prefix using field name (e.g. `contact_email`)
+- `#[sql(flatten, prefix = auto)]` same as above (explicit)
+- `#[sql(flatten, prefix = "c_")]` custom prefix
+
+### Notes
+
+- Changing prefix changes column names and requires a database migration.
+- Recommended max nesting depth: 2 levels (Postgres identifier limit is 63 chars).
+
 [`table`]: https://docs.rs/atmosphere/latest/atmosphere/attr.table.html
diff --git a/tests/flatten/auto_prefix.rs b/tests/flatten/auto_prefix.rs
new file mode 100644
index 0000000..1ef2120
--- /dev/null
+++ b/tests/flatten/auto_prefix.rs
@@ -0,0 +1,63 @@
+//! Tests for #[sql(flatten, prefix)] (auto prefix using field name)
+
+use atmosphere::prelude::*;
+
+use super::ContactInfo;
+
+/// A scientist with flattened contact info (auto prefix = contact_)
+#[derive(Debug, Clone, PartialEq, Eq)]
+#[table(name = "scientist_auto_prefix", schema = "public")]
+pub struct ScientistAutoPrefix {
+    #[sql(pk)]
+    pub id: i32,
+    pub name: String,
+    #[sql(flatten, prefix)]
+    pub contact: ContactInfo,
+}
+
+/// Test that the table metadata is correctly generated with auto prefix
+#[test]
+fn test_table_metadata() {
+    use atmosphere::Table;
+
+    assert_eq!(ScientistAutoPrefix::SCHEMA, "public");
+    assert_eq!(ScientistAutoPrefix::TABLE, "scientist_auto_prefix");
+    assert_eq!(ScientistAutoPrefix::FLATTEN_COLUMNS.len(), 1);
+    assert_eq!(ScientistAutoPrefix::FLATTEN_COLUMNS[0].field, "contact");
+    assert_eq!(
+        ScientistAutoPrefix::FLATTEN_COLUMNS[0].sql_prefix,
+        Some("contact_")
+    );
+}
+
+#[sqlx::test(migrations = "tests/flatten/migrations")]
+async fn create(pool: sqlx::PgPool) {
+    let mut scientist = ScientistAutoPrefix {
+        id: 1,
+        name: "Barbara McClintock".to_string(),
+        contact: ContactInfo::new("barbara@mcclintock.edu", Some("+1-555-0100")),
+    };
+
+    scientist.create(&pool).await.unwrap();
+
+    // Verify the record was created
+    let found = ScientistAutoPrefix::find(&pool, &1).await.unwrap().unwrap();
+    assert_eq!(found.name, "Barbara McClintock");
+    assert_eq!(found.contact.email, "barbara@mcclintock.edu");
+}
+
+#[sqlx::test(migrations = "tests/flatten/migrations")]
+async fn read(pool: sqlx::PgPool) {
+    // Insert a record first
+    let mut scientist = ScientistAutoPrefix {
+        id: 2,
+        name: "Lise Meitner".to_string(),
+        contact: ContactInfo::new("lise@meitner.at", None),
+    };
+    scientist.create(&pool).await.unwrap();
+
+    // Read it back
+    let found = ScientistAutoPrefix::find(&pool, &2).await.unwrap().unwrap();
+    assert_eq!(found.name, "Lise Meitner");
+    assert_eq!(found.contact.email, "lise@meitner.at");
+}
diff --git a/tests/flatten/custom_prefix.rs b/tests/flatten/custom_prefix.rs
new file mode 100644
index 0000000..e3c6844
--- /dev/null
+++ b/tests/flatten/custom_prefix.rs
@@ -0,0 +1,63 @@
+//! Tests for #[sql(flatten, prefix = "custom_")] (custom prefix)
+
+use atmosphere::prelude::*;
+
+use super::ContactInfo;
+
+/// A scientist with flattened contact info (custom prefix = c_)
+#[derive(Debug, Clone, PartialEq, Eq)]
+#[table(name = "scientist_custom_prefix", schema = "public")]
+pub struct ScientistCustomPrefix {
+    #[sql(pk)]
+    pub id: i32,
+    pub name: String,
+    #[sql(flatten, prefix = "c_")]
+    pub contact: ContactInfo,
+}
+
+/// Test that the table metadata is correctly generated with custom prefix
+#[test]
+fn test_table_metadata() {
+    use atmosphere::Table;
+
+    assert_eq!(ScientistCustomPrefix::SCHEMA, "public");
+    assert_eq!(ScientistCustomPrefix::TABLE, "scientist_custom_prefix");
+    assert_eq!(ScientistCustomPrefix::FLATTEN_COLUMNS.len(), 1);
+    assert_eq!(ScientistCustomPrefix::FLATTEN_COLUMNS[0].field, "contact");
+    assert_eq!(
+        ScientistCustomPrefix::FLATTEN_COLUMNS[0].sql_prefix,
+        Some("c_")
+    );
+}
+
+#[sqlx::test(migrations = "tests/flatten/migrations")]
+async fn create(pool: sqlx::PgPool) {
+    let mut scientist = ScientistCustomPrefix {
+        id: 1,
+        name: "Grace Hopper".to_string(),
+        contact: ContactInfo::new("grace@hopper.navy.mil", Some("+1-555-0200")),
+    };
+
+    scientist.create(&pool).await.unwrap();
+
+    // Verify the record was created
+    let found = ScientistCustomPrefix::find(&pool, &1).await.unwrap().unwrap();
+    assert_eq!(found.name, "Grace Hopper");
+    assert_eq!(found.contact.email, "grace@hopper.navy.mil");
+}
+
+#[sqlx::test(migrations = "tests/flatten/migrations")]
+async fn read(pool: sqlx::PgPool) {
+    // Insert a record first
+    let mut scientist = ScientistCustomPrefix {
+        id: 2,
+        name: "Hedy Lamarr".to_string(),
+        contact: ContactInfo::new("hedy@lamarr.hollywood.com", None),
+    };
+    scientist.create(&pool).await.unwrap();
+
+    // Read it back
+    let found = ScientistCustomPrefix::find(&pool, &2).await.unwrap().unwrap();
+    assert_eq!(found.name, "Hedy Lamarr");
+    assert_eq!(found.contact.email, "hedy@lamarr.hollywood.com");
+}
diff --git a/tests/flatten/migrations/1_flatten_tables.sql b/tests/flatten/migrations/1_flatten_tables.sql
new file mode 100644
index 0000000..b88a3f0
--- /dev/null
+++ b/tests/flatten/migrations/1_flatten_tables.sql
@@ -0,0 +1,28 @@
+-- Migration for flatten tests
+
+-- Table with flattened contact info (no prefix)
+CREATE TABLE public.scientist_no_prefix (
+    id INTEGER PRIMARY KEY,
+    name VARCHAR NOT NULL,
+    -- Flattened fields from ContactInfo
+    email VARCHAR NOT NULL,
+    phone VARCHAR
+);
+
+-- Table with flattened contact info (auto prefix)
+CREATE TABLE public.scientist_auto_prefix (
+    id INTEGER PRIMARY KEY,
+    name VARCHAR NOT NULL,
+    -- Flattened fields from ContactInfo with 'contact_' prefix
+    contact_email VARCHAR NOT NULL,
+    contact_phone VARCHAR
+);
+
+-- Table with flattened contact info (custom prefix)
+CREATE TABLE public.scientist_custom_prefix (
+    id INTEGER PRIMARY KEY,
+    name VARCHAR NOT NULL,
+    -- Flattened fields from ContactInfo with 'c_' prefix
+    c_email VARCHAR NOT NULL,
+    c_phone VARCHAR
+);
diff --git a/tests/flatten/mod.rs b/tests/flatten/mod.rs
new file mode 100644
index 0000000..85ee499
--- /dev/null
+++ b/tests/flatten/mod.rs
@@ -0,0 +1,35 @@
+//! Tests for #[sql(flatten)] attribute support
+
+use atmosphere::prelude::*;
+
+mod no_prefix;
+mod auto_prefix;
+mod custom_prefix;
+
+/// A simple contact info struct to be flattened into parent tables
+#[derive(Debug, Clone, PartialEq, Eq, FlattenFields, sqlx::FromRow)]
+pub struct ContactInfo {
+    pub email: String,
+    pub phone: Option<String>,
+}
+
+impl ContactInfo {
+    pub fn new(email: &str, phone: Option<&str>) -> Self {
+        Self {
+            email: email.to_owned(),
+            phone: phone.map(|s| s.to_owned()),
+        }
+    }
+}
+
+/// Verify FlattenFields derive generates correct COLUMN_NAMES
+#[test]
+fn test_flatten_fields_column_names() {
+    use atmosphere::FlattenFields;
+
+    assert_eq!(
+        ContactInfo::COLUMN_NAMES,
+        &["email", "phone"],
+        "ContactInfo should have email and phone columns"
+    );
+}
diff --git a/tests/flatten/no_prefix.rs b/tests/flatten/no_prefix.rs
new file mode 100644
index 0000000..137e266
--- /dev/null
+++ b/tests/flatten/no_prefix.rs
@@ -0,0 +1,103 @@
+//! Tests for #[sql(flatten)] without prefix
+
+use atmosphere::prelude::*;
+
+use super::ContactInfo;
+
+/// A scientist with flattened contact info (no prefix)
+#[derive(Debug, Clone, PartialEq, Eq)]
+#[table(name = "scientist_no_prefix", schema = "public")]
+pub struct ScientistNoPrefix {
+    #[sql(pk)]
+    pub id: i32,
+    pub name: String,
+    #[sql(flatten)]
+    pub contact: ContactInfo,
+}
+
+/// Test that the table metadata is correctly generated
+#[test]
+fn test_table_metadata() {
+    use atmosphere::Table;
+
+    assert_eq!(ScientistNoPrefix::SCHEMA, "public");
+    assert_eq!(ScientistNoPrefix::TABLE, "scientist_no_prefix");
+    assert_eq!(ScientistNoPrefix::FLATTEN_COLUMNS.len(), 1);
+    assert_eq!(ScientistNoPrefix::FLATTEN_COLUMNS[0].field, "contact");
+    assert!(ScientistNoPrefix::FLATTEN_COLUMNS[0].sql_prefix.is_none());
+}
+
+#[sqlx::test(migrations = "tests/flatten/migrations")]
+async fn create(pool: sqlx::PgPool) {
+    let mut scientist = ScientistNoPrefix {
+        id: 1,
+        name: "Marie Curie".to_string(),
+        contact: ContactInfo::new("marie@curie.edu", Some("+33-1-23456789")),
+    };
+
+    scientist.create(&pool).await.unwrap();
+
+    // Verify the record was created
+    let found = ScientistNoPrefix::find(&pool, &1).await.unwrap().unwrap();
+    assert_eq!(found.name, "Marie Curie");
+    assert_eq!(found.contact.email, "marie@curie.edu");
+    assert_eq!(found.contact.phone, Some("+33-1-23456789".to_string()));
+}
+
+#[sqlx::test(migrations = "tests/flatten/migrations")]
+async fn read(pool: sqlx::PgPool) {
+    // Insert a record first
+    let mut scientist = ScientistNoPrefix {
+        id: 2,
+        name: "Ada Lovelace".to_string(),
+        contact: ContactInfo::new("ada@lovelace.com", None),
+    };
+    scientist.create(&pool).await.unwrap();
+
+    // Read it back
+    let found = ScientistNoPrefix::find(&pool, &2).await.unwrap().unwrap();
+    assert_eq!(found.name, "Ada Lovelace");
+    assert_eq!(found.contact.email, "ada@lovelace.com");
+    assert_eq!(found.contact.phone, None);
+}
+
+#[sqlx::test(migrations = "tests/flatten/migrations")]
+async fn update(pool: sqlx::PgPool) {
+    // Insert a record first
+    let mut scientist = ScientistNoPrefix {
+        id: 3,
+        name: "Emmy Noether".to_string(),
+        contact: ContactInfo::new("emmy@noether.de", None),
+    };
+    scientist.create(&pool).await.unwrap();
+
+    // Update the record with new contact info
+    let mut updated_scientist = ScientistNoPrefix {
+        id: 3,
+        name: "Emmy Noether".to_string(),
+        contact: ContactInfo::new("emmy@noether.de", Some("+49-123-456789")),
+    };
+    updated_scientist.update(&pool).await.unwrap();
+
+    // Verify the update
+    let found = ScientistNoPrefix::find(&pool, &3).await.unwrap().unwrap();
+    assert_eq!(found.contact.phone, Some("+49-123-456789".to_string()));
+}
+
+#[sqlx::test(migrations = "tests/flatten/migrations")]
+async fn delete(pool: sqlx::PgPool) {
+    // Insert a record first
+    let mut scientist = ScientistNoPrefix {
+        id: 4,
+        name: "Rosalind Franklin".to_string(),
+        contact: ContactInfo::new("rosalind@franklin.uk", None),
+    };
+    scientist.create(&pool).await.unwrap();
+
+    // Delete it
+    scientist.delete(&pool).await.unwrap();
+
+    // Verify it's gone
+    let result = ScientistNoPrefix::find(&pool, &4).await.unwrap();
+    assert!(result.is_none());
+}
diff --git a/tests/lib.rs b/tests/lib.rs
index 27d0c3c..ae1ef13 100644
--- a/tests/lib.rs
+++ b/tests/lib.rs
@@ -4,3 +4,4 @@ mod db;
 mod postgis;
 
 mod json_attr;
+mod flatten;