diff --git a/.gitattributes b/.gitattributes
new file mode 100644
index 00000000..d72fd520
--- /dev/null
+++ b/.gitattributes
@@ -0,0 +1 @@
+*.pdf binary
diff --git a/codegen/src/common/model/column.rs b/codegen/src/common/model/column.rs
index 611c9d73..81ef9057 100644
--- a/codegen/src/common/model/column.rs
+++ b/codegen/src/common/model/column.rs
@@ -2,7 +2,7 @@ use indexmap::IndexMap;
 use std::collections::HashMap;
 
 use crate::common::model::index::Index;
-use crate::common::model::{GeneratorType, IndexBackend};
+use crate::common::model::{ColumnSlotIdType, ColumnarFieldConfig, ColumnarIndex, GeneratorType, IndexBackend};
 use proc_macro2::{Ident, TokenStream};
 use quote::quote;
 use syn::spanned::Spanned;
@@ -17,6 +17,9 @@ pub struct Columns {
     pub columns_map: HashMap<Ident, TokenStream>,
     pub field_positions: HashMap<Ident, usize>,
     pub indexes: IndexMap<Ident, Index>,
+    pub columnar_fields: IndexMap<Ident, ColumnarFieldConfig>,
+    pub columnar_indexes: IndexMap<Ident, ColumnarIndex>,
+    pub column_slot_id: ColumnSlotIdType,
     pub primary_keys: Vec<Ident>,
     pub primary_index_backend: IndexBackend,
     pub generator_type: GeneratorType,
@@ -30,6 +33,7 @@ pub struct Row {
     pub gen_type: GeneratorType,
     pub optional: bool,
     pub index_backend: Option<IndexBackend>,
+    pub columnar: Option<ColumnarFieldConfig>,
 }
 
 impl Columns {
@@ -40,6 +44,7 @@ impl Columns {
         let mut pk = vec![];
         let mut gen_type = None;
         let mut primary_index_backend = None;
+        let mut columnar_fields = IndexMap::new();
 
         for (pos, row) in rows.into_iter().enumerate() {
             let type_ = &row.type_;
@@ -53,6 +58,9 @@ impl Columns {
             };
             columns_map.insert(row.name.clone(), type_);
             field_positions.insert(row.name.clone(), pos);
+            if let Some(config) = row.columnar {
+                columnar_fields.insert(row.name.clone(), config);
+            }
 
             if row.is_primary_key {
                 if let Some(t) = gen_type {
@@ -90,6 +98,9 @@ impl Columns {
             is_sized: sized,
             columns_map,
             indexes: Default::default(),
+            columnar_fields,
+            columnar_indexes: Default::default(),
+            column_slot_id: Default::default(),
             primary_keys: pk,
             primary_index_backend: primary_index_backend.unwrap_or_default(),
             generator_type: gen_type.expect("set"),
diff --git a/codegen/src/common/model/columnar.rs b/codegen/src/common/model/columnar.rs
new file mode 100644
index 00000000..0508159a
--- /dev/null
+++ b/codegen/src/common/model/columnar.rs
@@ -0,0 +1,63 @@
+use proc_macro2::Ident;
+
+pub const DEFAULT_COLUMNAR_CHUNK_ROWS: usize = 65_536;
+
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
+pub enum ColumnSlotIdType {
+    U8,
+    U16,
+    #[default]
+    U32,
+    U64,
+}
+
+impl ColumnSlotIdType {
+    pub(crate) fn type_name(self) -> &'static str {
+        match self {
+            Self::U8 => "ColumnSlotId8",
+            Self::U16 => "ColumnSlotId16",
+            Self::U32 => "ColumnSlotId32",
+            Self::U64 => "ColumnSlotId64",
+        }
+    }
+}
+
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
+pub enum ColumnCompression {
+    #[default]
+    None,
+}
+
+impl ColumnCompression {
+    pub(crate) fn name(self) -> &'static str {
+        match self {
+            Self::None => "none",
+        }
+    }
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct ColumnarFieldConfig {
+    pub chunk_rows: Option<usize>,
+    pub compression: ColumnCompression,
+}
+
+impl Default for ColumnarFieldConfig {
+    fn default() -> Self {
+        Self {
+            chunk_rows: None,
+            compression: ColumnCompression::None,
+        }
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct ColumnarIndex {
+    pub name: Ident,
+    pub cluster_by: Vec<Ident>,
+}
+
+#[derive(Debug, Clone, Default, PartialEq, Eq)]
+pub struct ColumnarIndexes {
+    pub indexes: indexmap::IndexMap<Ident, ColumnarIndex>,
+}
diff --git a/codegen/src/common/model/config.rs b/codegen/src/common/model/config.rs
index f61c3e59..f3f23baa 100644
--- a/codegen/src/common/model/config.rs
+++ b/codegen/src/common/model/config.rs
@@ -1,7 +1,22 @@
 use proc_macro2::Ident;
 
-#[derive(Debug, Default)]
+use crate::common::model::{ColumnSlotIdType, DEFAULT_COLUMNAR_CHUNK_ROWS};
+
+#[derive(Debug)]
 pub struct Config {
     pub page_size: Option<u32>,
     pub row_derives: Vec<Ident>,
+    pub columnar_slot_id: ColumnSlotIdType,
+    pub columnar_chunk_rows: usize,
+}
+
+impl Default for Config {
+    fn default() -> Self {
+        Self {
+            page_size: None,
+            row_derives: Vec::new(),
+            columnar_slot_id: ColumnSlotIdType::default(),
+            columnar_chunk_rows: DEFAULT_COLUMNAR_CHUNK_ROWS,
+        }
+    }
 }
diff --git a/codegen/src/common/model/mod.rs b/codegen/src/common/model/mod.rs
index 93604295..91775eb5 100644
--- a/codegen/src/common/model/mod.rs
+++ b/codegen/src/common/model/mod.rs
@@ -1,4 +1,5 @@
 mod column;
+mod columnar;
 mod config;
 mod index;
 pub mod operation;
@@ -7,6 +8,10 @@ mod primary_key;
 mod queries;
 
 pub use column::{Columns, Row};
+pub use columnar::{
+    ColumnCompression, ColumnSlotIdType, ColumnarFieldConfig, ColumnarIndex, ColumnarIndexes,
+    DEFAULT_COLUMNAR_CHUNK_ROWS,
+};
 pub use config::Config;
 pub use index::{Index, IndexBackend};
 pub use operation::Operation;
diff --git a/codegen/src/common/parser/columnar.rs b/codegen/src/common/parser/columnar.rs
new file mode 100644
index 00000000..0a63cb29
--- /dev/null
+++ b/codegen/src/common/parser/columnar.rs
@@ -0,0 +1,327 @@
+use std::collections::HashSet;
+
+use indexmap::IndexMap;
+use proc_macro2::{Delimiter, Ident, TokenTree};
+use syn::spanned::Spanned as _;
+
+use crate::common::Parser;
+use crate::common::model::{ColumnCompression, ColumnarFieldConfig, ColumnarIndex, ColumnarIndexes};
+
+impl Parser {
+    pub(super) fn try_parse_columnar_field(&mut self) -> syn::Result<Option<ColumnarFieldConfig>> {
+        let Some(TokenTree::Ident(attribute)) = self.input_iter.peek() else {
+            return Ok(None);
+        };
+        if attribute != "columnar" {
+            return Ok(None);
+        }
+
+        let attribute_span = attribute.span();
+        self.input_iter.next();
+        let Some(TokenTree::Group(group)) = self.input_iter.peek() else {
+            return Ok(Some(ColumnarFieldConfig::default()));
+        };
+        let group = group.clone();
+        if group.delimiter() != Delimiter::Parenthesis {
+            return Err(syn::Error::new(group.span(), "expected `columnar(...)`"));
+        }
+        self.input_iter.next();
+
+        let mut config = ColumnarFieldConfig::default();
+        let mut saw_chunk_rows = false;
+        let mut saw_compression = false;
+        let mut parser = Parser::new(group.stream());
+
+        while parser.has_next() {
+            let option = parser
+                .input_iter
+                .next()
+                .ok_or_else(|| syn::Error::new(attribute_span, "expected a columnar field option"))?;
+            let TokenTree::Ident(option) = option else {
+                return Err(syn::Error::new(option.span(), "expected a columnar option name"));
+            };
+            let value = parser
+                .input_iter
+                .next()
+                .ok_or_else(|| syn::Error::new(option.span(), format!("expected `{option}(...)`")))?;
+            let TokenTree::Group(value) = value else {
+                return Err(syn::Error::new(value.span(), format!("expected `{option}(...)`")));
+            };
+            if value.delimiter() != Delimiter::Parenthesis {
+                return Err(syn::Error::new(value.span(), format!("expected `{option}(...)`")));
+            }
+
+            match option.to_string().as_str() {
+                "chunk_rows" => {
+                    if saw_chunk_rows {
+                        return Err(syn::Error::new(option.span(), "duplicate `chunk_rows` option"));
+                    }
+                    saw_chunk_rows = true;
+                    let mut values = value.stream().into_iter();
+                    let Some(TokenTree::Literal(rows)) = values.next() else {
+                        return Err(syn::Error::new(value.span(), "`chunk_rows` expects an integer"));
+                    };
+                    if values.next().is_some() {
+                        return Err(syn::Error::new(value.span(), "`chunk_rows` expects one integer"));
+                    }
+                    let parsed = rows
+                        .to_string()
+                        .replace('_', "")
+                        .parse::<usize>()
+                        .map_err(|_| syn::Error::new(rows.span(), "invalid `chunk_rows` integer"))?;
+                    if parsed == 0 {
+                        return Err(syn::Error::new(rows.span(), "`chunk_rows` must be greater than zero"));
+                    }
+                    config.chunk_rows = Some(parsed);
+                }
+                "compression" => {
+                    if saw_compression {
+                        return Err(syn::Error::new(option.span(), "duplicate `compression` option"));
+                    }
+                    saw_compression = true;
+                    let mut values = value.stream().into_iter();
+                    let Some(TokenTree::Ident(compression)) = values.next() else {
+                        return Err(syn::Error::new(value.span(), "`compression` expects a policy name"));
+                    };
+                    if values.next().is_some() {
+                        return Err(syn::Error::new(value.span(), "`compression` expects one policy"));
+                    }
+                    config.compression = match compression.to_string().as_str() {
+                        "none" => ColumnCompression::None,
+                        "auto" | "delta" | "rle" | "dictionary" => {
+                            return Err(syn::Error::new(
+                                compression.span(),
+                                format!(
+                                    "compression({compression}) is declared but not implemented in this release; only compression(none) is currently supported"
+                                ),
+                            ));
+                        }
+                        _ => {
+                            return Err(syn::Error::new(
+                                compression.span(),
+                                "unknown compression; only `none` is currently supported",
+                            ));
+                        }
+                    };
+                }
+                _ => {
+                    return Err(syn::Error::new(
+                        option.span(),
+                        "unknown columnar option; expected `chunk_rows` or `compression`",
+                    ));
+                }
+            }
+            parser.try_parse_comma()?;
+        }
+
+        Ok(Some(config))
+    }
+
+    pub fn parse_columnar_indexes(&mut self) -> syn::Result<ColumnarIndexes> {
+        let section = self
+            .input_iter
+            .next()
+            .ok_or_else(|| syn::Error::new(self.input.span(), "expected `columnar_indexes` section"))?;
+        let TokenTree::Ident(section) = section else {
+            return Err(syn::Error::new(section.span(), "expected `columnar_indexes`"));
+        };
+        if section != "columnar_indexes" {
+            return Err(syn::Error::new(section.span(), "expected `columnar_indexes`"));
+        }
+        self.parse_colon()?;
+
+        let body = self
+            .input_iter
+            .next()
+            .ok_or_else(|| syn::Error::new(section.span(), "expected `columnar_indexes: { ... }`"))?;
+        let TokenTree::Group(body) = body else {
+            return Err(syn::Error::new(body.span(), "expected `columnar_indexes: { ... }`"));
+        };
+        if body.delimiter() != Delimiter::Brace {
+            return Err(syn::Error::new(body.span(), "expected braces around columnar indexes"));
+        }
+
+        let mut parser = Parser::new(body.stream());
+        let mut indexes = IndexMap::new();
+        while parser.has_next() {
+            let name = parser
+                .input_iter
+                .next()
+                .ok_or_else(|| syn::Error::new(body.span(), "expected a columnar index name"))?;
+            let TokenTree::Ident(name) = name else {
+                return Err(syn::Error::new(name.span(), "expected a columnar index name"));
+            };
+            parser.parse_colon()?;
+
+            let definition = parser
+                .input_iter
+                .next()
+                .ok_or_else(|| syn::Error::new(name.span(), "expected a columnar index definition"))?;
+            let TokenTree::Group(definition) = definition else {
+                return Err(syn::Error::new(definition.span(), "expected `{ ... }`"));
+            };
+            if definition.delimiter() != Delimiter::Brace {
+                return Err(syn::Error::new(definition.span(), "expected `{ ... }`"));
+            }
+
+            let mut definition_parser = Parser::new(definition.stream());
+            let mut cluster_by = None;
+            while definition_parser.has_next() {
+                let property = definition_parser
+                    .input_iter
+                    .next()
+                    .ok_or_else(|| syn::Error::new(definition.span(), "expected a columnar index property"))?;
+                let TokenTree::Ident(property) = property else {
+                    return Err(syn::Error::new(property.span(), "expected `cluster_by`"));
+                };
+                definition_parser.parse_colon()?;
+                match property.to_string().as_str() {
+                    "cluster_by" if cluster_by.is_none() => {
+                        cluster_by = Some(parse_ident_list(&mut definition_parser, property.span())?)
+                    }
+                    "cluster_by" => {
+                        return Err(syn::Error::new(property.span(), "duplicate columnar index property"));
+                    }
+                    "columns" => {
+                        return Err(syn::Error::new(
+                            property.span(),
+                            "`columns` has no independent columnar-index semantics; remove it because projected fields are selected from base column stores",
+                        ));
+                    }
+                    "include" => {
+                        return Err(syn::Error::new(
+                            property.span(),
+                            "`include` is reserved for a future covering columnar projection and is not implemented",
+                        ));
+                    }
+                    _ => {
+                        return Err(syn::Error::new(
+                            property.span(),
+                            "unknown columnar index property; expected `cluster_by`",
+                        ));
+                    }
+                }
+                definition_parser.try_parse_comma()?;
+            }
+
+            let cluster_by = cluster_by
+                .ok_or_else(|| syn::Error::new(name.span(), "columnar index requires `cluster_by: [...]`"))?;
+            if cluster_by.is_empty() {
+                return Err(syn::Error::new(
+                    name.span(),
+                    "columnar index `cluster_by` cannot be empty",
+                ));
+            }
+            ensure_unique(&cluster_by, "columnar index `cluster_by` contains a duplicate")?;
+
+            if indexes.contains_key(&name) {
+                return Err(syn::Error::new(name.span(), "duplicate columnar index name"));
+            }
+            indexes.insert(name.clone(), ColumnarIndex { name, cluster_by });
+            parser.try_parse_comma()?;
+        }
+        self.try_parse_comma()?;
+        Ok(ColumnarIndexes { indexes })
+    }
+}
+
+fn parse_ident_list(parser: &mut Parser, span: proc_macro2::Span) -> syn::Result<Vec<Ident>> {
+    let list = parser
+        .input_iter
+        .next()
+        .ok_or_else(|| syn::Error::new(span, "expected `[field, ...]`"))?;
+    let TokenTree::Group(list) = list else {
+        return Err(syn::Error::new(list.span(), "expected `[field, ...]`"));
+    };
+    if list.delimiter() != Delimiter::Bracket {
+        return Err(syn::Error::new(list.span(), "expected `[field, ...]`"));
+    }
+    let mut values = Parser::new(list.stream());
+    let mut result = Vec::new();
+    while values.has_next() {
+        let field = values
+            .input_iter
+            .next()
+            .ok_or_else(|| syn::Error::new(list.span(), "expected a field identifier"))?;
+        let TokenTree::Ident(field) = field else {
+            return Err(syn::Error::new(field.span(), "expected a field identifier"));
+        };
+        result.push(field);
+        values.try_parse_comma()?;
+    }
+    Ok(result)
+}
+
+fn ensure_unique(values: &[Ident], message: &str) -> syn::Result<()> {
+    let mut seen = HashSet::new();
+    for value in values {
+        if !seen.insert(value.to_string()) {
+            return Err(syn::Error::new(value.span(), message));
+        }
+    }
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use quote::quote;
+
+    use crate::common::Parser;
+    use crate::common::model::{ColumnCompression, ColumnarFieldConfig};
+
+    #[test]
+    fn parses_columnar_field_options() {
+        let mut parser = Parser::new(quote! {
+            columnar(chunk_rows(65_536), compression(none))
+        });
+        let config = parser.try_parse_columnar_field().unwrap().unwrap();
+        assert_eq!(config.chunk_rows, Some(65_536));
+        assert_eq!(config.compression, ColumnCompression::None);
+    }
+
+    #[test]
+    fn empty_columnar_field_uses_defaults() {
+        let mut parser = Parser::new(quote! { columnar });
+        let config = parser.try_parse_columnar_field().unwrap().unwrap();
+        assert_eq!(config.chunk_rows, ColumnarFieldConfig::default().chunk_rows);
+        assert_eq!(config.compression, ColumnCompression::None);
+    }
+
+    #[test]
+    fn parses_columnar_indexes() {
+        let mut parser = Parser::new(quote! {
+            columnar_indexes: {
+                host_time: {
+                    cluster_by: [host_id, timestamp],
+                },
+            },
+        });
+        let indexes = parser.parse_columnar_indexes().unwrap();
+        let index = indexes.indexes.values().next().unwrap();
+        assert_eq!(
+            index.cluster_by.iter().map(ToString::to_string).collect::<Vec<_>>(),
+            ["host_id", "timestamp"]
+        );
+    }
+
+    #[test]
+    fn rejects_inert_columns_property() {
+        let mut parser = Parser::new(quote! {
+            columnar_indexes: {
+                host_time: {
+                    columns: [host_id, timestamp],
+                    cluster_by: [host_id, timestamp],
+                },
+            },
+        });
+        let error = parser.parse_columnar_indexes().unwrap_err();
+        assert!(error.to_string().contains("no independent columnar-index semantics"));
+    }
+
+    #[test]
+    fn rejects_unimplemented_compression() {
+        let mut parser = Parser::new(quote! { columnar(compression(dictionary)) });
+        let error = parser.try_parse_columnar_field().unwrap_err();
+        assert!(error.to_string().contains("not implemented"));
+    }
+}
diff --git a/codegen/src/common/parser/columns.rs b/codegen/src/common/parser/columns.rs
index 65126f64..52ce47ce 100644
--- a/codegen/src/common/parser/columns.rs
+++ b/codegen/src/common/parser/columns.rs
@@ -108,8 +108,18 @@ impl Parser {
             false
         };
 
+        let columnar = self.try_parse_columnar_field()?;
+
         let index_backend = self.try_parse_index_backend()?;
 
+        if let Some(next) = self.input_iter.peek()
+            && !matches!(next, TokenTree::Punct(punct) if punct.as_char() == ',')
+        {
+            return Err(syn::Error::new(
+                next.span(),
+                "unexpected column attribute; expected attributes in `primary_key`, generator, `optional`, `columnar`, `using` order",
+            ));
+        }
         self.try_parse_comma()?;
 
         Ok(Row {
@@ -119,6 +129,7 @@ impl Parser {
             gen_type,
             optional,
             index_backend,
+            columnar,
         })
     }
 }
@@ -255,7 +266,7 @@ mod tests {
 
         #[test]
         fn test_row_parse_no_comma() {
-            let row_tokens = quote! {id: i64 primary_key TreeIndex};
+            let row_tokens = quote! {id: i64 primary_key};
 
             let mut parser = Parser::new(row_tokens);
             let row = parser.parse_row();
@@ -324,6 +335,18 @@ mod tests {
             assert_eq!(row.index_backend, Some(crate::common::model::IndexBackend::Congee));
         }
 
+        #[test]
+        fn test_columnar_field_parse() {
+            let row_tokens = quote! {
+                host_id: u64 columnar(chunk_rows(65_536), compression(none)),
+            };
+            let mut parser = Parser::new(row_tokens);
+            let row = parser.parse_row().unwrap();
+            let config = row.columnar.unwrap();
+            assert_eq!(config.chunk_rows, Some(65_536));
+            assert_eq!(config.compression, crate::common::model::ColumnCompression::None);
+        }
+
         #[test]
         fn test_using_rejected_on_plain_column() {
             let tokens = quote! {columns: {
diff --git a/codegen/src/common/parser/config.rs b/codegen/src/common/parser/config.rs
index df7c35ef..ec21374c 100644
--- a/codegen/src/common/parser/config.rs
+++ b/codegen/src/common/parser/config.rs
@@ -1,10 +1,11 @@
+use std::collections::HashSet;
 use std::str::FromStr;
 
 use proc_macro2::{Delimiter, TokenTree};
 use syn::spanned::Spanned;
 
 use crate::common::Parser;
-use crate::common::model::Config;
+use crate::common::model::{ColumnSlotIdType, Config};
 
 const CONFIG_FIELD_NAME: &str = "config";
 
@@ -49,11 +50,13 @@ impl Parser {
         let mut parser = Parser::new(tt);
         let mut config = Config::default();
         parser.parse_config(&mut config)?;
+        self.try_parse_comma()?;
 
         Ok(config)
     }
 
     pub fn parse_config(&mut self, config: &mut Config) -> syn::Result<Option<()>> {
+        let mut seen = HashSet::new();
         while self.peek_next().is_some() {
             let Some(_) = self.input_iter.peek() else {
                 return Ok(None);
@@ -65,9 +68,17 @@ impl Parser {
                 return Err(syn::Error::new(ident.span(), "Expected identifier."));
             };
 
+            let name_string = name.to_string();
+            if !seen.insert(name_string.clone()) {
+                return Err(syn::Error::new(
+                    name.span(),
+                    format!("Duplicate `{name_string}` config"),
+                ));
+            }
+
             self.parse_colon()?;
 
-            match name.to_string().as_str() {
+            match name_string.as_str() {
                 "page_size" => {
                     let value = self.input_iter.next().ok_or(syn::Error::new(
                         self.input.span(),
@@ -86,8 +97,53 @@ impl Parser {
 
                     config.page_size = Some(u32::from_str(value.as_str()).unwrap())
                 }
+                "columnar_slot_id" => {
+                    let value = self.input_iter.next().ok_or(syn::Error::new(
+                        self.input.span(),
+                        "Expected ColumnSlotId8, ColumnSlotId16, ColumnSlotId32, or ColumnSlotId64",
+                    ))?;
+                    let TokenTree::Ident(value) = value else {
+                        return Err(syn::Error::new(value.span(), "Expected a column slot ID type."));
+                    };
+                    config.columnar_slot_id = match value.to_string().as_str() {
+                        "ColumnSlotId8" => ColumnSlotIdType::U8,
+                        "ColumnSlotId16" => ColumnSlotIdType::U16,
+                        "ColumnSlotId32" => ColumnSlotIdType::U32,
+                        "ColumnSlotId64" => ColumnSlotIdType::U64,
+                        _ => {
+                            return Err(syn::Error::new(
+                                value.span(),
+                                "Expected ColumnSlotId8, ColumnSlotId16, ColumnSlotId32, or ColumnSlotId64",
+                            ));
+                        }
+                    };
+                    self.try_parse_comma()?;
+                }
+                "columnar_chunk_rows" => {
+                    let value = self.input_iter.next().ok_or(syn::Error::new(
+                        self.input.span(),
+                        "Expected a positive columnar chunk row count",
+                    ))?;
+                    let TokenTree::Literal(value) = value else {
+                        return Err(syn::Error::new(value.span(), "Expected an integer."));
+                    };
+                    let parsed = value
+                        .to_string()
+                        .replace('_', "")
+                        .parse::<usize>()
+                        .map_err(|_| syn::Error::new(value.span(), "Invalid columnar chunk row count"))?;
+                    if parsed == 0 {
+                        return Err(syn::Error::new(
+                            value.span(),
+                            "columnar_chunk_rows must be greater than zero",
+                        ));
+                    }
+                    config.columnar_chunk_rows = parsed;
+                    self.try_parse_comma()?;
+                }
                 "row_derives" => {
-                    const CONFIG_VARIANTS: [&str; 2] = ["page_size", "row_derives"];
+                    const CONFIG_VARIANTS: [&str; 4] =
+                        ["page_size", "row_derives", "columnar_slot_id", "columnar_chunk_rows"];
 
                     let mut derives = vec![];
 
diff --git a/codegen/src/common/parser/mod.rs b/codegen/src/common/parser/mod.rs
index e2571e63..5dfa20a3 100644
--- a/codegen/src/common/parser/mod.rs
+++ b/codegen/src/common/parser/mod.rs
@@ -1,4 +1,5 @@
 mod attribute;
+mod columnar;
 mod columns;
 mod config;
 mod index;
diff --git a/codegen/src/generators/columnar.rs b/codegen/src/generators/columnar.rs
new file mode 100644
index 00000000..2193c1c9
--- /dev/null
+++ b/codegen/src/generators/columnar.rs
@@ -0,0 +1,498 @@
+use convert_case::{Case, Casing};
+use proc_macro2::{Ident, Literal, Span, TokenStream};
+use quote::{format_ident, quote};
+
+use crate::common::model::{ColumnCompression, Columns};
+use crate::common::name_generator::{WorktableNameGenerator, is_float};
+
+fn data_ident(table: &Ident) -> Ident {
+    format_ident!("{}ColumnarData", table)
+}
+
+fn column_field(field: &Ident) -> Ident {
+    format_ident!("column_{}", field)
+}
+
+fn index_field(index: &Ident) -> Ident {
+    format_ident!("columnar_index_{}", index)
+}
+
+fn slot_id_type(columns: &Columns) -> Ident {
+    Ident::new(columns.column_slot_id.type_name(), Span::mixed_site())
+}
+
+fn compression_variant(compression: ColumnCompression) -> Ident {
+    Ident::new(
+        &compression.name().from_case(Case::Snake).to_case(Case::Pascal),
+        Span::mixed_site(),
+    )
+}
+
+fn key_type(columns: &Columns, fields: &[Ident]) -> TokenStream {
+    let fields = fields.iter().map(|field| {
+        let ty = columns.columns_map.get(field).expect("validated columnar index field");
+        if is_float(&ty.to_string()) {
+            quote! { OrderedFloat<#ty> }
+        } else {
+            quote! { #ty }
+        }
+    });
+    quote! { (#(#fields,)*) }
+}
+
+fn row_key(columns: &Columns, fields: &[Ident], row: TokenStream) -> TokenStream {
+    let fields = fields.iter().map(|field| {
+        let ty = columns.columns_map.get(field).expect("validated columnar index field");
+        if is_float(&ty.to_string()) {
+            quote! { OrderedFloat(#row.#field) }
+        } else {
+            quote! { #row.#field.clone() }
+        }
+    });
+    quote! { (#(#fields,)*) }
+}
+
+pub(crate) fn index_struct_field(table: &Ident, columns: &Columns, persisted: bool) -> TokenStream {
+    if columns.columnar_fields.is_empty() {
+        return quote! {};
+    }
+    let data = data_ident(table);
+    let skip = persisted.then(|| quote! { #[index(skip)] });
+    quote! {
+        #skip
+        columnar: ParkingRwLock<#data>
+    }
+}
+
+pub(crate) fn index_default_field(columns: &Columns) -> TokenStream {
+    if columns.columnar_fields.is_empty() {
+        quote! {}
+    } else {
+        quote! { columnar: ParkingRwLock::new(Default::default()), }
+    }
+}
+
+pub(crate) fn save_row(columns: &Columns) -> TokenStream {
+    if columns.columnar_fields.is_empty() {
+        quote! {}
+    } else {
+        quote! {
+            if let Err(bits) = self.columnar.write().save_row(&row) {
+                return Err(IndexError::ColumnSlotIdExhausted {
+                    bits,
+                    inserted_already: inserted_indexes.clone(),
+                });
+            }
+        }
+    }
+}
+
+pub(crate) fn save_row_cdc(columns: &Columns) -> TokenStream {
+    if columns.columnar_fields.is_empty() {
+        quote! {}
+    } else {
+        quote! {
+            if let Err(bits) = self.columnar.write().save_row(&row) {
+                return (partial_events, Err(IndexError::ColumnSlotIdExhausted {
+                    bits,
+                    inserted_already: inserted_indexes.clone(),
+                }));
+            }
+        }
+    }
+}
+
+pub(crate) fn reinsert_row(columns: &Columns) -> TokenStream {
+    if columns.columnar_fields.is_empty() {
+        quote! {}
+    } else {
+        quote! {
+            if let Err(bits) = self.columnar.write().replace_row(&row_old, &row_new) {
+                return Err(IndexError::ColumnSlotIdExhausted {
+                    bits,
+                    inserted_already: inserted_indexes.clone(),
+                });
+            }
+        }
+    }
+}
+
+pub(crate) fn reinsert_row_cdc(columns: &Columns) -> TokenStream {
+    if columns.columnar_fields.is_empty() {
+        quote! {}
+    } else {
+        quote! {
+            if let Err(bits) = self.columnar.write().replace_row(&row_old, &row_new) {
+                return (partial_events, Err(IndexError::ColumnSlotIdExhausted {
+                    bits,
+                    inserted_already: inserted_indexes.clone(),
+                }));
+            }
+        }
+    }
+}
+
+pub(crate) fn delete_row(columns: &Columns) -> TokenStream {
+    if columns.columnar_fields.is_empty() {
+        quote! {}
+    } else {
+        quote! { self.columnar.write().delete_row(&row); }
+    }
+}
+
+pub(crate) fn mark_dirty(columns: &Columns) -> TokenStream {
+    if columns.columnar_fields.is_empty() {
+        quote! {}
+    } else {
+        quote! { self.columnar.write().mark_dirty(); }
+    }
+}
+
+pub(crate) fn table_mark_dirty(columns: &Columns) -> TokenStream {
+    if columns.columnar_fields.is_empty() {
+        quote! {}
+    } else {
+        quote! { self.0.indexes.columnar.write().mark_dirty(); }
+    }
+}
+
+pub(crate) fn definitions(table: &Ident, columns: &Columns) -> TokenStream {
+    if columns.columnar_fields.is_empty() {
+        return quote! {};
+    }
+
+    let names = WorktableNameGenerator::from_table_name(table.to_string());
+    let row = names.get_row_type_ident();
+    let pk = names.get_primary_key_type_ident();
+    let data = data_ident(table);
+    let slot_id = slot_id_type(columns);
+
+    let column_fields = columns.columnar_fields.iter().map(|(field, _)| {
+        let storage = column_field(field);
+        let ty = columns.columns_map.get(field).expect("columnar field exists");
+        quote! { #storage: ColumnarColumn<#ty>, }
+    });
+    let column_defaults = columns.columnar_fields.iter().map(|(field, config)| {
+        let storage = column_field(field);
+        let chunk_rows = Literal::usize_unsuffixed(config.chunk_rows.expect("columnar defaults applied"));
+        let compression = compression_variant(config.compression);
+        quote! { #storage: ColumnarColumn::new(#chunk_rows, ColumnCompression::#compression), }
+    });
+    let index_fields = columns.columnar_indexes.values().map(|index| {
+        let field = index_field(&index.name);
+        let ty = key_type(columns, &index.cluster_by);
+        quote! { #field: ClusteredColumnarIndex<#ty, #slot_id>, }
+    });
+    let index_defaults = columns.columnar_indexes.values().map(|index| {
+        let field = index_field(&index.name);
+        quote! { #field: Default::default(), }
+    });
+
+    let set_columns = columns.columnar_fields.keys().map(|field| {
+        let storage = column_field(field);
+        quote! { self.#storage.set(slot_id, row.#field.clone()); }
+    });
+    let remove_columns = columns.columnar_fields.keys().map(|field| {
+        let storage = column_field(field);
+        quote! { self.#storage.remove(slot_id); }
+    });
+    let insert_indexes = columns.columnar_indexes.values().map(|index| {
+        let field = index_field(&index.name);
+        let key = row_key(columns, &index.cluster_by, quote! { row });
+        quote! { self.#field.insert(#key, slot_id); }
+    });
+    let delete_indexes = columns.columnar_indexes.values().map(|index| {
+        let field = index_field(&index.name);
+        let key = row_key(columns, &index.cluster_by, quote! { row });
+        quote! { self.#field.remove(&#key, slot_id); }
+    });
+    let replace_remove_indexes = columns.columnar_indexes.values().map(|index| {
+        let field = index_field(&index.name);
+        let key = row_key(columns, &index.cluster_by, quote! { row_old });
+        quote! { self.#field.remove(&#key, slot_id); }
+    });
+    let replace_insert_indexes = columns.columnar_indexes.values().map(|index| {
+        let field = index_field(&index.name);
+        let key = row_key(columns, &index.cluster_by, quote! { row_new });
+        quote! { self.#field.insert(#key, slot_id); }
+    });
+    let replace_columns = columns.columnar_fields.keys().map(|field| {
+        let storage = column_field(field);
+        quote! { self.#storage.set(slot_id, row_new.#field.clone()); }
+    });
+
+    quote! {
+        #[derive(Debug, MemStat)]
+        struct #data {
+            next_slot_position: Option<u64>,
+            free_slot_ids: std::collections::BTreeSet<#slot_id>,
+            slot_generations: Vec<u64>,
+            incarnation: u64,
+            slots_high_water: usize,
+            dirty: bool,
+            slots: std::collections::BTreeMap<#pk, (#slot_id, u64)>,
+            primary_keys: ColumnarColumn<#pk>,
+            #(#column_fields)*
+            #(#index_fields)*
+        }
+
+        impl Default for #data {
+            fn default() -> Self {
+                Self {
+                    next_slot_position: Some(0),
+                    free_slot_ids: Default::default(),
+                    slot_generations: Default::default(),
+                    incarnation: next_columnar_incarnation(),
+                    slots_high_water: 0,
+                    dirty: true,
+                    slots: Default::default(),
+                    primary_keys: ColumnarColumn::new(65_536, ColumnCompression::None),
+                    #(#column_defaults)*
+                    #(#index_defaults)*
+                }
+            }
+        }
+
+        impl #data {
+            fn allocate_slot(&mut self) -> Result<(#slot_id, u64), u8> {
+                if let Some(slot_id) = self.free_slot_ids.pop_first() {
+                    let generation = self.slot_generations[slot_id.slot()];
+                    return Ok((slot_id, generation));
+                }
+                let position = self
+                    .next_slot_position
+                    .ok_or(<#slot_id as ColumnSlotId>::BITS)?;
+                let slot_id = <#slot_id as ColumnSlotId>::try_from_position(position)
+                    .ok_or(<#slot_id as ColumnSlotId>::BITS)?;
+                self.next_slot_position = position.checked_add(1);
+                let slot = slot_id.slot();
+                if self.slot_generations.len() <= slot {
+                    self.slot_generations.resize(slot + 1, 0);
+                }
+                Ok((slot_id, self.slot_generations[slot]))
+            }
+
+            fn save_row(&mut self, row: &#row) -> Result<(), u8> {
+                let primary_key = row.get_primary_key();
+                let (slot_id, _) = if let Some(slot) = self.slots.get(&primary_key).copied() {
+                    slot
+                } else {
+                    let slot = self.allocate_slot()?;
+                    self.slots.insert(primary_key.clone(), slot);
+                    self.primary_keys.set(slot.0, primary_key);
+                    self.slots_high_water = self.slots_high_water.max(self.slots.len());
+                    slot
+                };
+                #(#set_columns)*
+                #(#insert_indexes)*
+                Ok(())
+            }
+
+            fn delete_row(&mut self, row: &#row) {
+                let primary_key = row.get_primary_key();
+                let Some((slot_id, generation)) = self.slots.remove(&primary_key) else {
+                    return;
+                };
+                #(#delete_indexes)*
+                #(#remove_columns)*
+                self.primary_keys.remove(slot_id);
+                if let Some(next_generation) = generation.checked_add(1) {
+                    self.slot_generations[slot_id.slot()] = next_generation;
+                    self.free_slot_ids.insert(slot_id);
+                }
+            }
+
+            fn replace_row(&mut self, row_old: &#row, row_new: &#row) -> Result<(), u8> {
+                let old_primary_key = row_old.get_primary_key();
+                let new_primary_key = row_new.get_primary_key();
+                if old_primary_key != new_primary_key {
+                    self.delete_row(row_old);
+                    return self.save_row(row_new);
+                }
+                let Some((slot_id, _)) = self.slots.get(&old_primary_key).copied() else {
+                    return self.save_row(row_new);
+                };
+                #(#replace_remove_indexes)*
+                #(#replace_columns)*
+                #(#replace_insert_indexes)*
+                Ok(())
+            }
+
+            fn row_ref(&self, slot_id: #slot_id) -> Option<ColumnarRowRef<#pk, #slot_id>> {
+                let primary_key = self.primary_keys.get(slot_id)?.clone();
+                let (current_slot, generation) = self.slots.get(&primary_key).copied()?;
+                (current_slot == slot_id).then(|| {
+                    ColumnarRowRef::__new(primary_key, slot_id, generation, self.incarnation)
+                })
+            }
+
+            fn validates(&self, row_ref: &ColumnarRowRef<#pk, #slot_id>) -> bool {
+                row_ref.__incarnation() == self.incarnation
+                    && self.slots.get(row_ref.primary_key()).copied()
+                        == Some((row_ref.__slot_id(), row_ref.__generation()))
+            }
+
+            fn mark_dirty(&mut self) {
+                self.dirty = true;
+            }
+        }
+    }
+}
+
+pub(crate) fn table_methods(table: &Ident, columns: &Columns) -> TokenStream {
+    if columns.columnar_fields.is_empty() {
+        return quote! {};
+    }
+
+    let names = WorktableNameGenerator::from_table_name(table.to_string());
+    let row = names.get_row_type_ident();
+    let pk = names.get_primary_key_type_ident();
+    let data = data_ident(table);
+    let slot_id = slot_id_type(columns);
+    let row_ref = quote! { ColumnarRowRef<#pk, #slot_id> };
+
+    let field_methods = columns.columnar_fields.iter().map(|(field, _)| {
+        let storage = column_field(field);
+        let scan = format_ident!("columnar_scan_{}", field);
+        let project = format_ident!("columnar_project_{}", field);
+        let ty = columns.columns_map.get(field).expect("columnar field exists");
+        quote! {
+            pub fn #scan(&self) -> Result<Vec<(#row_ref, #ty)>, WorkTableError> {
+                loop {
+                    self.ensure_columnar_current()?;
+                    let columnar = self.0.indexes.columnar.read();
+                    if !columnar.dirty {
+                        return Ok(columnar.#storage.iter::<#slot_id>()
+                            .filter_map(|(slot_id, value)| {
+                                columnar.row_ref(slot_id).map(|row_ref| (row_ref, value.clone()))
+                            })
+                            .collect());
+                    }
+                }
+            }
+
+            pub fn #project(&self, rows: &[#row_ref]) -> Result<Vec<(#row_ref, #ty)>, WorkTableError> {
+                loop {
+                    self.ensure_columnar_current()?;
+                    let columnar = self.0.indexes.columnar.read();
+                    if !columnar.dirty {
+                        return Ok(rows.iter().filter_map(|row_ref| {
+                            columnar.validates(row_ref).then(|| {
+                                columnar.#storage.get(row_ref.__slot_id())
+                                    .cloned()
+                                    .map(|value| (row_ref.clone(), value))
+                            }).flatten()
+                        }).collect());
+                    }
+                }
+            }
+        }
+    });
+
+    let index_methods = columns.columnar_indexes.values().map(|index| {
+        let storage = index_field(&index.name);
+        let select = format_ident!("columnar_select_{}", index.name);
+        let scan = format_ident!("columnar_scan_{}", index.name);
+        let args = index.cluster_by.iter().map(|field| {
+            let ty = columns.columns_map.get(field).expect("columnar index field exists");
+            quote! { #field: #ty }
+        });
+        let key_fields = index.cluster_by.iter().map(|field| {
+            let ty = columns.columns_map.get(field).expect("columnar index field exists");
+            if is_float(&ty.to_string()) {
+                quote! { OrderedFloat(#field) }
+            } else {
+                quote! { #field }
+            }
+        });
+        quote! {
+            pub fn #select(&self, #(#args),*) -> Result<Vec<#row_ref>, WorkTableError> {
+                let key = (#(#key_fields,)*);
+                loop {
+                    self.ensure_columnar_current()?;
+                    let columnar = self.0.indexes.columnar.read();
+                    if !columnar.dirty {
+                        return Ok(columnar.#storage.exact(&key).into_iter()
+                            .filter_map(|slot_id| columnar.row_ref(slot_id))
+                            .collect());
+                    }
+                }
+            }
+
+            pub fn #scan(&self) -> Result<Vec<#row_ref>, WorkTableError> {
+                loop {
+                    self.ensure_columnar_current()?;
+                    let columnar = self.0.indexes.columnar.read();
+                    if !columnar.dirty {
+                        return Ok(columnar.#storage.ordered_slot_ids().into_iter()
+                            .filter_map(|slot_id| columnar.row_ref(slot_id))
+                            .collect());
+                    }
+                }
+            }
+        }
+    });
+
+    quote! {
+        fn ensure_columnar_current(&self) -> Result<(), WorkTableError> {
+            // Take the writer lock before reading authoritative rows. A row
+            // mutation publishes to this same lock after changing row storage,
+            // so it either lands in this rebuild or dirties/updates the replica
+            // after the rebuild.
+            let mut columnar = self.0.indexes.columnar.write();
+            if !columnar.dirty {
+                return Ok(());
+            }
+            let rows: Vec<#row> = {
+                let read_guard = self.0.data.read_guard();
+                self.0.primary_index.pk_map.iter_values().filter_map(|(_, link)| {
+                    let _read_guard = &read_guard;
+                    self.0.data.select_non_ghosted(link.0).ok()
+                }).collect()
+            };
+            // Retain every assigned primary-key/slot pair. A concurrent
+            // reinsert may temporarily publish a ghost link while waiting for
+            // this lock; absence from this scan is not proof of deletion.
+            let mut rebuilt: #data = Default::default();
+            rebuilt.next_slot_position = columnar.next_slot_position;
+            rebuilt.free_slot_ids = std::mem::take(&mut columnar.free_slot_ids);
+            rebuilt.slot_generations = std::mem::take(&mut columnar.slot_generations);
+            rebuilt.incarnation = columnar.incarnation;
+            rebuilt.slots_high_water = columnar.slots_high_water;
+            rebuilt.slots = std::mem::take(&mut columnar.slots);
+            rebuilt.primary_keys = std::mem::replace(
+                &mut columnar.primary_keys,
+                ColumnarColumn::new(65_536, ColumnCompression::None),
+            );
+            for row in &rows {
+                rebuilt.save_row(row).map_err(WorkTableError::ColumnSlotIdExhausted)?;
+            }
+            rebuilt.dirty = false;
+            *columnar = rebuilt;
+            Ok(())
+        }
+
+        pub fn columnar_slots_in_use(&self) -> usize {
+            self.0.indexes.columnar.read().slots.len()
+        }
+
+        pub fn columnar_slots_high_water(&self) -> usize {
+            self.0.indexes.columnar.read().slots_high_water
+        }
+
+        /// Returns whether a fallback mutation has invalidated the derived
+        /// columnar replica. The next columnar read rebuilds it automatically.
+        pub fn columnar_is_dirty(&self) -> bool {
+            self.0.indexes.columnar.read().dirty
+        }
+
+        /// Rebuilds a dirty derived columnar replica at an application-chosen
+        /// point instead of charging the first later columnar reader.
+        pub fn rebuild_columnar(&self) -> Result<(), WorkTableError> {
+            self.ensure_columnar_current()
+        }
+
+        #(#field_methods)*
+        #(#index_methods)*
+    }
+}
diff --git a/codegen/src/generators/in_memory/index/mod.rs b/codegen/src/generators/in_memory/index/mod.rs
index d912985f..d2ca8b31 100644
--- a/codegen/src/generators/in_memory/index/mod.rs
+++ b/codegen/src/generators/in_memory/index/mod.rs
@@ -12,6 +12,7 @@ use quote::quote;
 impl InMemoryGenerator {
     /// Generates index type and it's impls.
     pub fn gen_index_def(&mut self) -> syn::Result<TokenStream> {
+        let columnar_def = crate::generators::columnar::definitions(&self.name, &self.columns);
         let type_def = self.gen_type_def()?;
         let impl_def = self.gen_secondary_index_impl_def();
         let info_def = self.gen_secondary_index_info_impl_def();
@@ -24,6 +25,7 @@ impl InMemoryGenerator {
         let available_indexes = self.gen_available_indexes();
 
         Ok(quote! {
+            #columnar_def
             #type_def
             #impl_def
             #info_def
@@ -94,11 +96,13 @@ impl InMemoryGenerator {
                 #[derive(Debug, MemStat)]
             }
         };
+        let columnar_field = crate::generators::columnar::index_struct_field(&self.name, &self.columns, false);
 
         Ok(quote! {
             #derive
             pub struct #ident {
-                #(#index_rows),*
+                #(#index_rows,)*
+                #columnar_field
             }
         })
     }
@@ -161,12 +165,14 @@ impl InMemoryGenerator {
                 Ok::<_, syn::Error>(res)
             })
             .collect::<Result<Vec<_>, syn::Error>>()?;
+        let columnar_field = crate::generators::columnar::index_default_field(&self.columns);
 
         Ok(quote! {
             impl Default for #index_type_ident {
                 fn default() -> Self {
                     Self {
                         #(#index_rows)*
+                        #columnar_field
                     }
                 }
             }
diff --git a/codegen/src/generators/in_memory/index/usual.rs b/codegen/src/generators/in_memory/index/usual.rs
index 2c90abc7..4b4b06ea 100644
--- a/codegen/src/generators/in_memory/index/usual.rs
+++ b/codegen/src/generators/in_memory/index/usual.rs
@@ -73,11 +73,13 @@ impl InMemoryGenerator {
                 }
             })
             .collect::<Vec<_>>();
+        let columnar_save = crate::generators::columnar::save_row(&self.columns);
 
         quote! {
             fn save_row(&self, row: #row_type_ident, link: Link) -> core::result::Result<(), IndexError<#available_index_ident>> {
                 let mut inserted_indexes: Vec<#available_index_ident> = vec![];
                 #(#save_rows)*
+                #columnar_save
                 core::result::Result::Ok(())
             }
         }
@@ -151,6 +153,7 @@ impl InMemoryGenerator {
                 (insert, remove)
             })
             .unzip();
+        let columnar_reinsert = crate::generators::columnar::reinsert_row(&self.columns);
 
         quote! {
             fn reinsert_row(&self,
@@ -163,6 +166,7 @@ impl InMemoryGenerator {
                 let mut inserted_indexes: Vec<#available_index_ident> = vec![];
                 #(#insert_rows)*
                 #(#remove_rows)*
+                #columnar_reinsert
                 core::result::Result::Ok(())
             }
         }
@@ -196,10 +200,12 @@ impl InMemoryGenerator {
                 }
             })
             .collect::<Vec<_>>();
+        let columnar_delete = crate::generators::columnar::delete_row(&self.columns);
 
         quote! {
             fn delete_row(&self, row: #row_type_ident, link: Link) -> core::result::Result<(), IndexError<#available_index_ident>> {
                 #(#delete_rows)*
+                #columnar_delete
                 core::result::Result::Ok(())
             }
         }
@@ -240,6 +246,7 @@ impl InMemoryGenerator {
                 quote! {}
             }
         });
+        let columnar_dirty = crate::generators::columnar::mark_dirty(&self.columns);
 
         quote! {
             fn process_difference_remove(
@@ -248,6 +255,7 @@ impl InMemoryGenerator {
                 difference: std::collections::HashMap<&str, Difference<#avt_type_ident>>
             ) -> core::result::Result<(), IndexError<#avt_index_ident>> {
                 #(#process_difference_remove_rows)*
+                #columnar_dirty
                 core::result::Result::Ok(())
             }
         }
@@ -299,6 +307,7 @@ impl InMemoryGenerator {
                 quote! {}
             }
         });
+        let columnar_dirty = crate::generators::columnar::mark_dirty(&self.columns);
 
         quote! {
             fn process_difference_insert(
@@ -308,6 +317,7 @@ impl InMemoryGenerator {
             ) -> core::result::Result<(), IndexError<#avt_index_ident>> {
                 let mut inserted_indexes: Vec<#avt_index_ident> = vec![];
                 #(#process_difference_insert_rows)*
+                #columnar_dirty
                 core::result::Result::Ok(())
             }
         }
diff --git a/codegen/src/generators/in_memory/queries/in_place.rs b/codegen/src/generators/in_memory/queries/in_place.rs
index c0c38720..a450992a 100644
--- a/codegen/src/generators/in_memory/queries/in_place.rs
+++ b/codegen/src/generators/in_memory/queries/in_place.rs
@@ -95,6 +95,7 @@ impl InMemoryGenerator {
             }
         };
         let custom_lock = self.gen_custom_lock_for_update(lock_ident);
+        let columnar_dirty = crate::generators::columnar::table_mark_dirty(&self.columns);
 
         quote! {
             pub async fn #method_ident<Pk, F: FnMut(#column_types)>(
@@ -124,6 +125,7 @@ impl InMemoryGenerator {
                         .map_err(WorkTableError::PagesError)?
                     };
 
+                #columnar_dirty
                 Ok(())
             }
         }
diff --git a/codegen/src/generators/in_memory/queries/update.rs b/codegen/src/generators/in_memory/queries/update.rs
index 745ab635..28d85d4a 100644
--- a/codegen/src/generators/in_memory/queries/update.rs
+++ b/codegen/src/generators/in_memory/queries/update.rs
@@ -55,6 +55,7 @@ impl InMemoryGenerator {
         let persist_call = self.gen_persist_call();
         let persist_op = self.gen_persist_op();
         let full_row_lock = self.gen_full_lock_for_update();
+        let columnar_dirty = crate::generators::columnar::table_mark_dirty(&self.columns);
         // A full-row `update(row)` replaces EVERY column, so it inherently
         // rewrites every secondary index. The in-place fast path only applies
         // when no updated field is indexed (it emits no index diff), so a
@@ -88,6 +89,7 @@ impl InMemoryGenerator {
                 };
 
                 #diff_process_remove
+                #columnar_dirty
 
                 self.0.update_state.remove(&pk);
 
@@ -104,6 +106,7 @@ impl InMemoryGenerator {
                     self.0.data.update_in_place::<{ #const_name }>(row.clone(), link).is_ok()
                 };
                 if in_place_ok {
+                    #columnar_dirty
                     self.0.update_state.remove(&pk);
                     return core::result::Result::Ok(());
                 }
@@ -524,6 +527,28 @@ impl InMemoryGenerator {
 
                                 Err(WorkTableError::AlreadyExists(at.to_string_value()))
                             }
+                            IndexError::ColumnSlotIdExhausted {
+                                bits,
+                                inserted_already,
+                            } => {
+                                let (rollback_secondary_events, _): (#secondary_events_ident, _) = self.0.indexes.delete_from_indexes_cdc(
+                                    row_new.merge(row_old.clone()),
+                                    link,
+                                    inserted_already
+                                );
+
+                                let mut merged_events = secondary_events.clone();
+                                merged_events.extend(rollback_secondary_events);
+
+                                let ack_op = Operation::Acknowledge(AcknowledgeOperation {
+                                    id: OperationId::Single(uuid::Uuid::now_v7()),
+                                    primary_key_events: vec![],
+                                    secondary_keys_events: merged_events,
+                                });
+                                self.1.apply_operation(ack_op);
+
+                                Err(WorkTableError::ColumnSlotIdExhausted(bits))
+                            }
                             IndexError::NotFound => Err(WorkTableError::NotFound),
                         };
                     }
@@ -548,6 +573,15 @@ impl InMemoryGenerator {
 
                             Err(WorkTableError::AlreadyExists(at.to_string_value()))
                         }
+                        IndexError::ColumnSlotIdExhausted {
+                            bits,
+                            inserted_already,
+                        } => {
+                            self.0.indexes
+                                .delete_from_indexes(row_new.merge(row_old.clone()), link, inserted_already)?;
+
+                            Err(WorkTableError::ColumnSlotIdExhausted(bits))
+                        }
                         IndexError::NotFound => Err(WorkTableError::NotFound),
                     };
                 }
@@ -616,6 +650,7 @@ impl InMemoryGenerator {
         let persist_call = self.gen_persist_call();
         let persist_op = self.gen_persist_op();
         let custom_lock = self.gen_custom_lock_for_update(lock_ident);
+        let columnar_dirty = crate::generators::columnar::table_mark_dirty(&self.columns);
 
         let finish_update = if archived_swap_is_safe {
             quote! {
@@ -627,6 +662,7 @@ impl InMemoryGenerator {
                 }).map_err(WorkTableError::PagesError)? };
 
                 #diff_process_remove
+                #columnar_dirty
 
                 #persist_call
 
@@ -752,6 +788,7 @@ impl InMemoryGenerator {
             }
         };
         let custom_lock = self.gen_custom_lock_for_update(lock_ident);
+        let columnar_dirty = crate::generators::columnar::table_mark_dirty(&self.columns);
 
         quote! {
             pub async fn #method_ident(&self, row: #query_ident, by: #by_ident) -> core::result::Result<(), WorkTableError> {
@@ -828,6 +865,7 @@ impl InMemoryGenerator {
 
                     guards.remove(&pk);
                 }
+                #columnar_dirty
                 core::result::Result::Ok(())
             }
         }
@@ -871,6 +909,7 @@ impl InMemoryGenerator {
             }
         };
         let custom_lock = self.gen_custom_lock_for_update(lock_ident);
+        let columnar_dirty = crate::generators::columnar::table_mark_dirty(&self.columns);
 
         let finish_update = if self.columns.is_sized {
             quote! {
@@ -884,6 +923,7 @@ impl InMemoryGenerator {
                 }
 
                 #diff_process_remove
+                #columnar_dirty
 
                 #persist_call
 
diff --git a/codegen/src/generators/in_memory/table/mod.rs b/codegen/src/generators/in_memory/table/mod.rs
index ee1765ab..458efd40 100644
--- a/codegen/src/generators/in_memory/table/mod.rs
+++ b/codegen/src/generators/in_memory/table/mod.rs
@@ -18,6 +18,8 @@ impl InMemoryGenerator {
         let index_fns = self.gen_table_index_fns()?;
         let select_query_executor_impl = self.gen_table_select_query_executor_impl();
         let column_range_type = self.gen_table_column_range_type();
+        let columnar_methods = crate::generators::columnar::table_methods(&self.name, &self.columns);
+        let table_ident = WorktableNameGenerator::from_table_name(self.name.to_string()).get_work_table_ident();
 
         Ok(quote! {
             #page_size_consts
@@ -27,6 +29,9 @@ impl InMemoryGenerator {
             #index_fns
             #select_query_executor_impl
             #column_range_type
+            impl #table_ident {
+                #columnar_methods
+            }
         })
     }
 
diff --git a/codegen/src/generators/mod.rs b/codegen/src/generators/mod.rs
index 83bc1ecb..3f376633 100644
--- a/codegen/src/generators/mod.rs
+++ b/codegen/src/generators/mod.rs
@@ -1,3 +1,4 @@
+pub(crate) mod columnar;
 pub mod in_memory;
 pub(crate) mod index_backend;
 pub mod persist;
diff --git a/codegen/src/generators/persist/index/cdc.rs b/codegen/src/generators/persist/index/cdc.rs
index 86573cfd..59f1a1a8 100644
--- a/codegen/src/generators/persist/index/cdc.rs
+++ b/codegen/src/generators/persist/index/cdc.rs
@@ -68,6 +68,7 @@ impl PersistGenerator {
             })
             .collect::<Vec<_>>();
         let idents = self.columns.indexes.values().map(|idx| &idx.name).collect::<Vec<_>>();
+        let columnar_save = crate::generators::columnar::save_row_cdc(&self.columns);
 
         quote! {
             fn save_row_cdc(&self, row: #row_type_ident, link: Link) -> (#events_ident, Result<(), IndexError<#available_index_ident>>) {
@@ -75,6 +76,7 @@ impl PersistGenerator {
                 let mut partial_events = #events_ident::default();
 
                 #(#save_rows)*
+                #columnar_save
                 (#events_ident {
                     #(#idents,)*
                 }, Ok(()))
@@ -148,6 +150,7 @@ impl PersistGenerator {
             })
             .unzip();
         let idents = self.columns.indexes.values().map(|idx| &idx.name).collect::<Vec<_>>();
+        let columnar_reinsert = crate::generators::columnar::reinsert_row_cdc(&self.columns);
 
         quote! {
             fn reinsert_row_cdc(
@@ -162,6 +165,7 @@ impl PersistGenerator {
 
                 #(#insert_rows)*
                 #(#remove_rows)*
+                #columnar_reinsert
                 (#events_ident {
                     #(#idents,)*
                 }, Ok(()))
@@ -188,10 +192,12 @@ impl PersistGenerator {
             })
             .collect::<Vec<_>>();
         let idents = self.columns.indexes.values().map(|idx| &idx.name).collect::<Vec<_>>();
+        let columnar_delete = crate::generators::columnar::delete_row(&self.columns);
 
         quote! {
             fn delete_row_cdc(&self, row: #row_type_ident, link: Link) -> (#events_ident, Result<(), IndexError<#available_index_ident>>) {
                 #(#delete_rows)*
+                #columnar_delete
                 (#events_ident {
                     #(#idents,)*
                 }, Ok(()))
@@ -306,6 +312,7 @@ impl PersistGenerator {
             }
         });
         let idents = self.columns.indexes.values().map(|idx| &idx.name).collect::<Vec<_>>();
+        let columnar_dirty = crate::generators::columnar::mark_dirty(&self.columns);
 
         quote! {
             fn process_difference_remove_cdc(
@@ -314,6 +321,7 @@ impl PersistGenerator {
                 difference: std::collections::HashMap<&str, Difference<#avt_type_ident>>
             ) -> (#events_ident, Result<(), IndexError<#available_index_ident>>) {
                 #(#process_difference_rows)*
+                #columnar_dirty
                 (#events_ident {
                     #(#idents,)*
                 }, Ok(()))
@@ -376,6 +384,7 @@ impl PersistGenerator {
             }
         });
         let idents = self.columns.indexes.values().map(|idx| &idx.name).collect::<Vec<_>>();
+        let columnar_dirty = crate::generators::columnar::mark_dirty(&self.columns);
 
         quote! {
             fn process_difference_insert_cdc(
@@ -387,6 +396,7 @@ impl PersistGenerator {
                 let mut partial_events = #events_ident::default();
 
                 #(#process_difference_insert_rows)*
+                #columnar_dirty
                 (#events_ident {
                     #(#idents,)*
                 }, Ok(()))
diff --git a/codegen/src/generators/persist/index/mod.rs b/codegen/src/generators/persist/index/mod.rs
index 476fe724..1633c908 100644
--- a/codegen/src/generators/persist/index/mod.rs
+++ b/codegen/src/generators/persist/index/mod.rs
@@ -11,6 +11,7 @@ use quote::quote;
 
 impl PersistGenerator {
     pub fn gen_index_def(&mut self) -> syn::Result<TokenStream> {
+        let columnar_def = crate::generators::columnar::definitions(&self.name, &self.columns);
         let type_def = self.gen_type_def()?;
         let impl_def = self.gen_secondary_index_impl_def();
         let info_def = self.gen_secondary_index_info_impl_def();
@@ -19,6 +20,7 @@ impl PersistGenerator {
         let available_indexes = self.gen_available_indexes();
 
         Ok(quote! {
+            #columnar_def
             #type_def
             #impl_def
             #info_def
@@ -73,11 +75,13 @@ impl PersistGenerator {
         let derive = quote! {
             #[derive(Debug, MemStat, PersistIndex)]
         };
+        let columnar_field = crate::generators::columnar::index_struct_field(&self.name, &self.columns, true);
 
         Ok(quote! {
             #derive
             pub struct #ident {
-                #(#index_rows),*
+                #(#index_rows,)*
+                #columnar_field
             }
         })
     }
@@ -145,12 +149,14 @@ impl PersistGenerator {
                 Ok::<_, syn::Error>(res)
             })
             .collect::<Result<Vec<_>, syn::Error>>()?;
+        let columnar_field = crate::generators::columnar::index_default_field(&self.columns);
 
         Ok(quote! {
             impl Default for #index_type_ident {
                 fn default() -> Self {
                     Self {
                         #(#index_rows)*
+                        #columnar_field
                     }
                 }
             }
diff --git a/codegen/src/generators/persist/index/usual.rs b/codegen/src/generators/persist/index/usual.rs
index e8629cc8..23e0929b 100644
--- a/codegen/src/generators/persist/index/usual.rs
+++ b/codegen/src/generators/persist/index/usual.rs
@@ -69,11 +69,13 @@ impl PersistGenerator {
                 }
             })
             .collect::<Vec<_>>();
+        let columnar_save = crate::generators::columnar::save_row(&self.columns);
 
         quote! {
             fn save_row(&self, row: #row_type_ident, link: Link) -> core::result::Result<(), IndexError<#available_index_ident>> {
                 let mut inserted_indexes: Vec<#available_index_ident> = vec![];
                 #(#save_rows)*
+                #columnar_save
                 core::result::Result::Ok(())
             }
         }
@@ -147,6 +149,7 @@ impl PersistGenerator {
                 (insert, remove)
             })
             .unzip();
+        let columnar_reinsert = crate::generators::columnar::reinsert_row(&self.columns);
 
         quote! {
             fn reinsert_row(&self,
@@ -159,6 +162,7 @@ impl PersistGenerator {
                 let mut inserted_indexes: Vec<#available_index_ident> = vec![];
                 #(#insert_rows)*
                 #(#remove_rows)*
+                #columnar_reinsert
                 core::result::Result::Ok(())
             }
         }
@@ -189,10 +193,12 @@ impl PersistGenerator {
                 }
             })
             .collect::<Vec<_>>();
+        let columnar_delete = crate::generators::columnar::delete_row(&self.columns);
 
         quote! {
             fn delete_row(&self, row: #row_type_ident, link: Link) -> core::result::Result<(), IndexError<#available_index_ident>> {
                 #(#delete_rows)*
+                #columnar_delete
                 core::result::Result::Ok(())
             }
         }
@@ -231,6 +237,7 @@ impl PersistGenerator {
                 quote! {}
             }
         });
+        let columnar_dirty = crate::generators::columnar::mark_dirty(&self.columns);
 
         quote! {
             fn process_difference_remove(
@@ -239,6 +246,7 @@ impl PersistGenerator {
                 difference: std::collections::HashMap<&str, Difference<#avt_type_ident>>
             ) -> core::result::Result<(), IndexError<#avt_index_ident>> {
                 #(#process_difference_remove_rows)*
+                #columnar_dirty
                 core::result::Result::Ok(())
             }
         }
@@ -288,6 +296,7 @@ impl PersistGenerator {
                 quote! {}
             }
         });
+        let columnar_dirty = crate::generators::columnar::mark_dirty(&self.columns);
 
         quote! {
             fn process_difference_insert(
@@ -297,6 +306,7 @@ impl PersistGenerator {
             ) -> core::result::Result<(), IndexError<#avt_index_ident>> {
                 let mut inserted_indexes: Vec<#avt_index_ident> = vec![];
                 #(#process_difference_insert_rows)*
+                #columnar_dirty
                 core::result::Result::Ok(())
             }
         }
diff --git a/codegen/src/generators/persist/queries/in_place.rs b/codegen/src/generators/persist/queries/in_place.rs
index fcc066e8..0dd5a280 100644
--- a/codegen/src/generators/persist/queries/in_place.rs
+++ b/codegen/src/generators/persist/queries/in_place.rs
@@ -97,6 +97,7 @@ impl PersistGenerator {
             }
         };
         let custom_lock = self.gen_custom_lock_for_update(lock_ident);
+        let columnar_dirty = crate::generators::columnar::table_mark_dirty(&self.columns);
 
         quote! {
             pub async fn #method_ident<Pk, F: FnMut(#column_types)>(
@@ -126,6 +127,7 @@ impl PersistGenerator {
                         .map_err(WorkTableError::PagesError)?
                     };
 
+                #columnar_dirty
                 Ok(())
             }
         }
diff --git a/codegen/src/generators/persist/queries/update.rs b/codegen/src/generators/persist/queries/update.rs
index dbe0dfcb..00bb3b88 100644
--- a/codegen/src/generators/persist/queries/update.rs
+++ b/codegen/src/generators/persist/queries/update.rs
@@ -55,6 +55,7 @@ impl PersistGenerator {
         let persist_call = self.gen_persist_call();
         let persist_op = self.gen_persist_op();
         let full_row_lock = self.gen_full_lock_for_update();
+        let columnar_dirty = crate::generators::columnar::table_mark_dirty(&self.columns);
         let size_check = if self.columns.is_sized {
             quote! {}
         } else {
@@ -126,6 +127,7 @@ impl PersistGenerator {
                 }).map_err(WorkTableError::PagesError)? };
 
                 #diff_process_remove
+                #columnar_dirty
 
                 self.0.update_state.remove(&pk);
 
@@ -356,6 +358,28 @@ impl PersistGenerator {
 
                                 Err(WorkTableError::AlreadyExists(at.to_string_value()))
                             }
+                            IndexError::ColumnSlotIdExhausted {
+                                bits,
+                                inserted_already,
+                            } => {
+                                let (rollback_secondary_events, _): (#secondary_events_ident, _) = self.0.indexes.delete_from_indexes_cdc(
+                                    row_new.merge(row_old.clone()),
+                                    link,
+                                    inserted_already
+                                );
+
+                                let mut merged_events = secondary_events.clone();
+                                merged_events.extend(rollback_secondary_events);
+
+                                let ack_op = Operation::Acknowledge(AcknowledgeOperation {
+                                    id: OperationId::Single(uuid::Uuid::now_v7()),
+                                    primary_key_events: vec![],
+                                    secondary_keys_events: merged_events,
+                                });
+                                self.1.apply_operation(ack_op)?;
+
+                                Err(WorkTableError::ColumnSlotIdExhausted(bits))
+                            }
                             IndexError::NotFound => Err(WorkTableError::NotFound),
                         };
                     }
@@ -415,6 +439,7 @@ impl PersistGenerator {
         let persist_call = self.gen_persist_call();
         let persist_op = self.gen_persist_op();
         let custom_lock = self.gen_custom_lock_for_update(lock_ident);
+        let columnar_dirty = crate::generators::columnar::table_mark_dirty(&self.columns);
 
         quote! {
             pub async fn #method_ident<Pk>(&self, row: #query_ident, pk: Pk) -> core::result::Result<(), WorkTableError>
@@ -448,6 +473,7 @@ impl PersistGenerator {
                 }).map_err(WorkTableError::PagesError)? };
 
                 #diff_process_remove
+                #columnar_dirty
 
                 #persist_call
 
@@ -544,6 +570,7 @@ impl PersistGenerator {
             }
         };
         let custom_lock = self.gen_custom_lock_for_update(lock_ident);
+        let columnar_dirty = crate::generators::columnar::table_mark_dirty(&self.columns);
 
         quote! {
             pub async fn #method_ident(&self, row: #query_ident, by: #by_ident) -> core::result::Result<(), WorkTableError> {
@@ -621,6 +648,7 @@ impl PersistGenerator {
 
                     guards.remove(&pk);
                 }
+                #columnar_dirty
                 core::result::Result::Ok(())
             }
         }
@@ -664,6 +692,7 @@ impl PersistGenerator {
             }
         };
         let custom_lock = self.gen_custom_lock_for_update(lock_ident);
+        let columnar_dirty = crate::generators::columnar::table_mark_dirty(&self.columns);
 
         quote! {
             pub async fn #method_ident(&self, row: #query_ident, by: #by_ident) -> core::result::Result<(), WorkTableError> {
@@ -718,6 +747,7 @@ impl PersistGenerator {
                 }
 
                 #diff_process_remove
+                #columnar_dirty
 
                 #persist_call
 
diff --git a/codegen/src/generators/persist/table/mod.rs b/codegen/src/generators/persist/table/mod.rs
index 1d339ed3..7aefec4a 100644
--- a/codegen/src/generators/persist/table/mod.rs
+++ b/codegen/src/generators/persist/table/mod.rs
@@ -18,6 +18,8 @@ impl PersistGenerator {
         let index_fns = self.gen_table_index_fns()?;
         let select_query_executor_impl = self.gen_table_select_query_executor_impl();
         let column_range_type = self.gen_table_column_range_type();
+        let columnar_methods = crate::generators::columnar::table_methods(&self.name, &self.columns);
+        let table_ident = WorktableNameGenerator::from_table_name(self.name.to_string()).get_work_table_ident();
 
         Ok(quote! {
             #page_size_consts
@@ -27,6 +29,9 @@ impl PersistGenerator {
             #index_fns
             #select_query_executor_impl
             #column_range_type
+            impl #table_ident {
+                #columnar_methods
+            }
         })
     }
 
diff --git a/codegen/src/generators/read_only/index/mod.rs b/codegen/src/generators/read_only/index/mod.rs
index 048c679e..d8c4eac5 100644
--- a/codegen/src/generators/read_only/index/mod.rs
+++ b/codegen/src/generators/read_only/index/mod.rs
@@ -10,6 +10,7 @@ use quote::quote;
 
 impl ReadOnlyGenerator {
     pub fn gen_index_def(&mut self) -> syn::Result<TokenStream> {
+        let columnar_def = crate::generators::columnar::definitions(&self.name, &self.columns);
         let type_def = self.gen_type_def()?;
         let impl_def = self.gen_secondary_index_impl_def();
         let info_def = self.gen_secondary_index_info_impl_def();
@@ -17,6 +18,7 @@ impl ReadOnlyGenerator {
         let available_indexes = self.gen_available_indexes();
 
         Ok(quote! {
+            #columnar_def
             #type_def
             #impl_def
             #info_def
@@ -71,11 +73,13 @@ impl ReadOnlyGenerator {
             #[derive(Debug, MemStat, PersistIndex)]
             #[index(read_only)]
         };
+        let columnar_field = crate::generators::columnar::index_struct_field(&self.name, &self.columns, true);
 
         Ok(quote! {
             #derive
             pub struct #ident {
-                #(#index_rows),*
+                #(#index_rows,)*
+                #columnar_field
             }
         })
     }
@@ -138,12 +142,14 @@ impl ReadOnlyGenerator {
                 Ok::<_, syn::Error>(res)
             })
             .collect::<Result<Vec<_>, syn::Error>>()?;
+        let columnar_field = crate::generators::columnar::index_default_field(&self.columns);
 
         Ok(quote! {
             impl Default for #index_type_ident {
                 fn default() -> Self {
                     Self {
                         #(#index_rows)*
+                        #columnar_field
                     }
                 }
             }
diff --git a/codegen/src/generators/read_only/index/usual.rs b/codegen/src/generators/read_only/index/usual.rs
index 0af30608..6fa0b940 100644
--- a/codegen/src/generators/read_only/index/usual.rs
+++ b/codegen/src/generators/read_only/index/usual.rs
@@ -69,11 +69,13 @@ impl ReadOnlyGenerator {
                 }
             })
             .collect::<Vec<_>>();
+        let columnar_save = crate::generators::columnar::save_row(&self.columns);
 
         quote! {
             fn save_row(&self, row: #row_type_ident, link: Link) -> core::result::Result<(), IndexError<#available_index_ident>> {
                 let mut inserted_indexes: Vec<#available_index_ident> = vec![];
                 #(#save_rows)*
+                #columnar_save
                 core::result::Result::Ok(())
             }
         }
@@ -147,6 +149,7 @@ impl ReadOnlyGenerator {
                 (insert, remove)
             })
             .unzip();
+        let columnar_reinsert = crate::generators::columnar::reinsert_row(&self.columns);
 
         quote! {
             fn reinsert_row(&self,
@@ -159,6 +162,7 @@ impl ReadOnlyGenerator {
                 let mut inserted_indexes: Vec<#available_index_ident> = vec![];
                 #(#insert_rows)*
                 #(#remove_rows)*
+                #columnar_reinsert
                 core::result::Result::Ok(())
             }
         }
@@ -189,10 +193,12 @@ impl ReadOnlyGenerator {
                 }
             })
             .collect::<Vec<_>>();
+        let columnar_delete = crate::generators::columnar::delete_row(&self.columns);
 
         quote! {
             fn delete_row(&self, row: #row_type_ident, link: Link) -> core::result::Result<(), IndexError<#available_index_ident>> {
                 #(#delete_rows)*
+                #columnar_delete
                 core::result::Result::Ok(())
             }
         }
diff --git a/codegen/src/generators/read_only/table/mod.rs b/codegen/src/generators/read_only/table/mod.rs
index 72bc8c0d..ccfdcfe2 100644
--- a/codegen/src/generators/read_only/table/mod.rs
+++ b/codegen/src/generators/read_only/table/mod.rs
@@ -18,6 +18,8 @@ impl ReadOnlyGenerator {
         let index_fns = self.gen_table_index_fns()?;
         let select_query_executor_impl = self.gen_table_select_query_executor_impl();
         let column_range_type = self.gen_table_column_range_type();
+        let columnar_methods = crate::generators::columnar::table_methods(&self.name, &self.columns);
+        let table_ident = WorktableNameGenerator::from_table_name(self.name.to_string()).get_work_table_ident();
 
         Ok(quote! {
             #page_size_consts
@@ -27,6 +29,9 @@ impl ReadOnlyGenerator {
             #index_fns
             #select_query_executor_impl
             #column_range_type
+            impl #table_ident {
+                #columnar_methods
+            }
         })
     }
 
diff --git a/codegen/src/persist_index/generator.rs b/codegen/src/persist_index/generator.rs
index 9c536283..084fcffb 100644
--- a/codegen/src/persist_index/generator.rs
+++ b/codegen/src/persist_index/generator.rs
@@ -17,6 +17,7 @@ pub struct Generator {
     pub struct_def: ItemStruct,
     pub field_types: HashMap<Ident, TokenStream>,
     pub attributes: PersistIndexAttributes,
+    pub skipped_fields: Vec<Ident>,
 }
 
 pub(super) struct IndexLayout {
@@ -87,11 +88,29 @@ impl WorktableNameGenerator {
 }
 
 impl Generator {
-    pub fn with_attributes(struct_def: ItemStruct, attributes: PersistIndexAttributes) -> Self {
+    pub fn with_attributes(mut struct_def: ItemStruct, attributes: PersistIndexAttributes) -> Self {
         let mut fields = vec![];
         let mut types = vec![];
+        let mut skipped_fields = vec![];
 
         for field in &struct_def.fields {
+            let skipped = field.attrs.iter().any(|attribute| {
+                if !attribute.path().is_ident("index") {
+                    return false;
+                }
+                let mut skipped = false;
+                let _ = attribute.parse_nested_meta(|meta| {
+                    if meta.path.is_ident("skip") {
+                        skipped = true;
+                    }
+                    Ok(())
+                });
+                skipped
+            });
+            if skipped {
+                skipped_fields.push(field.ident.clone().expect("index fields should always be named fields"));
+                continue;
+            }
             fields.push(field.ident.clone().expect("index fields should always be named fields"));
 
             let syn::Type::Path(type_path) = &field.ty else {
@@ -119,12 +138,21 @@ impl Generator {
 
             types.push(ty.to_token_stream());
         }
+        if let syn::Fields::Named(named) = &mut struct_def.fields {
+            named.named = named
+                .named
+                .iter()
+                .filter(|field| !skipped_fields.iter().any(|ident| field.ident.as_ref() == Some(ident)))
+                .cloned()
+                .collect();
+        }
         let map = fields.into_iter().zip(types).collect::<HashMap<_, _>>();
 
         Self {
             struct_def,
             field_types: map,
             attributes,
+            skipped_fields,
         }
     }
 
@@ -577,6 +605,7 @@ impl Generator {
                 }
             })
             .collect::<syn::Result<Vec<_>>>()?;
+        let skipped_fields = &self.skipped_fields;
 
         Ok(quote! {
             fn from_persisted(persisted: Self::PersistedIndex) -> Self {
@@ -584,6 +613,7 @@ impl Generator {
 
                 Self {
                     #(#idents,)*
+                    #(#skipped_fields: Default::default(),)*
                 }
             }
         })
diff --git a/codegen/src/persist_index/mod.rs b/codegen/src/persist_index/mod.rs
index ef17c461..6063d218 100644
--- a/codegen/src/persist_index/mod.rs
+++ b/codegen/src/persist_index/mod.rs
@@ -83,4 +83,26 @@ mod tests {
             "read_only index should have from_persisted method"
         );
     }
+
+    #[test]
+    fn skipped_derived_field_is_not_part_of_persisted_index_format() {
+        let input = quote! {
+            #[derive(Debug, Default)]
+            pub struct DerivedIndex {
+                durable: TreeIndex<i64, Link>,
+                #[index(skip)]
+                columnar: ParkingRwLock<DerivedColumnarData>,
+            }
+        };
+
+        let output = expand(input).unwrap().to_string();
+        let persisted_type = output
+            .split("struct DerivedIndexPersisted")
+            .nth(1)
+            .expect("persisted index type");
+        let persisted_fields = persisted_type.split('}').next().unwrap();
+        assert!(persisted_fields.contains("durable"));
+        assert!(!persisted_fields.contains("columnar"));
+        assert!(output.contains("columnar : Default :: default"));
+    }
 }
diff --git a/codegen/src/worktable/mod.rs b/codegen/src/worktable/mod.rs
index f4567674..0313e5c7 100644
--- a/codegen/src/worktable/mod.rs
+++ b/codegen/src/worktable/mod.rs
@@ -8,6 +8,7 @@ pub fn expand(input: TokenStream) -> syn::Result<TokenStream> {
     let mut columns = None;
     let mut queries = None;
     let mut indexes = None;
+    let mut columnar_indexes = None;
     let mut config = None;
 
     let name = parser.parse_name()?;
@@ -23,6 +24,10 @@ pub fn expand(input: TokenStream) -> syn::Result<TokenStream> {
                 let res = parser.parse_indexes()?;
                 indexes = Some(res);
             }
+            "columnar_indexes" => {
+                let res = parser.parse_columnar_indexes()?;
+                columnar_indexes = Some(res);
+            }
             "queries" => {
                 let res = parser.parse_queries()?;
                 queries = Some(res)
@@ -51,8 +56,39 @@ pub fn expand(input: TokenStream) -> syn::Result<TokenStream> {
     if let Some(i) = indexes {
         columns.indexes = i
     }
+    if let Some(i) = columnar_indexes {
+        columns.columnar_indexes = i.indexes;
+    }
+
+    let columnar_chunk_rows = config
+        .as_ref()
+        .map(|config| config.columnar_chunk_rows)
+        .unwrap_or(crate::common::model::DEFAULT_COLUMNAR_CHUNK_ROWS);
+    columns.column_slot_id = config
+        .as_ref()
+        .map(|config| config.columnar_slot_id)
+        .unwrap_or_default();
+    for field in columns.columnar_fields.values_mut() {
+        let chunk_rows = field.chunk_rows.unwrap_or(columnar_chunk_rows);
+        let (smaller, larger) = if chunk_rows <= columnar_chunk_rows {
+            (chunk_rows, columnar_chunk_rows)
+        } else {
+            (columnar_chunk_rows, chunk_rows)
+        };
+        let nested = larger % smaller == 0 && (larger / smaller).is_power_of_two();
+        if !nested {
+            return Err(syn::Error::new(
+                proc_macro2::Span::call_site(),
+                format!(
+                    "columnar chunk_rows({chunk_rows}) must be a power-of-two multiple or divisor of config.columnar_chunk_rows ({columnar_chunk_rows})"
+                ),
+            ));
+        }
+        field.chunk_rows = Some(chunk_rows);
+    }
 
     validate_index_backends(&columns, persistence)?;
+    validate_columnar_indexes(&columns)?;
 
     if persistence.is_persisted() {
         crate::generators::persist::expand(name, columns, queries, config, version)
@@ -61,6 +97,54 @@ pub fn expand(input: TokenStream) -> syn::Result<TokenStream> {
     }
 }
 
+fn validate_columnar_indexes(columns: &Columns) -> syn::Result<()> {
+    for primary_key in &columns.primary_keys {
+        if columns.columnar_fields.contains_key(primary_key) {
+            return Err(syn::Error::new(
+                primary_key.span(),
+                "the primary key participates in columnar identity implicitly and must not declare `columnar`",
+            ));
+        }
+    }
+
+    if !columns.columnar_indexes.is_empty() && columns.columnar_fields.is_empty() {
+        return Err(syn::Error::new(
+            proc_macro2::Span::call_site(),
+            "`columnar_indexes` requires at least one field declaring `columnar`",
+        ));
+    }
+
+    for index in columns.columnar_indexes.values() {
+        if columns.columnar_fields.contains_key(&index.name) {
+            return Err(syn::Error::new(
+                index.name.span(),
+                format!(
+                    "columnar index `{}` conflicts with a columnar field name and would generate duplicate scan methods",
+                    index.name
+                ),
+            ));
+        }
+        for field in &index.cluster_by {
+            if !columns.columns_map.contains_key(field) {
+                return Err(syn::Error::new(
+                    field.span(),
+                    format!("columnar index `{}` references unknown field `{field}`", index.name),
+                ));
+            }
+            if !columns.columnar_fields.contains_key(field) {
+                return Err(syn::Error::new(
+                    field.span(),
+                    format!(
+                        "columnar index `{}` requires field `{field}` to declare `columnar(...)`",
+                        index.name
+                    ),
+                ));
+            }
+        }
+    }
+    Ok(())
+}
+
 fn validate_index_backends(columns: &Columns, persistence: Persistence) -> syn::Result<()> {
     let explicit_backend = if columns.primary_index_backend.requires_explicit_persistence() {
         Some((
@@ -160,6 +244,125 @@ mod tests {
         assert!(error.to_string().contains("keep `primary_key`"));
     }
 
+    #[test]
+    fn columnar_index_requires_columnar_fields() {
+        let error = expand(quote! {
+            name: InvalidColumnarIndex,
+            persist: false,
+            columns: {
+                id: u64 primary_key,
+                host_id: u64,
+            },
+            columnar_indexes: {
+                host_lookup: {
+                    cluster_by: [host_id],
+                },
+            },
+        })
+        .unwrap_err();
+
+        assert!(
+            error
+                .to_string()
+                .contains("requires at least one field declaring `columnar`")
+        );
+    }
+
+    #[test]
+    fn columnar_field_and_index_generate_scan_projection_and_lookup_apis() {
+        let output = expand(quote! {
+            name: ColumnarCodegen,
+            persist: false,
+            columns: {
+                id: u64 primary_key,
+                host_id: u64 columnar(chunk_rows(1024), compression(none)),
+                timestamp: i64 columnar(chunk_rows(2048), compression(none)),
+            },
+            columnar_indexes: {
+                host_time: {
+                    cluster_by: [host_id, timestamp],
+                },
+            },
+        })
+        .unwrap()
+        .to_string();
+
+        assert!(output.contains("columnar_scan_host_id"));
+        assert!(output.contains("columnar_project_timestamp"));
+        assert!(output.contains("columnar_select_host_time"));
+        assert!(output.contains("ColumnarColumn :: new (1024"));
+    }
+
+    #[test]
+    fn columnar_config_is_table_scoped_and_row_derives_stops_at_new_keys() {
+        let output = expand(quote! {
+            name: ColumnarConfig,
+            persist: false,
+            columns: {
+                id: u64 primary_key,
+                value: u64 columnar,
+            },
+            config: {
+                row_derives: Default,
+                columnar_slot_id: ColumnSlotId16,
+                columnar_chunk_rows: 1024,
+            },
+        })
+        .unwrap()
+        .to_string();
+
+        assert!(output.contains("ColumnSlotId16"));
+        assert!(output.contains("ColumnarColumn :: new (1024"));
+    }
+
+    #[test]
+    fn columnar_chunk_override_must_nest_with_table_default() {
+        let error = expand(quote! {
+            name: InvalidColumnarChunk,
+            persist: false,
+            columns: {
+                id: u64 primary_key,
+                value: u64 columnar(chunk_rows(50_000)),
+            },
+        })
+        .unwrap_err();
+
+        assert!(error.to_string().contains("power-of-two multiple or divisor"));
+    }
+
+    #[test]
+    fn primary_key_cannot_redeclare_columnar_identity() {
+        let error = expand(quote! {
+            name: InvalidColumnarPrimaryKey,
+            persist: false,
+            columns: {
+                id: u64 primary_key columnar,
+            },
+        })
+        .unwrap_err();
+
+        assert!(error.to_string().contains("must not declare `columnar`"));
+    }
+
+    #[test]
+    fn duplicate_columnar_config_is_rejected() {
+        let error = expand(quote! {
+            name: DuplicateColumnarConfig,
+            persist: false,
+            columns: {
+                id: u64 primary_key,
+                value: u64 columnar,
+            },
+            config: {
+                columnar_slot_id: ColumnSlotId16,
+                columnar_slot_id: ColumnSlotId32,
+            },
+        })
+        .unwrap_err();
+
+        assert!(error.to_string().contains("Duplicate `columnar_slot_id`"));
+    }
+
     fn assert_composite_primary_key_field_order(output: proc_macro2::TokenStream) {
         let output = output.to_string();
         let get_primary_key = output
diff --git a/codegen/src/worktable_version/mod.rs b/codegen/src/worktable_version/mod.rs
index ea7c1ed2..abe55ad9 100644
--- a/codegen/src/worktable_version/mod.rs
+++ b/codegen/src/worktable_version/mod.rs
@@ -16,6 +16,12 @@ pub fn expand(input: TokenStream) -> syn::Result<TokenStream> {
         match ident.to_string().as_str() {
             "columns" => columns = Some(parser.parse_columns()?),
             "indexes" => indexes = Some(parser.parse_indexes()?),
+            "columnar_indexes" => {
+                return Err(Error::new(
+                    ident.span(),
+                    "worktable_version! does not support columnar_indexes",
+                ));
+            }
             "queries" => {
                 return Err(Error::new(ident.span(), "worktable_version! does not support queries"));
             }
@@ -120,6 +126,25 @@ mod tests {
         assert!(res.is_err(), "should reject config section");
     }
 
+    #[test]
+    fn test_rejects_columnar_indexes_explicitly() {
+        let input = quote! {
+            name: UserV1,
+            columns: {
+                id: u64 primary_key,
+                value: u64,
+            },
+            columnar_indexes: {
+                value_idx: {
+                    cluster_by: [value],
+                },
+            },
+        };
+
+        let error = expand(input).unwrap_err();
+        assert!(error.to_string().contains("does not support columnar_indexes"));
+    }
+
     #[test]
     fn test_explicit_version() {
         let input = quote! {
diff --git a/docs/columnar-fields-and-indexes-guide-v3.md b/docs/columnar-fields-and-indexes-guide-v3.md
new file mode 100644
index 00000000..7fb66481
--- /dev/null
+++ b/docs/columnar-fields-and-indexes-guide-v3.md
@@ -0,0 +1,404 @@
+# WorkTable tabular + columnar side indexes
+
+> **v3 review boundary:** this guide describes the implemented tabular + columnar side-index
+> flavor, and labels the separate full-columnar roadmap explicitly.
+
+WorkTable can maintain selected fields in derived, chunked **side indexes** while its ordinary row
+store and primary key remain authoritative. Point-oriented application code keeps the existing
+WorkTable model, while analytical code gains cheaper columnar-flavored scans and projections over
+the duplicated selected values.
+
+The first implementation is intentionally format-compatible and uncompressed. It establishes the
+DSL, identity, mutation, validation, and recovery boundaries before adding sealed chunks,
+compression, a vector execution engine, or a new disk format.
+
+## The three WorkTable storage flavors
+
+| Flavor | Authoritative representation | What it provides | Status |
+|---|---|---|---|
+| **Tabular** | Existing WorkTable rows | Point lookup, mutation, ordinary indexes and persistence | Already well covered |
+| **Tabular + columnar side indexes** | Existing WorkTable rows, plus derived side structures | Cheap columnar-flavored field scans, projections and ordered lookup at the cost of duplicating selected values/index metadata | **This proposal and PR** |
+| **Columnar** | A true vector/column representation | Vector batches, encoded or compressed segments, columnar-native execution and persistence | Not covered |
+
+The middle flavor is not a full column store. Calling it one would erase the most useful property
+of the proposal: it adds a bounded, opt-in analytical acceleration structure without replacing the
+tabular engine. It is analogous to adding an index, except that a useful side index must duplicate
+the selected field values as well as ordering metadata. If it stored only keys and slot positions,
+every projection would still perform random gathers from tabular rows and lose most of its
+columnar flavor.
+
+## Complete schema
+
+```rust
+use worktable::prelude::*;
+use worktable::worktable;
+
+// WorkTable's current column parser accepts a single type identifier.
+type DiagnosticBlob = Vec<u8>;
+
+worktable!(
+    name: HistoricalCpu,
+    persist: true,
+
+    columns: {
+        id: u128 primary_key,
+
+        // Bare `columnar` uses the table defaults.
+        host_id: u64 columnar,
+        captured_at_ns: u64 columnar,
+        cpu_percent: f32 columnar,
+
+        // A non-indexed columnar field. It has contiguous field storage and
+        // can be scanned or projected even though no index clusters by it.
+        status: String columnar(chunk_rows(16_384)),
+
+        // An ordinary row-only field.
+        diagnostic_blob: DiagnosticBlob,
+    },
+
+    columnar_indexes: {
+        host_time: {
+            cluster_by: [host_id, captured_at_ns],
+        },
+    },
+
+    config: {
+        columnar_slot_id: ColumnSlotId32,
+        columnar_chunk_rows: 65_536,
+    },
+);
+```
+
+Both columnar `config` entries shown above are defaults and may be omitted.
+
+There is deliberately no table-level `layout: columnar`: this proposal does not select the third
+flavor. A field opts into a derived side index. There is also deliberately no `columns: [...]` list inside a columnar index: fields named
+by `cluster_by` are its key, and projected values come from base column stores.
+
+## Three independent choices
+
+### 1. Which fields have column storage?
+
+Add `columnar` to a non-primary-key field:
+
+```rust
+cpu_percent: f32 columnar,
+status: String columnar(chunk_rows(16_384)),
+latency_ns: u64 optional columnar(compression(none)),
+```
+
+This duplicates that field in a chunked derived side index. It does not create an ordered search index and it
+does not change row persistence. A field may be columnar without appearing in any
+`columnar_indexes` entry; `status` in the complete example is one such field.
+
+`columnar` occurs after `optional` and before `using`. The table default is 65,536 rows per chunk.
+A per-field `chunk_rows(N)` override must be a power-of-two multiple or divisor of the table
+default, which keeps cross-column chunk boundaries nestable.
+
+Mutable chunks are currently unencoded. Omitted compression and `compression(none)` mean the same
+thing. `auto`, `delta`, `rle`, and `dictionary` are reserved and fail macro expansion instead of
+silently behaving like `none`.
+
+### 2. Which access paths are ordered?
+
+```rust
+columnar_indexes: {
+    host_time: {
+        cluster_by: [host_id, captured_at_ns],
+    },
+},
+```
+
+`cluster_by` orders the index metadata, not the physical base columns. In the current
+implementation, the access path is a `BTreeMap` from the composite key to a set of column slot IDs.
+The base columns remain in canonical slot order.
+
+Every `cluster_by` field must itself declare `columnar`. `include` is reserved for a future genuine
+covering projection and is rejected today.
+
+### 3. How wide is the compact slot position?
+
+The table-wide setting is:
+
+```rust
+config: {
+    columnar_slot_id: ColumnSlotId16,
+},
+```
+
+Available types and theoretical live-slot capacities are:
+
+| Type | Positions | Typical reason to choose it |
+|---|---:|---|
+| `ColumnSlotId8` | 256 | tests or a strictly bounded tiny table |
+| `ColumnSlotId16` | 65,536 | embedded or hard-bounded live window |
+| `ColumnSlotId32` | 4,294,967,296 | default; broad capacity with compact metadata |
+| `ColumnSlotId64` | 18,446,744,073,709,551,616 | explicit very-large logical range |
+
+These are capacity bounds, not promises that the process can allocate that many rows. Address
+space, memory, and other table structures impose practical limits first—especially for 64-bit
+slots.
+
+Choosing a width that covers the maximum number of simultaneously live columnar rows is the
+schema author's responsibility. WorkTable does not silently widen, truncate, wrap, evict another
+row, or reinterpret the setting. A write beyond the selected range returns:
+
+```rust
+WorkTableError::ColumnSlotIdExhausted(bits)
+```
+
+The failed insert is rolled back and existing rows remain valid. Operators can monitor:
+
+```rust
+table.columnar_slots_in_use();
+table.columnar_slots_high_water();
+```
+
+## A slot is not an identity or sort key
+
+The primary key remains load-bearing. A `ColumnSlotId` is only a compact position shared by the
+derived field chunks and columnar indexes. It is not:
+
+- a replacement primary key;
+- the row's rank in `cluster_by` order;
+- a stable external identifier;
+- a durable identifier across restart; or
+- a public tuple value applications should store.
+
+Generated results carry an opaque reference:
+
+```rust
+pub struct ColumnarRowRef<PrimaryKey, SlotId> { /* private */ }
+
+impl<PrimaryKey, SlotId> ColumnarRowRef<PrimaryKey, SlotId> {
+    pub fn primary_key(&self) -> &PrimaryKey;
+}
+```
+
+`ColumnarRowRef` deliberately does not implement serialization. Durable application references
+must store the primary key.
+
+### Delete/reinsert and ABA safety
+
+A bounded slot allocator must reuse positions. Primary key plus slot alone is insufficient: delete
+and reinsert of the same primary key into the same slot would make an old reference appear valid.
+
+WorkTable therefore validates four pieces of state:
+
+```text
+primary key + slot position + u64 slot generation + table incarnation
+```
+
+The generation is separate from the configured slot width, so choosing `ColumnSlotId8` still gives
+256 live slots rather than carving generation bits out of those eight bits. A delete increments the
+slot's generation before reuse. Generation never wraps: if its `u64` counter is ever exhausted,
+that slot is permanently retired. A process-local table incarnation invalidates references created
+by another table instance or before a persisted table is reopened.
+
+This is stronger for retained owned references than relying only on an epoch grace period: an
+epoch protects active readers, but it cannot protect a reference an application stores after the
+read guard ends.
+
+## Current side-index physical model
+
+For each generated table with at least one columnar field, WorkTable keeps the tabular row and adds:
+
+```text
+authoritative primary key -> authoritative WorkTable row/link
+                          -> (ColumnSlotId, generation) directory
+                          -> chunked base field replicas
+                          -> zero or more clustered BTreeMap access paths
+```
+
+Each duplicated side field currently uses `Vec<Vec<Option<T>>>`. The outer vector holds chunks and the inner
+vector is indexed by the slot offset. A separate primary-key column supports reference validation.
+This first layout is deliberately generic:
+
+- fixed-width values are not yet exposed through SIMD/vector kernels;
+- optional fields currently store the Rust `Option<T>` representation rather than a validity
+  bitmap;
+- `String` values are owned values rather than offsets into a byte arena; and
+- chunks are mutable and uncompressed.
+
+These are implementation boundaries, not compression or vectorization claims.
+
+## Generated API in this implementation
+
+The current generated methods return owned collections:
+
+```rust
+// Exact lookup requires the complete composite clustered key.
+let refs = table.columnar_select_host_time(host_id, captured_at_ns)?;
+
+// Gather a selected field through validated opaque references.
+let cpu = table.columnar_project_cpu_percent(&refs)?;
+
+// A direct scan needs no columnar index.
+let statuses = table.columnar_scan_status()?;
+
+// Scan in the clustered index's key order.
+let ordered_refs = table.columnar_scan_host_time()?;
+```
+
+The owned `Vec` results keep locks out of the public return type and may be retained safely. They
+also materialize the result, so this API is not the final high-volume execution surface.
+
+Not yet implemented:
+
+- prefix equality and range predicates on a composite `cluster_by` key;
+- namespaced predicate/projection builders;
+- zero-copy or callback-based `scan_batches`;
+- a single multi-field projection API; and
+- row-gather fallback for row-only fields.
+
+Those remain the next query-API slice. Documentation and benchmarks must not present them as
+shipping behavior.
+
+## Mutation and consistency behavior
+
+Insert, ordinary update, delete, and reinsert maintain the derived directory, field chunks, and
+clustered metadata before the mutation call completes. A same-primary-key update retains its slot
+and generation. Vacuum may move the authoritative physical link without changing the columnar
+slot.
+
+The complete derived state is protected by one table-local read/write lock. Consequences:
+
+- an individual scan or projection observes one coherent columnar snapshot;
+- all columnar fields changed by one maintenance operation publish together;
+- concurrent columnar writers are table-serialized;
+- ordinary row-only selects do not take the columnar lock; and
+- two separate API calls are two snapshots, not a transactionally pinned multi-call view.
+
+Some archived in-place mutation paths cannot apply a typed column delta yet. They mark the
+derived replica dirty. The next columnar read rebuilds it from authoritative rows. Applications
+can see and schedule that cost explicitly:
+
+```rust
+if table.columnar_is_dirty() {
+    table.rebuild_columnar()?;
+}
+```
+
+The current rebuild is whole-table and holds the columnar writer lock. Per-chunk dirty tracking is
+planned; it is not implemented in this release.
+
+## Persistence and recovery
+
+For `persist: true`, ordinary WorkTable rows and indexes retain their existing formats. The
+columnar side index is marked derived, omitted from the persisted index structure, and rebuilt from
+authoritative rows after load. Therefore this change introduces no new on-disk columnar format.
+
+`ColumnSlotId` assignments are not promised to survive restart. The process/table incarnation in
+`ColumnarRowRef` prevents an old in-memory reference from being accepted by a reopened instance.
+Use primary keys for durable identity.
+
+Native columnar checkpoints are a later design choice. They should be added only if benchmarked
+restart time or row-store gather cost justifies another durable format and recovery protocol.
+
+## Relationship to SAP HANA's unified table
+
+Sikka et al.'s SAP HANA paper is the most relevant architectural contrast because it explains how
+a true column-oriented system can serve transactional and analytical work on one logical table.
+HANA primarily informs WorkTable's possible third flavor, while this PR deliberately implements
+the cheaper middle flavor. The similarity is the use of different physical representations for
+different access patterns; the authority and execution models differ.
+
+| Dimension | SAP HANA (SIGMOD 2012) | WorkTable side indexes / PR implementation |
+|---|---|---|
+| Logical goal | OLTP and OLAP through one unified table interface | Preserve the tabular engine and add opt-in columnar-flavored side indexes |
+| Write path | Uncompressed row-oriented L1 delta | Existing authoritative WorkTable row storage |
+| Intermediate form | Dictionary-encoded, unsorted column L2 delta | Mutable, uncompressed side-index vectors |
+| Read-optimized form | Compressed main store with sorted dictionaries and bit-packed values | Not implemented yet |
+| Record position | RowId created on entry; positional alignment across columns | Primary key is authoritative; opaque slot aligns derived columns |
+| Reorganization | Asynchronous L1→L2 and snapshot-safe L2→main merges | Synchronous maintenance; whole-table rebuild only for dirty fallback paths |
+| Readers during merge | Old/new versions retained until transactions using the old version finish | One table-local columnar `RwLock`; no versioned columnar merge yet |
+| Point access | Inverted indexes across delta and main structures | Generated `BTreeMap` clustered metadata plus the normal WorkTable indexes |
+| Execution | Row/column iterators and vectorized block-at-a-time operators | Owned `Vec` scans/projections in this first API |
+| Durability | REDO for incoming changes plus savepoints for column structures | Existing WorkTable persistence is authoritative; columnar state rebuilds after load |
+
+The closest honest analogy is: **WorkTable's tabular engine plays a role similar to HANA's
+write-optimized L1, while this PR adds optional uncompressed side indexes. It does not implement
+HANA's L2/main column-store lifecycle and does not claim the third, fully columnar flavor.**
+
+The HANA result points to a defensible next architecture for WorkTable:
+
+- keep a small mutable delta that accepts foreground mutations cheaply;
+- seal cold column chunks into immutable snapshots;
+- build dictionaries, bit packing, delta/RLE, and zone metadata off the foreground path;
+- publish a new manifest atomically while readers finish on the previous snapshot;
+- reclaim old snapshots only after the read grace period; and
+- checkpoint sealed chunks separately from the authoritative row representation only when the
+  recovery and performance measurements justify it.
+
+That architecture belongs to a future full-columnar effort and would also remove the current
+whole-table dirty-rebuild cliff. It is a roadmap informed by HANA's record-lifecycle design, not a
+performance claim for this PR or a requirement for the side-index flavor.
+
+Reference: Vishal Sikka, Franz Färber, Wolfgang Lehner, Sang Kyun Cha, Thomas Peh, and Christof
+Bornhövd. “Efficient Transaction Processing in SAP HANA Database: The End of a Column Store Myth.”
+SIGMOD 2012, pp. 731–741. DOI: 10.1145/2213836.2213946.
+
+## Compile-time validation
+
+The macro rejects:
+
+- `columnar` on a primary-key field;
+- an unknown or non-columnar field in `cluster_by`;
+- empty or duplicate `cluster_by` entries;
+- a columnar field/index generated-method name collision;
+- the removed `columns:` index property;
+- the reserved `include:` property;
+- unsupported compression policies;
+- zero or non-nesting chunk sizes;
+- duplicate table config entries;
+- unknown or out-of-order column attributes; and
+- `columnar_indexes` in `worktable_version!`.
+
+## Benchmark contract before performance claims
+
+For each supported primary-index backend (`WorkTablesIndex`, `congee-wt`, and `arctic-wt` where the
+`Using` and persistence rules permit it), measure:
+
+- row select throughput and p50/p95/p99 with no columnar declarations, fields only, and fields plus
+  clustered metadata;
+- insert, same-key update, indexed-key update, delete, and fixed-window delete/reinsert churn;
+- full field scan, exact clustered lookup, and projection gather;
+- single-thread and 1→core-count concurrent readers/writers;
+- memory amplification, allocation count, code size, and slot-directory overhead by width;
+- first-read dirty rebuild versus application-scheduled rebuild;
+- persisted reload and rebuild time; and
+- correctness counters alongside throughput, especially under slot reuse and concurrent mutation.
+
+Until those measurements exist, the safe statement is that the feature adds correctness-tested
+columnar side indexes—not that it is a full column store, faster for every workload, or ready for
+latency-sensitive HFT deployment.
+
+## Staged roadmap
+
+- **Query surface:** namespaced predicate builders, prefix/range selection, combined projection,
+  and bounded `scan_batches`.
+- **Incremental maintenance:** per-chunk dirtiness and typed in-place deltas.
+- **Column encodings:** validity bitmaps, fixed-width vector kernels, and offset/byte storage for
+  variable-width fields.
+- **Separate full-columnar design:** mutable delta, sealed immutable chunks, background merge,
+  versioned publication/reclamation, and vector execution. This is a different flavor, not a
+  silent expansion of the side-index feature.
+- **Compression:** type-checked dictionary, delta, RLE, bit packing, and an evidence-based `auto`.
+- **Optional native persistence:** manifests, checksums, recovery watermarks, and crash tests.
+- **Optimizer:** choose row lookup, ordinary index, clustered side-index lookup, or base-field scan
+  from measured costs.
+
+This ordering keeps correctness and compatibility ahead of compression claims while leaving the
+DSL stable for the later physical evolution.
+
+## Reviewer decision points
+
+- Is “tabular + columnar side indexes” the right permanent name for this middle flavor?
+- Is full-width `ColumnSlotId8|16|32|64` plus a separate `u64` generation preferable to hiding a
+  smaller slot/generation bit split inside the configured width?
+- Is the owned, fully materialized phase-one API acceptable if `scan_batches` is the next query
+  slice and no vector-execution claim is made now?
+- Is rebuild-on-load the correct compatibility choice until native side-index checkpoints show a
+  measured recovery benefit?
+- Should the future full-columnar flavor receive distinct DSL rather than changing the meaning of
+  today's field-level `columnar` attribute?
diff --git a/docs/columnar-index-plan.md b/docs/columnar-index-plan.md
new file mode 100644
index 00000000..c1b3ad40
--- /dev/null
+++ b/docs/columnar-index-plan.md
@@ -0,0 +1,142 @@
+# Columnar side-index implementation plan
+
+Status: phase-one implementation in `feat/columnar-fields-indexes`. The complete user and reviewer
+guide is [`columnar-fields-and-indexes-guide-v3.md`](columnar-fields-and-indexes-guide-v3.md).
+
+## Scope boundary
+
+WorkTable has three distinct storage flavors:
+
+1. **Tabular** — the existing authoritative row engine.
+2. **Tabular + columnar side indexes** — the scope of this branch. Selected field values and
+   clustered keys are duplicated into derived structures for cheaper columnar-flavored access.
+3. **Columnar** — an authoritative vector layout, vectorized execution, sealed/encoded segments,
+   and native columnar persistence. This is not implemented by this branch.
+
+The phase-one feature must not be marketed or documented as the third flavor.
+
+## Accepted DSL
+
+```rust
+worktable!(
+    name: HistoricalCpu,
+    persist: true,
+    columns: {
+        id: u128 primary_key,
+        host_id: u64 columnar,
+        captured_at_ns: u64 columnar,
+        temperature: i64 columnar(chunk_rows(32_768), compression(none)),
+        status: String columnar,
+        diagnostic_blob: DiagnosticBlob,
+    },
+    columnar_indexes: {
+        host_time: {
+            cluster_by: [host_id, captured_at_ns],
+        },
+    },
+    config: {
+        columnar_slot_id: ColumnSlotId32,
+        columnar_chunk_rows: 65_536,
+    },
+);
+```
+
+- Bare `columnar` uses defaults.
+- `compression(none)` is the only accepted policy until a codec exists.
+- `cluster_by` orders index metadata, not the side-field vectors.
+- `columns:` inside a columnar index has been removed as semantically redundant.
+- Slot width is a table setting, independent of whether the table declares any clustered side
+  index.
+
+## Identity and capacity
+
+`ColumnSlotId8|16|32|64` uses its complete unsigned range for side-index slot positions. It neither
+replaces the primary key nor represents sort rank.
+
+An opaque `ColumnarRowRef` carries:
+
+```text
+primary key + slot + separate u64 generation + table incarnation
+```
+
+Delete increments the generation before slot reuse. Generation never wraps; an exhausted slot is
+retired. Table incarnation invalidates retained refs across a new/reopened instance. The ref is not
+serializable and exposes only the authoritative primary key.
+
+The schema author is responsible for selecting a width that covers maximum simultaneously live
+side-indexed rows. Exceeding it returns `WorkTableError::ColumnSlotIdExhausted(bits)` and rolls back
+the insert. The implementation never widens, truncates, wraps, or evicts automatically.
+
+## Implemented side structures
+
+Generated tables maintain under one table-local `RwLock`:
+
+- primary-key → `(ColumnSlotId, generation)` directory;
+- reusable slot set and generation vector;
+- process/table incarnation;
+- chunked `Vec<Vec<Option<T>>>` for each opted-in field;
+- primary-key side column for ref validation; and
+- a `BTreeMap<composite key, BTreeSet<slot>>` for each `columnar_indexes` entry.
+
+Insert/update/delete/reinsert hooks maintain these after the authoritative row mutation. A vacuum
+link change does not change the slot. In-place paths without a typed delta mark the side indexes
+dirty; `rebuild_columnar()` lets applications pay the whole-table rebuild cost deliberately.
+
+Persisted tables skip these derived fields in their existing index disk format and rebuild them
+from authoritative rows after load. This branch adds no on-disk format.
+
+## Current generated operations
+
+```rust
+table.columnar_select_host_time(host_id, captured_at_ns)?;
+table.columnar_scan_host_time()?;
+table.columnar_scan_status()?;
+table.columnar_project_temperature(&row_refs)?;
+table.columnar_is_dirty();
+table.rebuild_columnar()?;
+table.columnar_slots_in_use();
+table.columnar_slots_high_water();
+```
+
+They return owned `Vec` collections. Full-key equality is the only clustered predicate in phase
+one.
+
+## Phase-one correctness gates
+
+- Same-primary-key delete/reinsert into the same slot must invalidate the old ref.
+- A ref from another table incarnation must fail validation.
+- Slot exhaustion must roll back the authoritative mutation.
+- All four slot widths must enforce their numeric range without wrapping.
+- Mutation paths must update or dirty side indexes before returning.
+- A dirty rebuild must preserve live slot/generation mappings.
+- Macro validation must reject primary-key `columnar`, unknown/non-columnar cluster keys, duplicate
+  keys/names/config, inert compression, non-nesting chunks, `columns:`, and reserved `include:`.
+- Persisted load must reconstruct side indexes without changing the current disk format.
+
+## Performance gates
+
+Before an HFT-facing claim or default:
+
+- compare tabular baseline against fields-only and fields-plus-clustered side indexes;
+- measure row select, insert, update, delete, churn, exact lookup, scan, and gather;
+- report p50/p95/p99, allocations, memory, and code size;
+- run 1→core-count concurrency with correctness counters;
+- measure first-reader and explicit dirty rebuild costs; and
+- repeat across supported WorkTablesIndex, congee-wt, and arctic-wt `Using` configurations.
+
+## Follow-up within the side-index flavor
+
+1. Namespaced builders and prefix/range predicates.
+2. Bounded `scan_batches` and one-lock multi-field projection.
+3. Per-chunk dirty tracking and typed in-place deltas.
+4. Validity bitmaps, fixed-width kernels, and variable-width offset buffers.
+5. Optional sealed side-index snapshots if benchmarks justify persistence.
+
+## Separate full-columnar flavor
+
+SAP HANA's unified-table record lifecycle is useful prior art for a future third flavor: an
+uncompressed row write delta, a column delta, a compressed main, asynchronous merge, old/new
+snapshot coexistence, and vector/block execution. That is a separate architecture and performance
+contract. It must not arrive by quietly changing what `columnar` side indexes mean.
+
+See the v3 guide for the detailed comparison and citation.
diff --git a/output/pdf/worktable-columnar-side-indexes-guide-v3.pdf b/output/pdf/worktable-columnar-side-indexes-guide-v3.pdf
new file mode 100644
index 00000000..38ac26c7
Binary files /dev/null and b/output/pdf/worktable-columnar-side-indexes-guide-v3.pdf differ
diff --git a/src/columnar.rs b/src/columnar.rs
new file mode 100644
index 00000000..cbad3c1e
--- /dev/null
+++ b/src/columnar.rs
@@ -0,0 +1,337 @@
+use std::collections::{BTreeMap, BTreeSet};
+use std::fmt::Debug;
+use std::hash::Hash;
+use std::sync::atomic::{AtomicU64, Ordering};
+
+use crate::mem_stat::MemStat;
+
+/// Compact position used by generated columnar storage.
+///
+/// This is supplemental metadata, never a replacement for a WorkTable primary
+/// key. Slot IDs are not sort keys and are not durable identities.
+pub trait ColumnSlotId: Copy + Debug + Eq + Ord + Hash + Send + Sync + MemStat + 'static {
+    const BITS: u8;
+
+    fn try_from_position(position: u64) -> Option<Self>;
+    fn position(self) -> u64;
+
+    fn slot(self) -> usize {
+        usize::try_from(self.position()).expect("column slot ID exceeds this target's address space")
+    }
+}
+
+macro_rules! column_slot_id {
+    ($name:ident, $inner:ty, $bits:literal) => {
+        #[derive(Debug, Default, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
+        pub struct $name($inner);
+
+        impl ColumnSlotId for $name {
+            const BITS: u8 = $bits;
+
+            fn try_from_position(position: u64) -> Option<Self> {
+                <$inner>::try_from(position).ok().map(Self)
+            }
+
+            fn position(self) -> u64 {
+                self.0 as u64
+            }
+        }
+
+        impl MemStat for $name {
+            fn heap_size(&self) -> usize {
+                0
+            }
+
+            fn used_size(&self) -> usize {
+                0
+            }
+        }
+    };
+}
+
+column_slot_id!(ColumnSlotId8, u8, 8);
+column_slot_id!(ColumnSlotId16, u16, 16);
+column_slot_id!(ColumnSlotId32, u32, 32);
+column_slot_id!(ColumnSlotId64, u64, 64);
+
+static NEXT_COLUMNAR_INCARNATION: AtomicU64 = AtomicU64::new(1);
+
+/// Returns a process-local table incarnation used to invalidate retained
+/// columnar references when a table is rebuilt or reopened.
+#[doc(hidden)]
+pub fn next_columnar_incarnation() -> u64 {
+    NEXT_COLUMNAR_INCARNATION
+        .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |value| value.checked_add(1))
+        .expect("columnar table incarnation space is exhausted")
+}
+
+/// Identity carried by generated columnar query results.
+///
+/// The primary key remains authoritative. The slot, generation, and table
+/// incarnation are private validation metadata and are deliberately not
+/// serializable or exposed as ordering keys.
+#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
+pub struct ColumnarRowRef<PrimaryKey, SlotId = ColumnSlotId32> {
+    primary_key: PrimaryKey,
+    slot_id: SlotId,
+    generation: u64,
+    incarnation: u64,
+}
+
+impl<PrimaryKey, SlotId> ColumnarRowRef<PrimaryKey, SlotId> {
+    /// Returns the authoritative WorkTable primary key.
+    pub fn primary_key(&self) -> &PrimaryKey {
+        &self.primary_key
+    }
+
+    /// Constructor used by generated WorkTable code.
+    #[doc(hidden)]
+    pub fn __new(primary_key: PrimaryKey, slot_id: SlotId, generation: u64, incarnation: u64) -> Self {
+        Self {
+            primary_key,
+            slot_id,
+            generation,
+            incarnation,
+        }
+    }
+
+    #[doc(hidden)]
+    pub fn __slot_id(&self) -> SlotId
+    where
+        SlotId: Copy,
+    {
+        self.slot_id
+    }
+
+    #[doc(hidden)]
+    pub fn __generation(&self) -> u64 {
+        self.generation
+    }
+
+    #[doc(hidden)]
+    pub fn __incarnation(&self) -> u64 {
+        self.incarnation
+    }
+}
+
+impl<PrimaryKey: MemStat, SlotId: MemStat> MemStat for ColumnarRowRef<PrimaryKey, SlotId> {
+    fn heap_size(&self) -> usize {
+        self.primary_key.heap_size() + self.slot_id.heap_size()
+    }
+
+    fn used_size(&self) -> usize {
+        self.primary_key.used_size() + self.slot_id.used_size()
+    }
+}
+
+/// Compression used by a generated columnar field.
+///
+/// Mutable chunks are currently unencoded; unsupported policies are rejected
+/// by the macro instead of being accepted as inert configuration.
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
+pub enum ColumnCompression {
+    #[default]
+    None,
+}
+
+impl ColumnCompression {
+    pub fn is_encoded(self) -> bool {
+        false
+    }
+}
+
+impl MemStat for ColumnCompression {
+    fn heap_size(&self) -> usize {
+        0
+    }
+
+    fn used_size(&self) -> usize {
+        0
+    }
+}
+
+/// Chunked storage for one generated columnar field.
+#[derive(Debug)]
+pub struct ColumnarColumn<T> {
+    chunk_rows: usize,
+    compression: ColumnCompression,
+    chunks: Vec<Vec<Option<T>>>,
+}
+
+impl<T> ColumnarColumn<T> {
+    pub fn new(chunk_rows: usize, compression: ColumnCompression) -> Self {
+        assert!(chunk_rows > 0, "columnar chunks cannot be empty");
+        Self {
+            chunk_rows,
+            compression,
+            chunks: Vec::new(),
+        }
+    }
+
+    pub fn chunk_rows(&self) -> usize {
+        self.chunk_rows
+    }
+
+    pub fn compression(&self) -> ColumnCompression {
+        self.compression
+    }
+
+    pub fn set<SlotId: ColumnSlotId>(&mut self, slot_id: SlotId, value: T) {
+        let row = slot_id.slot();
+        let chunk_index = row / self.chunk_rows;
+        let offset = row % self.chunk_rows;
+        while self.chunks.len() <= chunk_index {
+            self.chunks.push(Vec::new());
+        }
+        let chunk = &mut self.chunks[chunk_index];
+        if chunk.len() <= offset {
+            chunk.resize_with(offset + 1, || None);
+        }
+        chunk[offset] = Some(value);
+    }
+
+    pub fn remove<SlotId: ColumnSlotId>(&mut self, slot_id: SlotId) -> Option<T> {
+        let row = slot_id.slot();
+        self.chunks
+            .get_mut(row / self.chunk_rows)
+            .and_then(|chunk| chunk.get_mut(row % self.chunk_rows))
+            .and_then(Option::take)
+    }
+
+    pub fn get<SlotId: ColumnSlotId>(&self, slot_id: SlotId) -> Option<&T> {
+        let row = slot_id.slot();
+        self.chunks
+            .get(row / self.chunk_rows)
+            .and_then(|chunk| chunk.get(row % self.chunk_rows))
+            .and_then(Option::as_ref)
+    }
+
+    pub fn iter<SlotId: ColumnSlotId>(&self) -> impl Iterator<Item = (SlotId, &T)> {
+        let chunk_rows = self.chunk_rows;
+        self.chunks.iter().enumerate().flat_map(move |(chunk_index, chunk)| {
+            chunk.iter().enumerate().filter_map(move |(offset, value)| {
+                value.as_ref().map(|value| {
+                    let position = (chunk_index * chunk_rows + offset) as u64;
+                    let slot_id = SlotId::try_from_position(position)
+                        .expect("stored column position fits its configured column slot ID");
+                    (slot_id, value)
+                })
+            })
+        })
+    }
+}
+
+impl<T: MemStat> MemStat for ColumnarColumn<T> {
+    fn heap_size(&self) -> usize {
+        self.chunks.heap_size()
+    }
+
+    fn used_size(&self) -> usize {
+        self.chunks.used_size()
+    }
+}
+
+/// Ordered metadata for one generated `columnar_indexes` declaration.
+#[derive(Debug)]
+pub struct ClusteredColumnarIndex<K, SlotId = ColumnSlotId32> {
+    rows: BTreeMap<K, BTreeSet<SlotId>>,
+}
+
+impl<K, SlotId> Default for ClusteredColumnarIndex<K, SlotId> {
+    fn default() -> Self {
+        Self { rows: BTreeMap::new() }
+    }
+}
+
+impl<K: Ord, SlotId: Ord + Copy> ClusteredColumnarIndex<K, SlotId> {
+    pub fn insert(&mut self, key: K, slot_id: SlotId) {
+        self.rows.entry(key).or_default().insert(slot_id);
+    }
+
+    pub fn remove(&mut self, key: &K, slot_id: SlotId) {
+        let remove_key = self.rows.get_mut(key).is_some_and(|rows| {
+            rows.remove(&slot_id);
+            rows.is_empty()
+        });
+        if remove_key {
+            self.rows.remove(key);
+        }
+    }
+
+    pub fn exact(&self, key: &K) -> Vec<SlotId> {
+        self.rows
+            .get(key)
+            .map(|rows| rows.iter().copied().collect())
+            .unwrap_or_default()
+    }
+
+    pub fn ordered_slot_ids(&self) -> Vec<SlotId> {
+        self.rows.values().flat_map(|rows| rows.iter().copied()).collect()
+    }
+}
+
+impl<K: MemStat + Ord, SlotId: MemStat + Ord> MemStat for ClusteredColumnarIndex<K, SlotId> {
+    fn heap_size(&self) -> usize {
+        self.rows.heap_size()
+    }
+
+    fn used_size(&self) -> usize {
+        self.rows.used_size()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn chunks_are_addressed_by_configured_slot_id() {
+        let mut column = ColumnarColumn::new(2, ColumnCompression::None);
+        column.set(ColumnSlotId16(3), 30);
+        column.set(ColumnSlotId16(0), 10);
+
+        assert_eq!(column.chunk_rows(), 2);
+        assert_eq!(column.get(ColumnSlotId16(3)), Some(&30));
+        assert_eq!(
+            column
+                .iter::<ColumnSlotId16>()
+                .map(|(id, value)| (id.0, *value))
+                .collect::<Vec<_>>(),
+            [(0, 10), (3, 30)]
+        );
+
+        assert_eq!(column.remove(ColumnSlotId16(0)), Some(10));
+        assert!(column.get(ColumnSlotId16(0)).is_none());
+    }
+
+    #[test]
+    fn clustered_index_preserves_key_order() {
+        let mut index = ClusteredColumnarIndex::default();
+        index.insert((2, 1), ColumnSlotId8(1));
+        index.insert((1, 9), ColumnSlotId8(2));
+        index.insert((1, 9), ColumnSlotId8(0));
+
+        assert_eq!(index.exact(&(1, 9)), [ColumnSlotId8(0), ColumnSlotId8(2)]);
+        assert_eq!(
+            index.ordered_slot_ids(),
+            [ColumnSlotId8(0), ColumnSlotId8(2), ColumnSlotId8(1)]
+        );
+    }
+
+    #[test]
+    fn widths_have_expected_capacity_boundaries() {
+        assert_eq!(ColumnSlotId8::try_from_position(255), Some(ColumnSlotId8(255)));
+        assert_eq!(ColumnSlotId8::try_from_position(256), None);
+        assert_eq!(ColumnSlotId16::try_from_position(65_535), Some(ColumnSlotId16(65_535)));
+        assert_eq!(ColumnSlotId16::try_from_position(65_536), None);
+        assert_eq!(
+            ColumnSlotId32::try_from_position(u32::MAX as u64),
+            Some(ColumnSlotId32(u32::MAX))
+        );
+        assert_eq!(ColumnSlotId32::try_from_position(u32::MAX as u64 + 1), None);
+        assert_eq!(
+            ColumnSlotId64::try_from_position(u64::MAX),
+            Some(ColumnSlotId64(u64::MAX))
+        );
+    }
+}
diff --git a/src/index/table_secondary_index/mod.rs b/src/index/table_secondary_index/mod.rs
index 69434a82..5713c590 100644
--- a/src/index/table_secondary_index/mod.rs
+++ b/src/index/table_secondary_index/mod.rs
@@ -93,6 +93,10 @@ pub enum IndexError<IndexNameEnum> {
         at: IndexNameEnum,
         inserted_already: Vec<IndexNameEnum>,
     },
+    ColumnSlotIdExhausted {
+        bits: u8,
+        inserted_already: Vec<IndexNameEnum>,
+    },
     NotFound,
 }
 
@@ -106,6 +110,10 @@ where
                 at,
                 inserted_already: _,
             } => WorkTableError::AlreadyExists(at.to_string_value()),
+            IndexError::ColumnSlotIdExhausted {
+                bits,
+                inserted_already: _,
+            } => WorkTableError::ColumnSlotIdExhausted(bits),
             IndexError::NotFound => WorkTableError::NotFound,
         }
     }
diff --git a/src/lib.rs b/src/lib.rs
index 9e28c540..8c366d95 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -1,5 +1,6 @@
 #![doc = include_str!("../docs/crate.md")]
 
+mod columnar;
 pub mod in_memory;
 mod index;
 pub mod lock;
@@ -14,6 +15,10 @@ mod util;
 #[cfg(feature = "s3-support")]
 pub mod features;
 
+pub use columnar::{
+    ClusteredColumnarIndex, ColumnCompression, ColumnSlotId, ColumnSlotId8, ColumnSlotId16, ColumnSlotId32,
+    ColumnSlotId64, ColumnarColumn, ColumnarRowRef, next_columnar_incarnation,
+};
 pub use index::*;
 pub use persistence::{LoadMode, PersistedWorkTable, PersistenceConfig, PersistenceLoadError};
 pub use row::*;
@@ -47,12 +52,13 @@ pub mod prelude {
     pub use crate::table::system_info::{IndexInfo, IndexKind, SystemInfo};
     pub use crate::util::{OffsetEqLink, OrderedF32Def, OrderedF64Def};
     pub use crate::{
-        ArcticIndex, ArcticKey, AvailableIndex, CongeeIndex, CongeeKey, Difference, IndexError, IndexMap,
-        IndexMultiMap, MultiPairRecreate, PersistentArcticIndex, PersistentArtIndex, PersistentCongeeIndex,
-        PersistentWtiIndex, PrimaryIndex, TableIndex, TableIndexCdc, TableRow, TableSecondaryIndex,
-        TableSecondaryIndexCdc, TableSecondaryIndexEventsOps, TableSecondaryIndexInfo, UniqueIndex, UnsizedNode,
-        UpstreamIndexMap, UpstreamIndexPair, WorkTable, WorkTableError, vacuum::EmptyDataVacuum,
-        vacuum::VacuumPersistence, vacuum::WorkTableVacuum,
+        ArcticIndex, ArcticKey, AvailableIndex, ClusteredColumnarIndex, ColumnCompression, ColumnSlotId, ColumnSlotId8,
+        ColumnSlotId16, ColumnSlotId32, ColumnSlotId64, ColumnarColumn, ColumnarRowRef, CongeeIndex, CongeeKey,
+        Difference, IndexError, IndexMap, IndexMultiMap, MultiPairRecreate, PersistentArcticIndex, PersistentArtIndex,
+        PersistentCongeeIndex, PersistentWtiIndex, PrimaryIndex, TableIndex, TableIndexCdc, TableRow,
+        TableSecondaryIndex, TableSecondaryIndexCdc, TableSecondaryIndexEventsOps, TableSecondaryIndexInfo,
+        UniqueIndex, UnsizedNode, UpstreamIndexMap, UpstreamIndexPair, WorkTable, WorkTableError,
+        next_columnar_incarnation, vacuum::EmptyDataVacuum, vacuum::VacuumPersistence, vacuum::WorkTableVacuum,
     };
     pub use data_bucket::{
         DATA_VERSION, DataPage, GENERAL_HEADER_SIZE, GeneralHeader, GeneralPage, INNER_PAGE_SIZE, IndexPage, Interval,
diff --git a/src/mem_stat/mod.rs b/src/mem_stat/mod.rs
index 666d61d1..1bfad87e 100644
--- a/src/mem_stat/mod.rs
+++ b/src/mem_stat/mod.rs
@@ -1,6 +1,6 @@
 mod primitives;
 
-use std::collections::HashMap;
+use std::collections::{BTreeMap, BTreeSet, HashMap};
 use std::fmt::Debug;
 use std::rc::Rc;
 use std::sync::Arc;
@@ -31,6 +31,33 @@ pub trait MemStat {
     fn used_size(&self) -> usize;
 }
 
+macro_rules! impl_tuple_mem_stat {
+    ($($name:ident),+) => {
+        impl<$($name: MemStat),+> MemStat for ($($name,)+) {
+            fn heap_size(&self) -> usize {
+                #[allow(non_snake_case)]
+                let ($($name,)+) = self;
+                0usize $(+ $name.heap_size())+
+            }
+
+            fn used_size(&self) -> usize {
+                #[allow(non_snake_case)]
+                let ($($name,)+) = self;
+                0usize $(+ $name.used_size())+
+            }
+        }
+    };
+}
+
+impl_tuple_mem_stat!(A);
+impl_tuple_mem_stat!(A, B);
+impl_tuple_mem_stat!(A, B, C);
+impl_tuple_mem_stat!(A, B, C, D);
+impl_tuple_mem_stat!(A, B, C, D, E);
+impl_tuple_mem_stat!(A, B, C, D, E, F);
+impl_tuple_mem_stat!(A, B, C, D, E, F, G);
+impl_tuple_mem_stat!(A, B, C, D, E, F, G, H);
+
 impl<T: MemStat> MemStat for Option<T> {
     fn heap_size(&self) -> usize {
         self.as_ref().map_or(0, |v| v.heap_size())
@@ -228,6 +255,40 @@ impl<K: MemStat + Eq + std::hash::Hash, V: MemStat> MemStat for HashMap<K, V> {
     }
 }
 
+impl<K: MemStat + Ord, V: MemStat> MemStat for BTreeMap<K, V> {
+    fn heap_size(&self) -> usize {
+        self.len() * std::mem::size_of::<(K, V)>()
+            + self
+                .iter()
+                .map(|(key, value)| key.heap_size() + value.heap_size())
+                .sum::<usize>()
+    }
+
+    fn used_size(&self) -> usize {
+        self.heap_size()
+    }
+}
+
+impl<T: MemStat + Ord> MemStat for BTreeSet<T> {
+    fn heap_size(&self) -> usize {
+        self.len() * std::mem::size_of::<T>() + self.iter().map(MemStat::heap_size).sum::<usize>()
+    }
+
+    fn used_size(&self) -> usize {
+        self.heap_size()
+    }
+}
+
+impl<T: MemStat> MemStat for parking_lot::RwLock<T> {
+    fn heap_size(&self) -> usize {
+        self.read().heap_size()
+    }
+
+    fn used_size(&self) -> usize {
+        self.read().used_size()
+    }
+}
+
 impl<T> MemStat for OrderedFloat<T>
 where
     T: MemStat,
diff --git a/src/table/mod.rs b/src/table/mod.rs
index 3af0b187..1c3573e9 100644
--- a/src/table/mod.rs
+++ b/src/table/mod.rs
@@ -254,6 +254,13 @@ where
 
                     Err(WorkTableError::AlreadyExists(at.to_string_value()))
                 }
+                IndexError::ColumnSlotIdExhausted { bits, inserted_already } => {
+                    self.primary_index.remove(&pk, link);
+                    self.indexes.delete_from_indexes(row, link, inserted_already)?;
+                    self.data.delete(link).map_err(WorkTableError::PagesError)?;
+
+                    Err(WorkTableError::ColumnSlotIdExhausted(bits))
+                }
                 IndexError::NotFound => Err(WorkTableError::NotFound),
             };
         }
@@ -338,6 +345,32 @@ where
                         (ack_op, WorkTableError::AlreadyExists(at.to_string_value()))
                     }
                 }
+                IndexError::ColumnSlotIdExhausted { bits, inserted_already } => {
+                    let (_, rollback_pk_events) = self.primary_index.remove_cdc(pk.clone(), link);
+                    let rollback_pk_events = convert_change_events(rollback_pk_events);
+
+                    let (rollback_secondary_events, _) =
+                        self.indexes
+                            .delete_from_indexes_cdc(row.clone(), link, inserted_already);
+
+                    let mut merged_primary_events = primary_key_events.clone();
+                    merged_primary_events.extend(rollback_pk_events);
+
+                    let mut merged_secondary_events = secondary_events.clone();
+                    merged_secondary_events.extend(rollback_secondary_events);
+
+                    let ack_op = Operation::Acknowledge(AcknowledgeOperation {
+                        id: OperationId::Single(Uuid::now_v7()),
+                        primary_key_events: merged_primary_events,
+                        secondary_keys_events: merged_secondary_events,
+                    });
+
+                    if let Err(e) = self.data.delete(link) {
+                        (ack_op, WorkTableError::PagesError(e))
+                    } else {
+                        (ack_op, WorkTableError::ColumnSlotIdExhausted(bits))
+                    }
+                }
                 IndexError::NotFound => {
                     let ack_op = Operation::Acknowledge(AcknowledgeOperation {
                         id: OperationId::Single(Uuid::now_v7()),
@@ -439,6 +472,13 @@ where
 
                     Err(WorkTableError::AlreadyExists(at.to_string_value()))
                 }
+                IndexError::ColumnSlotIdExhausted { bits, inserted_already } => {
+                    self.primary_index.insert(pk.clone(), old_link);
+                    self.indexes.delete_from_indexes(row_new, new_link, inserted_already)?;
+                    self.data.delete(new_link).map_err(WorkTableError::PagesError)?;
+
+                    Err(WorkTableError::ColumnSlotIdExhausted(bits))
+                }
                 IndexError::NotFound => Err(WorkTableError::NotFound),
             };
         }
@@ -536,6 +576,32 @@ where
                         (ack_op, WorkTableError::AlreadyExists(at.to_string_value()))
                     }
                 }
+                IndexError::ColumnSlotIdExhausted { bits, inserted_already } => {
+                    let (_, rollback_pk_events) = self.primary_index.insert_cdc(pk.clone(), old_link);
+                    let rollback_pk_events = convert_change_events(rollback_pk_events);
+
+                    let (rollback_secondary_events, _) =
+                        self.indexes
+                            .delete_from_indexes_cdc(row_new, new_link, inserted_already);
+
+                    let mut merged_primary_events = primary_key_events.clone();
+                    merged_primary_events.extend(rollback_pk_events);
+
+                    let mut merged_secondary_events = secondary_events.clone();
+                    merged_secondary_events.extend(rollback_secondary_events);
+
+                    let ack_op = Operation::Acknowledge(AcknowledgeOperation {
+                        id: OperationId::Single(Uuid::now_v7()),
+                        primary_key_events: merged_primary_events,
+                        secondary_keys_events: merged_secondary_events,
+                    });
+
+                    if let Err(e) = self.data.delete(new_link) {
+                        (ack_op, WorkTableError::PagesError(e))
+                    } else {
+                        (ack_op, WorkTableError::ColumnSlotIdExhausted(bits))
+                    }
+                }
                 IndexError::NotFound => {
                     let ack_op = Operation::Acknowledge(AcknowledgeOperation {
                         id: OperationId::Single(Uuid::now_v7()),
@@ -591,6 +657,8 @@ pub enum WorkTableError {
     AlreadyExists(#[error(not(source))] String),
     #[display("Row with this primary key already exists")]
     PrimaryAlreadyExists,
+    #[display("ColumnSlotId{} capacity is exhausted", _0)]
+    ColumnSlotIdExhausted(#[error(not(source))] u8),
     SerializeError,
     SecondaryIndexError,
     PrimaryUpdateTry,
diff --git a/tests/worktable/columnar.rs b/tests/worktable/columnar.rs
new file mode 100644
index 00000000..fa12ec4c
--- /dev/null
+++ b/tests/worktable/columnar.rs
@@ -0,0 +1,298 @@
+use std::sync::Arc;
+use worktable::prelude::*;
+use worktable::worktable;
+
+worktable!(
+    name: ColumnarMetrics,
+    persist: false,
+    columns: {
+        id: u64 primary_key,
+        host_id: u64 columnar(chunk_rows(2), compression(none)),
+        timestamp: i64 columnar,
+        temperature: i64 columnar(chunk_rows(2)),
+        label: String,
+    },
+    config: {
+        columnar_slot_id: ColumnSlotId16,
+        columnar_chunk_rows: 4,
+    },
+    columnar_indexes: {
+        host_time: {
+            cluster_by: [host_id, timestamp],
+        },
+    },
+    queries: {
+        update: {
+            TemperatureById(temperature) by id,
+        },
+        in_place: {
+            TimestampById(timestamp) by id,
+        }
+    },
+);
+
+worktable!(
+    name: TinyColumnarIds,
+    persist: false,
+    columns: {
+        id: u16 primary_key,
+        value: u16 columnar(chunk_rows(32), compression(none)),
+    },
+    config: {
+        columnar_slot_id: ColumnSlotId8,
+        columnar_chunk_rows: 32,
+    },
+);
+
+// Compile coverage for the persisted derive path. The columnar replica is
+// intentionally skipped by the existing index file format and rebuilt from
+// authoritative rows after load.
+worktable!(
+    name: PersistedColumnarMetrics,
+    persist: true,
+    columns: {
+        id: u64 primary_key,
+        host_id: u64 columnar(chunk_rows(4), compression(none)),
+        timestamp: i64 columnar(chunk_rows(4), compression(none)),
+    },
+    columnar_indexes: {
+        host_time: {
+            cluster_by: [host_id, timestamp],
+        },
+    },
+);
+
+worktable!(
+    name: CongeeColumnarSideIndex,
+    persist: false,
+    columns: {
+        id: u64 primary_key using congee,
+        value: u64 columnar,
+    },
+    columnar_indexes: {
+        value_order: {
+            cluster_by: [value],
+        },
+    },
+);
+
+worktable!(
+    name: ArcticColumnarSideIndex,
+    persist: false,
+    columns: {
+        id: u64 primary_key using arctic,
+        value: u64 columnar,
+    },
+    columnar_indexes: {
+        value_order: {
+            cluster_by: [value],
+        },
+    },
+);
+
+#[tokio::test]
+async fn columnar_fields_and_clustered_index_follow_mutations() {
+    let table = ColumnarMetricsWorkTable::default();
+    table
+        .insert(ColumnarMetricsRow {
+            id: 1,
+            host_id: 2,
+            timestamp: 20,
+            temperature: 72,
+            label: "second".to_string(),
+        })
+        .unwrap();
+    table
+        .insert(ColumnarMetricsRow {
+            id: 2,
+            host_id: 1,
+            timestamp: 10,
+            temperature: 68,
+            label: "first".to_string(),
+        })
+        .unwrap();
+
+    let host_two = table.columnar_select_host_time(2, 20).unwrap();
+    assert_eq!(host_two.len(), 1);
+    assert_eq!(host_two[0].primary_key().0, 1);
+    assert_eq!(table.columnar_project_temperature(&host_two).unwrap()[0].1, 72);
+
+    let ordered = table.columnar_scan_host_time().unwrap();
+    let projected = table.columnar_project_host_id(&ordered).unwrap();
+    assert_eq!(projected.iter().map(|(_, value)| *value).collect::<Vec<_>>(), [1, 2]);
+
+    table
+        .update(ColumnarMetricsRow {
+            id: 1,
+            host_id: 3,
+            timestamp: 30,
+            temperature: 75,
+            label: "updated".to_string(),
+        })
+        .await
+        .unwrap();
+
+    assert!(table.columnar_select_host_time(2, 20).unwrap().is_empty());
+    let updated = table.columnar_select_host_time(3, 30).unwrap();
+    assert_eq!(updated, host_two, "row identity survives an update");
+    assert_eq!(table.columnar_project_temperature(&updated).unwrap()[0].1, 75);
+
+    table
+        .update_temperature_by_id(TemperatureByIdQuery { temperature: 76 }, 1)
+        .await
+        .unwrap();
+    assert_eq!(table.columnar_project_temperature(&updated).unwrap()[0].1, 76);
+
+    table
+        .update_timestamp_by_id_in_place(|value| *value = 40.into(), 1)
+        .await
+        .unwrap();
+    assert!(table.columnar_is_dirty());
+    table.rebuild_columnar().unwrap();
+    assert!(!table.columnar_is_dirty());
+    assert!(table.columnar_select_host_time(3, 30).unwrap().is_empty());
+    assert_eq!(table.columnar_select_host_time(3, 40).unwrap(), updated);
+
+    table.delete(2).await.unwrap();
+    assert_eq!(table.columnar_scan_host_id().unwrap().len(), 1);
+    assert_eq!(table.columnar_scan_host_time().unwrap(), updated);
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn concurrent_reinsert_and_columnar_refresh_preserve_row_identity() {
+    let table = Arc::new(ColumnarMetricsWorkTable::default());
+    table
+        .insert(ColumnarMetricsRow {
+            id: 7,
+            host_id: 1,
+            timestamp: 0,
+            temperature: 1,
+            label: "short".to_string(),
+        })
+        .unwrap();
+    let stable_id = table.columnar_select_host_time(1, 0).unwrap()[0].clone();
+
+    let updater = {
+        let table = Arc::clone(&table);
+        tokio::spawn(async move {
+            for value in 1..=200 {
+                table
+                    .update(ColumnarMetricsRow {
+                        id: 7,
+                        host_id: 1,
+                        timestamp: value,
+                        temperature: value,
+                        label: if value % 2 == 0 {
+                            "a much longer row value".to_string()
+                        } else {
+                            "tiny".to_string()
+                        },
+                    })
+                    .await
+                    .unwrap();
+            }
+        })
+    };
+
+    for _ in 0..200 {
+        for (row_id, _) in table.columnar_scan_timestamp().unwrap() {
+            assert_eq!(row_id, stable_id);
+        }
+    }
+    updater.await.unwrap();
+
+    assert_eq!(table.columnar_select_host_time(1, 200).unwrap(), [stable_id]);
+}
+
+#[tokio::test]
+async fn configured_slot_id_capacity_is_checked_and_deleted_slots_are_safe_to_reuse() {
+    let table = TinyColumnarIdsWorkTable::default();
+    for id in 0..=u8::MAX as u16 {
+        table.insert(TinyColumnarIdsRow { id, value: id }).unwrap();
+    }
+
+    let stale = table
+        .columnar_scan_value()
+        .unwrap()
+        .into_iter()
+        .find(|(row_ref, _)| row_ref.primary_key().0 == 7)
+        .unwrap()
+        .0;
+    let error = table.insert(TinyColumnarIdsRow { id: 256, value: 256 }).unwrap_err();
+    assert!(matches!(error, WorkTableError::ColumnSlotIdExhausted(8)));
+    assert!(
+        table.select(256).is_none(),
+        "capacity failure rolls back the authoritative row"
+    );
+
+    table.delete(7).await.unwrap();
+    table.insert(TinyColumnarIdsRow { id: 256, value: 256 }).unwrap();
+
+    let replacement = table
+        .columnar_scan_value()
+        .unwrap()
+        .into_iter()
+        .find(|(row_ref, _)| row_ref.primary_key().0 == 256)
+        .unwrap()
+        .0;
+    assert!(
+        table.columnar_project_value(&[stale]).unwrap().is_empty(),
+        "a recycled slot cannot alias a different primary key"
+    );
+
+    table.delete(256).await.unwrap();
+    table.insert(TinyColumnarIdsRow { id: 256, value: 999 }).unwrap();
+    assert!(
+        table.columnar_project_value(&[replacement]).unwrap().is_empty(),
+        "delete and reinsert of the same primary key cannot revive a stale row reference"
+    );
+    assert_eq!(table.columnar_slots_in_use(), 256);
+    assert_eq!(table.columnar_slots_high_water(), 256);
+}
+
+#[test]
+fn row_refs_are_scoped_to_one_table_incarnation() {
+    let first = TinyColumnarIdsWorkTable::default();
+    first.insert(TinyColumnarIdsRow { id: 1, value: 11 }).unwrap();
+    let retained = first.columnar_scan_value().unwrap()[0].0.clone();
+
+    let second = TinyColumnarIdsWorkTable::default();
+    second.insert(TinyColumnarIdsRow { id: 1, value: 22 }).unwrap();
+
+    assert!(
+        second.columnar_project_value(&[retained]).unwrap().is_empty(),
+        "a ref from another table instance must not alias the same primary key and slot"
+    );
+}
+
+#[tokio::test]
+async fn columnar_side_indexes_compose_with_congee_and_arctic_using_backends() {
+    macro_rules! exercise {
+        ($table:ident, $row:ident) => {{
+            let table = $table::default();
+            table.insert($row { id: 1, value: 20 }).unwrap();
+            table.insert($row { id: 2, value: 10 }).unwrap();
+
+            let ordered = table.columnar_scan_value_order().unwrap();
+            assert_eq!(
+                table
+                    .columnar_project_value(&ordered)
+                    .unwrap()
+                    .into_iter()
+                    .map(|(_, value)| value)
+                    .collect::<Vec<_>>(),
+                [10, 20]
+            );
+
+            table.update($row { id: 1, value: 5 }).await.unwrap();
+            assert_eq!(table.columnar_select_value_order(20).unwrap(), []);
+            assert_eq!(table.columnar_select_value_order(5).unwrap().len(), 1);
+
+            table.delete(2).await.unwrap();
+            assert_eq!(table.columnar_scan_value().unwrap().len(), 1);
+        }};
+    }
+
+    exercise!(CongeeColumnarSideIndexWorkTable, CongeeColumnarSideIndexRow);
+    exercise!(ArcticColumnarSideIndexWorkTable, ArcticColumnarSideIndexRow);
+}
diff --git a/tests/worktable/mod.rs b/tests/worktable/mod.rs
index 5a111ca3..bd0ce985 100644
--- a/tests/worktable/mod.rs
+++ b/tests/worktable/mod.rs
@@ -2,6 +2,7 @@ mod array;
 mod base;
 mod bench;
 mod borrowed_primary_key;
+mod columnar;
 mod config;
 mod count;
 mod custom_pk;