Enum validation

crates/data-dict-parquet/src/dictionary.rs

@@ -0,0 +1,164 @@
+//! Dictionary-page fast path for enum membership (D03).
+//!
+//! Enums are low-cardinality and so almost always dictionary-encoded, which
+//! means a column chunk's distinct values sit in one small dictionary page
+//! rather than being spread across every row. Reading those pages lets the
+//! common, conforming case be settled without decoding all the data.
+//!
+//! This proves *conformance* only. It returns `true` when every value is
+//! provably in the allowed set, and `false` at the first sign of doubt — a
+//! chunk that isn't fully dictionary-encoded, an unsupported physical type, or
+//! a dictionary entry outside the set — leaving the caller to fall back to the
+//! full value scan (which also reports the offending rows).
+
+use std::collections::HashSet;
+
+use parquet::basic::{Encoding, PageType, Type as PhysicalType};
+use parquet::column::page::{Page, PageReader};
+use parquet::file::metadata::ColumnChunkMetaData;
+use parquet::file::reader::FileReader;
+
+use crate::ParquetError;
+
+/// Whether every non-null value in the `leaf`th column is provably in `allowed`,
+/// determined from the row groups' dictionary pages alone. `false` means "not
+/// proven" (scan to be sure), never "definitely violates".
+///
+/// `allowed` holds the canonical string forms produced by `scan::field_key`;
+/// dictionary values are canonicalized the same way here.
+pub(crate) fn dictionary_conforms(
+    reader: &dyn FileReader,
+    leaf: usize,
+    allowed: &HashSet<String>,
+) -> Result<bool, ParquetError> {
+    let meta = reader.metadata();
+    for group in 0..meta.num_row_groups() {
+        let column = meta.row_group(group).column(leaf);
+        if column.num_values() == 0 {
+            continue;
+        }
+        if column.dictionary_page_offset().is_none() {
+            return Ok(false);
+        }
+        let row_group = reader.get_row_group(group)?;
+        let mut pages = row_group.get_column_page_reader(leaf)?;
+
+        // The dictionary page comes first; it holds every value the data pages
+        // reference. If any entry is outside the set, defer to the scan.
+        let Some(page @ Page::DictionaryPage { .. }) = pages.get_next_page()? else {
+            return Ok(false);
+        };
+        if !dictionary_in_set(&page, column.column_type(), allowed) {
+            return Ok(false);
+        }
+        if !data_pages_all_dictionary(column, &mut *pages)? {
+            return Ok(false);
+        }
+    }
+    Ok(true)
+}
+
+/// Whether every data page draws from the dictionary (so the dictionary page is
+/// exhaustive). Uses the footer's page encoding stats when the writer recorded
+/// them; otherwise inspects each page's encoding directly — cheap, since these
+/// pages hold only dictionary indices and are never decoded here.
+fn data_pages_all_dictionary(
+    column: &ColumnChunkMetaData,
+    pages: &mut dyn PageReader,
+) -> Result<bool, ParquetError> {
+    if let Some(stats) = column.page_encoding_stats() {
+        let mut data_pages = 0;
+        for stat in stats {
+            if matches!(stat.page_type, PageType::DATA_PAGE | PageType::DATA_PAGE_V2) {
+                if !is_dictionary(stat.encoding) {
+                    return Ok(false);
+                }
+                data_pages += stat.count;
+            }
+        }
+        return Ok(data_pages > 0);
+    }
+    let mut data_pages = 0;
+    while let Some(page) = pages.get_next_page()? {
+        if !is_dictionary(page.encoding()) {
+            return Ok(false);
+        }
+        data_pages += 1;
+    }
+    Ok(data_pages > 0)
+}
+
+fn is_dictionary(encoding: Encoding) -> bool {
+    matches!(
+        encoding,
+        Encoding::RLE_DICTIONARY | Encoding::PLAIN_DICTIONARY
+    )
+}
+
+/// Whether every value in a PLAIN-encoded dictionary page is in `allowed`.
+/// `false` for physical types an enum can't sensibly use, or a malformed buffer.
+fn dictionary_in_set(page: &Page, physical: PhysicalType, allowed: &HashSet<String>) -> bool {
+    let Page::DictionaryPage {
+        buf, num_values, ..
+    } = page
+    else {
+        return false;
+    };
+    let count = *num_values as usize;
+    match physical {
+        PhysicalType::BYTE_ARRAY => byte_arrays_in_set(buf, count, allowed),
+        PhysicalType::INT32 => fixed_in_set::<4>(buf, count, allowed, |b| {
+            i32::from_le_bytes(b).to_string()
+        }),
+        PhysicalType::INT64 => fixed_in_set::<8>(buf, count, allowed, |b| {
+            i64::from_le_bytes(b).to_string()
+        }),
+        PhysicalType::FLOAT => fixed_in_set::<4>(buf, count, allowed, |b| {
+            f32::from_le_bytes(b).to_string()
+        }),
+        PhysicalType::DOUBLE => fixed_in_set::<8>(buf, count, allowed, |b| {
+            f64::from_le_bytes(b).to_string()
+        }),
+        _ => false,
+    }
+}
+
+/// Decode `count` PLAIN byte-array values (`[u32 length][bytes]`), requiring each
+/// to be UTF-8 (matching `field_key`, which only keys `Field::Str`) and present
+/// in `allowed`.
+fn byte_arrays_in_set(buf: &[u8], count: usize, allowed: &HashSet<String>) -> bool {
+    let mut pos = 0;
+    for _ in 0..count {
+        let Some(len_bytes) = buf.get(pos..pos + 4) else {
+            return false;
+        };
+        let len = u32::from_le_bytes(len_bytes.try_into().unwrap()) as usize;
+        pos += 4;
+        let Some(value) = buf.get(pos..pos + len) else {
+            return false;
+        };
+        pos += len;
+        let Ok(text) = std::str::from_utf8(value) else {
+            return false;
+        };
+        if !allowed.contains(text) {
+            return false;
+        }
+    }
+    true
+}
+
+/// Decode `count` fixed-width PLAIN values, canonicalizing each with `key`.
+fn fixed_in_set<const N: usize>(
+    buf: &[u8],
+    count: usize,
+    allowed: &HashSet<String>,
+    key: impl Fn([u8; N]) -> String,
+) -> bool {
+    if buf.len() < count * N {
+        return false;
+    }
+    buf.chunks_exact(N)
+        .take(count)
+        .all(|chunk| allowed.contains(&key(chunk.try_into().unwrap())))
+}

crates/data-dict-parquet/src/lib.rs

@@ -1,5 +1,6 @@
 //! Parquet reader for data-dict.yaml validation.
 
+mod dictionary;
 mod metadata;
 mod scan;
 

crates/data-dict-parquet/src/scan.rs

@@ -1,4 +1,4 @@
-use std::collections::HashMap;
+use std::collections::{HashMap, HashSet};
 use std::fs::File;
 use std::path::Path;
 
@@ -13,16 +13,21 @@ use crate::ParquetError;
 pub struct ColumnNeeds {
     /// Count nulls and sample the row numbers where they occur.
     pub nulls: bool,
+    /// The set of allowed values (D03). When present, non-null values not in
+    /// the set are counted and sampled. Values are the canonical string form
+    /// produced by [`field_key`]; the caller must canonicalize its set to match.
+    pub allowed: Option<HashSet<String>>,
 }
 
 impl ColumnNeeds {
     pub fn any(&self) -> bool {
-        self.nulls
+        self.nulls || self.allowed.is_some()
     }
 
     pub fn merge(self, other: Self) -> Self {
         ColumnNeeds {
             nulls: self.nulls || other.nulls,
+            allowed: self.allowed.or(other.allowed),
         }
     }
 }
@@ -33,6 +38,13 @@ pub struct ColumnStats {
     pub null_count: usize,
     /// 1-based row numbers, capped by the caller's limit.
     pub null_rows: Vec<usize>,
+    /// Non-null values found outside the [`ColumnNeeds::allowed`] set.
+    pub outside_count: usize,
+    /// 1-based row numbers of outside values, capped by the caller's limit.
+    pub outside_rows: Vec<usize>,
+    /// Distinct offending values, capped by the caller's limit, in first-seen
+    /// order.
+    pub outside_values: Vec<String>,
 }
 
 /// Gather requested statistics in one projected, streaming pass over the file.
@@ -62,9 +74,27 @@ pub fn column_stats(
         .iter()
         .map(|(name, _, _)| (name.clone(), ColumnStats::default()))
         .collect();
+
+    // Fast path: settle the enum-membership need (D03) from dictionary pages
+    // where the data conforms, sparing those columns the value scan. A column
+    // still scanned for its nulls skips the redundant dictionary read.
+    let proven: HashSet<&str> = requested
+        .iter()
+        .filter(|(_, _, need)| need.allowed.is_some() && !need.nulls)
+        .filter_map(|(name, index, need)| {
+            let allowed = need.allowed.as_ref()?;
+            crate::dictionary::dictionary_conforms(&reader, *index, allowed)
+                .ok()
+                .filter(|&conforms| conforms)
+                .map(|_| name.as_str())
+        })
+        .collect();
+
     let to_scan: Vec<usize> = requested
         .iter()
-        .filter(|(_, _, need)| need.nulls)
+        .filter(|(name, _, need)| {
+            need.nulls || (need.allowed.is_some() && !proven.contains(name.as_str()))
+        })
         .map(|(_, index, _)| *index)
         .collect();
 
@@ -87,14 +117,52 @@ pub fn column_stats(
             let (Some(stat), Some(need)) = (stats.get_mut(name), needs.get(name)) else {
                 continue;
             };
-            if need.nulls && matches!(field, Field::Null) {
-                stat.null_count += 1;
-                if stat.null_rows.len() < limit {
-                    stat.null_rows.push(index + 1);
+            if matches!(field, Field::Null) {
+                if need.nulls {
+                    stat.null_count += 1;
+                    if stat.null_rows.len() < limit {
+                        stat.null_rows.push(index + 1);
+                    }
+                }
+                continue;
+            }
+            if let Some(allowed) = &need.allowed
+                && !proven.contains(name.as_str())
+                && let Some(key) = field_key(field)
+                && !allowed.contains(&key)
+            {
+                stat.outside_count += 1;
+                if stat.outside_rows.len() < limit {
+                    stat.outside_rows.push(index + 1);
+                }
+                if stat.outside_values.len() < limit && !stat.outside_values.contains(&key) {
+                    stat.outside_values.push(key);
                 }
             }
         }
     }
 
     Ok(stats)
 }
+
+/// The canonical string form of a scalar field value, for set membership (D03).
+/// `None` for kinds that can't be an `enum` value (a matching `enum` column
+/// would already be an `M01` type mismatch). Integer and float forms follow
+/// Rust's `Display`, matching `Scalar::value_key` on the dictionary side.
+fn field_key(field: &Field) -> Option<String> {
+    Some(match field {
+        Field::Bool(v) => v.to_string(),
+        Field::Byte(v) => v.to_string(),
+        Field::Short(v) => v.to_string(),
+        Field::Int(v) => v.to_string(),
+        Field::Long(v) => v.to_string(),
+        Field::UByte(v) => v.to_string(),
+        Field::UShort(v) => v.to_string(),
+        Field::UInt(v) => v.to_string(),
+        Field::ULong(v) => v.to_string(),
+        Field::Float(v) => v.to_string(),
+        Field::Double(v) => v.to_string(),
+        Field::Str(v) => v.clone(),
+        _ => return None,
+    })
+}

crates/data-dict/src/lower.rs

@@ -87,7 +87,7 @@ fn lower_column(node: &YamlWithSourceInfo) -> Option<Column> {
     let mut name: Option<Spanned<String>> = None;
     let mut constraints: Vec<Spanned<Constraint>> = Vec::new();
     let mut col_type: Option<Spanned<String>> = None;
-    let mut values: Option<SourceInfo> = None;
+    let mut values: Option<Representation> = None;
     let mut range: Option<Representation> = None;
     let mut examples: Option<Representation> = None;
     let mut units: Option<Spanned<String>> = None;
@@ -108,7 +108,12 @@ fn lower_column(node: &YamlWithSourceInfo) -> Option<Column> {
                     col_type = Some(Spanned::new(s.to_string(), entry.value_span.clone()));
                 }
             }
-            "values" => values = Some(entry.value_span.clone()),
+            "values" => {
+                values = Some(Representation {
+                    span: entry.value_span.clone(),
+                    items: lower_enum_values(&entry.value),
+                });
+            }
             "range" => {
                 range = Some(Representation {
                     span: entry.value_span.clone(),
@@ -157,6 +162,20 @@ fn lower_column(node: &YamlWithSourceInfo) -> Option<Column> {
     })
 }
 
+/// Lower an enum's `values` node into its allowed scalars with spans.
+fn lower_enum_values(node: &YamlWithSourceInfo) -> Vec<Spanned<Scalar>> {
+    if let Some(entries) = node.as_hash() {
+        // Map form: the keys are the values, the labels are dropped.
+        entries
+            .iter()
+            .map(|entry| Spanned::new(lower_scalar(&entry.key), entry.key.source_info.clone()))
+            .collect()
+    } else {
+        // List form (or a lone scalar, which the schema rejects upstream).
+        lower_scalars(node)
+    }
+}
+
 /// Lower a `range` or `examples` node into its scalar elements with spans.
 /// Non-array nodes yield an empty vector (the schema rejects them upstream).
 fn lower_scalars(node: &YamlWithSourceInfo) -> Vec<Spanned<Scalar>> {

crates/data-dict/src/model.rs

@@ -59,7 +59,9 @@ pub struct Column {
     pub name: Spanned<String>,
     pub constraints: Vec<Spanned<Constraint>>,
     pub col_type: Option<Spanned<String>>,
-    pub values: Option<SourceInfo>,
+    /// The allowed values of an `enum` column: the list items, or the keys of
+    /// the map form (whose labels are dropped — only the values are constrained).
+    pub values: Option<Representation>,
     pub range: Option<Representation>,
     pub examples: Option<Representation>,
     pub units: Option<Spanned<String>>,
@@ -93,6 +95,18 @@ impl Scalar {
             Scalar::Compound => "a list or map",
         }
     }
+
+    /// A canonical string form for value-equality comparison against data (D03).
+    /// `None` for kinds that can't appear as a data value (`null`, compound).
+    /// Must agree with the data side's canonicalization in `data-dict-parquet`.
+    pub fn value_key(&self) -> Option<String> {
+        match self {
+            Scalar::Number(n) => Some(n.to_string()),
+            Scalar::String(s) => Some(s.clone()),
+            Scalar::Bool(b) => Some(b.to_string()),
+            Scalar::Null | Scalar::Compound => None,
+        }
+    }
 }
 
 impl Column {
@@ -111,6 +125,10 @@ impl Column {
     pub fn is_required_implied(&self) -> bool {
         self.has(Constraint::Required) || self.has(Constraint::PrimaryKey)
     }
+
+    pub fn is_enum(&self) -> bool {
+        self.col_type.as_ref().is_some_and(|t| t.value == "enum")
+    }
 }
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]